上一篇把事实、摘要与原始事件分层。本篇讨论何时值得把一个循环拆成多个 Agent,以及如何用明确交接协议避免“多人聊天式”失控。多 Agent 的价值来自权限、上下文或能力隔离,不来自角色名字更多。

一、痛点:从“能演示”走向“可托管”

先问单 Agent 加普通函数能否完成。如果任务只是固定三步,工作流更简单。只有当子任务需要不同工具权限、独立上下文或可并行专业判断时,才值得拆分。每个角色必须拥有单一职责、输入契约、输出契约和停止条件;协调者负责路由与合并,不替专家编造结论。

真正可迁移的做法,是先写任务契约再选模型:输入从哪里来,成功产物是什么,哪些动作禁止自动执行,最多允许多少步、多少时间和多少费用,失败后由谁接管。契约一旦写进代码和测试,模型升级只是实现替换,不会悄悄改变业务责任边界。

二、原理:把概率判断放进确定性边界

交接包应包含 task_id、目标、已知事实引用、允许工具、交付格式和截止预算。专家输出结论、证据、置信说明与未解决问题。禁止把完整内部对话转发给下一角色,因为其中可能含无关敏感信息和提示注入。共享状态采用追加事件,冲突由预先定义的仲裁规则处理。

模型输出、工具数据和记忆内容都按不可信输入处理。每个边界都执行校验、授权、裁剪和审计;所有状态变化都有明确原因。这样做看似增加一些代码,却把“偶尔答错”拆成可定位的规划错误、参数错误、工具错误或策略错误,团队才能针对性改进。

三、实现:用独立程序跑通最小闭环

示例模拟研究员与审阅员的交接。协调器验证每个消息的 sender、recipient 与必填字段,并限制角色只能发送给允许的下一跳。这个确定性信封可套在任意模型外部,使协作拓扑不受模型自由发挥影响。

from dataclasses import dataclass

@dataclass(frozen=True)
class Message:
    task_id: str
    sender: str
    recipient: str
    evidence: tuple[str, ...]
    decision: str

allowed = {("researcher", "reviewer"), ("reviewer", "coordinator")}
messages = [
    Message("T-42", "researcher", "reviewer", ("doc:A", "doc:B"), "ready"),
    Message("T-42", "reviewer", "coordinator", ("doc:A", "doc:B"), "approved"),
]
evidence: set[str] = set()
status = "running"
for message in messages:
    if (message.sender, message.recipient) not in allowed:
        raise ValueError("illegal route")
    if not message.task_id or not message.evidence:
        raise ValueError("incomplete handoff")
    evidence.update(message.evidence)
    status = message.decision
    print(f"accepted={message.sender}->{message.recipient} task={message.task_id}")
print(f"final_status={status} evidence_count={len(evidence)}")

运行输出:

accepted=researcher->reviewer task=T-42
accepted=reviewer->coordinator task=T-42
final_status=approved evidence_count=2

并行并不自动更快。只有互不依赖的子任务才能并行;若审阅员必须等待研究证据,就应明确串行。合并阶段要保留冲突,而不是让协调者取平均。比如两位专家引用不同政策版本,应按生效日期和权威级别决断,无法决断就转人工。

第二个程序补上生产中最容易遗漏的控制点。它与前一个示例完全独立,可单独保存运行,不依赖本系列其他文件;示例数据是内存假实现,因此不会访问真实账户或产生外部副作用。

from dataclasses import dataclass

@dataclass(frozen=True)
class Evidence:
    claim: str
    value: str
    source: str
    effective_year: int

items = [
    Evidence("refund_window", "14_days", "policy_2025", 2025),
    Evidence("refund_window", "7_days", "policy_2026", 2026),
]
values = {item.value for item in items}
conflicts = max(0, len(values) - 1)
selected = max(items, key=lambda item: item.effective_year)
status = "resolved_by_newer_source" if conflicts else "consistent"
print(f"claim={selected.claim}")
print(f"selected={selected.value}")
print(f"source={selected.source}")
print(f"conflicts={conflicts}")
print(f"status={status}")

运行输出:

claim=refund_window
selected=7_days
source=policy_2026
conflicts=1
status=resolved_by_newer_source

落地时应把示例中的内存状态替换为事务型存储,把打印日志替换为结构化事件,但不要改变契约。事件至少包含 run_id、步骤、输入摘要、策略结果、耗时、错误类别和版本;密钥、完整个人信息及未经脱敏的提示不得进入日志。

四、踩坑:失败路径决定系统上限

常见反模式包括:多个角色访问同一组高权限工具,实际上没有隔离;角色互相礼貌确认却不产生证据,白白消耗 token;协调者转述专家结果时丢掉限制条件。控制方法是最小权限、最大交接次数、结构化输出和证据直引。角色失败时只重跑对应分支,不要让整组重新讨论。

还要防止把确定逻辑模型化。金额计算、权限判断、枚举路由和状态转移应使用普通代码;模型适合处理分类、抽取、歧义消解与证据综合。每减少一个无必要的模型决策点,就减少一处延迟、成本和不可复现性,同时让测试更容易覆盖。

五、验证:用轨迹和门槛验收

评测应同时与单 Agent 基线比较。若成功率没有提升,或延迟、成本增长超过预算,就不应采用多 Agent。轨迹检查还要覆盖非法下一跳、缺证据交接、冲突来源、子任务超时和部分完成。下一篇将把这些角色节点放入显式工作流,以持久化方式编排分支、审批与恢复。

发布顺序固定为离线回放、影子流量、小比例灰度和有限自治。每阶段先定义通过线和回滚条件,再看结果;不能在看到分数后移动门槛。关键样本要保存初始状态、每步动作、观察、最终产物与终止原因,模型、提示和工具契约版本也必须一起记录。

工作流会把‘谁在何时做什么’从提示词移到可检查的图结构中,多 Agent 只负责图中确实需要语义判断的节点。

参考来源


👍 觉得有用就点个 赞 + 收藏,方便回头查阅;有疑问直接在评论区留言,我看到都会回。

🚀 本文属于 《AI Agent 应用实战》 系列,持续更新,关注不迷路。

📌 文章里的代码都能直接跑。想要可直接 clone 的完整工程 + 配套部署脚本 / 踩坑清单?评论一声或发邮件到 cj2664@qq.com,我免费发你。
如果你正好在做类似系统、或有工程化难题想找人做,也欢迎邮件聊一句——我按实际情况评估,能落地的就接单或出方案。评论和邮件都能直接找到我,不用跳别的平台。

更多推荐