9장: 에이전트 프레임워크 비교 - LangGraph, CrewAI, OpenAI Agents SDK

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add
 
class AgentState(TypedDict):
    messages: Annotated[list, add]
    current_step: str
    results: dict
 
def research_node(state: AgentState) -> dict:
    """연구 단계를 수행하는 노드입니다."""
    # 연구 로직 구현
    research_result = perform_research(state["messages"][-1])
    return {
        "messages": [{"role": "assistant", "content": research_result}],
        "results": {"research": research_result},
    }
 
def analysis_node(state: AgentState) -> dict:
    """분석 단계를 수행하는 노드입니다."""
    research_data = state["results"].get("research", "")
    analysis = perform_analysis(research_data)
    return {
        "messages": [{"role": "assistant", "content": analysis}],
        "results": {"analysis": analysis},
    }
 
def should_continue(state: AgentState) -> str:
    """다음 단계를 결정하는 조건부 엣지입니다."""
    if needs_more_research(state):
        return "research"
    return "analysis"
 
# 그래프 구성
graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
 
graph.add_edge(START, "research")
graph.add_conditional_edges("research", should_continue)
graph.add_edge("analysis", END)
 
app = graph.compile()

from langgraph.checkpoint.memory import MemorySaver
 
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
 
# 실행 시 thread_id로 세션을 구분
config = {"configurable": {"thread_id": "user-session-1"}}
result = app.invoke(initial_state, config)
 
# 나중에 같은 thread_id로 이어서 실행
result = app.invoke(new_input, config)

from langgraph.graph import StateGraph
from langgraph.types import interrupt
 
def sensitive_action_node(state: AgentState) -> dict:
    """민감한 작업 전 사용자 확인을 요청합니다."""
    action = state["pending_action"]
 
    # 실행을 중단하고 사용자 확인을 요청
    approval = interrupt(
        f"다음 작업을 실행할까요?\n{action}\n승인(yes) 또는 거부(no)"
    )
 
    if approval == "yes":
        result = execute_action(action)
        return {"results": {"action": result}}
    else:
        return {"results": {"action": "사용자가 작업을 거부했습니다."}}

from crewai import Agent, Task, Crew, Process
 
# 에이전트 정의
researcher = Agent(
    role="기술 리서처",
    goal="주어진 주제에 대한 최신 기술 동향을 조사한다",
    backstory=(
        "당신은 10년 경력의 기술 리서처입니다. "
        "빠르게 변하는 기술 트렌드를 정확하게 파악하는 것이 "
        "당신의 전문 분야입니다."
    ),
    verbose=True,
    allow_delegation=False,
)
 
writer = Agent(
    role="기술 작가",
    goal="조사 결과를 바탕으로 명확하고 깊이 있는 기술 보고서를 작성한다",
    backstory=(
        "당신은 복잡한 기술 개념을 누구나 이해할 수 있는 "
        "언어로 전달하는 기술 작가입니다."
    ),
    verbose=True,
    allow_delegation=False,
)
 
# 작업 정의
research_task = Task(
    description=(
        "2026년 AI 에이전트 프레임워크의 현황을 조사하십시오. "
        "주요 프레임워크의 특징, 장단점, 사용 사례를 포함해야 합니다."
    ),
    expected_output="프레임워크별 분석이 포함된 조사 보고서",
    agent=researcher,
)
 
writing_task = Task(
    description=(
        "조사 결과를 바탕으로 기업 의사결정자를 위한 "
        "프레임워크 선택 가이드를 작성하십시오."
    ),
    expected_output="프레임워크 비교표와 선택 기준이 포함된 보고서",
    agent=writer,
)
 
# 팀 구성 및 실행
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True,
)
 
result = crew.kickoff()

# 계층적 프로세스: 매니저가 작업을 자동으로 분배
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[complex_task],
    process=Process.hierarchical,
    manager_llm="claude-sonnet-4-20250514",
)

from agents import Agent, Runner, function_tool
 
@function_tool
def search_web(query: str) -> str:
    """웹에서 정보를 검색합니다."""
    return f"검색 결과: {query}"
 
@function_tool
def calculate(expression: str) -> str:
    """수학 계산을 수행합니다."""
    return str(eval(expression))
 
agent = Agent(
    name="research_assistant",
    instructions=(
        "당신은 연구 보조 에이전트입니다. "
        "사용자의 질문에 대해 웹 검색과 계산을 활용하여 "
        "정확한 답변을 제공하십시오."
    ),
    tools=[search_web, calculate],
)
 
result = Runner.run_sync(agent, "파이썬 3.13의 주요 변경사항을 알려 주세요.")
print(result.final_output)

from agents import Agent, Runner
 
triage_agent = Agent(
    name="triage",
    instructions="사용자의 질문을 분류하여 적절한 전문가에게 전달합니다.",
    handoffs=["tech_expert", "business_expert"],
)
 
tech_expert = Agent(
    name="tech_expert",
    instructions="기술적인 질문에 대해 상세하게 답변합니다.",
    handoffs=["triage"],  # 필요시 다시 분류 에이전트로
)
 
business_expert = Agent(
    name="business_expert",
    instructions="비즈니스 관련 질문에 대해 답변합니다.",
    handoffs=["triage"],
)
 
result = Runner.run_sync(
    triage_agent,
    "AI 에이전트를 도입하면 ROI가 어떻게 되나요?"
)