最近在折腾多智能体系统时不得不面对一个问题单独一个智能体跑demo很顺一旦放进真实业务场景要么是不知道该找谁要数据要么是拿到了数据却不会用更别提在多个智能体之间来回传话。如果想做一套能支撑电商售后、库存查询、订单流转这类真实场景的智能体协作系统光靠把Prompt写长一点、把工具函数堆多一些根本撑不住。这就是我开始认真研究“Agent-Reach”的动机——它解决的核心问题不是“让一个智能体更聪明”而是让智能体之间的触达、路由和消息协作变得像调用内部服务一样可靠。如果你也在做AI Agent相关的应用尤其是多个智能体需要协同完成任务的场景这篇文章值得看完。我会先讲Agent-Reach要解决的问题和核心设计思路再走一遍从零接入的完整过程最后把我在真实运行中踩过坑、调过优的实战经验分享出来。1. 为什么“单兵智能体”撑不起生产系统Agent-Reach 的立项背景先说个我自己的经历。今年上半年团队接到一个需求在电商平台场景里做一个“售后处理助手”。直觉方案是让一个大模型智能体直接接管售后工单给它配上订单查询、退货退款、库存查询这些工具。听起来很简单真正跑起来就发现不对劲。单一智能体的问题在于它会迅速变成一个“什么都会一点、什么都做不精”的大杂烩。模型要同时理解售后政策、订单状态流转、库存余量、物流时效工具列表越来越长Prompt几经膨胀最终的结果是该触达库存系统的时候它在凭猜测回复该同步订单状态的时候它在自己编。不是说大模型能力不够而是把太多职责塞进一个上下文窗口里本身就违背了软件工程里“单一职责”的基本原则。1.1 多智能体协作的真正难点不是“智能”是“触达”我后来的思路转变很简单与其把所有能力塞进一个智能体不如拆成多个专职智能体——一个管售后规则一个管库存一个管订单。每个智能体只做自己那一摊事通过“触达”彼此来获得信息。但拆完之后真正的难点立刻暴露出来了售后智能体怎么知道该找哪个智能体要数据它发出请求之后对方以什么格式返回如果一个智能体暂时不可用是重试、降级还是直接告诉用户“稍后再试”多个智能体之间的请求会不会互相干扰这些问题全部都和模型智力无关而是“触达”层面的问题。用更直白的话说智能体之间要有一套标准化的“通讯录”和“话术”否则就是一群高手在鸡同鸭讲。1.2 “Agent-Reach”是什么一个智能体触达与协作的中间层Agent-Reach 就是我为了解决上述问题做的轻量级中间层。它的定位很明确不替任何智能体做决策只负责解决“触达”这件事——让一个智能体能快速找到对的另一个智能体或工具并且用统一的协议安全、可靠地完成信息交换。整个系统抽象成三个核心概念注册表Registry所有智能体在启动时向注册表声明自己的能力、输入输出格式和调用地址。路由器Router请求进来后根据目标智能体的语义描述做匹配决定这条消息应该交给谁。消息总线Message Bus负责在智能体之间传递请求和响应支持同步和异步两种模式。把这三件事落实了多智能体协作的代码才写得下去而不是靠Prompt碰运气。2. Agent-Reach 的核心设计注册表、语义路由与消息契约这一节是全文的重点。Agent-Reach 最关键的设计决策有三个每个都对应一类我曾经踩过的问题。2.1 注册表给每个智能体一张机器可读的“能力名片”人类之间协作靠的是互相知道“你负责什么”。智能体也一样。但智能体不能靠读文档来理解对方它需要一张机器可读的“能力名片”挂在注册表里让其它智能体在需要的时候主动发现。我这里的注册表实现并不复杂核心就是一个带元数据的数据结构dataclass class AgentSpec: agent_id: str # 告诉路由器“这个智能体是干嘛的”用于语义匹配 description: str # 输入参数声明用 JSON Schema方便校验 input_schema: dict output_schema: dict # 调用地址可以是 HTTP API、gRPC 或消息队列主题 endpoint: str # 同步接口还是异步事件 transport: str http # 粗粒度限流配置 max_concurrency: int 10 timeout_ms: int 5000以库存智能体为例它注册出来的名片大概是这样的agent_id: inventory-agent description: 查询商品实时库存、锁定库存、释放库存。适合回答库存余量、缺货状态相关的问题。 input_schema: type: object properties: sku_ids: type: array items: string action: type: string enum: [query, lock, release] output_schema: type: object properties: sku_stock: type: object additionalProperties: type: integer endpoint: http://inventory-service:8080/agent transport: http max_concurrency: 20 timeout_ms: 3000这张“名片”解决了两个问题一是让路由系统知道这个智能体擅长什么二是调用方不用关心对方的内部实现只要按这套schema发数据就行。2.2 语义路由不写死流程让请求自己找对门有了能力名片下一步就是怎么让一个请求找对门。选项有两个一是硬编码工作流比如“售后工单先查库存再查订单”但这会很脆业务一变就要改代码二是用语义匹配让请求自己去匹配最合适的智能体。Agent-Reach 采用的是语义匹配为主、规则覆盖为辅的方案。具体做法拆成三步把用户输入/智能体发出的请求向量化成一个query向量。在注册表里对每个智能体的 description 向量化计算 query 与各 agent 描述之间的余弦相似度。如果最高相似度超过设定阈值我常用的值是0.75就路由给对应智能体否则打回让调用方补充信息或换一种说法。这里有一个我知道很多初学者会踩的坑别把阈值设得太低。设到0.6什么请求都会命中同一个描述比较宽泛的智能体我最后定在0.75~0.8之间宁可漏掉也不乱配。漏发请求最多让人工介入一次错发请求会导致一连串错误操作。当语义匹配不靠谱的时候规则覆盖兜底。比如 “退款金额不能大于订单金额”这种硬约束就直接写成规则放在消息总线的消息前置校验阶段不依赖模型理解。2.3 消息契约同步请求与异步事件别混在一起智能体之间传消息最容易犯的错误是把所有交互都设计成同步请求。售后智能体需要调用库存智能体查库存同步就该没问题但假如有一个“生成每日报表智能体”它可能要跑几分钟同步调用就会把调用方活活卡死。Agent-Reach 在消息契约上做了明确区分同步请求Request/Reply适用于查询类、短耗时操作走 HTTP超时时间在注册表里声明。异步事件Event适用于通知类、长耗时操作发到消息队列由订阅方自行消费。回调Callback异步任务完成后结果通过回调接口主动推给调用方或者写入共享存储由调用方轮询。# 同步调用示例Agent-Reach client 的简化用法 from agent_reach import ReachClient client ReachClient(registry_urlhttp://agent-reach-registry:8080) # 这句会经历查询注册表 - 语义路由 - 协议转换 - 调用目标智能体 resp client.reach( querySKU-1001 目前库存还剩多少, target_typeinventory-agent, payload{sku_ids: [SKU-1001], action: query}, ) print(resp.data) # 异步事件示例 client.publish_event( topicorder.completed, payload{order_id: ORD-2024-0001}, )每个同步调用都带超时和重试参数。超时设置不能只看智能体平均耗时要看P95否则大量请求都会在边缘超时。3. 实战接入售后工单与库存系统两个 Agent 怎么协作概念讲了一堆没有落地代码就是空中楼阁。我用一个实际场景走一遍完整的接入过程用户发起一个售后工单内容为“我买的SKU-1001还没发货申请退款”。售后智能体必须先确认库存状态才能判断是直接退款还是先锁库存再走退货流程。3.1 启动注册表和消息总线Agent-Reach 本身不包含消息总线实现它对接的是外部组件。我本地测试时用的是 Redis Stream FastAPI生产环境则换成了合适的消息队列。先把最基础的注册表服务起来docker run -d --name agent-reach-registry -p 8080:8080 \ -e STORAGE_TYPEredis -e REDIS_URLredis://localhost:6379 agent-reach/registry:0.4.1 docker run -d --name agent-reach-bus -p 6380:6379 redis:7注册表服务本质上是把所有 AgentSpec 存在一个带索引的存储里并提供 HTTP 接口做注册和查询。消息队列则独立部署避免和业务数据库挤在一起。3.2 注册库存智能体库存智能体是一个 FastAPI 应用先把能力名片注册到 Agent-Reach 注册表# inventory_agent.py简化示例 from fastapi import FastAPI from agent_reach import register_agent app FastAPI() app.post(/agent/inventory/query) def query_inventory(payload: dict): sku_ids payload[sku_ids] # 真实场景这里查数据库或缓存 return {sku_stock: {sku_id: 42 for sku_id in sku_ids}} # 服务启动时注册 app.on_event(startup) def startup(): register_agent( agent_idinventory-agent, description查询商品实时库存、锁定库存、释放库存。适合回答库存余量、缺货状态的问题。, endpointhttp://localhost:8000/agent/inventory/query, input_schema{...}, # 用 JSON Schema output_schema{...}, transporthttp, )这里有几个细节要重点说一下description 写得好不好直接决定语义路由准不准。要写“这个智能体接受什么类型的问题擅长解决什么”不要写API层执行细节。endpoint 不要写成内网IP要写可路由的域名否则容器化部署时经常出现调用失败。注册表里保留心跳机制库存智能体每隔30秒上报一次健康状态路由时优先选健康的节点。3.3 售后智能体里发起触达售后智能体本质是一个 LLM 智能体它收到用户工单后决定自己先处理还是触达其他智能体。关键代码在“判断要不要触达”这一步# after_sale_agent.py简化示例 from agent_reach import ReachClient client ReachClient() async def handle_ticket(ticket_text: str): # 把工单文本交给售后策略模型模型会产出两种结果 # 1. 直接生成回复政策明确 # 2. 生成一个“需要触达其它智能体”的plan plan await agent_plan(ticket_text) if plan.action query_inventory: inv await client.reach( query查询库存是否充足, target_typeinventory-agent, payload{sku_ids: plan.sku_ids, action: query}, ) if inv.is_success: # 根据库存结果再走下一步退款还是锁库存 ... return final_reply“让大模型决定要不要触达”这步比烂大街的 function calling 多了一个好处Agent-Reach 提供了client.reach()这个统一的入口模型输出的 action 不再绑定到具体的API函数而是绑定到一个“能力描述”。以后新增一个智能体不用改售后服务代码只要在注册表里挂新能力模型通过语义描述就能自己找到。3.4 跑通场景后的效果跑通之后售后智能体再遇到“申请退款”的工单会走一遍“路由到库存智能体 → 查库存 → 结果回流到售后智能体 → 生成退款/锁库存决策”的链路。整个过程大概在1到2秒内完成快的时候不到500毫秒。不过话说回来demo跑通只是第一步。真正让人头皮发麻的是上线之后遇到的那些问题下面这章我展开讲讲。4. 上线后踩过的四个典型事故排查链路与补救方案Agent-Reach 这套体系运行了两个月先后触发过好几次线上事故。我把其中最有代表性的四个完整排查过程写出来希望能帮你省下几晚的排查时间。4.1 并发风暴一个库存智能体被50个售后工单同时打垮现象很直接库存智能体的接口在晚高峰时段大面积超时售后工单越积越多。我一开始怀疑是数据库慢查询看了一圈数据库负载很低。后来去看注册表里的调用日志才发现问题的根源库存智能体的max_concurrency写得太大注册成了20而且售后智能体在生成 plan 时经常对一个工单里的多个 SKU 分别发起查询而不是合并成一个批量请求。结果是50个工单一秒内发出了近200个并发请求把库存智能体的线程池直接打满。补救动作分两步售后智能体侧做聚合一个工单里的多个SKU合并成一个reach()请求传sku_ids数组一次拿回结果。Agent-Reach 注册表侧加“入口限流”每个智能体节点周边配一个并发信号量超过max_concurrency的请求直接进等待队列而不是打到目标服务上。# 库存智能体限流配置 agents: - id: inventory-agent max_concurrency: 10 queue_size: 50 reject_policy: timeout_in_queue经过这次之后我养成了一个习惯所有智能体默认并发度先设保守值再压测后往上调。宁可牺牲一点吞吐也不因为响应风暴拖垮核心服务。4.2 上下文爆满把全量查询结果塞给了模型的教训第二次事故更像是一个“成本事故”。售后智能体在调用完库存智能体后会把所有返回结果都拼进模型上下文。单个SKU的库存数据不多但一次查询可能返回几百个SKU的库存列表再加上物流状态、订单记录上下文轻轻松松超过几万token。表现是售后智能体的响应时间从原来的1秒变成10秒而且模型开始“忘事”最后给出的退款理由前后矛盾。排查过程先看调用链耗时发现路由和库存响应都正常多出来的时间全在模型推理。把传给模型的最终 prompt 打印出来发现里面塞了完整库存结果。回看代码确实没做一步处理就直接str()全量塞进去了。修复方案是在 Agent-Reach 层加了一个“结果裁剪器Result Trimmer”对上游返回的 output_schema 做白名单字段提取只留下游真正需要的字段。库存查询只返回汇总的“可售库存总数”和“缺货SKU数”明细不进入模型上下文。# 结果裁剪白名单保留 trim_rule { inventory-agent:query: {allow: [total_available, out_of_stock_skus]} }这个坑提醒我多智能体系统的上下文预算跟代码里一个未释放的循环引用一样致命。不仅要控制“发给智能体的内容”还要控制“智能体返回后怎么注入下游上下文”。4.3 路由误判相似语义的智能体之间“抢单”第三个事故比较隐蔽。系统里有一个“订单状态咨询智能体”和一个“售后政策智能体”前者的描述里有“退换货状态”后者的描述里有“退换货条件”。当用户投诉“我什么时候能收到退货退款”query 与两个智能体的描述相似度都超过了0.75路由器随机选了一个。运气不好时它选中了售后政策智能体结果给用户回复了一大段退换货政策完全没解答“钱什么时候到账”。排查链路看路由日志发现同一条query有时候进A有时候进B被路由到A或B的概率接近五五开。看两个智能体的 AgentSpec.description相似度确实高。临时方案把语义路由阈值从0.75调到0.82并引入“top-2拦截”如果前两名相似度差值小于0.05则请求转人工决策不让模型猜。长期方案在 AgentSpec 里新增“intent_examples”字段用几个典型问题辅助路由匹配而不是只靠一段描述文字开盲盒。agent_id: order-status-agent description: 查询订单物流状态、退款到账进度、延误原因。 intent_examples: - 我的退款到账了吗 - 订单现在显示什么状态 - 物流卡住了怎么回事加了intent_examples之后路由准确率上了一个台阶。经验是语义路由的准确率取决于你对每个智能体的“边界描述”做了多少沉淀。描述越长、例子越具体误判率越低。4.4 工具幻觉智能体“嘴上说做了”实际根本没权限最后这个事故说到底是Prompt工程和Agent-Reach的权限模型没有对齐。售后智能体在处理“用户申请改价”的请求时模型在回复里写了“已为您申请改价成功”但实际上它根本没有调用改价智能体的权限Agent-Reach 的路由层已经把这次reach()拒绝了。排查后发现一个尴尬的事实售后智能体的系统提示词里并没有说明“自己不能触达改价服务”模型以为所有事情都能干。修复方案有两层Agent-Reach 路由层在响应里返回permission_denied错误码调用方必须识别这个错误码并改变话术而不是默认成功。把“可用能力清单”动态注入售后智能体的系统提示词。清单由 Agent-Reach 的注册表实时生成LLM 每次处理请求前都明确知道“此刻我可以触达谁”。# 动态构造prompt capabilities await registry.get_available_agents() sys_prompt f 你是售后处理助手。你当前可以协作的智能体包括 {capabilities} 任何不在列表中的能力一律回复『目前我无法处理该操作请咨询人工客服』。 这起事故给我的启发很大权限边界不能只写在配置文件里必须落到每次会话的提示词里。模型看不到的能力就是不存在的能力。5. 生产化调优限流、降级、可观测性一个都不能少稳定运行两个月后我把 Agent-Reach 的配置清单整理成了一份可复用的模板。这里不是兜售银弹而是给一套能照着抄的底线。5.1 为每个智能体建立“三层降级链”我现在的标准配置是每个智能体节点背后都有三条路。主路径正常调用。降级路径目标智能体超时后调近义词智能体或查缓存结果。比如库存查询挂了就用“最近一次定时同步的库存快照”。兜底路径以上都不行返回标准化文案让人工介入。对应 Agent-Reach 的配置fallback: inventory-agent: cache_ttl: 60s fallback_agent_id: warehouse-report-agent manual_fallback_text: 库存信息暂时不可用请稍后重试注意兜底文案要由业务方确认不要自己拍脑袋写“系统繁忙”有些售后服务团队会有自己的话术要求。5.2 可观测性一定要有统一trace不然事故永远只能靠猜多智能体系统的排障难度比单体应用高一个量级。一条用户请求可能经过售后智能体、库存智能体、订单智能体三个节点任何一个环节出错都会表现为“最终答案不对”。Agent-Reach 从第一天就把统一trace纳入核心能力。每个reach()请求都带一个request_id链路日志按这个ID串联# 日志示例 request_idreq_87q3xa trace_routeafter_sale_agent - inventory-agent statusok耗时230ms request_idreq_87q3xa trace_routeafter_sale_agent - order-status-agent statuspermission_denied 耗时8ms实际查线上问题时我第一件事永远是按request_id拉全链路的日志先看“到底调了谁”再谈“为什么结果不对”。没有统一trace之前排查一个问题要开三个控制台效率太低了。5.3 上下线流程注册表要有“灰度”和“摘流”的能力智能体更新依赖时不能直接改完就重启。Agent-Reach 的注册表里支持一个weight字段用来做流量灰度新版本先注册成weight: 10旧版本保持weight: 90。观察半小时新版本无误提高到weight: 50。最终旧版本weight: 0摘流下线。这个过程不需要改上游调用方代码只动注册表配置。多智能体之间的依赖关系本来就复杂能用配置解决的绝不发版。6. 写在最后多智能体协作的真正边界感Agent-Reach 从立项到现在最大的收获不是某个技术方案而是让我重新理解了“触达”在多智能体系统里的分量。模型本身的推理能力当然重要但当你的系统里有五个智能体、二十个工具、上百种消息类型时真正决定系统稳不稳的是它们在“触达”层面有没有一套清晰的契约。如果你在做一个多智能体 PoC别一上来就堆 LangChain、AutoGPT 这类重型编排框架。先用一个轻量注册表 语义路由 消息契约把链路串起来跑通一个双 Agent 场景再说。等到遇到真实的并发、误判、权限问题再逐步增加限流、降级、可观测性组件。最后分享一个小技巧把注册表里的智能体描述当成“团队部门简介”来写——它不在于字多而在于把“什么需求该找谁”写得一清二楚。这套系统的健康度一半取决于代码另一半取决于你写描述时有没有把边界感拿捏准。