class CustomerServiceAgent:
    def handle_request(self, message: str) -> str:
        intent = self.classify_intent(message)
        context = self.search_knowledge_base(intent)
        account = self.get_account_info()
        response = self.generate_response(context, account)
        
        if self.should_escalate(response):
            return self.escalate_to_human()
        return response

┌──────────────────────────────────────┐
│     오케스트레이터 에이전트           │
│  요청을 전문가들에게 라우팅           │
└──────┬──────────────────┬────────────┘
       │                  │
┌──────▼──────┐   ┌──────▼──────────┐
│ 분류기      │   │ 지식베이스      │
│ 에이전트    │   │ 에이전트        │
└──────┬──────┘   └──────┬──────────┘
       │                  │
┌──────▼──────┐   ┌──────▼──────────┐
│ 계정        │   │ 응답            │
│ 에이전트    │   │ 생성기          │
└─────────────┘   └─────────────────┘

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1단계: 공유 상태 정의
class AgentState(TypedDict):
    messages: list
    intent: str
    context: str
    should_escalate: bool

# 2단계: 노드 함수 정의
def classify_intent(state: AgentState) -> AgentState:
    llm = ChatOpenAI(model="gpt-4o")
    response = llm.invoke([{
        "role": "system", 
        "content": "의도 분류: 청구, 기술지원, 불만"
    }])
    return {"intent": response.content.strip()}

def retrieve_knowledge(state: AgentState) -> AgentState:
    knowledge_map = {
        "청구": "30일 내 환불...",
        "기술지원": "먼저 재부팅...",
    }
    return {"context": knowledge_map.get(state["intent"], "")}

def check_escalation(state: AgentState) -> AgentState:
    should_escalate = state["intent"] == "불만"
    return {"should_escalate": should_escalate}

# 3단계: 그래프 구축
def build_graph():
    workflow = StateGraph(AgentState)
    
    workflow.add_node("classify", classify_intent)
    workflow.add_node("retrieve", retrieve_knowledge)
    workflow.add_node("check", check_escalation)
    
    workflow.add_edge(START, "classify")
    workflow.add_edge("classify", "retrieve")
    workflow.add_edge("retrieve", "check")
    workflow.add_edge("check", END)
    
    return workflow.compile()

from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = build_graph().compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user-123"}}
result = graph.invoke({"messages": [...]}, config)

# 나중에 재개
result = graph.invoke({"messages": [new_message]}, config)

from langgraph.types import interrupt

def human_approval_node(state: AgentState):
    if state["requires_approval"]:
        approval = interrupt("관리자 승인 대기 중")
        return {"approved": approval}
    return state

from crewai import Agent, Task, Crew, Process

# 에이전트 정의 (팀 멤버)
classifier = Agent(
    role="고객 의도 분류기",
    goal="문의를 정확하게 분류",
    backstory="""수년간 고객 서비스 경험을 가진 
    고객 니즈 이해 전문가.""",
    verbose=True
)

researcher = Agent(
    role="지식 베이스 연구원",
    goal="관련 정보 찾기",
    backstory="""회사 정책을 속속들이 아는 
    꼼꼼한 연구원.""",
    tools=[SerperDevTool()],
    verbose=True
)

# 작업 정의
classification_task = Task(
    description="분석: {message}. 분류: 청구/기술지원/불만",
    expected_output="긴급도 수준이 포함된 분류",
    agent=classifier
)

research_task = Task(
    description="{classification}에 대한 지식 베이스 검색",
    expected_output="관련 정책 및 솔루션",
    agent=researcher,
    context=[classification_task]  # 분류에 의존
)

# 크루 구성
crew = Crew(
    agents=[classifier, researcher],
    tasks=[classification_task, research_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(inputs={"message": "청구서가 잘못됐어요!"})

crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.hierarchical,  # 매니저가 조정
    manager_llm=ChatOpenAI(model="gpt-4o")
)

crew = Crew(
    agents=[...],
    memory=True,  # 과거 실행에서 학습 활성화
    embedder={"provider": "openai"}
)

from crewai_tools import (
    SerperDevTool,      # 웹 검색
    ScrapeWebsiteTool,  # 스크래핑
    CodeInterpreterTool # 코드 실행
)

from autogen import AssistantAgent, GroupChat, GroupChatManager

# LLM 설정
config = [{"model": "gpt-4o", "api_key": os.environ["OPENAI_API_KEY"]}]

# 대화형 에이전트 생성
classifier = AssistantAgent(
    name="분류기",
    system_message="고객 의도를 분류하세요. 간결하게.",
    llm_config={"config_list": config}
)

researcher = AssistantAgent(
    name="연구원",
    system_message="관련 정책을 연구하세요.",
    llm_config={"config_list": config}
)

# 그룹 채팅 설정
group_chat = GroupChat(
    agents=[classifier, researcher],
    messages=[],
    max_round=10,
    speaker_selection_method="auto"  # LLM이 누가 말할지 결정
)

manager = GroupChatManager(groupchat=group_chat, llm_config={"config_list": config})

# 대화 시작
result = classifier.initiate_chat(
    manager, 
    message="청구서가 잘못됐어요!"
)

executor = UserProxyAgent(
    name="실행자",
    code_execution_config={
        "work_dir": "sandbox",
        "use_docker": True  # 샌드박스
    }
)

# 코더가 작성 → 실행자가 실행 → 코더가 개선
executor.initiate_chat(
    coder, 
    message="복리 이자를 계산하고 테스트해줘"
)

human = UserProxyAgent(
    name="인간",
    human_input_mode="ALWAYS"  # 또는 "TERMINATE" 또는 "NEVER"
)

# 에이전트가 파일 읽고, 웹 검색하고, 코드 작성
$ openclaw "AI 에이전트 프레임워크 조사하고 비교 글 써줘"

# 자동 도구 사용:
# 1. 최신 정보를 위한 web_search
# 2. 심층 분석을 위한 web_fetch
# 3. 마크다운 파일 작성
# 4. 검증을 위한 exec

# 민감한 작업에 대한 인간 승인
⚠️  50개 파일을 삭제하려고 합니다. 진행하시겠습니까? [y/N]

def supervisor_node(state):
    decision = llm.invoke([{
        "role": "system", 
        "content": """다음 액션 결정:
        - 'research': 더 많은 정보 필요
        - 'respond': 응답 준비 완료
        - 'escalate': 인간 필요
        - 'complete': 완료"""
    }])
    return {"next_action": decision.content}

crew = Crew(
    agents=[researcher, writer, editor, publisher],
    tasks=[research_task, write_task, edit_task, publish_task],
    process=Process.sequential
)

optimist = AssistantAgent(name="낙관주의자", ...)
critic = AssistantAgent(name="비평가", ...)
synthesizer = AssistantAgent(name="통합자", ...)

group_chat = GroupChat(agents=[optimist, critic, synthesizer])

# 명시적 종료 설정
graph.invoke(state, config={"recursion_limit": 25})
crew = Crew(agents=[...], max_iter=10)
group_chat = GroupChat(max_round=10)

# 에이전트 간 컨텍스트 요약
def summarize_context(state):
    summary = cheap_llm.invoke(f"요약: {state['context']}")
    return {"context": summary.content}

def safe_node(func):
    def wrapper(state):
        try:
            return func(state)
        except Exception as e:
            return {"error": str(e), "fallback": "오류가 발생했습니다"}
    return wrapper

시작
 │
 ▼
세밀한 제어 필요? ─예→ LangGraph
 │아니오
 ▼
.NET 앱 구축? ─예→ Microsoft Agent Framework
 │아니오
 ▼
Human-in-loop CLI 필요? ─예→ OpenClaw
 │아니오
 ▼
워크플로우가 팀 역할에 매핑? ─예→ CrewAI
 │아니오
 ▼
코드 실행 필요? ─예→ AutoGen
 │아니오
 ▼
빠른 프로토타이핑 우선? ─예→ CrewAI
 │아니오
 ▼
기본 선택 → CrewAI부터 시작 (진입 장벽 최소)