LLM Agent Framework 비교: AutoGen vs CrewAI vs LangGraph 실전 선택 가이드

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat, SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 1. LLM 클라이언트 설정
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    api_key="your-api-key",
)

# 2. 도구 정의
async def search_web(query: str) -> str:
    """웹에서 정보를 검색합니다."""
    # 실제로는 Tavily, SerpAPI 등 사용
    return f"'{query}'에 대한 검색 결과: [검색 결과 요약]"

async def analyze_data(data: str) -> str:
    """데이터를 분석하고 인사이트를 추출합니다."""
    return f"분석 결과: {data[:100]}에 대한 심층 분석 완료"

# 3. 에이전트 정의
researcher = AssistantAgent(
    name="Researcher",
    model_client=model_client,
    system_message="""당신은 전문 리서처입니다.
    주어진 주제에 대해 웹 검색을 수행하고 관련 정보를 수집합니다.
    수집한 정보를 구조화하여 Analyst에게 전달합니다.""",
    tools=[search_web],
)

analyst = AssistantAgent(
    name="Analyst",
    model_client=model_client,
    system_message="""당신은 데이터 분석가입니다.
    Researcher가 수집한 정보를 분석하고 핵심 인사이트를 추출합니다.
    분석 결과를 Writer에게 전달합니다.""",
    tools=[analyze_data],
)

writer = AssistantAgent(
    name="Writer",
    model_client=model_client,
    system_message="""당신은 보고서 작성자입니다.
    Analyst의 분석 결과를 바탕으로 최종 보고서를 작성합니다.
    보고서 작성이 완료되면 'TERMINATE'를 출력합니다.""",
)

# 4. 팀 구성 및 실행
termination = MaxMessageTermination(max_messages=15) | TextMentionTermination("TERMINATE")

# SelectorGroupChat: LLM이 다음 발언자를 선택
team = SelectorGroupChat(
    participants=[researcher, analyst, writer],
    model_client=model_client,
    termination_condition=termination,
    selector_prompt="""다음 발언자를 선택하세요.
    정보 수집이 필요하면 Researcher, 분석이 필요하면 Analyst,
    최종 보고서 작성이 필요하면 Writer를 선택합니다.""",
)

async def main():
    result = await team.run(
        task="2026년 AI 에이전트 프레임워크 시장 동향을 조사하고 보고서를 작성하세요."
    )
    for message in result.messages:
        print(f"[{message.source}]: {message.content[:200]}")
    print(f"\n총 메시지 수: {len(result.messages)}")

asyncio.run(main())

from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from langchain_openai import ChatOpenAI

# 1. LLM 설정
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 2. 커스텀 도구 정의
@tool("Search Tool")
def search_tool(query: str) -> str:
    """주어진 쿼리로 웹 검색을 수행합니다."""
    # 실제로는 Tavily, SerpAPI 등 사용
    return f"'{query}' 검색 결과: [관련 정보 요약]"

@tool("SEO Analyzer")
def seo_analyzer(content: str) -> str:
    """콘텐츠의 SEO 점수를 분석합니다."""
    word_count = len(content.split())
    return f"단어 수: {word_count}, SEO 점수: {'양호' if word_count > 500 else '부족'}"

# 3. 에이전트 정의
research_agent = Agent(
    role="기술 리서처",
    goal="최신 기술 트렌드를 조사하고 정확한 정보를 수집한다",
    backstory="""10년 경력의 기술 저널리스트로, AI와 소프트웨어 엔지니어링
    분야의 최신 동향을 항상 파악하고 있습니다.""",
    tools=[search_tool],
    llm=llm,
    verbose=True,
    allow_delegation=True,  # 다른 에이전트에게 작업 위임 가능
)

writer_agent = Agent(
    role="기술 블로그 작성자",
    goal="독자가 쉽게 이해할 수 있는 깊이 있는 기술 블로그를 작성한다",
    backstory="""기술 블로그 전문 작가로, 복잡한 기술 개념을
    명확하고 실용적인 글로 풀어내는 것에 능숙합니다.""",
    llm=llm,
    verbose=True,
)

editor_agent = Agent(
    role="편집장",
    goal="콘텐츠의 품질, 정확성, SEO 최적화를 검수한다",
    backstory="""기술 미디어 편집장으로, 콘텐츠의 기술적 정확성과
    가독성, SEO 최적화를 동시에 추구합니다.""",
    tools=[seo_analyzer],
    llm=llm,
    verbose=True,
)

# 4. 태스크 정의
research_task = Task(
    description="""'{topic}'에 대한 종합 리서치를 수행합니다.
    - 최신 동향 3가지 이상 수집
    - 핵심 기술 개념 정리
    - 실제 사용 사례 2건 이상 포함""",
    expected_output="구조화된 리서치 보고서 (마크다운 형식)",
    agent=research_agent,
)

writing_task = Task(
    description="""리서치 결과를 바탕으로 기술 블로그 포스트를 작성합니다.
    - 2000자 이상의 본문
    - 코드 예제 2개 이상 포함
    - 실무자 관점의 인사이트 포함""",
    expected_output="완성된 블로그 포스트 (마크다운 형식)",
    agent=writer_agent,
    context=[research_task],  # research_task 결과를 컨텍스트로 받음
)

editing_task = Task(
    description="""작성된 블로그 포스트를 검수합니다.
    - 기술적 정확성 확인
    - 가독성 개선
    - SEO 분석 수행
    - 최종 수정본 출력""",
    expected_output="검수 완료된 최종 블로그 포스트",
    agent=editor_agent,
    context=[writing_task],
)

# 5. Crew 구성 및 실행
crew = Crew(
    agents=[research_agent, writer_agent, editor_agent],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,  # 순차 실행
    verbose=True,
    memory=True,                 # 단기 메모리 활성화
    max_rpm=10,                  # API 호출 속도 제한
)

# 실행
result = crew.kickoff(inputs={"topic": "LLM 에이전트 프레임워크 비교"})
print(f"최종 결과:\n{result}")
print(f"토큰 사용량: {crew.usage_metrics}")

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage

# 1. 상태 정의
class SupportState(TypedDict):
    messages: Annotated[list, add_messages]
    category: str           # 문의 카테고리 (billing, technical, general)
    sentiment: str          # 고객 감정 (positive, neutral, negative)
    escalated: bool         # 상위 담당자 에스컬레이션 여부
    resolution: str         # 해결 상태

# 2. 도구 정의
@tool
def lookup_order(order_id: str) -> str:
    """주문 정보를 조회합니다."""
    orders = {
        "ORD-001": {"status": "배송중", "eta": "2026-03-11"},
        "ORD-002": {"status": "배송완료", "delivered": "2026-03-07"},
    }
    if order_id in orders:
        return f"주문 {order_id}: {orders[order_id]}"
    return f"주문 {order_id}를 찾을 수 없습니다."

@tool
def create_ticket(summary: str, priority: str) -> str:
    """고객 지원 티켓을 생성합니다."""
    return f"티켓 생성 완료 - 요약: {summary}, 우선순위: {priority}, 티켓 ID: TKT-{hash(summary) % 10000}"

@tool
def refund_order(order_id: str, reason: str) -> str:
    """주문 환불을 처리합니다. (승인 필요)"""
    return f"환불 요청 접수: {order_id}, 사유: {reason}. 관리자 승인 대기중."

tools = [lookup_order, create_ticket, refund_order]
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)

# 3. 노드 함수 정의
def classify_intent(state: SupportState) -> dict:
    """고객 문의를 분류합니다."""
    classifier_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    last_message = state["messages"][-1].content

    response = classifier_llm.invoke([
        SystemMessage(content="""고객 문의를 분류하세요.
        카테고리: billing, technical, general
        감정: positive, neutral, negative
        JSON 형식으로 응답: {"category": "...", "sentiment": "..."}"""),
        HumanMessage(content=last_message)
    ])

    import json
    try:
        result = json.loads(response.content)
    except json.JSONDecodeError:
        result = {"category": "general", "sentiment": "neutral"}

    return {
        "category": result.get("category", "general"),
        "sentiment": result.get("sentiment", "neutral"),
    }

def agent_respond(state: SupportState) -> dict:
    """에이전트가 응답합니다."""
    system_prompt = f"""당신은 고객 지원 에이전트입니다.
    현재 문의 카테고리: {state.get('category', 'unknown')}
    고객 감정: {state.get('sentiment', 'unknown')}
    에스컬레이션 여부: {state.get('escalated', False)}

    고객에게 친절하고 정확하게 응답하세요.
    필요한 경우 도구를 사용하여 정보를 조회하거나 티켓을 생성하세요."""

    messages = [SystemMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

def check_escalation(state: SupportState) -> dict:
    """에스컬레이션 필요 여부를 판단합니다."""
    if state.get("sentiment") == "negative" and state.get("category") == "billing":
        return {"escalated": True}
    return {"escalated": False}

def human_escalation(state: SupportState) -> dict:
    """상위 담당자에게 에스컬레이션합니다."""
    return {
        "messages": [SystemMessage(content="[시스템] 이 문의는 상위 담당자에게 전달되었습니다. 잠시 기다려주세요.")],
        "resolution": "escalated_to_human",
    }

# 4. 라우팅 함수
def route_after_classification(state: SupportState) -> Literal["check_escalation", "agent_respond"]:
    """분류 결과에 따라 라우팅합니다."""
    if state.get("sentiment") == "negative":
        return "check_escalation"
    return "agent_respond"

def route_after_escalation_check(state: SupportState) -> Literal["human_escalation", "agent_respond"]:
    """에스컬레이션 판단 결과에 따라 라우팅합니다."""
    if state.get("escalated"):
        return "human_escalation"
    return "agent_respond"

# 5. 그래프 구성
workflow = StateGraph(SupportState)

# 노드 추가
workflow.add_node("classify_intent", classify_intent)
workflow.add_node("check_escalation", check_escalation)
workflow.add_node("agent_respond", agent_respond)
workflow.add_node("tools", ToolNode(tools))
workflow.add_node("human_escalation", human_escalation)

# 엣지 정의
workflow.add_edge(START, "classify_intent")
workflow.add_conditional_edges("classify_intent", route_after_classification)
workflow.add_conditional_edges("check_escalation", route_after_escalation_check)
workflow.add_conditional_edges("agent_respond", tools_condition, {"tools": "tools", END: END})
workflow.add_edge("tools", "agent_respond")
workflow.add_edge("human_escalation", END)

# 체크포인터 (상태 영속화)
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# 6. 실행
config = {"configurable": {"thread_id": "customer-123"}}
result = app.invoke(
    {"messages": [HumanMessage(content="주문 ORD-001 배송이 너무 늦어요. 환불해주세요.")]},
    config=config,
)

for msg in result["messages"]:
    print(f"[{msg.__class__.__name__}]: {msg.content[:200]}")
print(f"카테고리: {result['category']}, 감정: {result['sentiment']}")

# LangGraph에서의 오류 복구 패턴 예시
from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_core.messages import AIMessage

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))
def resilient_agent_node(state: SupportState) -> dict:
    """재시도 로직이 포함된 에이전트 노드."""
    try:
        response = llm.invoke(state["messages"])
        return {"messages": [response]}
    except Exception as e:
        # 폴백: 미리 정의된 응답 반환
        fallback_response = AIMessage(
            content="죄송합니다. 일시적인 시스템 오류가 발생했습니다. "
                    "잠시 후 다시 시도해주세요. 긴급한 경우 고객센터 "
                    "1588-0000으로 연락해주세요."
        )
        return {
            "messages": [fallback_response],
            "resolution": f"error_fallback: {str(e)}",
        }