CrewAI vs AutoGen vs LangGraph:多 Agent 框架对比

概述

多 Agent 系统是当前 AI 工程最活跃的方向之一。三大主流框架各有哲学:

  • AutoGen(微软):Agent 之间通过对话协作
  • LangGraph(LangChain):将 Agent 行为建模为状态机
  • CrewAI:以"团队角色"为核心的任务编排

本文通过同一个任务在三个框架中的实现,直观对比它们的设计理念、开发体验和适用场景。

框架概览

维度 AutoGen LangGraph CrewAI
开发方 Microsoft Research LangChain CrewAI Inc
核心抽象 Agent + 对话 State + Graph Agent + Task + Crew
编排方式 对话驱动 状态机 任务驱动
循环支持 对话循环 原生图循环 任务循环(有限)
代码执行 内置(Docker/Local) 需自行集成 不内置
持久化 无原生 Checkpointer 无原生
Human-in-Loop 内置三种模式 原生中断点 有限
学习曲线 最低
适合场景 代码生成/多角色讨论 复杂工作流/生产部署 快速原型/角色协作

任务定义

我们用同一个任务在三个框架中实现:

任务:研究一家公司的竞争格局,编写分析报告。包含三个子任务:

  1. 搜索该公司及其竞争对手的最新信息
  2. 分析竞争优势和劣势
  3. 撰写结构化的分析报告

AutoGen 实现

import autogen

config_list = [{"model": "gpt-4o", "api_key": "sk-xxx"}]
llm_config = {"config_list": config_list, "temperature": 0}

# 定义三个专业 Agent
researcher = autogen.AssistantAgent(
    name="researcher",
    system_message="""你是一个行业研究员。
你的职责是搜索和收集公司及其竞争对手的关键信息。
关注:市场份额、产品线、最新动态、财务指标。
输出结构化的调研数据。""",
    llm_config=llm_config,
)

analyst = autogen.AssistantAgent(
    name="analyst",
    system_message="""你是一个竞争分析师。
基于研究员提供的数据,进行深度分析:
1. SWOT 分析
2. 竞争矩阵
3. 市场定位对比
输出结论明确、有数据支撑的分析结果。""",
    llm_config=llm_config,
)

writer = autogen.AssistantAgent(
    name="writer",
    system_message="""你是一个商业报告撰写专家。
基于分析师的结论,撰写一份专业的竞争分析报告。
报告结构:
1. 执行摘要
2. 公司概况
3. 竞争格局分析
4. SWOT 分析
5. 战略建议
完成后以 TERMINATE 结尾。""",
    llm_config=llm_config,
)

user_proxy = autogen.UserProxyAgent(
    name="admin",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: "TERMINATE" in x.get("content", ""),
)

# 群聊模式
groupchat = autogen.GroupChat(
    agents=[user_proxy, researcher, analyst, writer],
    messages=[],
    max_round=15,
    speaker_selection_method="auto",
)

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

# 执行
user_proxy.initiate_chat(
    manager,
    message="请对特斯拉(Tesla)进行竞争格局分析。"
            "重点关注电动汽车市场的竞争态势。",
)

AutoGen 特点

  • 代码量少,设置直观
  • Agent 之间自由对话,LLM 决定谁发言
  • 灵活但不够可控,对话可能跑偏
  • 适合探索性任务和多角色讨论

LangGraph 实现

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 定义状态
class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    company: str
    research_data: str
    analysis: str
    report: str
    current_phase: str

# 节点定义
def research_node(state: ResearchState) -> dict:
    """研究阶段:收集信息"""
    prompt = f"""作为行业研究员,搜索并整理以下公司的竞争信息:
公司:{state['company']}

请收集:
1. 主要竞争对手(至少3个)
2. 各公司市场份额
3. 核心产品对比
4. 最新战略动态

输出结构化的调研数据。"""

    response = llm.invoke([
        SystemMessage(content="你是行业研究员,专注收集准确的市场数据。"),
        HumanMessage(content=prompt),
    ])

    return {
        "research_data": response.content,
        "current_phase": "research_done",
        "messages": [AIMessage(content=f"[Research] {response.content}")],
    }

def analysis_node(state: ResearchState) -> dict:
    """分析阶段:深度分析"""
    prompt = f"""基于以下调研数据,进行竞争分析:

{state['research_data']}

请输出:
1. SWOT 分析(针对 {state['company']})
2. 竞争矩阵(与主要竞争对手对比)
3. 竞争优势和风险
4. 市场趋势预判"""

    response = llm.invoke([
        SystemMessage(content="你是竞争分析专家。"),
        HumanMessage(content=prompt),
    ])

    return {
        "analysis": response.content,
        "current_phase": "analysis_done",
        "messages": [AIMessage(content=f"[Analysis] {response.content}")],
    }

def quality_check_node(state: ResearchState) -> dict:
    """质量检查:判断分析是否充分"""
    prompt = f"""审查以下竞争分析,判断质量是否达标:

{state['analysis']}

检查点:
1. 是否包含 SWOT 四个维度
2. 是否有数据支撑
3. 是否覆盖至少3个竞争对手
4. 是否有可行的战略建议

如果质量达标,回复 "PASS"。
如果需要补充,回复 "NEEDS_IMPROVEMENT" 并说明原因。"""

    response = llm.invoke([HumanMessage(content=prompt)])
    return {
        "current_phase": "pass" if "PASS" in response.content else "needs_improvement",
        "messages": [AIMessage(content=f"[QA] {response.content}")],
    }

def report_node(state: ResearchState) -> dict:
    """报告生成阶段"""
    prompt = f"""基于以下分析结果,撰写专业的竞争分析报告:

调研数据:
{state['research_data']}

分析结论:
{state['analysis']}

报告结构:
# {state['company']} 竞争格局分析报告
## 1. 执行摘要
## 2. 公司概况
## 3. 竞争格局
## 4. SWOT 分析
## 5. 战略建议"""

    response = llm.invoke([
        SystemMessage(content="你是商业报告撰写专家。"),
        HumanMessage(content=prompt),
    ])

    return {
        "report": response.content,
        "current_phase": "complete",
        "messages": [AIMessage(content=f"[Report] {response.content}")],
    }

# 构建图
builder = StateGraph(ResearchState)

builder.add_node("research", research_node)
builder.add_node("analyze", analysis_node)
builder.add_node("quality_check", quality_check_node)
builder.add_node("report", report_node)

builder.add_edge(START, "research")
builder.add_edge("research", "analyze")
builder.add_edge("analyze", "quality_check")

# 条件路由:质量检查通过则生成报告,否则重新分析
def quality_router(state: ResearchState) -> str:
    if state["current_phase"] == "pass":
        return "report"
    return "analyze"  # 循环:回到分析阶段

builder.add_conditional_edges("quality_check", quality_router)
builder.add_edge("report", END)

graph = builder.compile()

# 执行
result = graph.invoke({
    "messages": [],
    "company": "特斯拉(Tesla)",
    "research_data": "",
    "analysis": "",
    "report": "",
    "current_phase": "start",
})

print(result["report"])

LangGraph 特点

  • 流程完全可控,每步做什么一目了然
  • 支持条件路由和循环(质量不达标自动重做)
  • 状态显式传递,可审计可调试
  • 代码量最多,但可控性最强
  • 适合生产级工作流和复杂业务逻辑

CrewAI 实现

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

# 工具
search_tool = SerperDevTool()

# 定义 Agent(角色)
researcher = Agent(
    role="行业研究员",
    goal="收集{company}及其竞争对手的全面市场信息",
    backstory="你是一位资深的行业研究员,擅长从公开信息中"
              "提取关键市场数据和竞争情报。你的调研以数据为驱动,"
              "注重信息的准确性和时效性。",
    tools=[search_tool],
    verbose=True,
    llm="gpt-4o",
)

analyst = Agent(
    role="竞争分析师",
    goal="对{company}进行深度竞争分析,识别关键优势和风险",
    backstory="你是一位波士顿咨询级别的竞争分析师,"
              "擅长 SWOT 分析、波特五力分析和竞争矩阵构建。"
              "你的分析以数据为基础,结论清晰明确。",
    verbose=True,
    llm="gpt-4o",
)

writer = Agent(
    role="商业报告作者",
    goal="将分析结果撰写为专业的竞争分析报告",
    backstory="你是一位优秀的商业写作专家,"
              "擅长将复杂的分析结果转化为清晰易懂的报告。"
              "你的报告结构严谨、论据充分、结论可操作。",
    verbose=True,
    llm="gpt-4o",
)

# 定义任务
research_task = Task(
    description="""调研{company}的竞争格局:
    1. 识别主要竞争对手(至少3个)
    2. 收集各公司市场份额数据
    3. 对比核心产品和技术路线
    4. 整理最新战略动态和融资信息
    输出结构化的调研报告。""",
    expected_output="包含竞争对手信息、市场数据和最新动态的结构化调研报告",
    agent=researcher,
)

analysis_task = Task(
    description="""基于调研数据,对{company}进行深度竞争分析:
    1. SWOT 分析
    2. 竞争矩阵(产品/技术/市场/财务)
    3. 核心竞争优势识别
    4. 关键风险和威胁评估
    5. 市场趋势预判""",
    expected_output="包含 SWOT 分析、竞争矩阵和战略洞察的分析报告",
    agent=analyst,
    context=[research_task],  # 依赖研究任务的输出
)

report_task = Task(
    description="""基于调研和分析结果,撰写完整的竞争分析报告:
    结构:
    1. 执行摘要(300字以内)
    2. 公司概况
    3. 竞争格局分析
    4. SWOT 分析详解
    5. 战略建议(至少3条可行建议)""",
    expected_output="专业的竞争分析报告,包含执行摘要和战略建议",
    agent=writer,
    context=[research_task, analysis_task],
    output_file="competition_report.md",  # 自动保存到文件
)

# 创建 Crew 并执行
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.sequential,  # 顺序执行
    verbose=True,
)

result = crew.kickoff(
    inputs={"company": "特斯拉(Tesla)"}
)

print(result)

CrewAI 特点

  • 代码最简洁,概念最直观
  • "角色 + 任务 + 团队"的比喻非常容易理解
  • backstory 让角色定义更丰富
  • 任务之间的 context 自动传递上下文
  • 适合快速原型和业务人员理解

深度对比

控制流

模式 AutoGen LangGraph CrewAI
顺序执行 initiate_chats 线性边 Process.sequential
并行执行 不支持 并行节点 不支持
条件分支 LLM 隐式决策 条件边(显式) 不支持
循环 对话循环 图循环 不支持
嵌套 嵌套对话 子图 不支持
人工审批 human_input_mode interrupt_before 有限

状态管理

AutoGen:    消息历史(隐式)
            messages = [msg1, msg2, ...]
            无法自定义额外状态

LangGraph:  TypedDict(显式)
            {messages, company, research_data, analysis, ...}
            完全自定义,reducer 控制更新方式

CrewAI:     任务输出(隐式)
            task.output -> next_task.context
            自动传递,不可精确控制

可观测性

维度 AutoGen LangGraph CrewAI
对话日志 verbose=True stream events verbose=True
状态查看 消息列表 状态快照 任务输出
可视化 Mermaid 图
回溯/重放 不支持 Checkpointer 不支持
LangSmith 需集成 原生支持 需集成

扩展性

能力 AutoGen LangGraph CrewAI
自定义 Agent 继承 ConversableAgent 自定义节点函数 继承 Agent
自定义工具 register_function @tool 装饰器 crewai_tools
外部集成 需手动 LangChain 全生态 crewai_tools
部署方案 无官方 LangGraph Platform CrewAI Enterprise

选型指南

选 AutoGen 当你需要

  • 多角色自由讨论/辩论
  • 代码生成和自动执行
  • 快速验证多 Agent 协作的想法
  • 人工灵活介入对话

选 LangGraph 当你需要

  • 生产级可控的工作流
  • 复杂条件分支和循环
  • 状态持久化和断点恢复
  • 与 LangChain 生态深度集成
  • 精确的流程可视化和调试

选 CrewAI 当你需要

  • 最快的原型开发速度
  • 业务团队也能理解的代码
  • 角色驱动的任务分配
  • 简单的顺序/层级任务流

决策矩阵

                    开发速度
                      |
              CrewAI  *
                      |
          AutoGen *   |
                      |
                      |   * LangGraph
                      |
     ─────────────────┼─────────────── 可控性/生产就绪
                      |

混合使用

实际项目中,可以混合使用:

# LangGraph 做顶层编排,内部节点用 CrewAI 或 AutoGen
from langgraph.graph import StateGraph

def crew_research_node(state):
    """用 CrewAI 执行研究子任务"""
    crew = Crew(agents=[researcher], tasks=[research_task])
    result = crew.kickoff(inputs={"company": state["company"]})
    return {"research_data": str(result)}

def autogen_analysis_node(state):
    """用 AutoGen 执行分析讨论"""
    # ... AutoGen 群聊讨论
    return {"analysis": discussion_result}

# LangGraph 编排全局流程
builder = StateGraph(State)
builder.add_node("research", crew_research_node)
builder.add_node("analysis", autogen_analysis_node)
builder.add_edge("research", "analysis")

总结

没有"最好"的框架,只有"最适合"的选择。理解每个框架的核心假设和设计取舍,根据你的具体需求(开发速度 vs 可控性 vs 生产就绪)做出选择。对于大多数生产项目,建议从 LangGraph 开始;对于快速原型,CrewAI 是最快路径;对于需要自由讨论的场景,AutoGen 最为自然。


Maurice | maurice_wen@proton.me