1. 从"能跑"到"跑得好":Agent 编排到底在解决什么
如果你已经用大模型搭过几个 Agent,大概率经历过这个阶段:单个 Agent 跑起来挺惊艳,一旦把两三个串起来,整个系统就开始"发疯"——要么死循环烧 token,要么中间某一步输出格式不对导致后面全崩,要么多个 Agent 互相"踢皮球"谁都不干活。这不是模型不行,而是**编排(Orchestration)**这一层没做好。
Agent 编排,说白了就是决定"谁在什么时候、按什么顺序、拿什么信息、去干什么事"的那套调度逻辑。它介于底层模型调用和上层业务目标之间,是让一堆孤立的 Agent 变成一个能协作完成复杂任务的系统的关键。没有编排,你手里只有一堆零件;有了编排,你才有一台能运转的机器。
这篇内容适合三类人:一是刚把单个 Agent 跑通、准备做多 Agent 协作的开发者;二是被多步任务流程折磨过、想搞清楚编排套路的工程同学;三是想系统理解 Agent 架构、但被各种框架名词绕晕的技术负责人。我会从编排的核心矛盾讲起,拆解几种主流编排模式,给出可落地的实现思路,再重点聊聊那些文档里不写、只有踩过才知道的坑。
先给一个我自己的判断:编排的本质是"控制流 + 数据流 + 状态"三件事的协同设计。控制流决定执行顺序,数据流决定信息怎么传递,状态决定上下文怎么保持。绝大多数编排翻车,都是这三者里至少有一个没想清楚。下面我们逐个拆。
2. 编排的三种主流范式:顺序、路由与循环
在动手写代码之前,得先想清楚你的任务适合哪种编排范式。我见过太多人一上来就上"多 Agent 自由协作",结果任务简单到根本不需要,白白增加了调试成本。范式选错,后面全是坑。
2.1 顺序编排:最朴素也最容易被低估
顺序编排就是 A 做完交给 B,B 做完交给 C。听起来简单到不值一提,但它其实是生产环境里用得最多、最稳的范式。原因很直接:执行路径确定,出问题好定位,token 消耗可预测。
顺序编排的关键不在"顺序"本身,而在步骤之间的数据契约。我踩过的一个典型坑:第一步 Agent 输出一段自然语言,第二步 Agent 期望的是结构化 JSON,结果第二步直接解析失败。后来我强制要求每一步的输出都走 schema 校验,问题才消失。
一个实用的顺序编排骨架大概长这样:
from pydantic import BaseModel class StepOutput(BaseModel): result: str confidence: float next_action: str def run_pipeline(input_data, steps): context = {"input": input_data} for step in steps: raw = step.agent.invoke(context) parsed = StepOutput.model_validate_json(raw) # 强制校验 context[step.name] = parsed.result if parsed.confidence < 0.5: raise RuntimeError(f"步骤 {step.name} 置信度过低,中断") return context注意那个confidence字段和中断逻辑。顺序编排最怕的就是"一步错、步步错",所以在关键步骤设置质量闸门是必须的。我一般会在信息抽取、格式转换这类容易出错的步骤后面加校验,校验不过就重试或中断,而不是硬着头皮往下走。
2.2 路由编排:让对的 Agent 干对的事
路由编排解决的是"分诊"问题:来了一个请求,先判断它属于哪一类,再分发给对应的专家 Agent。这就像医院挂号,先分科室再找医生。
路由的核心是分类器的准确性。这里有个反直觉的经验:不要用大模型做路由分类,除非类别语义非常微妙。我实测下来,对于类别边界清晰的场景(比如"退款/咨询/投诉"),一个微调过的小模型甚至关键词规则,准确率和速度都吊打大模型路由,成本还低一个数量级。
路由编排的另一个坑是兜底策略。分类器总有分错的时候,如果分错了直接报错,用户体验很差。我的做法是加一个"通用 Agent"作为兜底,路由置信度低于阈值时交给它处理,同时记录日志用于后续优化分类器。
| 路由方式 | 适用场景 | 准确率 | 成本 | 延迟 |
|---|---|---|---|---|
| 关键词规则 | 类别边界清晰 | 中 | 极低 | 极低 |
| 小模型分类 | 类别较多、语义中等 | 高 | 低 | 低 |
| 大模型分类 | 语义微妙、类别模糊 | 很高 | 高 | 高 |
| 混合策略 | 生产环境推荐 | 很高 | 中 | 中 |
2.3 循环编排:最强大也最危险
循环编排让 Agent 能反复执行直到满足某个条件,这是实现"自我修正""迭代优化"的基础。比如让 Agent 写代码、跑测试、根据报错改代码,循环直到测试通过。
但循环编排是烧钱和死循环的重灾区。我见过一个没设上限的循环,一晚上烧掉了几百块。所以循环编排有三条铁律:
第一,必须有最大迭代次数。我一般设 5 到 10 次,超过就强制退出并报警。第二,必须有明确的终止条件,而且这个条件要能被程序判定,不能是"感觉差不多了"。第三,每轮循环要能感知进展,如果连续两轮输出几乎一样,说明卡住了,应该主动跳出而不是继续空转。
def loop_orchestration(task, max_iter=8): history = [] for i in range(max_iter): result = agent.invoke({"task": task, "history": history}) if is_satisfied(result): # 明确的程序化终止条件 return result if is_stuck(history, result): # 检测停滞 raise RuntimeError("检测到循环停滞,主动中断") history.append(result) raise RuntimeError(f"达到最大迭代次数 {max_iter} 仍未完成")is_stuck这个函数是我强烈建议加的。实现很简单:比较最近两轮输出的相似度,超过阈值就判定停滞。这个小小的检查,帮我省下了大量无谓的 token 消耗。
3. 多 Agent 协作:什么时候该拆,什么时候不该拆
"多 Agent"是这两年最热的词之一,但我必须泼盆冷水:大多数任务根本不需要多 Agent。把一个大 Agent 拆成五个小 Agent 互相通信,往往不是提升了能力,而是把一个问题变成了五个问题外加四个通信问题。
3.1 拆分的真正理由:上下文隔离与职责专一
那什么时候该拆?我的判断标准只有两条。
第一条是上下文隔离需求。当一个任务的不同部分需要完全不同的知识背景,且这些背景塞进一个上下文会互相干扰时,就该拆。比如一个系统既要处理法律条款又要处理技术文档,这两类知识放一起会让模型注意力分散,拆成两个专家 Agent 效果明显更好。
第二条是职责专一到需要不同工具集。如果一个环节需要调用数据库,另一个环节需要调用外部 API,且工具描述很长,塞进一个 Agent 会让工具选择准确率下降,这时候拆开是合理的。
反过来,如果两个 Agent 用的是同一套工具、同一类知识,只是"人设"不同,那基本没必要拆。我试过给同一个模型套不同 system prompt 假装成多个 Agent,效果和单 Agent 加角色切换几乎没差别,纯属自欺欺人。
3.2 通信开销:被严重低估的成本
多 Agent 系统里,Agent 之间的通信不是免费的。每一次消息传递都意味着:序列化、传输、反序列化、重新构建上下文。更麻烦的是信息在传递中会失真。
我做过一个实验:让 Agent A 把一段结构化信息传给 Agent B,中间经过三次转述,最后 B 拿到的信息准确率下降了将近 30%。这就是为什么我坚持Agent 之间传递结构化数据,而不是自然语言。自然语言适合人读,不适合 Agent 之间传。
# 不推荐:自然语言传递,容易失真 message = "刚才那个用户说他想要退款,订单号好像是12345,金额大概100块" # 推荐:结构化传递,字段明确 message = { "intent": "refund", "order_id": "12345", "amount": 100.0, "confidence": 0.92 }结构化传递还有个好处:接收方可以直接做字段校验,缺字段就报错,而不是靠模型"猜"。
3.3 中心化 vs 去中心化:别被"自治"忽悠了
多 Agent 架构分两种:中心化(有一个协调者 Agent 统一调度)和去中心化(Agent 之间平等协商)。去中心化听起来很酷,像一群智能体开会讨论,但生产环境我几乎只用中心化。
原因很现实:去中心化系统的行为难以预测和调试。当五个 Agent 自由对话时,你很难复现一个 bug,因为每次对话路径都可能不同。而中心化架构里,协调者的决策逻辑是明确的,出了问题能定位到具体哪一步。
去中心化唯一值得用的场景是探索性任务,比如头脑风暴、方案发散。这种任务本来就不追求确定性,让 Agent 自由碰撞反而能出好点子。但一旦进入执行阶段,立刻切回中心化。
4. 状态管理:编排里最容易翻车的地方
如果说编排有哪个环节最容易出问题,我的答案一定是状态管理。控制流和数据流相对好设计,但状态——也就是"系统当前处于什么阶段、已经知道了什么、还缺什么"——是最容易失控的。
4.1 无状态设计的诱惑与陷阱
很多人一开始会想:让每个 Agent 无状态,所有信息通过参数传递,多干净。这个想法在小规模下没问题,但一旦任务步骤变多,参数会爆炸式增长,而且跨步骤的隐含状态无处安放。
比如一个客服 Agent,用户第一轮说"我要退款",第二轮说"就那个上周买的"。这个"上周买的"指代什么,需要状态来记录。如果完全无状态,每一轮都要重新理解全部历史,成本和准确率都受不了。
我的做法是显式状态机 + 状态快照。用一个明确的状态对象记录当前阶段和已知信息,每次 Agent 调用前后都更新这个状态,并定期快照以便恢复。
class OrchestrationState(BaseModel): stage: str # 当前阶段 collected: dict # 已收集的信息 history: list # 关键决策历史 retry_count: int = 0 def transition(self, next_stage): self.history.append({"from": self.stage, "to": next_stage}) self.stage = next_stage4.2 状态爆炸:当上下文越来越长
状态管理的另一个大坑是上下文无限增长。多轮对话或多步任务里,如果把所有历史都塞进上下文,很快就会超出窗口,而且模型注意力会被稀释,效果反而下降。
我的经验是分层保留:最近的几轮完整保留,较早的历史做摘要压缩,更早的只保留关键结论。摘要压缩这一步本身可以交给一个专门的"记忆管理 Agent"来做,但要注意摘要会丢信息,所以关键的结构化数据(订单号、金额这类)永远不要压缩,原样保留。
提示:上下文压缩的阈值不要设得太激进。我一般保留最近 5 到 8 轮完整对话,再往前才压缩。压得太早,模型会"失忆",用户会觉得系统很蠢。
4.3 并发与竞态:多 Agent 同时干活时的暗雷
当多个 Agent 并行执行时,状态管理会变得棘手。两个 Agent 同时读写同一个状态对象,很容易出现覆盖。我踩过一次:两个 Agent 并行收集信息,结果后写的把先写的覆盖了,丢了一半数据。
解决办法有两个:一是给状态加锁,串行化写操作;二是让每个 Agent 写自己的命名空间,最后再合并。我倾向后者,因为加锁会拖慢并行度。合并的时候要注意冲突检测,如果两个 Agent 对同一字段写了不同值,得有个仲裁逻辑。
5. 错误处理与可观测性:让编排系统"死得明白"
编排系统最怕的不是出错,而是出错了不知道错在哪。一个多 Agent 系统如果缺乏可观测性,调试起来就是噩梦。这一节聊聊怎么让系统"死得明白"。
5.1 重试不是万能药:区分可重试与不可重试
很多人一遇到错误就无脑重试,结果把不可重试的错误重试了十遍,浪费资源还掩盖了真正的问题。我的分类标准是:
- 可重试:网络超时、限流、临时性服务不可用。这类错误重试往往能成功。
- 不可重试:参数错误、schema 校验失败、权限不足。这类错误重试一百遍还是错,应该直接失败并报警。
- 需修正后重试:模型输出格式不对。这类错误不能原样重试,要把错误信息反馈给模型让它修正,再重试。
def smart_retry(fn, max_retry=3): for i in range(max_retry): try: return fn() except RetryableError: time.sleep(2 ** i) # 指数退避 except FixableError as e: fn = lambda: fn_with_feedback(e) # 带反馈重试 except FatalError: raise # 直接抛出,不重试 raise RuntimeError("重试次数耗尽")指数退避这个细节别省。我见过有人重试间隔固定 100ms,结果把下游服务打挂了。退避加上随机抖动,能有效避免重试风暴。
5.2 追踪链路:每个决策都要留痕
一个可观测的编排系统,应该能回答这些问题:这个请求经过了哪些 Agent?每个 Agent 的输入输出是什么?在哪一步耗时最长?在哪一步失败了?
我的做法是给每个请求分配一个 trace_id,所有 Agent 调用都带上它,日志按 trace_id 聚合。这样排查问题时,一条命令就能拉出完整链路。别小看这个,没有 trace_id 的时候,我在日志海里捞针捞到怀疑人生。
| 观测维度 | 记录内容 | 用途 |
|---|---|---|
| 链路 | trace_id、步骤序列 | 复现完整执行路径 |
| 输入输出 | 每步的 prompt 和 response | 定位是哪步出的问题 |
| 耗时 | 每步开始结束时间 | 找性能瓶颈 |
| 成本 | 每步 token 消耗 | 控制预算 |
| 决策 | 路由选择、终止判断 | 理解系统行为 |
5.3 降级策略:优雅地失败
编排系统要有降级预案。当某个 Agent 不可用时,是直接报错,还是走备用路径?我的原则是核心路径必须有降级方案。
比如一个依赖外部搜索的 Agent 挂了,可以降级到用缓存结果或返回"暂时无法查询"。降级不追求完美,追求的是"用户还能用",而不是整个系统瘫痪。降级逻辑要提前设计好,别等出事了临时加,那时候手忙脚乱容易出更大的错。
6. 我踩过的那些编排坑:一份实战避坑清单
前面讲了不少原理,这一节集中说说那些文档里不写、只有真跑过才知道的坑。这些都是我或者身边同行实际踩过的,每一条都对应着真金白银的教训。
6.1 提示词里的"隐形指令冲突"
多 Agent 系统里,每个 Agent 都有自己的 system prompt。当这些 prompt 里的指令互相冲突时,系统行为会变得诡异。我遇到过一次:协调者 Agent 的 prompt 说"尽快给出答案",而专家 Agent 的 prompt 说"必须充分验证后再回答",结果专家 Agent 一直在验证,协调者一直催,系统卡在中间反复横跳。
解决办法是建立全局的指令优先级,明确哪些指令是硬约束、哪些是软偏好。硬约束(如安全、格式)所有 Agent 必须遵守,软偏好(如风格、速度)可以按场景调整。写 prompt 的时候,把硬约束放在最前面,并且用明确的措辞标注。
6.2 工具描述太长导致的"选择困难"
给 Agent 挂太多工具,或者工具描述写得太啰嗦,会让模型在工具选择上犯迷糊。我实测过,工具数量超过 15 个之后,选择准确率明显下降。
对策是按场景动态加载工具。不要把所有工具一次性塞给 Agent,而是根据当前任务阶段,只加载相关的那几个。比如信息收集阶段只给搜索和读取工具,执行阶段才给写入工具。这样既减少了干扰,也降低了 prompt 长度。
6.3 温度参数在编排中的微妙影响
温度(temperature)这个参数在单 Agent 场景下影响有限,但在编排场景下会被放大。协调者 Agent 如果温度太高,路由决策会不稳定,同样的输入两次可能走不同路径,调试起来极其痛苦。
我的经验是:协调者、路由者、校验者这类"决策型" Agent,温度设低(0 到 0.3);创意生成、方案发散这类"生产型" Agent,温度可以高一些(0.7 到 1.0)。别全局用一个温度,那是偷懒。
6.4 别忽视"空输入"和"边界输入"
编排系统上线后,最容易被忽略的是边界情况。用户什么都不输入直接回车、输入超长文本、输入一堆特殊字符,这些都会让 Agent 行为异常。我见过一个系统,用户输入一个空字符串,结果触发了默认路由,跑了一整套流程,白白烧钱。
对策是在编排入口加输入预处理和校验。空输入直接拦截,超长输入截断或分段,特殊字符清洗。这些看起来是小事,但能避免大量无谓的调用。
6.5 版本管理:Agent 也会"升级翻车"
Agent 的 prompt、模型版本、工具定义都会变。如果这些变更没有版本管理,某天你改了协调者的 prompt,结果整个系统行为大变,而你根本不知道是哪次改动导致的。
我的做法是把 prompt 和配置当代码管理,每次变更都走版本控制,并且记录变更前后的效果对比。有条件的话,搞一套回归测试集,每次改动前跑一遍,确保核心场景不退化。这个投入在系统复杂到一定程度后,回报是巨大的。
7. 从零搭一个编排骨架:可复用的最小实现
讲了这么多原理和坑,最后给一个我常用的最小编排骨架。它不是某个框架的教程,而是一套与框架无关的设计思路,你可以用任何语言、任何框架去实现。
7.1 核心抽象:三个对象搞定编排
我的骨架只有三个核心对象:State(状态)、Step(步骤)、Orchestrator(编排器)。
State 负责承载所有跨步骤的信息,Step 是单个可执行单元(可以是一个 Agent 调用,也可以是一个纯函数),Orchestrator 负责按规则驱动 Step 执行并更新 State。
class Step: name: str def run(self, state: State) -> State: raise NotImplementedError class Orchestrator: def __init__(self, steps, router=None, max_iter=10): self.steps = steps self.router = router self.max_iter = max_iter def execute(self, state: State) -> State: for _ in range(self.max_iter): step = self.router.select(state) if self.router else self.steps[0] if step is None: break # 终止条件满足 state = step.run(state) return state这个骨架的好处是职责清晰。State 只管数据,Step 只管干活,Orchestrator 只管调度。想加日志?在 Orchestrator 里加。想加校验?在 Step 里加。想换路由策略?换 router 就行。各模块互不干扰。
7.2 路由器的可插拔设计
路由器决定了下一步走哪个 Step,它是编排的"大脑"。我把它设计成可插拔的,因为不同场景需要不同的路由策略。
class Router: def select(self, state: State): raise NotImplementedError class SequentialRouter(Router): def select(self, state): idx = len(state.history) return self.steps[idx] if idx < len(self.steps) else None class ConditionalRouter(Router): def select(self, state): # 根据 state 内容决定下一步 if state.stage == "collect": return self.collect_step elif state.stage == "execute": return self.execute_step return None顺序路由适合固定流程,条件路由适合有分支的场景。你甚至可以实现一个"模型路由",让大模型来决定下一步,但要记得加缓存和兜底,别每次都问模型。
7.3 状态快照与恢复:让系统能"断点续跑"
生产系统难免中断,如果每次中断都从头开始,成本受不了。所以状态要能快照和恢复。
import json def save_snapshot(state: State, path: str): with open(path, "w") as f: f.write(state.model_dump_json()) def load_snapshot(path: str) -> State: with open(path) as f: return State.model_validate_json(f.read())快照的粒度要权衡:太频繁影响性能,太稀疏丢的进度多。我一般在每个 Step 完成后快照一次,这样最多丢一个 Step 的进度。快照存储用文件、数据库还是对象存储都行,看你的部署环境。
7.4 一个完整的编排示例:多步信息处理任务
把上面的零件拼起来,看一个完整例子。假设任务是:接收一段用户描述,抽取关键信息,查询相关资料,最后生成一份报告。
class ExtractStep(Step): name = "extract" def run(self, state): raw = llm.invoke(f"从以下描述抽取结构化信息:{state.input}") state.collected = parse_json(raw) state.transition("query") return state class QueryStep(Step): name = "query" def run(self, state): results = search(state.collected["keywords"]) state.collected["references"] = results state.transition("report") return state class ReportStep(Step): name = "report" def run(self, state): state.report = llm.invoke( f"基于{state.collected}生成报告" ) state.transition("done") return state这个例子里,每个 Step 只做一件事,通过 State 传递数据,通过 transition 记录阶段。想加一步校验?插一个 ValidateStep 进去就行。想改流程顺序?调整 steps 列表。这种结构在需求变化时特别省心。
8. 编排方案选型:自研、框架还是平台
最后一个绕不开的问题:编排到底该自己写,还是用现成框架?我的答案取决于你的阶段和需求。
8.1 什么时候该自研
如果你的编排逻辑比较固定、步骤不多(五步以内)、团队对可控性要求高,自研是更优解。自研的代码量其实不大,前面那个骨架一两百行就能跑起来,而且完全可控,出问题好定位,没有框架的黑盒。
自研的代价是要自己处理重试、日志、状态这些基础设施。但如果你的场景简单,这些基础设施也不需要太复杂,反而比引入一个重框架更轻快。
8.2 什么时候该用框架
当你的编排开始涉及复杂的图结构、动态分支、大量 Agent 协作时,框架能帮你省下不少重复劳动。框架通常提供了现成的状态管理、可视化调试、并发控制等能力。
但用框架要有心理准备:框架的抽象会带来学习成本和调试难度。当框架行为不符合预期时,你得深入源码才能搞明白。我见过团队为了用某个框架,花了两周才把它的状态机制搞清楚,这两周自研都写完了。
8.3 选型的三个判断维度
我一般从三个维度判断:
| 维度 | 自研 | 框架 |
|---|---|---|
| 流程复杂度 | 低(线性、少量分支) | 高(图结构、动态路由) |
| 团队规模 | 小团队、快速迭代 | 大团队、需要规范 |
| 可控性要求 | 极高、需要深度定制 | 中等、接受框架约束 |
没有绝对的好坏,只有适不适合。我的建议是:先用自研跑通核心逻辑,等复杂度真的上来了,再考虑引入框架。过早引入框架,往往是被框架的抽象绑架,而不是被它帮助。
8.4 一个务实的演进路径
如果你现在正要开始一个 Agent 编排项目,我会建议这样的路径:第一阶段,用最简单的顺序编排把主流程跑通,验证核心价值;第二阶段,根据实际需求引入路由和循环,同时把状态管理和日志补齐;第三阶段,如果多 Agent 协作成为刚需,再考虑框架或更复杂的架构。
这个路径的好处是每一步都有真实的反馈驱动,而不是一开始就过度设计。我见过太多项目,一上来就搭了个复杂无比的多 Agent 系统,结果核心任务还没跑通,全在调试架构了。编排是为任务服务的,别本末倒置。
我在实际项目里最深的一个体会是:编排的复杂度应该匹配任务的复杂度,而不是匹配你的技术野心。一个能稳定跑通的简单编排,价值远大于一个炫酷但三天两头出问题的复杂系统。先把简单的事情做扎实,复杂的事情才有基础。