多智能体编排的五个反模式:为什么系统越"聪明"越难维护
前言
第一次把多个 Agent 拼成一个"团队"的时候,多数人都会经历同一条曲线:Demo 阶段惊艳,接入真实任务后迅速失控。表现是会话记录越来越长、Agent 之间互相刷消息、同一个任务被重复执行、出问题无法定位。
这些问题很少是模型能力不够,几乎都出在编排层。下面五个反模式是我在阅读和拆解多智能体项目时反复见到的,每一个都给出症状、根因和改法。
反模式一:把"广播"当协调
症状:消息日志里同一条状态更新出现几十次,Agent 之间来回确认"你收到了吗",token 消耗随 Agent 数量呈平方级增长。
根因:默认用了 broadcast 分发。三个 Agent 时广播是无害的,五个以上就变成了噪声放大器——每个 Agent 都以为别人需要知道,于是所有人都被唤醒,然后互相唤醒。
改法:把通信收敛成"点对点 + 显式订阅"。
- 每个 Agent 声明自己关心哪几类事件,运行时分发器按订阅表路由,未订阅者根本不收到消息。
- 状态变更只由"当前持锁的 Agent"发出,其他 Agent 只读不写。
- 给同一条业务对象的连续更新做合并窗口(例如 300ms 内的多次状态变更合并成一次通知)。
一个可验证的判断标准:任意任务结束后,消息总数应该和任务步骤数同阶。如果消息数远超步骤数,说明存在广播放大。
反模式二:让 LLM 当路由器,却没有 schema 兜底
症状:跑一百次成功九十五次,剩下五次莫名其妙走进了错误的 Agent,或者返回一个谁都不认识的 next_agent 值,整个流程直接卡死。
根因:把"下一步该谁做"完全交给模型自由生成。模型输出的是自然语言,而调度器需要的是枚举值——中间缺一层强制约束。
改法:路由决策必须走结构化输出,并且做白名单校验。
from enum import Enum
from pydantic import BaseModel, ValidationError
class Route(str, Enum):
PLANNER = "planner"
CODER = "coder"
REVIEWER = "reviewer"
DONE = "done"
class Decision(BaseModel):
route: Route
reason: str
def decide(raw_model_output: str) -> Route:
try:
return Decision.model_validate_json(raw_model_output).route
except ValidationError:
# 兜底:不猜,交回给 Planner 重新规划
return Route.PLANNER
关键在于兜底分支要显式设计。校验失败时回退到一个确定的安全状态,而不是抛异常或者猜一个最近似的值。
反模式三:状态机藏在 prompt 里
症状:偶尔出现"顺序颠倒"——评审 Agent 在代码还没生成时就开始评审;或者任务明明完成了又从头跑一遍。
根因:把流程约束写进了系统提示词,例如"你是第三步,只在前两步完成后开始"。提示词是软约束,上下文一长就会被稀释;而并发、重试、超时这些真实条件是提示词覆盖不到的。
改法:状态机写到代码里,Agent 只负责"在当前状态内做判断"。
TRANSITIONS = {
"planning": {"coded"},
"coded": {"reviewing", "fixing"},
"reviewing": {"done", "fixing"},
"fixing": {"reviewing"},
"done": set(),
}
def advance(state: str, event: str) -> str:
if event not in TRANSITIONS[state]:
raise IllegalTransition(f"{state} -> {event}")
return event
状态机落在代码里之后,非法跳转会在第一时间报错,而不是在半小时后表现为"结果不对"。
反模式四:工具返回值不做归一化
症状:同一个工具,有时候返回 {"ok": true, "data": [...]},有时候返回裸数组,超时时返回 null,报错时返回一段中文提示。下游 Agent 面对五种形状,只能靠模型去"理解",于是行为不可复现。
根因:把工具的原始响应直接塞进上下文。
改法:所有工具出口统一包一层结果对象。
from dataclasses import dataclass, field
@dataclass
class ToolResult:
ok: bool
data: object = None
error: str | None = None
meta: dict = field(default_factory=dict)
def to_context(self) -> str:
if self.ok:
return f"SUCCESS: {self.data}"
return f"FAILED: {self.error} (retryable={self.meta.get('retryable', False)})"
模型需要知道的只有三件事:成功还是失败、数据是什么、能不能重试。其余字段留在 trace 里给工程师看,不要占用上下文。
反模式五:没有可回放的 trace
症状:线上偶发失败,日志里只有"任务失败",没有 Agent 决策链路,无法复现,只能猜。
根因:只在出入口打日志,中间每一步的输入、输出、决策理由全部丢弃。
改法:把每一次 Agent 调用当作一条结构化事件落盘。
必备字段:
| 字段 | 说明 |
|---|---|
trace_id |
贯穿整个任务的唯一 ID |
step |
单调递增的步骤号 |
agent |
哪个 Agent |
input_digest |
输入的哈希或前 N 字,避免存超长上下文 |
output |
完整输出(超过阈值截断并落对象存储) |
route_decided |
本次路由决策 |
latency_ms / tokens |
成本与耗时 |
有了这张表,"某次任务为什么走错分支"可以从 trace 里直接读出来,而不是重跑十次碰运气。
一张自查清单
上线前问自己五个问题:
- 任意任务的消息总数,是否和步骤数同阶?
- 路由输出是否走了 schema 校验,且兜底分支是确定的安全状态?
- 流程约束是在代码里,还是在提示词里?
- 工具返回值有几种形状,下游是否只需要认识一种?
- 出问题时,能不能只凭 trace 复现一次完整决策?
五个都能答"是",基本就脱离了"Demo 能跑、生产不敢上"的阶段。
多智能体系统的复杂度不在于 Agent 有多聪明,而在于编排层有没有把不确定性收敛到可控的几个点上。把广播换成订阅、把自由生成换成 schema、把提示词换成状态机、把裸返回值换成统一结果、把日志换成 trace——这五件事都不需要换模型,但决定了系统能不能上生产。
- 点赞
- 收藏
- 关注作者
评论(0)