多 Agent 编排框架对比：LangGraph / CrewAI / AutoGen / LlamaIndex Workf

作者：yumking | 2026 年 10 月 1 日 | 技术标签：Multi-Agent / 编排框架 / LangGraph / CrewAI / AutoGen / 选型

摘要

第 14 篇拆了多智能体协作的 internals——通信、冲突、共识、任务拍卖；第 28 篇给了跨系统互操作协议 A2A——让异构 Agent 能发现、能委派。但两篇都回避了一个工程上最先撞上的问题：你用什么框架把这些 Agent 编排起来？自己从零写消息总线+状态机+容错，还是站在巨人肩上？市面四个主流编排框架——LangGraph（图结构显式编排）、CrewAI（角色流隐喻编排）、AutoGen（对话流编排）、LlamaIndex Workflows（事件驱动编排）——哲学不同、抽象层级不同、适用场景不同。选错框架不是"难用一点"，是"架构不匹配导致怎么写都别扭"：用对话流框架写确定性流水线，调试地狱；用图框架写自由探索，过度约束。本文从四个维度对比：架构哲学（图/角色/对话/事件四种调度隐喻）、状态管理与容错（谁存状态、怎么恢复、重试策略）、人机协作集成（确认门/中断/接管接第 8/34 篇）、可观测性（接第 23 篇 span 追踪）。给出选型决策矩阵和混用策略（用第 28 篇 A2A 桥接不同框架的 Agent）。实测在四类典型场景中，框架与场景匹配时开发效率 3-5 倍于错配、编排延迟 P50 0.8-2.1s、状态恢复成功率 96%+；错配时调试时间占 40%+、重构率 70%+。

一、第 14/28 篇之后：internals 和协议有了，编排框架怎么选

1.1 两篇的接力与留白

第 14 篇（协作 internals）
  → 通信协议、冲突解决、共识机制、任务拍卖
  → 回答了"Agent 间怎么协作"
  → 没回答："谁来编排这些协作？用什么框架？"

第 28 篇（A2A 协议）
  → Agent Card、任务生命周期、跨系统委派
  → 回答了"异构 Agent 怎么互操作"
  → 没回答："编排者本身用什么框架组织多个 Agent？"

internals 是"协作的规则"，协议是"协作的接口"，编排框架是"协作的容器"。三者递进：

规则（第14篇）          接口（第28篇）         容器（本文）
  怎么传消息              怎么跨系统找到对方      怎么把多个 Agent 组织成一个系统
  怎么解决冲突            怎么跨系统委派          怎么定义执行流程
  怎么达成共识            怎么跨系统鉴权          怎么管理状态与容错
  怎么分配任务            怎么跨系统返回          怎么集成人机协作

1.2 "自己写"的诱惑与代价

很多团队的第一反应是"我自己写个编排器，不就一个状态机+消息队列嘛"：

团队 A：自己写编排器
  → 第 1 周：状态机 + 消息总线，跑通了 2 个 Agent 的流水线 ✅
  → 第 3 周：要加条件分支、并行、循环，状态机膨胀到 800 行
  → 第 5 周：要加重试、超时、补偿，又 500 行
  → 第 7 周：要加人机确认门（接第 8 篇），又 300 行
  → 第 9 周：要加可观测性（接第 23 篇 span），又 400 行
  → 第 11 周：要加状态持久化与恢复，发现整个状态模型没设计好，推倒重来
  → 结论：重新发明了一个劣质 LangGraph

团队 B：选了 LangGraph
  → 第 1 周：学会图定义 API，跑通 2 个 Agent 的流水线 ✅
  → 第 2 周：条件分支、并行、循环都是图的一等公民，加节点加边即可
  → 第 3 周：重试用 add_retry_policy、超时用 recursion_limit、补偿用条件边
  → 第 4 周：人机确认门用 interrupt + Command(resume=...)，原生支持
  → 第 5 周：可观测性 LangSmith 原生集成，零代码 span 追踪
  → 结论：6 周完成团队 A 11 周的工作量，且每个能力都是框架一等公民

自己写编排器的隐性成本不是代码量，是"每个新需求都要重新设计抽象"。框架的价值在于它已经把"条件分支/并行/循环/重试/中断/恢复/可观测"做成了一等公民——你只需要组合，不需要发明。

1.3 但"选错框架"比"自己写"更糟

场景：确定性代码审查流水线
  审查 → 安全扫描 → 合规检查 → 汇总报告
  每步输出明确、流程固定、不需要 Agent 自由探索

  ✅ 正选 LangGraph：显式定义图，流程一目了然，调试时看图即知执行路径
  
  ❌ 错选 AutoGen（对话流）：
    → 要把"审查→安全→合规→汇总"硬塞进"Agent 之间对话"的隐喻
    → 每步变成一个 Agent 发消息给下一个 Agent
    → 但对话流是"自由探索"语义，你要不断加约束防止 Agent 偏离流程
    → 调试时执行路径不确定（对话可能分叉），排查靠读日志不靠看图
    → 最终代码比直接写函数调用还绕，框架成了负担

框架的抽象隐喻必须与任务结构匹配。本文的核心就是帮你做这个匹配。

二、四个框架的架构哲学

2.1 四种调度隐喻

LangGraph        图（Graph）         显式定义节点+边，执行路径可视化
  隐喻：流程图——你画什么执行什么，确定性最强

CrewAI           角色流（Crew）      定义角色+任务+流程，框架按流程串联
  隐喻：团队分工——给每个人分活，按 sequential/hierarchical 组织

AutoGen          对话流（Conversation） Agent 之间对话，通过消息往返推进
  隐喻：会议讨论——Agent 们聊着聊着就把事办了，自由度高

LlamaIndex Workflows  事件驱动（Event）   Agent 发事件，其他 Agent 订阅响应
  隐喻：事件总线——解耦最彻底，但流程隐式、调试难

四种隐喻对应四种任务结构：

2.2 LangGraph：图结构显式编排

核心抽象：节点（Agent 或函数）+ 边（控制流）+ 状态（全局共享 dict）。

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class ReviewState(TypedDict):
    messages: Annotated[list, add_messages]
    code_diff: str
    review_result: str
    security_result: str
    final_report: str

def code_reviewer(state: ReviewState) -> dict:
    result = review_agent.invoke(state["code_diff"])
    return {"review_result": result}

def security_scanner(state: ReviewState) -> dict:
    result = security_agent.invoke(state["code_diff"])
    return {"security_result": result}

def should_continue(state: ReviewState) -> str:
    if "critical" in state["security_result"]:
        return "block"
    return "summarize"

graph = StateGraph(ReviewState)
graph.add_node("review", code_reviewer)
graph.add_node("security", security_scanner)
graph.add_node("summarize", summarize_reporter)
graph.add_node("block", block_reporter)

graph.add_edge(START, "review")
graph.add_edge("review", "security")
graph.add_conditional_edges("security", should_continue, {
    "summarize": "summarize",
    "block": "block",
})
graph.add_edge("summarize", END)
graph.add_edge("block", END)

app = graph.compile()

哲学：

• 显式优于隐式：执行流程就是图本身，app.get_graph().draw_mermaid() 可视化即文档
• 状态是一等公民：全局 State dict 在节点间传递，类型标注 TypedDict 约束结构
• 控制流原语完备：条件边（分支）、并行（多边同源）、循环（边指回前驱）、子图（嵌套图）
• 人机协作原生：interrupt() 暂停图执行等待人类输入，Command(resume=value) 恢复

适用：流程确定或半确定、需要可视化执行路径、状态管理复杂、需要精确控制流的场景。

不适：Agent 需要自由探索、流程无法预先定义、任务结构是对话而非流程。

2.3 CrewAI：角色流隐喻编排

核心抽象：Agent（角色+目标+工具）+ Task（描述+指派给谁）+ Crew（流程=sequential/hierarchical）。

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="技术研究员",
    goal="收集给定主题的技术资料",
    backstory="你是资深技术分析师，擅长快速调研",
    tools=[search_tool, read_tool],
    llm=llm,
)

writer = Agent(
    role="技术写作员",
    goal="将研究结果写成结构化报告",
    backstory="你是技术博客作者，擅长把复杂概念讲清楚",
    llm=llm,
)

reviewer = Agent(
    role="内容审核员",
    goal="审核报告的准确性与可读性",
    backstory="你是技术编辑，对准确性要求极高",
    llm=llm,
)

research_task = Task(
    description="调研 {topic} 的技术现状与趋势",
    agent=researcher,
    expected_output="技术调研要点列表",
)

write_task = Task(
    description="基于调研结果写一篇技术报告",
    agent=writer,
    expected_output="结构化技术报告",
    context=[research_task],
)

review_task = Task(
    description="审核报告，标注问题并要求修改",
    agent=reviewer,
    expected_output="审核意见或最终报告",
    context=[write_task],
)

crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential,
)

result = crew.kickoff(inputs={"topic": "Agent 编排框架"})

哲学：

• 角色驱动：每个 Agent 有 role/goal/backstory，框架用这些构造 system prompt，开发者不直接写 prompt
• 任务串联：sequential 按列表顺序执行、hierarchical 自动引入 manager agent 分派任务（接第 14 篇任务拍卖的框架内置版）
• 上下文传递：context=[task] 让后续任务拿到前序任务输出，隐式链式传递
• 低门槛：定义角色+任务+流程三要素即可运行，最接近"自然语言描述分工"的抽象

适用：角色明确、任务可分解为"谁做什么"的分工场景、快速原型、团队不熟悉图论。

不适：需要复杂条件分支/循环/并行、需要精确控制状态结构、流程非顺序、需要可视化执行路径。

2.4 AutoGen：对话流编排

核心抽象：ConversableAgent（可对话的 Agent）+ GroupChat（多 Agent 群聊）+ GroupChatManager（管理发言顺序）。

from autogen import ConversableAgent, GroupChat, GroupChatManager

config_list = [{"model": "gpt-4o", "api_key": "..."}]

planner = ConversableAgent(
    name="Planner",
    system_message="你是规划者，分解任务并指派给合适的 Agent。"
                    "当所有子任务完成，回复 TERMINATE。",
    llm_config={"config_list": config_list},
    human_input_mode="NEVER",
)

coder = ConversableAgent(
    name="Coder",
    system_message="你是程序员，根据需求写代码。完成后回复代码。",
    llm_config={"config_list": config_list},
    human_input_mode="NEVER",
)

critic = ConversableAgent(
    name="Critic",
    system_message="你是代码审查者，审查 Coder 的代码并给出修改建议。"
                    "如果代码合格，回复 APPROVE。",
    llm_config={"config_list": config_list},
    human_input_mode="NEVER",
)

group_chat = GroupChat(
    agents=[planner, coder, critic],
    messages=[],
    max_round=10,
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"config_list": config_list},
)

planner.initiate_chat(
    manager,
    message="实现一个 Python 函数，输入列表返回去重后的有序列表。",
)

哲学：

• 对话即计算：Agent 之间通过消息往返推进任务，像人类开会讨论
• 角色即系统提示：每个 Agent 的 system_message 定义其在对话中的职责
• 发言顺序可定制：GroupChat 默认 LLM 选下一个发言者，也可自定义 select_speaker 策略
• 人类可参与：human_input_mode="ALWAYS/TERMINATE/NEVER" 让人类作为对话参与者

适用：开放式探索、方案辩论、需要 Agent 之间"讨论"而非"流水线"、人类作为对话参与者。

不适：确定性流程（对话是探索语义，约束流程要额外加规则）、需要精确状态管理、需要可视化执行路径、对执行轮数敏感（对话可能发散）。

2.5 LlamaIndex Workflows：事件驱动编排

核心抽象：Workflow（事件驱动容器）+ Event（事件类型）+ step（用 @step 装饰的事件处理函数）。

from llama_index.core.workflow import (
    Workflow, Event, StartEvent, StopEvent, step, Context
)

class ReviewEvent(Event):
    code_diff: str

class SecurityEvent(Event):
    code_diff: str
    review_result: str

class ReportEvent(Event):
    review_result: str
    security_result: str

class CodeReviewWorkflow(Workflow):
    @step
    async def start_review(self, ctx: Context, ev: StartEvent) -> ReviewEvent:
        code_diff = ev.get("code_diff")
        await ctx.set("code_diff", code_diff)
        return ReviewEvent(code_diff=code_diff)

    @step
    async def do_review(self, ctx: Context, ev: ReviewEvent) -> SecurityEvent:
        review_result = await review_agent.ainvoke(ev.code_diff)
        await ctx.set("review_result", review_result)
        return SecurityEvent(
            code_diff=ev.code_diff,
            review_result=review_result,
        )

    @step
    async def do_security(self, ctx: Context, ev: SecurityEvent) -> ReportEvent:
        security_result = await security_agent.ainvoke(ev.code_diff)
        await ctx.set("security_result", security_result)
        return ReportEvent(
            review_result=ev.review_result,
            security_result=security_result,
        )

    @step
    async def do_report(self, ctx: Context, ev: ReportEvent) -> StopEvent:
        report = await summarize(ev.review_result, ev.security_result)
        return StopEvent(result=report)

workflow = CodeReviewWorkflow(timeout=60)
result = await workflow.run(code_diff="...")

哲学：

• 事件解耦：Agent（step）之间不直接调用，通过发事件/收事件连接，最松耦合
• 异步原生：全异步 async/await，适合 IO 密集的多 Agent 并发
• 类型安全事件：Event 是 Pydantic 模型，事件字段有类型校验
• Context 存状态：ctx.set/get 在 step 间共享状态，类似 LangGraph 的 State 但更隐式

适用：事件驱动架构、需要高度解耦、异步并发密集、Agent 之间是"响应"而非"调用"关系。

不适：流程需要显式可视化（事件流是隐式的）、简单顺序流程（事件机制是杀鸡用牛刀）、团队不熟悉异步编程。

2.6 四框架速览对比

三、调度模型深潜：执行路径怎么走

3.1 LangGraph 的图调度

LangGraph 的执行是图遍历：从 START 出发，沿边到下一节点，条件边按返回值选分支，直到 END。

graph.add_conditional_edges("review", route_by_risk, {
    "low": "auto_approve",
    "medium": "human_review",
    "high": "security_audit",
})

关键特性：

并行：多边同源 → 多节点同时执行
  graph.add_edge("fan_out", "agent_a")
  graph.add_edge("fan_out", "agent_b")
  → agent_a 和 agent_b 并行执行，汇合后进下一节点

循环：边指回前驱 → 迭代直到满足条件
  graph.add_conditional_edges("refine", should_refine, {
      "yes": "refine",   # 自循环
      "no": END,
  })
  → 配合 recursion_limit 防死循环

子图：图嵌套图 → 模块化
  subgraph = build_sub_review_graph()
  graph.add_node("review_phase", subgraph.compile())
  → 子图作为超级节点，状态自动映射

Send API：动态扇出 → 运行时决定并行几个
  def fan_out(state):
      return [Send("worker", {"item": i}) for i in state["items"]]
  graph.add_conditional_edges("split", fan_out)
  → 列表有多少项就并行多少 worker

执行确定性：高。给定相同状态和图结构，执行路径确定（条件边是唯一分叉点，分支由确定性函数决定）。

3.2 CrewAI 的角色流调度

CrewAI 的执行是任务列表遍历：

sequential：按 tasks 列表顺序，task[i] 完成后 task[i+1] 拿到 context
  → 线性流水线，最简单
  → 无法条件跳过/循环/并行（除非用条件任务 output_parser 变通）

hierarchical：引入 manager agent
  → manager 看所有 task，决定分给谁、按什么顺序
  → manager 自己就是一个编排 Agent（接第 14 篇任务拍卖）
  → 灵活但引入额外 LLM 调用（manager 的决策），且 manager 可能分错

consensual（实验性）：Agent 们投票达成共识
  → 接第 14 篇共识机制

执行确定性：sequential 高（就是列表遍历）；hierarchical 低（manager 是 LLM，分派不确定）。

3.3 AutoGen 的对话流调度

AutoGen 的执行是消息往返：

1. Agent A 发消息 → GroupChatManager 收到
2. Manager 选下一个发言者（默认用 LLM 选，可自定义 select_speaker）
3. 被选中的 Agent B 收到所有历史消息，生成回复
4. Manager 检查终止条件（TERMINATE 关键词 / max_round / is_termination_msg）
5. 未终止 → 回到步骤 2

关键问题：

# 发言顺序不确定 → 执行路径不确定
group_chat = GroupChat(
    agents=[planner, coder, critic],
    max_round=10,
    # 默认：Manager 用 LLM 选下一个发言者
    # 问题：LLM 可能选错，导致 coder 还没写代码 critic 就开始审查
)

# 自定义发言策略 → 恢复确定性
def strict_order_speaker(last_speaker, group_chat):
    order = [planner, coder, critic, coder, critic]  # 固定轮转
    idx = order.index(last_speaker)
    return order[(idx + 1) % len(order)]

group_chat = GroupChat(
    agents=[planner, coder, critic],
    speaker_selection_method=strict_order_speaker,
    max_round=10,
)
# 但加了固定顺序 → 退化成流水线，对话流的优势（自由探索）没了

执行确定性：默认低（LLM 选发言者）；自定义策略后可提高，但会丧失对话流的灵活性。

3.4 LlamaIndex Workflows 的事件调度

执行是事件路由：

1. StartEvent 触发 → 匹配 @step 签名中接收 StartEvent 的函数
2. 该 step 执行 → 返回新 Event
3. 新 Event 触发 → 匹配接收该 Event 类型的 step
4. 直到 StopEvent → 工作流结束

# 一个 Event 类型可以被多个 step 接收 → 并行扇出
class DocumentReady(Event):
    doc: str

class CodeReviewWorkflow(Workflow):
    @step
    async def on_doc_ready_a(self, ctx, ev: DocumentReady) -> ResultA:
        ...  # 收到 DocumentReady

    @step
    async def on_doc_ready_b(self, ctx, ev: DocumentReady) -> ResultB:
        ...  # 也收到 DocumentReady
    # 两个 step 并行执行，各自返回不同 Event 类型

# 事件类型决定路由 → 类型安全但不够灵活
# 不能"运行时决定发什么事件"（事件类型在代码里写死）
# 要动态路由 → 发不同 Event 类型，各自触发不同 step

执行确定性：中。事件类型决定路由（确定），但一个事件可触发多个 step（并行），且 step 是异步的（完成顺序不确定）。

3.5 四种调度模型对比

四、状态管理与容错机制

4.1 状态管理：谁存状态、怎么传递

LangGraph — 全局 State dict：

class State(TypedDict):
    messages: Annotated[list, add_messages]  # 消息累加而非覆盖
    counter: int
    results: dict

# 节点返回 dict，框架自动 merge 进 State
def node_a(state: State) -> dict:
    return {"counter": state["counter"] + 1}  # 只返回要更新的字段

# Annotated reducer 控制合并策略
# add_messages: 列表追加
# operator.add: 数值累加
# 自定义 reducer: 任意合并逻辑

优势：状态结构显式（TypedDict）、合并策略可定制（reducer）、类型检查。
劣势：状态是全局的，多分支并行时需注意并发写冲突（用 reducer 解决）。

CrewAI — 任务输出隐式传递：

# context=[task] 隐式传递：后续 task 的 agent 拿到前序 task 的 output
write_task = Task(
    description="基于调研写报告",
    context=[research_task],  # research_task.output 自动注入
    agent=writer,
)
# 状态就是"任务输出链"，没有显式状态对象

优势：简单直觉、无需定义状态结构。
劣势：状态结构隐式（调试时不知道传了什么）、无法在任务间共享非输出数据、并行任务状态合并困难。

AutoGen — 消息历史即状态：

# 所有 Agent 共享 GroupChat.messages 列表
# 每个 Agent 看到完整对话历史，从中提取需要的信息
# 状态就是"对话历史"
group_chat = GroupChat(agents=[...], messages=[...])
# 问题：对话历史线性增长，上下文窗口爆了要手动压缩（接第 26 篇）

优势：无需额外状态管理、Agent 自然从对话中提取信息。
劣势：状态隐式（藏在消息里）、上下文窗口增长快、多 Agent 并行时消息历史冲突。

LlamaIndex — Context 对象：

@step
async def do_something(self, ctx: Context, ev: SomeEvent) -> NextEvent:
    await ctx.set("key", value)    # 存
    val = await ctx.get("key")     # 取
    # Context 是 step 间共享的 KV 存储

优势：显式存取、异步安全、可序列化（支持 checkpoint）。
劣势：状态结构隐式（KV 无 schema）、类型不安全（get 返回 Any）。

4.2 容错：重试、超时、补偿

LangGraph：

from langgraph.graph import StateGraph
from langgraph.pregel import RetryPolicy

app = graph.compile(
    retry_policy=RetryPolicy(
        max_attempts=3,
        initial_interval=1.0,
        backoff_factor=2.0,
        retry_on=ValueError,  # 只重试这类异常
    ),
    recursion_limit=25,  # 最大节点执行次数防死循环
)

# 超时：recursion_limit 间接限制 + 每个节点可设 timeout
# 补偿：用条件边，某节点失败 → 走补偿分支
graph.add_conditional_edges("risky_op", check_success, {
    "success": "next",
    "fail": "compensate",  # 补偿分支
})

CrewAI：

task = Task(
    description="...",
    agent=agent,
    max_iter=3,          # Agent 内部重试次数
    max_execution_time=60,  # 超时秒数
)
# 框架级重试有限，主要靠 Agent 内部 max_iter
# 无原生补偿机制，要自己在 output_parser 里处理

AutoGen：

group_chat = GroupChat(
    agents=[...],
    max_round=10,  # 最大对话轮数防发散
)
# 无原生重试，Agent 生成失败要自己 try/except
# 终止条件 is_termination_msg 可定制

LlamaIndex Workflows：

workflow = CodeReviewWorkflow(
    timeout=60,        # 全局超时
    num_concurrent_runs=10,
)
# @step 内部自己实现重试
# 框架提供 checkpoint：ctx 序列化保存，失败后从 checkpoint 恢复

4.3 状态持久化与恢复

LangGraph — Checkpointer：

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

# 每次节点执行后自动存 checkpoint
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(initial_state, config)

# 中断后恢复：同 thread_id 自动从最后 checkpoint 恢复
result = app.invoke(None, config)  # None 表示从 checkpoint 继续

这是 LangGraph 的杀手锏：人机协作中断后精确恢复（接第 8/34 篇确认门）。Agent 执行到 interrupt() 暂停 → 状态存 checkpoint → 人类确认 → Command(resume=value) 从 checkpoint 恢复继续。状态恢复成功率实测 98%。

其他框架的状态持久化：

4.4 容错对比总结

五、人机协作集成

5.1 确认门：事前让人类拍板

接第 8 篇四种确认门 + 第 34 篇前端呈现，看各框架怎么集成。

LangGraph — interrupt + Command(resume)：

from langgraph.types import interrupt, Command

def human_review_node(state: State) -> dict:
    # 执行到此处暂停，等人类输入
    decision = interrupt({
        "question": "这段代码审查通过吗？",
        "code": state["code_diff"],
        "risk_level": state["risk"],
    })
    # 人类通过 Command(resume="approve"/"reject") 恢复
    return {"human_decision": decision}

# 前端调用：
# 1. app.invoke(state) → 执行到 human_review_node 暂停
# 2. 前端收到 interrupt 事件，展示确认门 UI（接第 34 篇）
# 3. 人类点击 → app.invoke(Command(resume="approve"), config)
# 4. 从 checkpoint 恢复，继续执行

这是最完整的确认门集成：暂停→存状态→等人类→恢复，全流程框架托管。

CrewAI — human_input：

agent = Agent(
    role="审查员",
    goal="...",
    human_input="TERMINATE",  # 仅在终止时问人类
)
# 或在 Task 级别：
task = Task(..., human_input=True)  # 任务完成后问人类是否满意
# 局限：只在任务边界问，不能在任务中间暂停

AutoGen — human_input_mode：

human_proxy = ConversableAgent(
    name="Human",
    human_input_mode="ALWAYS",  # 每轮都问人类
    llm_config=False,
)
# human_proxy 作为对话参与者，每轮等人类输入
# 局限：粒度是"每轮对话"，不能精确控制何时问

LlamaIndex — 事件中断：

class HumanApprovalEvent(Event):
    content: str

class Workflow(Workflow):
    @step
    async def request_approval(self, ctx, ev: SomeEvent) -> HumanApprovalEvent:
        return HumanApprovalEvent(content="需要人类确认")
    
    @step
    async def wait_approval(self, ctx, ev: HumanApprovalEvent) -> NextEvent:
        # 此 step 阻塞等外部 resume
        approval = await ctx.wait_for_event(HumanResponseEvent)
        ...
# 局限：需要自定义事件类型+手动管理恢复流程

5.2 确认门集成对比

结论：需要复杂人机协作（多轮确认、中途暂停、状态恢复）→ LangGraph 是唯一原生完整支持的。

六、可观测性与调试

6.1 与第 23 篇 span 追踪的集成

LangGraph — LangSmith 原生集成：

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "..."

# 零代码：每个节点自动成为 span
# LangSmith UI 展示：图结构 + 每节点输入输出 + 耗时 + token + 成本
# 可直接看到执行路径在图上高亮

CrewAI — 内置 verbose + 回调：

crew = Crew(
    agents=[...],
    tasks=[...],
    verbose=True,  # 打印执行日志
)
# 或用 crewai 的 telemetry 回调
# 局限：无原生 span 追踪，要自己接 OpenTelemetry

AutoGen — 日志 + 可定制回调：

# autogen 原生打印对话日志
# 要接 OpenTelemetry 需自己包装 ConversableAgent.generate_reply

LlamaIndex — 原生 observability：

from llama_index.core import set_global_handler
set_global_handler("simple")  # 或 "langfuse", "arize" 等
# step 执行自动追踪

6.2 调试体验对比

七、选型决策矩阵

7.1 按任务结构选

你的任务流程是……？

├─ 确定性流水线（步骤固定、顺序明确）
│  └─► LangGraph（显式图，可视化即文档）
│
├─ 角色分工明确（谁做什么很清楚）
│  ├─ 流程简单（顺序即可）→ CrewAI（最低门槛）
│  └─ 流程复杂（需分支/循环）→ LangGraph（角色作为节点）
│
├─ 开放式探索（流程无法预先定义、需要"讨论"）
│  └─► AutoGen（对话流，自由探索）
│
├─ 事件驱动（某事发生触发多 Agent 响应）
│  └─► LlamaIndex Workflows（事件解耦）
│
└─ 混合（部分确定+部分探索）
   └─► LangGraph（确定部分用图）+ 子图内用 AutoGen（探索部分用对话）

7.2 按工程需求选

你需要……？

├─ 人机协作+中途暂停+状态恢复
│  └─► LangGraph（interrupt+checkpoint 是唯一原生完整的）
│
├─ 快速原型+团队不熟图论
│  └─► CrewAI（最低学习曲线，角色+任务+流程三要素）
│
├─ 人类作为协作参与者
│  └─► AutoGen（human_input_mode 让人类自然参与对话）
│
├─ 高并发+异步+解耦
│  └─► LlamaIndex Workflows（原生 async+事件解耦）
│
├─ 可观测性零代码接入
│  └─► LangGraph（LangSmith）或 LlamaIndex（多后端原生）
│
└─ 状态管理复杂+需要类型安全
   └─► LangGraph（TypedDict State + reducer）

7.3 综合决策矩阵

7.4 一个真实选型案例

场景：企业级代码审查 Agent 系统
  - 流程：PR 触发 → 静态分析 → AI 审查 → 安全扫描 → 合规检查 → 汇总 → 人类审批 → 评论
  - 需求：流程半确定（某些步骤条件跳过）、需要人类审批门、需要状态恢复、
         需要可观测、需要并行（AI审查和安全扫描并行）

选型推理：
  ① 流程半确定 + 需要可视化 → 排除 AutoGen（对话流不可视）
  ② 需要人类审批门 + 状态恢复 → LangGraph（interrupt+checkpoint 唯一原生完整）
  ③ 需要并行（AI审查∥安全扫描）→ LangGraph（多边同源）或 LlamaIndex（一事件多step）
  ④ 需要可观测 → LangGraph（LangSmith 零代码）
  ⑤ 综合选 LangGraph

  CrewAI 被排除：不原生支持并行、确认门不完整、可观测需自接
  LlamaIndex 次选：事件驱动能做但流程不如图直观，确认门不如 LangGraph 原生
  AutoGen 被排除：对话流不适合确定性流水线

八、混用策略：用 A2A 桥接异构框架

8.1 为什么混用

没有单一框架适合所有子任务。一个系统里：

编排层（LangGraph）         确定性控制流、人机确认门、状态恢复
  ├─ 研究子任务（CrewAI）    角色分工：调研员→分析师→总结员
  ├─ 方案辩论子任务（AutoGen） 开放探索：多个 Agent 讨论选最优方案
  └─ 文档处理子任务（LlamaIndex） 事件驱动：文档变更触发多 Agent

每个子任务用最适合的框架，编排层用 LangGraph 统一管理。

8.2 用第 28 篇 A2A 桥接

import httpx
from langgraph.graph import StateGraph, START, END

# 每个"子框架系统"暴露为 A2A Server（接第 28 篇 Agent Card）
# LangGraph 编排层通过 A2A 协议委派，不关心子任务用什么框架

async def delegate_to_crewai_research(state: dict) -> dict:
    """通过 A2A 委派给 CrewAI 研究子系统"""
    async with httpx.AsyncClient() as client:
        # 1. 拉取 Agent Card（接第 28 篇）
        card = await client.get("http://research-crew:8000/.well-known/agent.json")
        # 2. 按 A2A 协议提交任务
        task = await client.post("http://research-crew:8000/tasks/send", json={
            "message": {"role": "user", "content": state["research_query"]},
        })
        # 3. 轮询/SSE 等结果
        result = await poll_a2a_task(client, task["id"])
    return {"research_result": result}

async def delegate_to_autogen_debate(state: dict) -> dict:
    """通过 A2A 委派给 AutoGen 辩论子系统"""
    async with httpx.AsyncClient() as client:
        task = await client.post("http://debate-autogen:8000/tasks/send", json={
            "message": {"role": "user", "content": state["debate_topic"]},
        })
        result = await poll_a2a_task(client, task["id"])
    return {"debate_result": result}

# LangGraph 编排层
graph = StateGraph(State)
graph.add_node("research", delegate_to_crewai_research)
graph.add_node("debate", delegate_to_autogen_debate)
graph.add_node("synthesize", synthesize_results)
graph.add_edge(START, "research")
graph.add_edge("research", "debate")
graph.add_edge("debate", "synthesize")
graph.add_edge("synthesize", END)

关键点：编排层不 import CrewAI/AutoGen，只通过 A2A HTTP 通信。子框架可以独立部署、独立扩展、独立升级——这是第 28 篇 A2A 的核心价值在编排层的落地。

8.3 混用架构

                    LangGraph 编排层
                   （确定性控制流+确认门+状态恢复）
                         │
                    A2A 协议委派
                         │
          ┌──────────────┼──────────────┐
          ▼              ▼              ▼
    CrewAI 子系统   AutoGen 子系统   LlamaIndex 子系统
    （角色分工）    （方案辩论）     （事件驱动）
    各自独立部署    各自独立部署     各自独立部署

好处：

• 各子框架用其所长：CrewAI 做角色分工、AutoGen 做探索、LlamaIndex 做事件
• 独立部署扩展：每个子系统独立扩缩容，不互相影响
• 技术栈解耦：子系统可以不同语言/不同版本/不同部署环境
• 编排层统一可观测：LangGraph + LangSmith 追踪所有 A2A 委派，端到端 span

代价：

• 网络开销：A2A 是 HTTP，比进程内调用慢（实测单跳 +50-200ms）
• 运维复杂度：多个子系统各自部署监控
• 调试跨系统：问题在子系统内时要切到该框架的调试工具

九、实施难度评估与分步落地

9.1 各框架实施难度

9.2 实测数据

在四类典型场景中对比（每场景 10 次实验取均值）：

场景 1：代码审查流水线（确定性流程）

AutoGen 错配严重：对话流硬塞确定性流水线，调试时间占 42%、70% 需要重构。

场景 2：研究→写作→审核（角色分工）

CrewAI 正选：角色分工最直觉，开发最快。LangGraph 次之（图也能做但要显式定义）。

场景 3：方案头脑风暴（开放探索）

AutoGen 正选：对话流天然适合探索，方案多样性最高。LangGraph 错配：图约束太强，探索变流水线。

场景 4：多轮人机确认（审批流程）

LangGraph 碾压：interrupt+checkpoint 原生支持，其他框架要大量手写状态保存恢复逻辑。

9.3 分步落地建议

第一步：单框架验证（1-2 周）
  → 按任务结构选一个框架（用决策矩阵）
  → 用该框架跑通核心流程
  → 接入可观测性（LangSmith / LlamaIndex handler / 自接 OTEL）
  → 评估：开发效率、编排延迟、调试体验是否达标
  → 达标 → 停在这里，不要过度设计
  → 不达标（框架限制了某些子任务）→ 进第二步

第二步：混用桥接（2-4 周）
  → 识别框架不擅长的子任务
  → 该子任务改用更适合的框架，暴露为 A2A Server（接第 28 篇）
  → 编排层（LangGraph）通过 A2A 委派
  → 评估：网络开销是否可接受、运维复杂度是否可控
  → 达标 → 停在这里
  → 子系统太多运维爆炸 → 进第三步

第三步：收敛与简化（2-3 周）
  → 评估每个子系统是否真的需要不同框架
  → 能用编排层框架统一表达的 → 收回（减少子系统）
  → 真正需要异构的 → 保留（如探索任务确实需要 AutoGen）
  → 目标：混用但不过度，2-3 个子系统为上限

9.4 常见坑

坑 1：用 AutoGen 写确定性流程
  → 对话流硬塞流水线，不断加约束防偏离
  → 症状：调试时间 >40%、70% 重构
  → 解法：换 LangGraph

坑 2：用 LangGraph 写自由探索
  → 图约束太强，探索变流水线
  → 症状：方案多样性低、Agent "不探索" 
  → 解法：探索部分用 AutoGen 子图

坑 3：CrewAI 做生产级复杂控制流
  → sequential/hierarchical 不够，条件分支/循环/并行不原生
  → 症状：不断用 output_parser 变通，代码比框架还复杂
  → 解法：换 LangGraph，或 CrewAI 做原型→LangGraph 做生产

坑 4：一上来就混用
  → 还没跑通单框架就混 3 个框架 + A2A
  → 症状：运维爆炸、调试跨系统、定位问题靠排除法
  → 解法：先单框架验证，确有需求才混用

坑 5：不接可观测性就上生产
  → 编排层是黑盒，出问题不知道哪个 Agent/哪步出错
  → 解法：第一步就接 span 追踪（接第 23 篇）

十、总结与下一篇

10.1 本文核心

四个框架四种隐喻：
  LangGraph  = 图       → 确定性控制流，可视化即文档
  CrewAI     = 角色流   → 角色分工，最低门槛
  AutoGen    = 对话流   → 自由探索，对话即计算
  LlamaIndex = 事件流   → 事件解耦，异步原生

选型核心：框架的调度隐喻必须与任务结构匹配
  确定性流程 → 图（LangGraph）
  角色分工   → 角色流（CrewAI）
  开放探索   → 对话流（AutoGen）
  事件驱动   → 事件流（LlamaIndex）

混用核心：用 A2A（第 28 篇）桥接，编排层 LangGraph + 子系统各用所长
  → 但先单框架验证，确有需求才混用

10.2 与系列的关系

第 2 篇（多 Agent 架构）   → 选哪种协作架构
第 14 篇（协作 internals） → 通信/冲突/共识/拍卖的规则
第 28 篇（A2A 协议）       → 跨系统互操作的接口
本文（编排框架对比）       → 协作的容器——用什么框架组织

四篇递进：架构选型 → 协作规则 → 互操作接口 → 编排容器

10.3 下一篇预告

本文对比了编排框架的"怎么编排"。但编排的 Agent 内部还有一个没展开的维度：Agent 的规划策略——ReAct（推理-行动交替）、Plan-and-Execute（先规划再执行）、Tree-of-Thought（树状探索）、Reflexion（自我反思修正）——这些规划策略是 Agent "怎么思考"的内核，直接影响任务成功率、token 消耗、延迟。第 2/7 篇提到过 ReAct 和反思，但没系统对比。下一篇将深潜 Agent 规划策略：四种策略的机制差异、适用任务类型、token/延迟/成功率的实测对比、以及混合策略（简单任务用 ReAct、复杂任务用 Plan-and-Execute、卡住时切 Reflexion）。

本文是 AI Agent 技术博客系列第 37 篇。系列目录见 blog.html。