Multi-Agent 시스템 완전 비교: AutoGen, CrewAI, LangGraph 중 무엇을 선택하는가

import autogen

# LLM 설정
llm_config = {
    "model": "gpt-4",
    "api_key": "your-api-key"
}

# 에이전트 정의
coder = autogen.AssistantAgent(
    name="Coder",
    llm_config=llm_config,
    system_message=(
        "You are a Python expert. Write clean, well-tested code. "
        "Always include error handling and type hints. "
        "When you finish, say 'TERMINATE'."
    )
)

reviewer = autogen.AssistantAgent(
    name="Reviewer",
    llm_config=llm_config,
    system_message=(
        "You are a senior software engineer. Review code for: "
        "1. Bugs and edge cases "
        "2. Security vulnerabilities "
        "3. Performance issues "
        "4. Code style and maintainability "
        "Provide specific, actionable feedback."
    )
)

# UserProxyAgent: 실제 코드 실행 담당
user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="NEVER",       # 완전 자동화
    max_consecutive_auto_reply=10,
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False  # 프로덕션에서는 True 권장
    },
    is_termination_msg=lambda x: "TERMINATE" in x.get("content", "")
)

# 대화 시작
user_proxy.initiate_chat(
    coder,
    message="날씨 데이터를 가져와서 matplotlib으로 시각화하는 Python 스크립트를 작성해줘"
)

# 여러 에이전트가 하나의 그룹 대화에 참여
groupchat = autogen.GroupChat(
    agents=[user_proxy, coder, reviewer],
    messages=[],
    max_round=20,
    speaker_selection_method="auto"  # LLM이 다음 발언자 결정
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config
)

user_proxy.initiate_chat(
    manager,
    message="REST API 클라이언트 라이브러리를 설계하고 구현해줘"
)

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

# 검색 도구 설정
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# 에이전트 정의: 역할과 목표가 핵심
researcher = Agent(
    role="Research Analyst",
    goal="Find accurate, comprehensive, and up-to-date information on any topic",
    backstory=(
        "You are an expert researcher with 10 years of experience. "
        "You always verify information from multiple sources and cite your findings."
    ),
    tools=[search_tool, web_tool],
    llm="gpt-4",
    verbose=True
)

analyst = Agent(
    role="Data Analyst",
    goal="Analyze information and identify key trends and insights",
    backstory=(
        "You are a data analyst who excels at finding patterns and drawing "
        "actionable conclusions from complex information."
    ),
    llm="gpt-4",
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Write clear, engaging, and well-structured content",
    backstory=(
        "You are a technical writer who makes complex topics accessible. "
        "You write for engineers who value precision and clarity."
    ),
    llm="gpt-4",
    verbose=True
)

# 작업 정의: 의존성이 중요
research_task = Task(
    description=(
        "AI 에이전트 분야의 최신 트렌드 5가지를 조사하세요. "
        "각 트렌드에 대해 구체적인 예시와 출처를 포함하세요."
    ),
    agent=researcher,
    expected_output="5가지 트렌드, 각각 설명 2-3문장 + 출처"
)

analysis_task = Task(
    description=(
        "조사된 트렌드를 분석하여 가장 영향력 있는 트렌드를 "
        "엔지니어 관점에서 순위를 매기고, 각각의 실용적 의미를 설명하세요."
    ),
    agent=analyst,
    expected_output="순위가 매겨진 트렌드 분석, 각각의 실용적 의미 포함",
    context=[research_task]  # research_task 결과를 context로 사용
)

writing_task = Task(
    description=(
        "분석 결과를 바탕으로 1500자 분량의 기술 블로그 포스트를 작성하세요. "
        "실제 엔지니어가 당장 활용할 수 있는 실용적인 내용으로 구성하세요."
    ),
    agent=writer,
    expected_output="마크다운 형식의 블로그 포스트, 제목/소제목/코드 예시 포함",
    context=[research_task, analysis_task]
)

# Crew 실행
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,  # 순차 실행 (또는 Process.hierarchical)
    verbose=True
)

result = crew.kickoff()
print(result)

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
import operator

# 상태 정의: 그래프 전체에서 공유되는 상태
class ResearchState(TypedDict):
    messages: Annotated[List[str], operator.add]  # 메시지 누적
    research_done: bool
    analysis_done: bool
    draft: str
    final_report: str

# 그래프 생성
workflow = StateGraph(ResearchState)

# 노드 함수 정의
def research_node(state: ResearchState) -> dict:
    """인터넷에서 정보 수집"""
    results = search_web(state["messages"][-1])
    return {
        "messages": [f"Research results: {results}"],
        "research_done": True
    }

def analysis_node(state: ResearchState) -> dict:
    """수집된 정보 분석"""
    research = [m for m in state["messages"] if "Research results:" in m]
    analysis = analyze_data(research)
    return {
        "messages": [f"Analysis: {analysis}"],
        "analysis_done": True
    }

def writing_node(state: ResearchState) -> dict:
    """최종 보고서 작성"""
    all_context = "\n".join(state["messages"])
    draft = write_report(all_context)
    return {"draft": draft}

def review_node(state: ResearchState) -> dict:
    """초안 검토 및 개선"""
    reviewed = review_and_improve(state["draft"])
    return {"final_report": reviewed}

# 조건부 라우팅 함수
def route_after_research(state: ResearchState) -> str:
    """조사 결과에 따라 다음 노드 결정"""
    if len(state["messages"]) > 3:
        return "analysis"
    else:
        return "research"  # 더 많은 조사가 필요

# 노드 추가
workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("writing", writing_node)
workflow.add_node("review", review_node)

# 엣지 추가 (흐름 정의)
workflow.set_entry_point("research")
workflow.add_conditional_edges(
    "research",
    route_after_research,
    {
        "analysis": "analysis",
        "research": "research"  # 자기 자신으로 돌아감 (루프)
    }
)
workflow.add_edge("analysis", "writing")
workflow.add_edge("writing", "review")
workflow.add_edge("review", END)

# 컴파일 및 실행
app = workflow.compile()

initial_state = {
    "messages": ["AI 에이전트 최신 트렌드 조사"],
    "research_done": False,
    "analysis_done": False,
    "draft": "",
    "final_report": ""
}

result = app.invoke(initial_state)
print(result["final_report"])

from langgraph.checkpoint.sqlite import SqliteSaver

# 체크포인트 설정 — 중간 상태 저장/복원
memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)

# 스레드 ID로 실행 — 같은 대화 이어서 진행 가능
config = {"configurable": {"thread_id": "session-123"}}

# 첫 번째 실행
result = app.invoke(initial_state, config=config)

# 나중에 같은 스레드로 이어서 실행
follow_up = app.invoke(
    {"messages": ["더 자세한 분석을 추가해줘"]},
    config=config
)

# LangGraph는 그래프를 Mermaid 형식으로 출력 가능
print(app.get_graph().draw_mermaid())

# 출력 예시:
# graph TD
#     __start__ --> research
#     research -->|더 많은 조사 필요| research
#     research -->|충분| analysis
#     analysis --> writing
#     writing --> review
#     review --> __end__

# 반드시 비용 추적 추가
import tiktoken

def estimate_cost(messages, model="gpt-4"):
    enc = tiktoken.encoding_for_model(model)
    total_tokens = sum(len(enc.encode(m["content"])) for m in messages)
    cost_per_1k = 0.03  # GPT-4 기준 (입력)
    return (total_tokens / 1000) * cost_per_1k

import asyncio

async def run_with_timeout(crew, timeout=300):
    try:
        return await asyncio.wait_for(
            asyncio.to_thread(crew.kickoff),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        raise RuntimeError(f"Crew execution timed out after {timeout}s")