多Agent协作的工程实践：消息协议、编排器与上下文管理-北京尧图网络

2. 通信与上下文Agent之间的“共同语言”很多人以为多Agent协作就是让多个Agent拉一个群聊你一句我一句大模型自己会搞定。早期我真的见过这么做的团队结果非常惨两个Agent能为了一个订单是否超时来回吵十轮到最后谁也没拍板主流程直接卡死。核心原因在于Agent之间并没有真正的“共同语言”它们只是在用自然语言对话而自然语言是有歧义的。企业里的Agent协作最佳的通信方式不是自由对话而是结构化的消息协议。每条消息要有明确的类型、来源、目标、内容体和唯一ID。比如类型可以是task_assign、task_result、query_request、query_response、error_report每个类型有固定的字段规范和执行语义。这样设计之后Agent的每个行为都是可预期的编排层可以判断当前流程走到哪一步出了错也能追溯到具体是哪条消息触发了异常。我习惯用Pydantic或者TypedDict来定义消息结构这样在代码层面就能卡住格式错误而不是等到运行时才靠Prompt去纠正。通信协议定好之后下一个问题是消息怎么路由是点对点直连还是走事件总线还是统一交给编排器转发。我的经验和教训是初期阶段不要引入太重的基础设施。消息总线看着没有专用网关优雅但调试、过滤、重放消息都会更麻烦。早期使用一条简单的内存队列或者用Task表在数据库里存状态够用且直观。真正到了多个服务、多个团队共用一套Agent能力的时候才需要引入独立的Event Bus或者消息中间件那时候每个Agent也是一个发布者/订阅者协作关系就变成了事件流驱动。还有一点容易忽略的是上下文续传。Agent协作过程中下游Agent完成的任务需要把结论回传而不只是把一堆工具的原始输出往上丢。比如库存Agent查到了一个SKU有现货它回传的应该是“库存充足预计发货时效24小时”这样的结论而不是一个原始JSON对象。这看上去很简单但我见过不少项目忽略了这一步导致下一环节的Agent每次都要用一个很长的上下文重新理解上游给的是什么token消耗和错误率一起涨上去。比较好的做法是每个Agent在输出里带上一个summary字段只写结论和关键依据详细数据放在附件里按需读取。关于上下文共享还要记住一点共享不是越多越好。你给Agent的是业务上下文而不是把所有人的对话历史都塞给它。很多单Agent时代养成的习惯——把完整对话历史塞进Prompt——到了多Agent协作场景必须改掉。协作场景里每个Agent只需要看到自己职责范围内的上下文和当前任务的上下文。多给的部分不会让它更聪明只会让它在无关信息里“游泳”。3. 实操一个Agent协作工作流的搭建过程3.1 场景选型用订单履约流程做例子理论说了这么多还是上手搭建一个看得见摸得着的协作流程最实在。我选一个几乎所有企业都会碰到的场景订单履约。从用户下单开始系统需要校验订单信息、检查库存、计算运费、生成履约指令如果遇到异常还要给客服提供处理建议。单Agent也能做但你会发现它会特别臃肿一个Prompt里塞了五六个职责动不动就“忘记”自己是干什么的。拆开后这个流程的每一步都变得清爽每个Agent负责一个环节专注于自己的工具和数据。订单履约流程我拆成四个角色订单校验Agent负责检查支付信息和收件信息是否完整合法库存确认Agent负责调用库存系统的接口确认可售库存和锁库物流测算Agent负责根据商品重量体积和收件地址计算运费和时效履约编排Agent作为最上层的协调者接收前三个Agent的输出决定整个单子是正常流转还是进入人工审核。这四个角色放在一起就是一个最小但完整的Agentic Collaboration系统。这里有一个选型要点协作场景不要一开始就挑那种特别复杂的业务比如多部门审批流、跨系统对账。先从“线性度高、环节边界清晰”的流程跑起来团队才能积累协作模式的经验以后再去碰那些网状协作的复杂场景才知道瓶颈在哪里。3.2 架构设计与Agent定义跑通之前先把架构图画稳。我习惯从两个角度做设计静态角度看角色和消息动态角度看执行时机与回退条件。静态上我要把每个Agent的职责边界、可调用工具、输出消息类型都写清楚职责边界这条线最难画画粗了下游看不懂画细了哪个Agent都干不完整只能靠实际跑几轮再慢慢调整。动态上我需要定义流程何时结束、何时重试、何时转人工这比让Agent自由发挥重要得多。Agent的自由度要体现在中间过程而不是体现在流程死活上。画架构图的时候我会刻意问自己几个问题如果库存Agent挂了主流程能不能临时用缓存数据顶一阵如果物流测算结果异常系统是重试还是直接转人工如果多Agent之间的结论互相矛盾谁有最终解释权这些问题的答案最终都写在编排器的代码里而不是写在Prompt里。新团队做这类系统最大的坑就是把所有决策都交给模型结果整个流程像一团浆糊出了问题还要翻对话记录才能猜出发生了什么。Agent定义这个环节我给大家一个非常朴素的公式角色定位可用工具列表输入消息类型输出消息格式决策边界。五个要素缺一不可。缺了决策边界的Agent会“越权”缺了输出消息格式的Agent会让下游看不懂。我们给这个订单履约系统里每个Agent写角色文档时把“决策边界”单独加粗例如库存确认Agent有权在商品缺货时提出建议但无权直接取消订单这个权力只在履约编排Agent手里。3.3 关键代码与消息协议示例代码部分我尽量精简用最小可运行的逻辑展示协作骨架。这里我用Python写一个简化版本只展示结构不粘贴公司内部代码。核心思路是编排器调度Agent、Agent消费消息输出消息。from enum import Enum from dataclasses import dataclass, field from typing import Optional, Callable class MessageType(str, Enum): TASK_ASSIGN task_assign TASK_RESULT task_result QUERY_REQUEST query_request QUERY_RESPONSE query_response ERROR_REPORT error_report dataclass class Message: msg_id: str msg_type: MessageType source: str target: str payload: dict trace_id: str status: str pending dataclass class AgentResult: success: bool summary: str detail: Optional[dict] None error: Optional[str] None class BaseAgent: name: str def __init__(self, name: str, tools: dict): self.name name self.tools tools def call_tool(self, tool_name: str, **kwargs): if tool_name not in self.tools: return {error: funknown tool {tool_name}} return self.tools[tool_name](**kwargs) def run(self, task: Message) - Message: raise NotImplementedError class OrderValidator(BaseAgent): def run(self, task: Message) - Message: payload task.payload order_id payload[order_id] if not payload.get(pay_status) or not payload.get(address): return self._result( task, AgentResult(False, 订单信息不完整, {order_id: order_id}) ) # 调用校验工具 check_resp self.call_tool(validate_address, addresspayload[address]) if not check_resp.get(valid): return self._result( task, AgentResult(False, 收件地址不合法, check_resp) ) return self._result( task, AgentResult(True, 订单信息校验通过, {order_id: order_id}) ) def _result(self, task, result): return Message( msg_idf{task.trace_id}-res-{self.name}, msg_typeMessageType.TASK_RESULT, sourceself.name, targetfulfillment_orchestrator, payload{order_id: task.payload[order_id], result: result.__dict__}, trace_idtask.trace_id, )这个代码段里的核心不是每个Agent内部的逻辑而是整个协作结构每个Agent继承BaseAgent通过统一的run(task)入口消费任务符合“统一接口”原则所有结果都打包成Message返回携带task.trace_id这条链路清晰可追踪Agent不允许直接调用另一个Agent的私有方法它只能发消息这就是“Agent之间只认协议不认对象”的体现。编排器这边核心是三步循环取出阶段要执行的任务找到对应Agent把结果交给MessageRouter决定下一步。这里甚至不需要Agent之间彼此知道对方的存在协作关系全部由编排器维护后期加Agent、换Agent都不动整体结构。关键的控制逻辑可以简化成一句判断如果第一阶段的所有Agent都返回success就进入下一步如果任何一个返回failure就走人工审核分支不管后面的Agent是否已经准备好了。3.4 联调过程与参数调优流程跑通之后有一段联调期这段时间里不只是“功能能不能用”的问题还会暴露一堆参数调优的坑。先说模型温度Agent协作场景里不要用太高的随机性一般temperature控制在0.1到0.2之间比较合适。如果某个Agent经常在format、判断逻辑上飘忽不定可以考虑直接降到0让模型输出更确定性。而那些处理开放的、有创造性的任务比如生成个性化话术的Agent可以单独调高一点根据职责分开配参一个模型服务可以同时服务多个Agent但参数各自独立。超时和重试参数也很容易踩坑。我见过一个团队把HTTP超时设成30秒结果某个Agent因为上游工具响应慢锁住了整条流水线。后来我们把超时改成了动态策略外部接口调用最长等待8秒如果超时则触发降级动作直接返回“不确定”上游Agent拿到这个结果后转给人工审核。别小看这个“不确定”状态有些Agent协作系统死在“所有Agent都必须给出确定性答案”这个假设上。还有重试次数Agent调用工具失败后不要让它无脑重试三次应该结合错误类型来判断。比如库存接口返回“参数错误”重试一百次也没用直接报错如果是“连接超时”重试一次或两次是合理的。这个经验直接写在Agent的tool call策略里比依靠模型自己的判断更省心省力。参数调优过程中我强烈建议做一次“混沌测试”手动把某一环节的中间结果改成异常值比如篡改库存Agent返回的数据为负数看看整个协作流会怎么反应。真正好的协作系统在这个环节应该能识别出异常值并且有明确的后续动作而不是把异常值一路传下去最后生成一个荒谬的履约指令。4. 协作深水区常见问题与排查实录4.1 问题速查从症状到根因Agent协作系统上线之后排查问题的方式和传统开发完全不同。传统开发是查代码Agent协作系统是查“对话史状态消息流”。我把过去踩过的坑整理成了速查表典型症状可能根因排查思路流程卡在某个环节不往下走编排器的状态机没覆盖该消息类型或Agent超时未返回查trace_id对应的消息流确认最终状态Agent之间反复确认相同问题上游Agent的summary没给结论下游Agent不信任检查summary字段是否只输出结论和依据某个Agent频繁调用错误工具工具列表太长模型选择困难精简工具列表或用函数式编排降低单选压力结果时好时坏temperature过高Prompt边界不清调低temperature检查Agent角色文档一个Agent出错后坏结果传到全链路缺少校验环节Agent输出没有schema校验在编排器增加Message校验层响应变慢且token消耗激增上下文缓存没生效重复注入大段背景上下文改按需加载不要全量塞Prompt这里我想强调一个观点协作系统里问题往往不在单个Agent的推理能力上而是在消息设计和流程控制上。每次排查问题第一反应不要调Prompt先去看消息流看每一步的输入输出是否符合既定规范。十个问题里至少六个是消息不规范导致的剩下三个是流程控制有漏洞真正需要靠模型调Prompt才能解决的只有一个。4.2 两条务实建议先做减法再协作、人格别太多聊到最后我一定要给正在计划做Agentic Collaboration的团队两条建议都是拿实际操作换来的教训。第一条先做减法。协作的初衷不是把简单流程拆得又碎又多而是让复杂流程的每个环节可以有独立的判断空间。我见过最夸张的一个项目刚启动就把采购流程拆成了12个Agent结果连一个最小闭环都跑不通每天都在协调Agent之间的关系。后来我们把这12个砍到4个流程反而顺了。新增Agent的成本不只是多写一个角色文档还意味着消息复杂度、异常路径、权限管理的成本一起上升。正确的节奏是从一个最少可用协作闭环开始比如“一个编排器加两个执行Agent”跑稳定了再逐步扩。第二条人格别太多。给Agent起名字、设人设这个习惯在客服聊天机器人领域是合理的但在后台执行类Agent那里往往只会带来灾难。一个Agent如果被定义了“友好热情”的人格它可能为了凑一句热情满满的回复把消息格式结尾都改了然后下游Agent就解析失败。企业里的执行力Agent是一台处理消息和工具的机器它需要的是标准和效率不是人格魅力。我在涉及这类Agent的Prompt里都会写上一句“不要寒暄不要评价直接给出结构化结果”。这两条建议本质上是同一个原则Agentic Collaboration的价值在于把复杂问题拆解成更可管理、更可观测的小环节而不是把不确定性叠加起来。协作本身需要严密的工程纪律比任何单Agent技术都更需要纪律。最后再分享一个我最近在用的实操技巧给协作流程加一个“最小逃生动作”。就是无论协作多复杂始终有一条物理链路可以让用户或运营人员一键跳过所有Agent直接进入人工处理。这个逃生通道平时可能很少被触发但它会给你一种底气Agent协作出了问题不会变成灾难因为业务永远有一条人工兜底路线。有了这条线你在推进技术架构改造的时候会更务实也更从容。

多Agent协作的工程实践：消息协议、编排器与上下文管理

相关推荐

GG3M 战略文献总览：从战略纲领、文明维度跃迁到国家战略与 C2 文明蓝图——GG3M 战略认知、技术路线、全球合作、资本投资与文明操作系统汇编

基于Java的Spring Boot项目管理平台：毕设源码四件套实战解析

蓝桥杯Java模拟赛解题策略与实战技巧：填空题、编程大题与赛前准备

更多资讯

Python机器学习人脸识别系统：从期末大作业到门禁机实战

基于Python的单通道脑电睡眠分期：从数据到模型全流程实战

Python机器学习实战：从环境配置到模型训练复现的完整流程

供应链断裂与重构测度：基于Python的级联失效与替代路径实现

Python车牌识别计费系统实战：从OCR识别到状态机与跨天计费

南开编译原理作业实战：简单C语言编译器源码解析与避坑指南

今日更新

Susi AI API实战指南：susi_alexa_skill如何用Node.js调用chat.json获取智能回答

换新电脑了？KeyStats 恢复码数据找回完全指南，端到端加密统计一键重建

重建二叉树：由前序与中序遍历序列还原二叉树的原理与 Java 实现（YCBlogs 剑指 Offer 精讲）

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？