Agent 编排引擎:LangGraph vs CrewAI vs AutoGen
原创
灵阙教研团队
S 精选 进阶 |
约 11 分钟阅读
更新于 2026-02-28 AI 导读
Agent 编排引擎:LangGraph vs CrewAI vs AutoGen 三大主流 Agent 编排框架的架构对比与选型指南 编排引擎解决什么问题 Agent 编排引擎负责管理 Agent 的生命周期、工作流控制、状态管理和工具调度。核心要回答的问题: 流程控制:Agent 的执行路径如何定义?是固定的还是动态的? 状态管理:多步骤执行中的中间状态如何持久化和传递? 多 Agent...
Agent 编排引擎:LangGraph vs CrewAI vs AutoGen
三大主流 Agent 编排框架的架构对比与选型指南
编排引擎解决什么问题
Agent 编排引擎负责管理 Agent 的生命周期、工作流控制、状态管理和工具调度。核心要回答的问题:
- 流程控制:Agent 的执行路径如何定义?是固定的还是动态的?
- 状态管理:多步骤执行中的中间状态如何持久化和传递?
- 多 Agent 协调:多个 Agent 之间如何通信和协作?
- 容错恢复:步骤失败时如何重试、回退或降级?
编排引擎的核心职责:
任务输入 ──→ [编排引擎] ──→ 最终输出
│
┌───────┼───────┐
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)
工程实践建议
- 从场景出发选框架:不要因为某个框架"热门"就选它,匹配场景最重要
- 先单 Agent 后多 Agent:很多任务单 Agent + 好的 Prompt 就够了
- LangGraph 学习曲线投资:如果项目需要生产级部署,LangGraph 的学习投资值得
- CrewAI 适合快速验证:团队不熟悉 Agent 编排时,用 CrewAI 最快出效果
- AutoGen 适合研究场景:需要灵活的多 Agent 对话实验时,AutoGen 最自由
- 关注版本迭代:三个框架都在快速迭代(尤其是 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