Agent-harness组内协议设计：多Agent协作的消息模型与排障实战

组内协议这四个字我前后琢磨了将近一个月才算真正把它从“概念”变成了“代码”。刚接触 Agent-harness 的时候我犯过所有新手都会犯的错误以为重点在 Agent 本身网上教程也几乎清一色在教怎么让单个 Agent 调工具、写代码、查资料。结果真把两三个 Agent 放进同一个组里跑协作任务问题立刻变成了另外一套任务由谁分配、先后顺序怎么定、结果用什么格式传递、出了冲突谁来仲裁。这些事单个 Agent 根本管不了恰好是 Agent-harness 和组内协议要解决的。这篇学习笔记围绕一个核心问题在 Agent-harness 的框架里一组 Agent 之间到底需要什么样的“对话规矩”才能协作顺畅而不互相踩踏。我会从设计思路讲起拆解消息模型和协作模式选型再给出一套可以直接改来用的最小协议实现最后把我踩过的坑整理成排查手册。适合正在做多 Agent 协作、刚上手 Agent-harness、或者想让 Agent 组不再“各说各话”的开发者和算法工程师参考。1. Agent-harness 在管什么先理解边界再谈协议1.1 没有组内协议时协作到底乱在哪我拿一个很常见的场景举例给一个内容产出组配上“调研 Agent”和“写作 Agent”让它们共同完成一篇文章。因为没有协议两边都直接操作同一个共享文档。结果写作 Agent 拿到一份还没写完整的调研资料就开始动笔调研 Agent 更新完资料后反手就把文档覆盖了。更离谱的是两个 Agent 都觉得自己应该承担“总结”这一步于是同一个任务被同时执行了两遍。这不是 Agent 能力不行而是它们之间缺少一套明确的交接规则。类似的问题在代码生成组、客服工单组、数据分析组里都会复现。我做了一段时间之后总结出一个判断多 Agent 协作里绝大多数故障都不是单个 Agent 跑飞了而是“交接”出了问题——信息没传到位、顺序没人保证、结果没人确认。组内协议的本质就是给这些交接动作立法让每一步都说得清、查得到、能重来。1.2 Harness 管生命周期协议管说话方式这里要先理清一个边界不然代码很容易写成一大坨。Agent-harness 真正要管的是四类事Agent 实例的创建与销毁也就是生命周期管理消息的投递与路由也就是通信基础设施共享上下文和记忆的读写也就是状态管理还有外部工具的权限控制以及运行日志、链路追踪这类可观测性能力。而组内协议是跑在 harness 之上的“会话规则”它规定消息长什么样、每种消息代表什么含义、任务状态如何流转。我的实操经验是harness 层只做“传输和存储”不要掺和业务判断比如别在 harness 里写死“当消息类型是 X 时就该做什么任务”那是 Agent 层的职责协议层只做“约定和校验”不越权去调度谁先做谁后做Agent 层才专注自己的任务逻辑。这个三层层级一旦混乱排查问题时你会同时面对路由错误、协议错误和业务错误谁也分不清锅该给谁。先记住这条边界后面所有设计才有位置放。2. 组内协议设计原则动手写代码前先定规矩2.1 显式化所有协作动作都必须是消息第一条原则是“显式化”。两个 Agent 之间不要靠共享变量的隐式约定来协作比如“我把结果写到某个固定路径你自己去读”。一切动作——你接了我的任务、你开始做了、你做到一半、你做完了、你失败了——都要通过一条明确的消息来表达。原因非常现实隐式协作出问题时无据可查你连是谁动了共享数据都说不清楚而显式消息让每一步都可回放、可测试、可 mock。我做的一个小而关键的落地动作是每个 Agent 的代码里只允许通过 harness 提供的 send 方法发消息不允许直接写共享存储。这样所有协作动作都会在消息日志里留下记录回头抽日志复盘时每一步都清清楚楚。这看起来多写了一行代码但省掉的是无数个凌晨排查“数据怎么又没了”的时间。2.2 可追溯trace_id 是排障的命根子第二条原则是“可追溯”。我见过太多 Agent 组出问题时日志里全是零散的消息看不到一条完整链路根本没法回答“这个任务到底走到哪一步才断的”。所以协议里必须带上 trace_id一次任务从发起、分配、执行到返回结果过程中产生的所有相关消息共享同一个 trace_id。这样一来出问题时一条 grep 命令就能把全链路拉出来配合时间戳就能看出卡点在哪里。我强烈建议把这条原则放在协议设计的最高优先级因为大多数项目刚上线时不会出问题等出问题的时候没有 trace_id 的协议约等于没有协议你只能靠猜。在我自己的实现里trace_id 由调度方在创建任务时生成后续子任务继承同一个 trace_id再通过子任务 id 区分分支这样既能看整条链路又能定位到具体某一步。2.3 可恢复每条关键消息都要配超时和重试第三条原则是“可恢复”。生产环境跑 Agent 组超时是常态而不是意外。LLM 调用不稳定外部工具可能挂消息队列可能抖动你不能假设“发出去的消息对方一定会收到并响应”。协议必须提前规定任务下发后多少秒内必须 ack进度汇报的间隔上限是多少失败后最大重试几次重试采用什么退避策略。这些参数要写进协议文档而不是指望每个 Agent 自觉处理。不然每次换一个新的 Agent 进组协作质量就完全取决于这个 Agent 的开发者有没有想到要处理超时。我在 5.4 小节会给一组可以直接抄的参数和代码。先记住大方向没有超时和重试机制的组内协议只适合 demo不适合任何需要长期运行的服务。2.4 克制协议不是越重越好第四条原则可能最反直觉——“克制”。不要一上来就设计三十种消息类型不要给每条消息加二十个字段。我见过团队把协议设计得非常完备结果实现成本极高组里每个人都在写协议转换代码最后没人愿意维护协议成了一纸空文。新手阶段的合理做法是先定义七种左右的消息进组、能力声明、任务、确认、进度、结果、错误配上统一信封和一个简单的任务状态机完整跑通一条链路后再按需扩展。协议是给协作立规矩不是给自己造枷锁。能用一个字段解决的问题不要用三个能用一种消息类型表达的语义不要拆成两种。后面要扩展很容易但协议臃肿到没人能看懂时想瘦身就难了。3. 消息模型设计把对话变成可计算的数据3.1 统一信封杜绝“各说各话”的散装字段消息模型是协议的地基。先给出我实际在用的信封结构用 JSON 表示大概是这个样子的{ version: 1.0, message_id: b7e3c2a1-4f1d-4c99-9e0a-8f6d2b0e1234, timestamp: 2025-06-01T10:15:30Z, trace_id: task-20250601-001, sender: scheduler, recipient: researcher, message_type: task, ttl: 60, payload: { task_id: t-101, instruction: 调研行业近三个月动态并输出要点, priority: 3 } }为什么要求统一信封因为只要每条消息都带 sender、recipient、message_type、trace_id 这四个字段harness 就能实现通用的路由、鉴权和追踪完全不需要为每个 Agent 定制适配器。这个道理跟寄信一样大家统一用带地址的信封邮局才能只开一个分拣口要是每个人用各不相同的包裹分拣员只能天天写 if-else那必然是灾难。信封字段的核心原则是路由相关字段放顶层业务相关字段一律放 payload。这样 harness 不需要关心业务语义也能正确投递Agent 也不需要理解路由细节只解析 payload 就行。谁改业务都不影响对方耦合度最低。3.2 消息类型与任务状态机我建议从下面这套消息类型起步它覆盖了协作场景里绝大多数需求消息类型发送方 → 接收方含义join新 Agent → 组内声明进组capability新 Agent → 组内广播能力清单task调度方 → 执行方下发任务ack执行方 → 调度方确认接收任务progress执行方 → 调度方任务进度更新result执行方 → 调度方返回最终结果error任意 → 相关方上报错误cancel调度方 → 执行方取消任务配合这套消息类型每个任务的状态机可以这样定义pending已创建→ assigned已分配→ running执行中/已 ack→ succeeded / failed / canceled。这条状态链是协议的地基中的地基因为整个 Agent 组的运行情况和健康度最终都反映在每个任务的状态分布上。我后面实现 StateStore 时就是围绕这个状态机来设计的。3.3 进组与能力注册新 Agent 的第一条消息一个 Agent 要加入组第一步不是干活而是“报到”。报到消息里要带能力清单比如[web_search, code_runner, writing]。harness 收到后会写入能力注册表后续任务路由就根据能力清单来匹配执行方。这个步骤很容易被忽略但等到任务被分给一个根本没有对应能力的 Agent 时你就会发现当初省掉的这一步有多关键。能力声明虽然是 Agent 自己说了算但 harness 必须做校验和黑名单控制。我见过一个 Agent 到处声明自己会“写代码”结果只会在回答里贴代码片段根本不会真跑。所以在能力注册时harness 应该同时登记能力对应的工具权限并在路由时做二次校验。做不到完全校验也没关系至少要在日志里标记“该能力声明来自 Agent 自称”给下游一个提醒。进组这件事做扎实了后续路由和权限管理都会省心非常多。4. 协作模式怎么选导演制、对等制还是黑板制4.1 三种模式的适用场景对比组内协议不是只存在于单一协作形态里不同团队常用的其实就是三种模式各有各的适配场景。先看一张总表再展开讲我的选型判断。对比维度导演制orchestrator-worker对等制peer-to-peer黑板制blackboard调度方式中心化的调度 Agent 统一分派无固定中心Agent 互相请求共享黑板按事件驱动取用通信路径全部汇聚到调度方两两直达都走黑板读写适合规模组内 Agent 少5 个以内任意但对控制能力要求极高Agent 数量多、并行度高排障复杂度低链路清晰高链路发散中依赖黑板状态的正确性新手友好度最友好不建议新手直接上需要额外的并发控制设计从表格也能看出来导演制的链路最清晰对等制的链路最发散黑板制则把重心转移到了共享状态上。没有绝对的优劣只有适不适合当前的任务形态。4.2 我的选型判断标准我的判断逻辑很简单。组内 Agent 少于等于 5 个、任务有明显依赖顺序优先导演制因为排查链路短、心智负担小调度方自己能看清全局。Agent 数量大、任务天然可以并行拆分的优先黑板制加事件订阅让各 Agent 专注处理自己关心的那部分数据。对等制看起来最灵活但工程上最难收敛一旦链路循环起来排查成本会让你怀疑人生新手不要轻易上。其实多数真实业务是混合形态主链路用导演制中间产物放黑板上共享。比如调研 Agent 把资料写到共享区写作 Agent 订阅“资料已更新”事件后开始动笔主流程由调度方控制。这种混合设计实战效果很好协议层只要把消息类型和路由规则定义清楚就能同时支撑两种协作形态不需要为每一种模式重写一套协议。5. 实操落地一套最小可用的组内协议5.1 第一步用 Python 定义协议层为了让协议不流于口头我用 Python 写了一套最小实现拆成三个模块。首先是协议层定义信封结构和消息类型枚举# protocol.py from dataclasses import dataclass, field from enum import Enum from datetime import datetime, timezone import uuid class MsgType(str, Enum): JOIN join CAPABILITY capability TASK task ACK ack PROGRESS progress RESULT result ERROR error CANCEL cancel class TaskState(str, Enum): PENDING pending ASSIGNED assigned RUNNING running SUCCEEDED succeeded FAILED failed CANCELED canceled dataclass class Envelope: version: str 1.0 message_id: str field(default_factorylambda: str(uuid.uuid4())) timestamp: str field(default_factorylambda: datetime.now(timezone.utc).isoformat()) trace_id: str sender: str recipient: str message_type: str ttl: int 60 payload: dict field(default_factorydict) def is_expired(self) - bool: # 简化处理生产环境用 timestamp ttl 判断 return False这个模块很薄但它定义了整套协议的“宪法”。每个 Agent 和 harness 都依赖这个模块所以字段越稳定越好。我把 ttl 放在信封顶层而不是 payload 里因为它是路由层关心的属性不属于业务。想加字段时也优先加 payload别破坏信封结构否则所有下游都要同步改。5.2 第二步实现组注册与路由接下来是 harness 层的核心注册表加投递逻辑。这里的关键是只做传输和校验不做业务决策。# harness.py class GroupRegistry: def __init__(self): self.members {} # name - channel self.capabilities {} # name - [capability] def join(self, agent_name, channel, caps): self.members[agent_name] channel self.capabilities[agent_name] caps self.broadcast(Envelope( senderharness, recipient*, message_typeMsgType.JOIN, payload{name: agent_name, capabilities: caps}, )) def send(self, envelope: Envelope) - bool: if envelope.is_expired(): return False channel self.members.get(envelope.recipient) if channel is None: return False channel.deliver(envelope) return True def route_by_capability(self, capability: str, envelope: Envelope) - bool: for name, caps in self.capabilities.items(): if capability in caps: envelope.recipient name return self.send(envelope) return False def broadcast(self, envelope: Envelope): for name, channel in self.members.items(): envelope.recipient name channel.deliver(envelope)route_by_capability 是路由层的核心方法它把“哪个 Agent 具备某能力”这件事收敛在 harness 内部。这样调度方不用知道组内具体有谁只需要声明“我需要一个能写代码的 Agent”剩下的匹配交给 harness。这个设计让 Agent 组可以动态增减成员调度方代码完全不用改。5.3 第三步状态存储与去重然后是需要重点关注的状态存储层。它的职责是跟踪每个任务的状态同时防止重复执行# state_store.py class StateStore: def __init__(self): self.tasks {} # task_id - TaskState self.results {} # task_id - payload def create_task(self, task_id: str) - bool: if task_id in self.tasks: return False # 已存在说明是重复请求 self.tasks[task_id] TaskState.PENDING return True def transition(self, task_id: str, new_state: TaskState) - bool: if task_id not in self.tasks: return False allowed { TaskState.PENDING: {TaskState.ASSIGNED}, TaskState.ASSIGNED: {TaskState.RUNNING, TaskState.FAILED}, TaskState.RUNNING: {TaskState.SUCCEEDED, TaskState.FAILED, TaskState.CANCELED}, } if new_state in allowed.get(self.tasks[task_id], set()): self.tasks[task_id] new_state return True return False # 非法状态流转这里 Theres one key trick: StateStore 必须先查重再创建任务。如果调度方因为超时重发了一次 task同一个 task_id 会在 create_task 时被拦下不会重复执行。状态机用白名单方式做合法性校验任何非法流转都会被拒绝这能最早暴露协议使用错误而不是等到结果错乱再去翻日志。我强烈建议状态流转校验放在这一步别偷懒。5.4 第四步超时、重试与链路追踪最后是超时和重试逻辑。推荐一组我实测下来稳定的初始参数task 下发后 30 秒内必须收到 ackprogress 上报间隔上限 60 秒失败最多重试 3 次重试采用指数退避 1 秒、3 秒、9 秒。核心代码是# scheduler.py import time def dispatch_with_retry(harness, envelope, max_retry3): for attempt in range(max_retry): if harness.send(envelope): return True time.sleep(3 ** attempt) # 指数退避1s, 3s, 9s harness.report_unresolved(envelope) # 进入补偿流程如人工接管 return False链路追踪的做法则很轻量调度方创建任务时生成 trace_id写入信封子 Agent 需要再拆分任务时把父任务的 trace_id 原样带到子任务里。这样最终日志里所有相关消息共享同一个 trace_id排查时一条 grep 就能拉出全链路。我还习惯在每次消息收发时打一行结构化日志包含 trace_id、sender、message_type 三个字段这是后期排障最重要的基础设施。6. 常见问题与排查实录6.1 消息风暴把上下文窗口炸了我遇到得最多的问题是消息风暴典型表现是某个 Agent 每分钟发一条 progress组里消息堆积如山最后所有人的上下文都被无关的中间状态占满真正重要的信息反而被挤掉了。这个问题的根源不是 Agent 太啰嗦而是协议里没有规定进度上报的密度上限。我的解决方式是三管齐下协议里明确规定 progress 的汇报间隔不能小于 30 秒一次harness 层加一个简单的限流器超频的 progress 消息直接吞掉只留最近一条调度方消费 progress 时只更新状态不和完整内容写进上下文需要详情的时候再按需拉取。另外一个实用技巧是定期把 progress 消息聚合成一条摘要比如“已完成 3/5 步当前卡在数据清洗”而不是让 50 条原始状态刷屏。6.2 两个 Agent 互踢皮球循环依赖第二个高发问题是循环依赖。我有一次调试一个“分析 Agent 和规划 Agent”的双 Agent 协作发现它们在协议上确实都遵守了消息格式但彼此反复发送“请你先做 X”的请求形成死循环直到把消息队列堵死。日志里 trace_id 相同时间戳一路递增一条链路上全是相同的两条消息重复出现。排查思路分两步先看是不是任务拆分有问题比如 A 需要 B 的结果B 又需要 A 的结果这是设计层面的循环依赖必须在拆任务时就拆成有向无环的结构再看是不是消息层面失控比如 Agent 收到无法处理的任务时协议里没有定义“拒绝”消息导致它只能回一句“你先做”。我的解决办法是给协议增加一个 hop 计数器每经过一条消息加一超过 10 就强制终止并把全链路标记为 failed。这样即使设计有漏洞也不会把系统拖垮。6.3 重复执行与结果互相覆盖第三个问题最隐蔽结果覆盖。两个 worker 并行处理同一个任务的不同部分因为共享存储没有分区后写入的覆盖了先写入的。这类问题靠看日志很难发现因为每一步看起来都正常直到最终结果验证时才发现数据对不上。根治办法是双保险。第一StateStore 层面保证同一个 task_id 只能被执行一次重复下发的 task 直接丢弃这是 5.3 小节查重逻辑的价值。第二写入共享存储时加版本号写入方必须携带它读取时拿到的版本号harness 发现版本不一致就拒绝写入并返回冲突错误让上游 Agent 重新拉取最新数据再合并。实践中我发现版本号机制能够把“静默错误”变成“显式冲突”后者至少可以被捕获和重试而不是等到最终结果验证才暴露。6.4 排查速查表把上面这些经验整理成一张速查表建议贴在工位旁边故障现象可能原因优先检查项任务没人执行能力注册缺失或路由匹配失败harness 的能力注册表、route_by_capability 命中情况任务重复执行调度方超时重发缺少查重StateStore.create_task 的幂等逻辑上下文被刷爆progress 消息过密限流参数、progress 聚合策略Agent 间互相卡住任务依赖成环trace_id 链路、hop 计数器、任务拆分设计结果数据被覆盖并行写共享存储缺少版本控制版本号字段、写入冲突日志任务迟迟不结束超时参数不匹配ack 超时、progress 间隔、ttl 配置这些坑基本覆盖了我实践早期遇到的大部分问题每一条背后都有真实的凌晨排查经历。协议这东西设计时总觉得简单真正放到生产环境跑起来才知道每一个字段、每一条超时规则都是必要的。最后再分享一个小经验组内协议不是写一次就能一劳永逸的我建议每跑通一个真实任务就回头把协议文档更新一次。协议文档不是写给上级看的汇报而是写给下个月熬夜排障的自己看的。文档里最该写的不是“我们设计了什么”而是“当初为什么这么设计、踩过什么坑”这样后来人改代码时才不会把你避开过的坑重新踩一遍。我自己就是在文档里补了一句“不要在 progress 消息里携带完整上下文”之后团队再也没有在上下文超限上翻过车。