CrewAI vs AutoGen vs LangGraph：多 Agent 框架终决

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role='资深市场研究员',
    goal='收集并分析 {topic} 的市场数据',
    backstory='你是一位拥有 15 年经验的市场研究专家，擅长数据分析和趋势预测',
    tools=[search_tool, calculator_tool],
    llm='gpt-4o',
    verbose=True,
    max_iter=10,
    memory=True,
    allow_delegation=True
)

analyst = Agent(
    role='金融分析师',
    goal='基于研究数据生成投资建议',
    backstory='你是华尔街资深分析师，专注于科技行业投资',
    llm='claude-3.5-sonnet',
    verbose=True
)

writer = Agent(
    role='技术报告撰写人',
    goal='将分析结果转化为易读的投资报告',
    backstory='你是前彭博社记者，擅长将复杂金融概念通俗化',
    llm='gpt-4o'
)

research_task = Task(
    description='研究 {topic} 市场规模、竞争格局和增长趋势',
    agent=researcher,
    expected_output='包含数据图表的市场研究报告'
)

analysis_task = Task(
    description='基于研究报告生成投资建议',
    agent=analyst,
    expected_output='投资评级和目标价分析',
    context=[research_task]
)

writing_task = Task(
    description='撰写最终投资报告',
    agent=writer,
    expected_output='格式完整的投资分析报告',
    context=[research_task, analysis_task]
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,
    verbose=True,
    memory=True,
    embedder={'provider': 'openai', 'config': {'model': 'text-embedding-3-small'}}
)

result = crew.kickoff(inputs={'topic': 'AGI 芯片'})

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# 创建多个专业 Agent
data_scientist = AssistantAgent(
    name="DataScientist",
    system_message="你是数据科学家，负责数据分析和建模。每次回复以 'DATA:' 开头。",
    llm_config={"model": "gpt-4o", "temperature": 0.1}
)

domain_expert = AssistantAgent(
    name="DomainExpert",
    system_message="你是行业专家，提供领域知识。每次回复以 'EXPERT:' 开头。",
    llm_config={"model": "claude-3.5-sonnet"}
)

reviewer = AssistantAgent(
    name="Reviewer",
    system_message="你是质量审查员，检查分析结果的准确性。回复以 'REVIEW:' 开头。如果结果满意，说 'APPROVED'。",
    llm_config={"model": "gpt-4o"}
)

user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=5,
    code_execution_config={"work_dir": "workspace", "use_docker": True}
)

group_chat = GroupChat(
    agents=[user_proxy, data_scientist, domain_expert, reviewer],
    messages=[],
    max_round=20,
    speaker_selection_method="auto"  # 自动选择下一个发言者
)

manager = GroupChatManager(groupchat=group_chat, llm_config={"model": "gpt-4o"})

user_proxy.initiate_chat(
    manager,
    message="分析 2026 年 AGI 行业投资机会"
)

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

class ResearchState(TypedDict):
    query: str
    documents: list
    analysis: str
    report: str
    review_passed: bool
    iteration: int

def research_node(state: ResearchState) -> ResearchState:
    docs = retriever.invoke(state["query"])
    return {"documents": docs, "iteration": state["iteration"] + 1}

def analyze_node(state: ResearchState) -> ResearchState:
    analysis = analyzer_llm.invoke(state["documents"])
    return {"analysis": analysis}

def review_node(state: ResearchState) -> ResearchState:
    review = reviewer_llm.invoke(state["analysis"])
    return {"review_passed": review.is_approved}

def route_review(state: ResearchState) -> str:
    if state["review_passed"] or state["iteration"] >= 3:
        return "write"
    return "research"

graph = StateGraph(ResearchState)
graph.add_node("research", research_node)
graph.add_node("analyze", analyze_node)
graph.add_node("review", review_node)
graph.add_node("write", write_node)

graph.set_entry_point("research")
graph.add_edge("research", "analyze")
graph.add_edge("analyze", "review")
graph.add_conditional_edges("review", route_review)
graph.add_edge("write", END)

app = graph.compile(checkpointer=PostgresSaver(...))