Agent 编排引擎:LangGraph vs CrewAI vs AutoGen

三大主流 Agent 编排框架的架构对比与选型指南


编排引擎解决什么问题

Agent 编排引擎负责管理 Agent 的生命周期、工作流控制、状态管理和工具调度。核心要回答的问题:

  1. 流程控制:Agent 的执行路径如何定义?是固定的还是动态的?
  2. 状态管理:多步骤执行中的中间状态如何持久化和传递?
  3. 多 Agent 协调:多个 Agent 之间如何通信和协作?
  4. 容错恢复:步骤失败时如何重试、回退或降级?
编排引擎的核心职责:

  任务输入 ──→ [编排引擎] ──→ 最终输出
                  │
          ┌───────┼───────┐
          v       v       v
        流程    状态    Agent
        控制    管理    生命周期
        │       │       │
        v       v       v
      分支     持久化   创建/销毁
      循环     快照     工具注入
      并行     恢复     上下文管理

三大框架总览

维度 LangGraph CrewAI AutoGen
开发者 LangChain CrewAI Inc. Microsoft
核心抽象 图(Graph) 角色(Role/Crew) 对话(Conversation)
状态管理 显式状态图 隐式(任务上下文) 消息历史
流程控制 节点+边(条件路由) 顺序/层级 对话驱动
学习曲线 中-高
灵活性 极高
生产就绪度
适用场景 复杂工作流 角色协作任务 多 Agent 对话

一、LangGraph

核心概念

LangGraph 的核心抽象是有向图:节点(Node)代表计算步骤,边(Edge)代表执行路径,状态(State)在图中流转。

LangGraph 的思维模型:

  状态 ──→ 节点A ──→ 条件边 ──→ 节点B ──→ 节点C ──→ 输出
                        │
                        └──→ 节点D ──→ 节点C ──→ 输出

架构设计

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

# 1. 定义状态 Schema
class AgentState(TypedDict):
    task: str
    plan: list[str]
    current_step: int
    results: Annotated[list[str], add]  # 使用 reducer 合并结果
    final_answer: str
    errors: list[str]

# 2. 定义节点(每个节点是一个函数)
def planner(state: AgentState) -> dict:
    """规划节点:将任务分解为步骤"""
    plan = llm.generate(
        f"将以下任务分解为执行步骤:{state['task']}"
    )
    return {"plan": plan.steps, "current_step": 0}

def executor(state: AgentState) -> dict:
    """执行节点:执行当前步骤"""
    step = state["plan"][state["current_step"]]
    result = llm_with_tools.run(step)
    return {
        "results": [result],
        "current_step": state["current_step"] + 1
    }

def reviewer(state: AgentState) -> dict:
    """审查节点:检查执行结果"""
    review = llm.generate(
        f"审查以下执行结果是否满足要求:\n"
        f"任务:{state['task']}\n"
        f"结果:{state['results']}"
    )
    return {"final_answer": review}

# 3. 定义条件路由
def should_continue(state: AgentState) -> str:
    """条件边:判断是否需要继续执行"""
    if state["current_step"] < len(state["plan"]):
        return "executor"    # 继续执行下一步
    else:
        return "reviewer"    # 所有步骤完成,进入审查

# 4. 构建图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("planner", planner)
graph.add_node("executor", executor)
graph.add_node("reviewer", reviewer)

# 添加边
graph.set_entry_point("planner")
graph.add_conditional_edges(
    "planner",
    lambda s: "executor" if s["plan"] else END
)
graph.add_conditional_edges("executor", should_continue)
graph.add_edge("reviewer", END)

# 编译
app = graph.compile()

# 5. 执行
result = app.invoke({"task": "分析项目代码质量"})

LangGraph 的核心优势

1. 持久化与检查点

from langgraph.checkpoint.sqlite import SqliteSaver

# 启用检查点持久化
checkpointer = SqliteSaver.from_conn_string("./checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

# 执行时指定 thread_id,支持中断和恢复
config = {"configurable": {"thread_id": "task-001"}}
result = app.invoke(initial_state, config=config)

# 中断后恢复
result = app.invoke(None, config=config)  # 从检查点恢复

2. Human-in-the-Loop

from langgraph.prebuilt import ToolNode

# 在特定节点设置中断点
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["executor"],  # 在执行前暂停等待人工审批
)

# 执行到中断点
result = app.invoke(state, config)
# result.status == "interrupted"

# 人工审批后继续
app.invoke(
    {"approved": True},  # 人工输入
    config
)

3. 子图复用

# 定义可复用的子图
research_subgraph = StateGraph(ResearchState)
research_subgraph.add_node("search", web_search)
research_subgraph.add_node("analyze", analyze_results)
# ...
research_compiled = research_subgraph.compile()

# 在主图中嵌入子图
main_graph = StateGraph(MainState)
main_graph.add_node("research", research_compiled)
main_graph.add_node("implement", implement_node)
main_graph.add_edge("research", "implement")

二、CrewAI

核心概念

CrewAI 的核心抽象是角色(Agent)和团队(Crew):每个 Agent 有明确的角色、目标和工具,Crew 定义 Agent 之间的协作方式。

CrewAI 的思维模型:

  Crew(团队)
    │
    ├── Agent: 研究员(Role + Goal + Backstory + Tools)
    ├── Agent: 分析师(Role + Goal + Backstory + Tools)
    └── Agent: 撰稿人(Role + Goal + Backstory + Tools)
         │
         └── Tasks: 按顺序/并行分配给 Agent

基础使用

from crewai import Agent, Task, Crew, Process

# 1. 定义 Agent(角色)
researcher = Agent(
    role="高级研究员",
    goal="搜集和分析关于 {topic} 的最新信息",
    backstory=(
        "你是一位经验丰富的技术研究员,"
        "擅长从大量信息中提取关键洞察。"
    ),
    tools=[web_search_tool, document_reader_tool],
    llm="claude-opus-4-6",
    verbose=True,
    memory=True,          # 启用记忆
    max_iter=5,           # 最大迭代次数
    allow_delegation=True # 允许委派任务给其他 Agent
)

analyst = Agent(
    role="数据分析师",
    goal="对研究数据进行深入分析,提取可操作的洞察",
    backstory="你是一位数据科学专家,擅长统计分析和趋势预测。",
    tools=[python_tool, chart_tool],
    llm="claude-opus-4-6",
)

writer = Agent(
    role="技术撰稿人",
    goal="将分析结果撰写成清晰、有说服力的报告",
    backstory="你是一位技术写作专家,能将复杂概念表达得简洁易懂。",
    tools=[file_write_tool],
    llm="claude-opus-4-6",
)

# 2. 定义任务
research_task = Task(
    description="搜集关于 {topic} 的最新技术进展和市场动态",
    expected_output="结构化的研究报告,包含 5-10 个关键发现",
    agent=researcher,
)

analysis_task = Task(
    description="分析研究数据,识别趋势和机会",
    expected_output="数据分析报告,包含图表和关键指标",
    agent=analyst,
    context=[research_task],  # 依赖 research_task 的输出
)

report_task = Task(
    description="撰写最终的综合分析报告",
    expected_output="2000 字的专业分析报告",
    agent=writer,
    context=[research_task, analysis_task],
    output_file="report.md",
)

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

# 4. 执行
result = crew.kickoff(inputs={"topic": "AI Agent 编排引擎"})

CrewAI 的核心特性

1. 流程模式

# 顺序执行:Task 按定义顺序执行
crew = Crew(process=Process.sequential, ...)

# 层级执行:Manager Agent 自动协调
crew = Crew(
    process=Process.hierarchical,
    manager_llm="claude-opus-4-6",
    ...
)
# Manager 会自动决定哪个 Agent 执行哪个任务

2. 委派与协作

# Agent 可以将任务委派给其他 Agent
researcher = Agent(
    role="高级研究员",
    allow_delegation=True,  # 允许委派
    # 当研究员遇到需要数据分析的部分,会自动委派给分析师
)

3. 记忆系统

crew = Crew(
    memory=True,
    # CrewAI 内置三种记忆:
    # - Short-term: 当前任务的对话历史
    # - Long-term: 跨任务的经验积累
    # - Entity: 实体知识(人物、组织、概念)
)

三、AutoGen

核心概念

AutoGen 的核心抽象是对话(Conversation):Agent 之间通过消息传递进行协作,对话模式定义了谁可以和谁说话。

AutoGen 的思维模型:

  GroupChat(群聊)
    │
    ├── AssistantAgent(AI 助手)
    ├── UserProxyAgent(用户代理)
    └── CustomAgent(自定义 Agent)
         │
         └── 通过消息传递协作,Speaker Selection 决定谁发言

基础使用(AutoGen 0.4+)

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 1. 定义模型客户端
model_client = OpenAIChatCompletionClient(
    model="claude-opus-4-6",
    api_key="...",
)

# 2. 定义 Agent
planner = AssistantAgent(
    name="planner",
    model_client=model_client,
    system_message=(
        "你是一个项目规划专家。"
        "将任务分解为具体可执行的步骤。"
    ),
)

coder = AssistantAgent(
    name="coder",
    model_client=model_client,
    system_message=(
        "你是一个编程专家。"
        "根据规划编写高质量的代码。"
    ),
    tools=[code_execution_tool],
)

reviewer = AssistantAgent(
    name="reviewer",
    model_client=model_client,
    system_message=(
        "你是一个代码审查专家。"
        "审查代码质量并提出改进建议。"
        "如果代码质量满足要求,回复 APPROVE。"
    ),
)

# 3. 组建群聊
team = RoundRobinGroupChat(
    participants=[planner, coder, reviewer],
    max_turns=10,
)

# 4. 执行
from autogen_agentchat.conditions import TextMentionTermination
termination = TextMentionTermination("APPROVE")

result = await team.run(
    task="编写一个 Python 脚本,分析 CSV 文件中的销售数据",
    termination_condition=termination,
)

AutoGen 的核心特性

1. 灵活的对话模式

# 轮询模式:Agent 轮流发言
team = RoundRobinGroupChat(participants=[a, b, c])

# 选择器模式:LLM 决定谁发言
from autogen_agentchat.teams import SelectorGroupChat
team = SelectorGroupChat(
    participants=[planner, coder, reviewer],
    model_client=model_client,
    # LLM 根据对话上下文选择下一个发言者
)

# Swarm 模式:Agent 自主切换
from autogen_agentchat.teams import Swarm
team = Swarm(
    participants=[agent_a, agent_b, agent_c],
    # Agent 通过 HandoffMessage 自主切换控制权
)

2. 代码执行

from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor

code_executor = DockerCommandLineCodeExecutor(
    image="python:3.12-slim",
    timeout=60,
    work_dir="./workspace",
)

coder = AssistantAgent(
    name="coder",
    code_executor=code_executor,  # 生成的代码在 Docker 中执行
)

3. 嵌套对话

# 外层团队可以包含内层团队
inner_team = RoundRobinGroupChat(
    participants=[coder, reviewer]
)

outer_team = RoundRobinGroupChat(
    participants=[planner, inner_team, deployer]
)
# planner 规划 -> inner_team 编码+审查 -> deployer 部署

深度对比

状态管理

LangGraph:
  - 显式状态 Schema(TypedDict)
  - 状态在图的边上传递
  - 内置持久化(SQLite/PostgreSQL)
  - 支持检查点和恢复
  - 最精确的状态控制

CrewAI:
  - 隐式状态(任务上下文自动传递)
  - context 参数声明任务依赖
  - 内置记忆系统(短期/长期/实体)
  - 状态管理对用户透明
  - 简单但灵活性有限

AutoGen:
  - 消息历史即状态
  - 每个 Agent 维护自己的消息列表
  - 群聊消息对所有参与者可见
  - 可通过自定义 Agent 实现复杂状态
  - 灵活但需要手动管理

流程控制

LangGraph:
  ┌─────────┐     ┌─────────┐     ┌─────────┐
  │ Node A  │──→──│ Node B  │──→──│ Node C  │
  └─────────┘  │  └─────────┘     └─────────┘
               │
               └──→──┌─────────┐
                     │ Node D  │
                     └─────────┘

  特点:完全自定义的 DAG,条件分支,循环,子图
  适合:复杂的、需要精确控制的工作流

CrewAI:
  Task1 ──→ Task2 ──→ Task3 ──→ Task4
  (顺序)    或     Manager 动态分配(层级)

  特点:简单直观,角色驱动
  适合:以角色分工为核心的任务

AutoGen:
  Agent A ←──对话──→ Agent B
       ↑                ↓
       └──── Agent C ───┘

  特点:对话驱动,灵活的发言选择
  适合:需要多轮讨论和迭代的场景

扩展性

维度 LangGraph CrewAI AutoGen
自定义节点 任意 Python 函数 Agent + Task 继承 Agent 类
工具集成 LangChain Tools + 自定义 CrewAI Tools + LangChain Function Calling + 自定义
模型支持 任意 LLM OpenAI/Anthropic/本地 OpenAI/Anthropic/本地
子图/嵌套 原生支持 不支持 嵌套群聊
流式输出 原生支持 有限 原生支持
持久化 SQLite/PostgreSQL 内置记忆 需自定义

选型决策矩阵

你的场景是什么?
     │
     ├── 复杂工作流(条件分支、循环、子图)
     │    └── LangGraph
     │
     ├── 角色协作(研究员+分析师+撰稿人)
     │    └── CrewAI
     │
     ├── 多轮对话式协作(讨论+迭代)
     │    └── AutoGen
     │
     ├── 需要精确状态控制和持久化
     │    └── LangGraph
     │
     ├── 快速原型验证
     │    └── CrewAI(最快上手)
     │
     ├── 代码生成+执行场景
     │    └── AutoGen(内置代码执行器)
     │
     └── 生产级部署
          └── LangGraph(LangGraph Platform)
               或根据团队技术栈选择

不同场景的推荐选择

场景 首选 理由
客服路由系统 LangGraph 需要精确的状态机和条件路由
内容创作团队 CrewAI 角色协作模式天然适配
代码审查系统 AutoGen 多轮对话讨论代码质量
数据处理流水线 LangGraph DAG 模式 + 持久化 + 恢复
研究助手 CrewAI 多角色分工,简单直接
RAG 增强系统 LangGraph 需要复杂的检索-推理-验证循环
教学模拟 AutoGen 多角色对话模拟

混合使用策略

实际项目中,三个框架可以互补使用:

外层编排(LangGraph):负责整体工作流控制
  │
  ├── 研究阶段 ──→ CrewAI Crew(研究员+分析师)
  │
  ├── 讨论阶段 ──→ AutoGen GroupChat(多角色辩论)
  │
  └── 实现阶段 ──→ LangGraph 子图(代码生成+测试+部署)
# 在 LangGraph 节点中调用 CrewAI Crew
def research_node(state):
    crew = Crew(agents=[researcher, analyst], ...)
    result = crew.kickoff(inputs={"topic": state["topic"]})
    return {"research_results": result}

# 在 LangGraph 节点中调用 AutoGen GroupChat
async def discussion_node(state):
    team = SelectorGroupChat(participants=[...])
    result = await team.run(task=state["discussion_topic"])
    return {"discussion_outcome": result}

# LangGraph 主图
graph = StateGraph(State)
graph.add_node("research", research_node)
graph.add_node("discuss", discussion_node)
graph.add_node("implement", implement_node)

工程实践建议

  1. 从场景出发选框架:不要因为某个框架"热门"就选它,匹配场景最重要
  2. 先单 Agent 后多 Agent:很多任务单 Agent + 好的 Prompt 就够了
  3. LangGraph 学习曲线投资:如果项目需要生产级部署,LangGraph 的学习投资值得
  4. CrewAI 适合快速验证:团队不熟悉 Agent 编排时,用 CrewAI 最快出效果
  5. AutoGen 适合研究场景:需要灵活的多 Agent 对话实验时,AutoGen 最自由
  6. 关注版本迭代:三个框架都在快速迭代(尤其是 AutoGen 0.4 重写),注意跟进最新版本

参考资料

  • LangGraph 官方文档:langchain-ai.github.io/langgraph/
  • CrewAI 官方文档:docs.crewai.com
  • AutoGen 官方文档:microsoft.github.io/autogen/
  • LangChain 博客:多框架对比分析
  • Andrew Ng 的 AI Agent 设计模式课程

Maurice | maurice_wen@proton.me