多Agent协作的3种编排模式：串行、并行与DAG

为什么你需要关心Agent编排

如果你只做过单Agent的demo，你会发现一个规律：只要业务稍微复杂一点，一个Agent就扛不住了。

比如一个"智能报工系统"，客户需求是这样的：

这里面涉及4个完全不同的能力：文档理解、产能计算、规则匹配、报告生成。你不可能让一个Agent同时精通这4件事——上下文太长，Prompt太复杂，Token成本爆炸。

正确的做法是拆成多个Agent，每个Agent专注一件事，然后用编排层把它们串起来。

但怎么串？这里有3种基本模式，每种都有明确的适用场景。

模式一：串行编排（Sequential）

核心逻辑

Agent A的输出，作为Agent B的输入。像流水线一样，一个接一个处理。

[输入] → Agent A → Agent B → Agent C → [输出]

适用场景

业务流程是线性的，每一步的输出都是下一步的输入，且各步骤之间强依赖。

典型案例：

• 文档解析 → 信息提取 → 数据入库
• 需求分析 → 代码生成 → 代码审查
• 客户提问 → 意图识别 → 知识检索 → 回答生成

代码结构

class SequentialOrchestrator:
"""串行编排器：按顺序执行Agent链"""

def __init__(self, agents: list[Agent]):
self.agents = agents

async def run(self, input_data: dict) –> dict:
context = input_data

for i, agent in enumerate(self.agents):
print(f"[Step {i+1}/{len(self.agents)}] 执行: {agent.name}")
context = await agent.execute(context)

# 可选：每步结果落盘，方便排查
await self.save_checkpoint(i, context)

return context

工程取舍

优点：

• 实现简单，调试容易（每一步的输入输出都能单独看）
• 适合长流程、强依赖的业务
• 每一步可以单独替换和升级

缺点：

• 总延迟 = 各步骤延迟之和（串行 = 慢）
• 一个环节出错，整条链路中断
• 不适合有并行需求的场景

踩坑点

串行编排最常见的坑是上下文膨胀。每经过一个Agent，context里就多一堆中间结果。到第三个Agent时，context可能已经有几万字，Token成本飙升。

解法： 每个Agent只传递下一步需要的字段，不要把全量context往下透传。设计一个明确的output_schema，每个Agent只输出下游需要的结构化数据。

class DocumentParserAgent:
"""只输出下游需要的字段，而不是把整个文档内容往透传"""

output_schema = {
"customer_name": str,
"order_id": str,
"products": list[dict], # [{name, quantity, spec}]
"delivery_date": str,
}

模式二：并行编排（Parallel）

核心逻辑

多个Agent同时执行，各自处理不同的任务，最后汇总结果。

┌→ Agent A →┐
[输入] ──┤→ Agent B → ├── [汇总] → [输出]
└→ Agent C →┘

适用场景

任务之间相互独立，没有依赖关系，可以同时执行。

典型案例：

• 用户同时问了一个包含产品咨询、价格查询、售后政策的复合问题 → 3个Agent并行回答，最后合并
• 竞品监控：同时抓取5个平台的价格数据
• 报告生成：并行生成摘要、数据分析、风险评估三个章节

代码结构

import asyncio

class ParallelOrchestrator:
"""并行编排器：同时执行多个Agent，汇总结果"""

def __init__(self, agents: list[Agent], aggregator: Agent | None = None):
self.agents = agents
self.aggregator = aggregator # 可选的汇总Agent

async def run(self, input_data: dict) –> dict:
# 所有Agent同时启动
tasks = [agent.execute(input_data) for agent in self.agents]
results = await asyncio.gather(*tasks, return_exceptions=True)

# 处理异常（一个Agent挂了不影响其他）
valid_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"[ERROR] {self.agents[i].name} 执行失败: {result}")
else:
valid_results.append(result)

# 可选：用汇总Agent合并结果
if self.aggregator:
return await self.aggregator.execute({
"partial_results": valid_results,
"input": input_data
})

return {"results": valid_results}

工程取舍

优点：

• 总延迟 = 最慢那个Agent的延迟（而不是所有之和）
• 容错性好：一个Agent挂了，其他Agent的结果仍然可用
• 适合需要同时获取多维度信息的场景

缺点：

• 不适合有依赖关系的任务
• 汇总逻辑可能复杂（多个Agent的结果格式、粒度可能不一致）
• 并发高时对API调用配额有压力

踩坑点

并行编排最常见的坑是汇总质量差。把多个Agent的回答简单拼接，往往会出现重复、矛盾、格式混乱。

解法： 加一个专门的"汇总Agent"，用LLM做一次合并去重和冲突消解。这个汇总Agent的Prompt要明确告诉它如何处理矛盾：

AGGREGATOR_PROMPT = """
你是一个结果汇总助手。你会收到多个Agent的回答，请执行以下操作：

1. 去重：多个Agent回答了同一个问题，只保留最准确的那个
2. 消解冲突：如果Agent A和Agent B的答案矛盾，优先采信带有数据/引用来源的那个
3. 格式化：输出统一的Markdown格式

注意：不要遗漏任何有价值的信息，但也不要简单拼接。
"""

模式三：DAG编排（有向无环图）

核心逻辑

Agent之间的依赖关系不是简单的线性或并行，而是一个有向无环图（DAG）——有些步骤可以并行，有些步骤必须等前置步骤完成后才能开始。

┌→ Agent A ──────→┐
[输入] ──┤ ├── Agent D → [输出]
└→ Agent B → Agent C ┘

上面这个DAG里：

• Agent A 和 Agent B 可以并行（无依赖）
• Agent C 必须等 Agent B 完成
• Agent D 必须等 Agent A 和 Agent C 都完成

适用场景

业务流程是混合依赖的——部分步骤可以并行，部分步骤有严格的先后顺序。

这是生产环境中最常见的模式，因为真实业务几乎不可能只有纯线性或纯并行。

典型案例：

• 智能客服系统：
  • 并行：意图识别 + 用户身份验证
  • 串行：意图识别完成后 → 路由到对应Agent
  • 汇总：回答生成 + 工单创建 → 合并推送
• 供应链决策：
  • 并行：需求预测 + 库存查询 + 供应商报价采集
  • 依赖：采购方案 = f(需求预测, 库存, 报价)
  • 最终：生成采购报告

代码结构

DAG编排的核心是一个任务调度器，它根据依赖关系决定哪些Agent可以并行执行、哪些需要等待。

from collections import defaultdict
import asyncio

class DAGOrchestrator:
"""DAG编排器：按依赖关系调度Agent执行"""

def __init__(self):
self.agents = {} # name -> Agent
self.dependencies = defaultdict(set) # name -> {前置依赖}
self.results = {} # name -> result

def add_agent(self, name: str, agent: Agent, depends_on: list[str] = None):
self.agents[name] = agent
if depends_on:
self.dependencies[name] = set(depends_on)

def _get_ready_agentss(self) –> list[str]:
"""获取所有前置依赖已完成的Agent"""
ready = []
for name in self.agents:
if name in self.results:
continue # 已执行
if self.dependencies[name].issubset(set(self.results.keys())):
ready.append(name)
return ready

async def run(self, input_data: dict) –> dict:
self.results = {"__input__": input_data}

while len(self.results) < len(self.agents) + 1:
ready = self._get_ready_agents()

if not ready:
raise RuntimeError("DAG中存在循环依赖或死锁")

# 所有就绪的Agent并行执行
tasks = []
for name in ready:
# 收集该Agent需要的前置结果
dep_results = {
dep: self.results[dep]
for dep in self.dependencies[name]
}
tasks.append(self._execute_agent(name, dep_results, input_data))

await asyncio.gather(*tasks)

return self.results

async def _execute_agent(self, name: str, dep_results: dict, input_data: dict):
agent = self.agents[name]
result = await agent.execute({
"input": input_data,
"dependencies": dep_results
})
self.results[name] = result

使用示例

用上面的"智能报工系统"举例：

dag = DAGOrchestrator()

# Step 1: 文档解析（无依赖，最先执行）
dag.add_agent("doc_parser", DocumentParserAgent())

# Step 2: 产能查询（无依赖，和文档解析并行）
dag.add_agent("capacity_query", CapacityQueryAgent())

# Step 3: 排产方案（依赖文档解析 + 产能查询）
dag.add_agent("scheduling", SchedulingAgent(), depends_on=["doc_parser", "capacity_query"])

# Step 4: 质检标准（依赖排产方案）
dag.add_agent("quality_check", QualityCheckAgent(), depends_on=["scheduling"])

# Step 5: 报告生成（依赖排产方案 + 质检标准）
dag.add_agent("report_gen", ReportGenAgent(), depends_on=["scheduling", "quality_check"])

# 执行
result = await dag.run(order_data)

执行时序：

时间轴 →
doc_parser: ████████
capacity_query: ████████
↓ 两者都完成
scheduling: ████████████
↓ 完成
quality_check: ████████
↓ scheduling + quality_check都完成
report_gen: ██████████

工程取舍

优点：

• 最灵活，能表达任意复杂的依赖关系
• 最大化并行度（只要依赖满足就立即执行）
• 生产环境最实用的编排模式

缺点：

• 实现复杂度高
• 调试困难（需要可视化DAG图才能看清执行路径）
• 状态管理复杂（需要处理部分完成、部分失败的情况）

踩坑点

坑1：循环依赖。 Agent A依赖B，B依赖C，C又依赖A——死锁。DAG编排器必须在注册时做拓扑排序，检测并拒绝循环依赖。

坑2：错误传播。 一个Agent挂了，依赖它的所有下游Agent怎么办？需要设计明确的错误处理策略：

• 跳过（下游Agent也跳过，标记为"因上游失败未执行"）
• 降级（下游Agent用默认值/缓存值继续执行）
• 重试（上游Agent重试N次，超过后走降级）

class ErrorStrategy:
SKIP = "skip" # 跳过下游
FALLBACK = "fallback" # 用默认值继续
RETRY = "retry" # 重试上游

坑3：结果传递的数据量。 DAG里的中间结果可能很大（比如一个Agent生成了10页的排产方案），全部传递给下游会导致内存和Token成本暴涨。解法是只传递下游需要的摘要，而不是全量结果。

三种模式的选型指南

一个经验法则： 如果你的Agent协作逻辑超过了5步，几乎一定需要DAG。因为真实业务流程很少有纯线性的，总会有"这两步可以同时做"或"这一步要等那两步都完成"的情况。

关于编排框架的选择

目前主流的Agent编排框架：

• LangGraph：LangChain生态，基于图（Graph）的编排，原生支持DAG。适合已经用LangChain的团队。
• CrewAI：基于角色的多Agent框架，内置串行/并行/层级三种模式。上手快，但复杂场景不够灵活。
• AutoGen（微软）：基于对话的多Agent框架，Agent之间通过消息传递协作。适合需要Agent间"讨论"的场景。
• 自研编排层：如果业务足够复杂，建议自己写编排层。上面的代码就是一个最小可用的骨架，生产环境加上持久化、监控、重试机制即可。

框架不是银弹。理解底层逻辑，才能在框架出问题时知道怎么修。

结语

多Agent协作不是"把几个Agent拼在一起"那么简单。编排模式的选择，本质上是在延迟、容错、复杂度三者之间做取舍。

串行最简单但最慢，并行最快但只适合独立任务，DAG最灵活但实现最复杂。生产环境中，大部分业务最终都会走向DAG——因为真实世界的业务流程，本来就不是线性的。

理解了这三种模式，再去选框架、做设计，才不会迷路。

栈上月明（ZSoftYM）是一家专注于AI智能体开发与多Agent系统架构的西安技术团队。