去年下半年我一直被两个Agent项目来回折腾一个做客服工单分拣一个做内部知识库问答。单独跑的时候都挺聪明模型推理稳定回答质量也过得去。可一旦接到让Agent A去调用Agent B完成一单活儿这种协作流程场面就开始失控——B要么答非所问要么迟迟不给回应消息像掉进了黑洞A这边等不到结果就疯狂重试重试又导致B重复执行数据越搞越乱。排查到最后我才意识到问题根本不出在模型智商上而出在多Agent之间的触达上。Agent-Reach 就是在这个背景下做出来的一个轻量级触达层框架。它的定位很明确不解决大模型怎么推理只解决多Agent系统里谁能干活、怎么找对干活的人、活干完怎么确认、干砸了怎么收拾这四件事。简单说它负责让Agent之间找得到、接得住、回得来、兜得住。这篇文章我会把背景、核心设计、最小实现和踩坑实录都摊开讲适合正要搭多Agent协作系统、或者已经在调试Agent互相调用的朋友参考。1. 为什么多Agent系统里触达比智能更先翻车1.1 单Agent没问题多Agent全乱套的诡异现场我先还原一个特别典型的故障现场。客服工单Agent收到用户投诉后需要向知识库Agent查询退款政策再决定怎么回复。单测的时候两个Agent各自都正常。接上线以后出现了三次让我头疼的情况第一次是知识库Agent压根没收到请求。查了半天发现工单Agent发出的消息只是写进了自己的日志并没有真正发出去——因为我图省事直接在代码里用函数调用代替了消息传递知识库Agent升级重启后之前的调用自然就失效了。第二次是知识库Agent收到了请求也做了查询但结果回不来。原因更蠢我把回传地址写死成了一个临时队列名队列在测试环境被回收了。这就像寄快递时写了个已经注销的地址包裹到了门口又被退回。第三次是知识库Agent既收了请求也回了结果但工单Agent因为等待超时把同一单任务又发了一遍结果知识库Agent把同一份退款申请创建了两次。这三件事有一个共同点模型推理完全没有参与全是传输层和状态管理的问题。这让我下决心多Agent协作不能靠函数调用碰运气必须有一个专门管触达的中间层。1.2 把Agent协作拆成三个朴素问题我后来把所有症状归纳成三类第一类是找得到。A要干活怎么知道B能接这个活如果系统里有20个Agent有的管数据清洗有的管文本生成有的管流程审批A总不能把所有Agent挨个问一遍。第二类是说得清。A给B传的任务长什么样是一段自然语言还是一个结构化参数B如果理解错了A的需求就算消息送到了也等于白送。第三类是回得来。B干完活之后怎么告诉A不仅要说干完了还得说干到了什么程度甚至要说干砸了原因是资源不足还是输入不合法。你可以把它想象成一个公司里找人办事的流程你得先知道谁负责什么通讯录然后写清楚需求工单表单对方收到后要给你一个收到预计什么时候处理完的回应回执办完还得把结果回到你手里交付。很多Agent系统的翻车都是因为跳过了中间任意一环直接让两个AI脑补对方的需求。1.3 为什么我选了触达层而不是中心化总控最早考虑方案的时候有人建议我做一个大一统的总控Agent所有任务都先汇总到它那里由它集中决策、集中调度。这个方案的优点很直观所有流程一目了然改逻辑也方便。但仔细想就发现不对劲。总控Agent实际上变成了一个上帝视角的单点。它需要理解所有下游Agent的能力边界一旦某个Agent升级了功能总控的知识就要跟着改否则它就不知道这个活该发给谁。更麻烦的是全部流量都过总控它很快就成了性能瓶颈和故障单点总控一挂全系统瘫痪。Agent-Reach用的模式更像快递中转站或者工单系统中间层只负责投递、暂存、退回和回执登记不替业务Agent做任何决策。A说要找能算销售额的人触达层就去注册表里查查到有Agent声明了我支持销售数据聚合就把任务投过去等着对方回执。至于这个活怎么干、结果准不准是B自己的事触达层不干预。这个取舍背后有个很现实的原因业务Agent的数量和能力会不断变化但快递公司不需要知道包裹里的商品该怎么加工只需要知道寄给谁、怎么确保签收。中心化总控把业务和路由耦合在一起后期维护成本会指数级上升而触达层把路由和业务拆开两边都能独立演化。2. 核心概念注册表、信封、回执一个都不能少2.1 Agent Registry每个Agent先报个到Agent-Reach里有一个全局注册表每个业务Agent在启动时都要向注册表声明自己的能力和当前状态。这个声明不是写一个函数签名那么简单而是一份能力描述我习惯用类似这样的结构{ agent_id: knowledge-base-01, name: 知识库检索Agent, capabilities: [ { skill: policy_query, description: 根据问题查询退款、物流、售后政策, input_schema: { type: object, properties: { question: {type: string}, category: {type: string, enum: [refund, shipping, after_sales]} }, required: [question] }, output_schema: { type: object, properties: { answer: {type: string}, source_ids: {type: array, items: {type: string}} } } } ], status: ready, endpoint: queue://kb_worker_01 }前几天我用的时候还踩过一个坑最开始我把能力定义成纯自然语言描述比如我可以查询政策结果路由时候靠关键词匹配经常把给我查物流匹配到退款查询上。后来我改成自然语言描述 结构化输入输出Schema双轨制路由匹配看Schema细粒度确认看描述准确率明显提升。注册表还有个隐藏作用健康检查。每个Agent每隔几秒要发一次心跳如果触达层发现某个Agent已经没有心跳了就不会再把新任务投给它。这就避免了消息发出去才发现对方早死了的尴尬。2.2 Message Envelope任务不能裸奔必须装进信封Agent之间传递的任务我不会直接丢一个裸的自然语言字符串而是统一封装成Envelope信封。信封里除了业务数据还有一串路由和追踪用的元信息。下面是我在Agent-Reach里实际用的字段字段作用备注request_id全局唯一请求号生成规则dateagent_iduuid后缀全链路追踪的依据trace_id链路追踪ID整个业务链路共享方便把多个请求串成一条线from发送方Agent ID用于回执路由to接收方Agent ID不填则走能力路由按skill匹配task任务内容遵循接收方的input_schemacontext_pointer上下文指针指向知识片段ID或会话摘要不直接携带全量内容callback回执地址接收方把结果投递到这个地址ttl存活时间超过TTL未完成则触发超时处理为什么用context_pointer而不是直接把上下文塞进信封这是我在实际项目里被上下文爆炸毒打之后才想明白的。有一次Agent A要把一段8000字的文档转给Agent B做摘要再加上任务描述和元信息消息体直接堆到近1万token传输慢不说B的模型上下文窗口也被占了一大半。后来改成传文档ID页码范围B需要的时候再按需拉取整个消息体立刻小了两个数量级。2.3 Receipt与任务状态机干活必须留痕投递只是开始我最看重的设计是回执Receipt机制。每个任务在Agent-Reach里都要走一遍明确的状态机pending已投递待确认、running接收方已确认开始处理、succeeded处理成功、failed处理失败、timeout超过TTL仍无结果、compensated触发补偿流程。这里有个经验接收方收到任务后要先用一个轻量确认告诉发送方我收到了正在处理而不是等整个任务干完才回话。这就是回执先行模式。如果你等全部干完才回执发送方在等待期间根本不知道任务是还在跑、还是已经死了而先回一个running状态的确认发送方就能安心等到succeeded或者timeout。回执消息本身也带着状态码和错误信息比如input_invalidresource_busycontext_too_long这些错误码能让发送方做出不同的补偿动作而不是一律傻傻重试。我见过很多系统把错误码全省略了只回一个false结果排查问题全靠猜。3. 一个最小可用的 Agent-Reach 实现3.1 先画清楚三个模块的职责动手写代码前先别急着上框架。一个最小可用的Agent-Reach只需要三个模块第一个是Registry负责存Agent能力和健康状态。第二个是Dispatcher负责接收信封、按to或按skill路由、把任务投给目标Agent。第三个是ResultStore负责记录每个request_id的状态、回执内容、重试次数相当于一个轻量级状态库。这三个模块在Demo阶段完全可以都放在一个Python进程里用内存字典和asyncio队列实现。等以后规模上来了再拆成独立服务。不要一上来就搞微服务我吃过这个亏第一版架构图画得很宏大结果两个月都跑不起来后来砍掉一半才顺利上线。3.2 核心代码一个可以在本机直接跑的版本下面是我整理的最小实现保留了Agent-Reach最核心的逻辑注册、发送、回执、超时重试、幂等去重。你可以直接把它存成一个脚本跑起来感受一下。import asyncio import uuid from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Dict, Optional, Callable, Awaitable dataclass class Envelope: request_id: str from_agent: str to_agent: str task: dict ttl_seconds: int 30 callback: Optional[str] None dataclass class Receipt: request_id: str status: str # running / succeeded / failed result: dict field(default_factorydict) error_code: str class AgentRegistry: 注册表维护Agent能力说明和在线状态 def __init__(self): self._agents: Dict[str, dict] {} def register(self, agent_id: str, capabilities: list, heartbeat: float 5.0): self._agents[agent_id] { capabilities: capabilities, heartbeat: heartbeat, last_beat: datetime.now(), online: True } def beat(self, agent_id: str): if agent_id in self._agents: self._agents[agent_id][last_beat] datetime.now() self._agents[agent_id][online] True def find_by_skill(self, skill: str) - Optional[str]: for agent_id, info in self._agents.items(): for cap in info[capabilities]: if cap[skill] skill and info[online]: return agent_id return None class ResultStore: 结果存储记录每个request_id的状态用于回执和幂等 def __init__(self): self._records: Dict[str, dict] {} def create(self, request_id: str, envelope: Envelope, ttl: int): self._records[request_id] { envelope: envelope, status: pending, expire_at: datetime.now() timedelta(secondsttl), retry_count: 0, receipt: None } def has(self, request_id: str) - bool: return request_id in self._records def update(self, request_id: str, status: str, receipt: Optional[Receipt] None): if request_id not in self._records: return self._records[request_id][status] status if receipt: self._records[request_id][receipt] receipt def get(self, request_id: str) - dict: return self._records.get(request_id) class AgentReach: 触达层投递、回执、重试、幂等 def __init__(self, registry: AgentRegistry, store: ResultStore, max_retries: int 3): self.registry registry self.store store self.max_retries max_retries self._workers: Dict[str, Callable[[Envelope], Awaitable[Receipt]]] {} def register_worker(self, agent_id: str, handler: Callable[[Envelope], Awaitable[Receipt]]): 业务Agent在这里注册自己的处理函数 self._workers[agent_id] handler async def send(self, from_agent: str, to_agent: str, task: dict, ttl: int 30) - str: 发送任务返回request_id request_id f{datetime.now().strftime(%Y%m%d%H%M%S)}-{uuid.uuid4().hex[:8]} envelope Envelope(request_idrequest_id, from_agentfrom_agent, to_agentto_agent, tasktask, ttl_secondsttl) self.store.create(request_id, envelope, ttl) asyncio.create_task(self._dispatch_with_retry(envelope)) return request_id async def _dispatch_with_retry(self, envelope: Envelope): 投递重试逻辑 record self.store.get(envelope.request_id) retry_base 1.0 for attempt in range(self.max_retries): # 幂等检查如果这个request_id已经执行成功过直接跳过 if self.store.get(envelope.request_id)[status] in (succeeded, failed): return await self._dispatch_once(envelope) record self.store.get(envelope.request_id) # 已经拿到最终回执就不再重试 if record[status] in (succeeded, failed): return # 超时后按指数退避重试 await asyncio.sleep(retry_base * (2 ** attempt)) record[retry_count] 1 # 重试耗尽进入超时状态 self.store.update(envelope.request_id, timeout) async def _dispatch_once(self, envelope: Envelope): handler self._workers.get(envelope.to_agent) if not handler: self.store.update(envelope.request_id, failed, Receipt(envelope.request_id, failed, error_codeagent_not_found)) return try: # 先给发送方一个已收到的回执避免发送方误判 self.store.update(envelope.request_id, running) receipt await handler(envelope) self.store.update(envelope.request_id, receipt.status, receipt) except Exception as e: self.store.update(envelope.request_id, failed, Receipt(envelope.request_id, failed, error_codestr(e))) async def main(): registry AgentRegistry() registry.register(kb_agent, [{skill: policy_query, input_schema: {}}]) store ResultStore() reach AgentReach(registry, store) async def kb_handler(envelope: Envelope): # 模拟业务Agent处理 await asyncio.sleep(0.1) return Receipt(envelope.request_id, succeeded, result{answer: 退款政策..}) reach.register_worker(kb_agent, kb_handler) request_id await reach.send(ticket_agent, kb_agent, {question: 退款多久到账}) await asyncio.sleep(0.3) print(store.get(request_id)) if __name__ __main__: asyncio.run(main())这段代码的核心就三句话发出去的每个任务都要有一个request_id接收方先回一个running回执拿着request_id做幂等和重试。你会发现我故意没有加任何智能排队或自适应路由的东西因为一个触达层能否抗住生产环境先看这三点有没有做扎实。3.3 参数怎么定TTL、重试次数、退避策略很多朋友拿到类似代码后最爱问一句话这些参数到底填多少我给的默认值是根据实际压测调出来的说下思路。TTL不要拍脑袋定。查询类任务比如查一下这个用户有没有开票资质我给10秒内容生成类任务比如生成一版30页的产品白皮书摘要我给10分钟批量数据处理类任务可以放宽到30分钟。原则很简单TTL要比P95处理时长的两倍略大留出调度和网络波动余量。重试次数和退避策略也要配套。指数退避的公式我用的是sleep_time base * (2 ** attempt) random_jitterbase设1秒jitter区间取0到0.3秒。加jitter是防止多个任务同时超时后在同一秒内全部重试把下游Agent直接打爆。这个现象我在没有加jitter的时候遇到过那次下游Agent的CPU直接拉满活没干成反而把整个集群拖垮了。幂等键就是request_id发送方的重试必须带同一个request_id接收方在执行前先检查这个id是不是已经跑过了。代码里的幂等检查关键就在execute前看RecordStore里是否已经有succeeded或failed状态。这一步不能省。3.4 用适配器把现有Agent接进触达层我自己的项目里有很多Agent不是用Agent-Reach的Worker函数写的而是已经跑在别的服务里的HTTP接口甚至有的只是一个人工操作面板。为了让它们能被统一触达我写了一个非常薄的适配器。适配器的思路是把任意Agent封装成标准形态输入是Envelope输出是Receipt。如果你有一个HTTP Agent适配器就在函数里把envelope.taskPOST到它的接口然后把响应包装成Receipt。如果你有一个基于Function Calling的LLM Agent适配器就把envelope.task转成tool的参数让模型决定怎么处理再把结果转回Receipt。真正花时间的不是包装函数而是状态回传。我建议适配器一定要把下游Agent的原始状态码、原始错误信息完整透传回来不要只转成success和fail。否则排查时你永远要两头对照日志效率极低。4. 常见问题排查实录4.1 消息神秘消失三处最常见的丢失点我遇到的第一个大问题是消息莫名其妙没送到。排查到最后基本都出在三个位置。第一是回调地址失效。接收方处理好任务以后要按callback地址把回执发回去如果这个地址是临时队列名或者没有持久化回执就发不出去发送方只能干等超时。解决办法是把callback设计成可查询的持久化主题名而不是内存中的临时对象。第二是任务内容太大导致序列化和传输失败。有些Agent会把完整的爬虫结果全文塞进task一个信封几十MB中间件默认消息上限一卡整条消息就被静默丢弃。解决办法是消息里只放内容指针大数据走共享存储。第三是Agent所在服务在投递瞬间刚好崩溃。接收方还没落到处理逻辑请求就断了发送方如果没开启重试这条任务就永远消失在黑名单里。所以至少要保留一次重试并且用requestId去重。排查这类问题时我靠的是全链路日志。Agent-Reach里从信封创建、投递、回执回传的每一步都会打出一条带上requestId和traceId的结构化日志。你只要拿出丢单的requestId一查立刻能看到最后停在了哪一步。4.2 重复执行回执丢了的连锁反应重复执行是比消息丢失更隐蔽的坑。表面现象是用户收到两条相同短信、账户被扣了两次钱、数据库里出现两条相同订单。追根溯源几乎都是同一个场景接收方已经成功执行了任务但succeeded回执在回传路上丢了发送方按超时重试把同一个requestId又发了过来。如果接收方没有做幂等检查它就会再执行一遍。解决办法分两层。接收方这层必须在执行业务逻辑前查一下这个request_id之前是不是已经跑成功了如果已经成功直接返回上次的结果哪怕返回的结果在本地只保留最近一小时也行成本很低。发送方这层重试时必须复用原request_id不能每次生成新的。很多重复执行的案子就是因为重试时重新生成了requestId接收方根本认不出来。另外还有一个容易被忽略的场景接收方执行成功了回执也顺利发出但接收方自己进程随后重启内存里的幂等表清空了。等发送方因为网络抖动再次重试接收方一看本地没有这个id的记录又执行了一遍。所以幂等表一定要落库哪怕用SQLite都行内存表只在Demo里能用。4.3 上下文爆炸全量传递不是好习惯这个坑我们在设计Message Envelope时已经提到过但实际中几乎每个接进来的Agent都会踩一遍。最开始我把上下文理解成让B更懂业务背景所以习惯性把A的历史对话、检索到的文档片段全部塞进去。结果就是任务消息越传越大模型侧prompt越来越长响应越来越慢token成本也一路飙升。更麻烦的是超长上下文中真正有用的信息往往只有一小块其余全是噪声直接拉低生成质量。Agent-Reach的处理思路是显式分离指令和资料。指令放在task里必须短小精悍资料放在共享的上下文仓库里用context_pointer引用。B拿到指针后按需拉取。如果B发现指针引用的内容已经过期还可以返回一个context_not_found错误码让A进行补充。这样消息体轻了上下文也可追溯了排查时还能知道B到底读了哪些资料一举三得。4.4 循环委派A找BB又找回A循环委派是真实存在的而且特别难排查。场景是Agent A发现自己能力不足以完成任务于是把活委派给BB尝试处理发现这个活其实需要最初A那边的某个权限或领域知识于是又把活委派回AA接到后判断自己依然处理不了再次发给B……直到消息无限循环把两个Agent的算力全部耗尽。我见过不止一次这种死循环而且它经常不是代码逻辑Bug而是Agent自己觉得应该委派造成的。Agent-Reach里我用两个手段控制第一是给Envelope加一个hop_count字段每经过一次委派就加一超过上限建议3到5跳就强制拒绝并把错误码定为circular_dispatch_detected。第二是在Envelope里记录visited链接收方在决定是否接单前先看自己是否已经在这个链上如果已经在链上就直接拒绝接单避免在同一批Agent之间反复绕圈。这两个机制加起来循环委派基本能被拦在门外。4.5 排查速查表症状可能原因优先检查解决建议消息没送达callback地址失效、消息体超上限、接收方崩溃全链路日志中requestId最后出现位置用持久化主题、内容指针、开启重试重复执行回执丢失导致重试、幂等表未落库接收方是否以request_id查重幂等表持久化重试复用原request_id回执迟迟不到业务处理过慢、模型调用超时看是否先回了running回执回执先行TTL按P95时长两倍设置Agent互相踢皮球委派逻辑不严谨、能力声明过泛看visited链和hop_count限制跳数细化能力声明上下文爆炸全量传文档、无指针机制检查Envelope大小用context_pointer引用资料5. 进阶玩法与我的几点体会5.1 在注册表里做语义路由基础版是写死to_agent进阶版是按能力声明做路由。你可以在注册表里把每个Agent的能力描述转成向量发送方只要声明我想找一个能做销售数据聚合的Agent触达层拿这个描述去匹配最接近的Agent。这个玩法让新Agent接入变得非常方便新Agent启动时自己上报能力描述不用改任何路由表后面的任务自然就能找到它。逻辑上有点像搜索引擎只是索引的对象从网页变成了能干活的人。但要提醒一句语义路由只适合宽松匹配的场景。如果你要的是强流程控制比如财务审批必须先于采购执行那就别指望语义路由老老实实写业务流程编排。语义路由负责的是动态找合适的人流程编排负责的是固定环节不能乱序两者可以共存但责任边界要分清楚。5.2 给Agent-Reach加一个人在回路环节有些任务不适合Agent完全自动完成比如给客户发调价通知、删除用户数据、批量修改生产环境配置。Agent-Reach里我在状态机里加了一种叫awaiting_approval的状态Agent执行到某个环节时返回这个状态附带待审批的内容摘要和审批链接等人工确认后再触发后续流程。这个功能我强烈建议所有要做自动化Agent系统的人尽早加上不要等技术债务堆到非加不可再做。原因很简单Agent系统一旦接上真实业务一定会出现模型误判或边界案例完全没有兜底的话一次错误操作就可能比人工成本贵得多。5.3 踩过这么多次坑我的几条实操体会第一个体会是回执先行永远比干完再汇报稳。哪怕你的接收方只需要1秒就能干完也要在收到任务的那一刻先回一个running把已收到这件事固定下来。这不是多余动作而是给发送方一个最朴素的确定性信号。第二个体会是日志一定要带requestId并全链路透传。没有requestId的日志等于没写排查问题时要拿着ID去每一环找记录如果某段日志缺ID那段过程就是一个黑盒。我后来给自己定的规矩是触达层内部任何一个分支判断都要输出带ID的日志哪怕只是decisionskip_retry这种一行字。第三个体会是不要过度设计。最早的Agent-Reach设计文档里有复杂的优先级队列、智能负载均衡、跨地域多活结果项目一拖三个月没进展。后来我把所有花哨功能砍掉只留注册、路由、回执、重试四件事一周就跑通了核心流程。很多系统的关键不是功能多而是基础通信可靠。先把找得到、回得来做到位再考虑更聪明的事。Agent-Reach目前在我这边已经稳定跑了数个线上流程每天处理的请求不算多但每一笔都有据可查、出错能快速定位。如果你也在折腾多Agent协作我建议别急着上线花哨的框架先把自己系统里的注册、信封、回执这三件事做扎实很多灵异问题会自己消失。