多Agent编排的三种模式：流水线、并发、条件分支

前面七篇文章构建了从单 Agent tool calling 到 Agentic RAG 再到规划设计的完整链条。但所有这些都在单 Agent 系统内运作——即使 Lesson 07 的 Planning + Concierge 也是规划-执行的一对一协作。

当系统需要三个、五个、十个 Agent 时，问题从"Agent 能做什么"变成了"Agent 之间怎么协作"。这就是多Agent编排（Multi-Agent Orchestration）。

📦 相关链接

项目仓库：Building Agent from Scratch
本课代码：08-multi-agent/code_samples/
LangGraph 版：08-python-agent-framework.py
qwen-agent 版：08-qwen-agent-framework.py

🎯 编排的三个维度

多Agent编排回答三个问题：

维度	问题	对应模式
顺序	"谁先谁后？"	流水线式 (Sequential)
并行	"哪些可以同时做？"	并发式 (Concurrent)
分支	"中间结果不同，下一步走哪？"	条件式 (Conditional)

这三种不是互斥的——真实系统通常是三种的组合。但拆开分别理解是组合它们的前提。

🔗 模式 1：流水线式 — Sales → Price → Quote

场景：客户描述了一个客厅，需要推荐家具 → 定价 → 生成报价单。三步有严格的先后顺序——没推荐完就定价是乱猜，没定价就出报价单是糊弄。

flowchart LR
    User["客户需求"] --> Sales["Sales-Agent<br/>家具推荐"]
    Sales --> Price["Price-Agent<br/>定价分析"]
    Price --> Quote["Quote-Agent<br/>生成报价单"]
    Quote --> Output["📋 正式报价单"]

LangGraph 实现：`add_edge` 链

from langgraph.graph import StateGraph, END
 
class SequentialState(TypedDict):
    user_query: str
    sales_recommendation: Optional[str]
    price_analysis: Optional[str]
    quote: Optional[str]
 
graph = StateGraph(SequentialState)
graph.add_node("sales", sales_node)     # LLM 调用，输出 → state["sales_recommendation"]
graph.add_node("price", price_node)     # 读取 sales_recommendation → 输出 price_analysis
graph.add_node("quote", quote_node)     # 读取前两步 → 输出 quote
 
graph.add_edge("sales", "price")
graph.add_edge("price", "quote")
graph.add_edge("quote", END)
graph.set_entry_point("sales")
 
app = graph.compile()
for _ in app.stream(initial_state):
    pass  # 每个节点内部流式打印 LLM 输出

每个节点是一个 (state) -> dict 函数——从 state 读取上游输出，调用 LLM，返回部分 state 更新。LangGraph 负责按边定义的顺序执行，节点之间通过 AgentState 共享数据。

qwen-agent 实现：手动链式调用

sales = Assistant(llm=llm_cfg, name="Sales-Agent", system_message=SALES_INSTRUCTIONS)
price = Assistant(llm=llm_cfg, name="Price-Agent", system_message=PRICE_INSTRUCTIONS)
quote = Assistant(llm=llm_cfg, name="Quote-Agent", system_message=QUOTE_INSTRUCTIONS)
 
rec = run_agent(sales, user_query, "Sales-Agent")
analysis = run_agent(price, f"推荐：\n{rec}\n\n请定价。", "Price-Agent")
run_agent(quote, f"推荐：\n{rec}\n\n定价：\n{analysis}\n\n请生成报价单。", "Quote-Agent")

没有图引擎——数据流向完全由代码的变量传递决定。三个 Agent 各自独立，通过自然语言文本交换信息。

⚡ 模式 2：并发式 — Researcher ‖ Planner

场景：规划东京旅行。研究和规划可以同时进行——研究员查景点、天气、文化，规划师排行程、交通、餐饮。两者互不依赖，总耗时约等于较慢的那个。

flowchart TD
    Dispatcher["Dispatcher"] --> Researcher["Researcher-Agent<br/>景点/文化/天气"]
    Dispatcher --> Planner["Plan-Agent<br/>行程/交通/餐饮"]
    Researcher --> Aggregate["Aggregate<br/>汇总结果"]
    Planner --> Aggregate

LangGraph 实现：`Send` API fan-out

from langgraph.graph import Send
 
class ConcurrentState(TypedDict):
    user_query: str
    results: Annotated[list, operator.add]  # reducer 合并并行结果
 
def fan_out_to_agents(state) -> list[Send]:
    return [
        Send("researcher", {"user_query": state["user_query"], "results": []}),
        Send("planner",    {"user_query": state["user_query"], "results": []}),
    ]
 
graph.add_conditional_edges("dispatcher", fan_out_to_agents)
graph.add_edge("researcher", "aggregate")
graph.add_edge("planner", "aggregate")

关键设计：

Send(target, state) — 每个 Send 创建一个独立的执行分支，target 节点收到一份 state 副本
Annotated[list, operator.add] — 当多个并行分支都返回 {"results": [...]} 时，reducer 自动把它们合并成一个 list
汇聚 — researcher 和 planner 都指向 aggregate，graph 等两者都完成才执行 aggregate

qwen-agent 实现：`asyncio.gather`

import asyncio
 
async def run_agent_async(agent, query):
    messages = [{"role": "user", "content": query}]
    result = ""
    for responses in agent.run(messages=messages):
        if responses:
            last = responses[-1]
            if last.get("role") == "assistant" and last.get("content"):
                result = last["content"]
    return result
 
async def run_both():
    return await asyncio.gather(
        run_agent_async(researcher, query),
        run_agent_async(planner, query),
    )
 
research_result, plan_result = asyncio.run(run_both())

没有 graph、没有 Send、没有 reducer。两个 Assistant.run() 在同一个 event loop 中并发执行，结果通过 gather 收集。代码量只有 LangGraph 版的 1/3，但失去了可视化、错误隔离、状态追踪等 graph 引擎提供的特性。

🔀 模式 3：条件式 — Writer ⇄ Reviewer → Publisher

场景：内容审核流水线。Writer 写稿 → Reviewer 审核（是否超过 200 字）→ 通过则 Publisher 发布，不通过则回到 Writer 重写。这是一个带循环的 DAG。

flowchart TD
    Writer["Writer-Agent<br/>撰写草稿"] --> Reviewer["Reviewer-Agent<br/>审核字数"]
    Reviewer -->|PASS| Publisher["Publisher-Agent<br/>发布内容"]
    Reviewer -->|REVISE| Writer
    Publisher --> Output["✅ 已发布"]

LangGraph 实现：`add_conditional_edges`

def route_after_review(state: ConditionalState) -> str:
    if state["review_result"] == "PASS":
        return "publisher"
    if state["iteration"] >= 3:
        return "publisher"  # 最多 3 轮，强制发布
    return "writer"         # 返回重写
 
graph.add_conditional_edges(
    "reviewer",
    route_after_review,
    {"writer": "writer", "publisher": "publisher"},
)

route_after_review 是一个纯函数——输入 state，输出下一个节点名。LangGraph 在运行时调用它来决定走哪条边。循环是天然的——writer → reviewer → writer → reviewer → ... 直到路由函数返回 publisher。

qwen-agent 实现：`for` 循环 + `if/else`

for iteration in range(1, max_iterations + 1):
    draft = run_agent(writer, prompt, f"Writer-Agent (第{iteration}轮)")
    review = run_agent(reviewer, f"审核稿件：\n{draft}", "Reviewer-Agent")
 
    if "PASS" in review.strip().split("\n")[0].upper():
        run_agent(publisher, f"发布：\n{draft}", "Publisher-Agent")
        break
    else:
        prompt = f"上一稿被驳回。修改重写：\n原稿：\n{draft}"
else:
    run_agent(publisher, f"强制发布：\n{draft}", "Publisher-Agent")

本质上和 LangGraph 做的是同一件事——解析中间结果 → 决定下一步 → 可能循环。区别是控制流显式写在代码里（for + if），而不是通过 graph 引擎的路由函数。

📊 三种模式的对比

维度	流水线	并发	条件
Agent 关系	串行依赖	并行独立	动态路由
数据流	上游输出→下游输入	同源分发→独立输出→汇聚	根据分支走向不同下游
LangGraph 原语	`add_edge`	`Send` + `operator.add`	`add_conditional_edges`
qwen-agent 原语	变量传递	`asyncio.gather`	`for` + `if/else`
循环	无	无	有（审核-重写）
总耗时	T₁ + T₂ + T₃	max(T₁, T₂)	取决于重写轮数

🔑 LangGraph vs 手动编排：什么时候用哪个

LangGraph 的优势：

可视化 — graph.get_graph().draw_mermaid() 直接生成架构图
状态管理 — TypedDict + reducer 自动合并并行结果，不需要手动收集
错误隔离 — 一个节点失败不影响其他分支（并发模式下）
复杂条件 — 多路分支、嵌套路由不需要嵌套 if/else
持久化 — 内置 checkpoint，可以暂停/恢复/回溯

手动编排的优势：

零抽象 — 控制流就是代码流，不需要理解 graph 引擎的语义
调试简单 — 断点直接打在调用点，不需要深入框架内部
依赖少 — 不需要装 langgraph
Agent 少时够用 — 3 个 Agent 以下，手动编排的代码量更少

选择建议：Agent 数 ≥ 4 或需要条件分支/循环 → LangGraph。Agent 数 ≤ 3 且纯线性 → 手动编排即可。

🚀 运行

cd 08-multi-agent/code_samples
 
# LangGraph 版（三种模式依次运行）
python 08-python-agent-framework.py
 
# qwen-agent 版
python 08-qwen-agent-framework.py

🔮 下一篇

下一篇进入 元认知（Metacognition）——Agent 如何自我反思、检测错误、在工具失败时自动回退到备用方案。这是从"Agent 能做事"到"Agent 能自己判断做得好不好"的关键一步。

✍️ 结语

多Agent编排的本质是控制流的显式化。

单 Agent 系统里，LLM 内部隐式地决定先调哪个工具、后调哪个工具——你只能看到结果，看不到过程。多 Agent 系统把控制流从 LLM 的黑箱里拉出来，变成显式的 graph 或代码——你可以看到每一步谁在做、数据怎么流、哪个分支被触发。

MAF 的 WorkflowBuilder、LangGraph 的 StateGraph、甚至最原始的手动 run_agent(a1) → run_agent(a2)——都是在做同一件事：让 Agent 之间的协作变得可见、可调试、可修改。

工具是形式，控制流是实质。

目录