去年我在做 AI 客服系统的时候一开始只接了一个智能体让它负责查订单、退换货。后来业务方觉得效果不错又塞进来一个负责写营销文案的一个负责做数据分析的甚至还有个专门处理用户情绪安抚的。单看每个智能体都很能打可一旦它们互相之间需要传递信息问题就来了A 拿到结果不知道该发给谁B 需要的上下文被另一个智能体改得面目全非C 的接口只认自己的 JSON 格式D 直接超时崩掉。我当时的第一个念头是——我是不是在建一座巴别塔。后来我整理了一个轻量的协作层没想到一路用到现在顺手给它起了个名字叫 Agent-Reach。Agent-Reach 本质上是一个面向 AI 代理的注册、发现、路由和协作框架也可以叫多智能体编排中间层。它解决的问题很简单当一个系统里有多个智能体同时存在时如何让它们互相发现、按规则路由请求、在统一的上下文里协作并且不把代码写成一团乱麻。这篇文章我就把当时的完整思路、技术选型、实操过程和踩坑记录都摊开来讲。如果你也在折腾多智能体或者正准备把几个 AI Agent 接入同一个系统这篇应该能给你省下不少时间。1. Agent-Reach这个项目到底在干什么1.1 它不是工作流引擎是连接层开始动手之前我花了两个周末去看各种现成的编排方案。市面上很多框架强调的是“流程”先让 Agent A 去做第一步然后把结果喂给 Agent B 做第二步最后汇总到 Agent C。这种思路适合业务链路非常固定的场景比如订单审核、内容生成流水线。但一旦你的智能体数量变多、能力交叉你真正缺的其实是另一层东西连接层。连接层负责三件事。第一让每个智能体都能被其他智能体“找到”。如果没有注册中心Agent A 怎么知道 Agent B 还在不在运行它的接口地址是什么它的能力描述是什么全靠写死配置是不现实的。Agent-Reach 做的就是服务注册与发现的活只不过服务对象从普通微服务换成了 AI Agent。第二把请求按规则送到该去的地方。一个用户问题进来可能涉及多个智能体。这时候需要一个路由层判断这个问题该交给客服主 Agent还是先交给数据分析 Agent 做个预处理判断的依据可以是关键词、意图识别结果、当前负载、用户等级、甚至 Agent 返回的置信度。Agent-Reach 把这些都做成可配置的策略而不是把路由判断塞进每个智能体的 Prompt 里。第三屏蔽掉不同 Agent 之间的协议差异。有的 Agent 是基于 OpenAI Function Calling 做的有的自己实现了 ReAct 循环有的干脆就是一个简单 HTTP 服务。它们传给下游的消息格式五花八门。Agent-Reach 规定了一套最小消息信封所有内部通信都走统一格式再在接入层做格式转换。这样新增一个 Agent 的时候不需要改动其他 Agent 的代码。我见过很多团队把多 Agent 协作写成了“神仙打架”每个 Agent 直接调对方的接口上下文互相污染一个字段对不上就崩。Agent-Reach 的价值不是替你实现高阶智能而是让协作这件事变得有秩序、可观测、可回滚。1.2 真正适合 Agent-Reach 的场景什么样的项目才值得引入类似的协作层我总结了几类典型场景。多任务分发的客服系统是最典型的。用户的问题往往不是一个 Agent 能搞定的需要先意图识别再分给售后、销售、技术支持等多个专用 Agent。Agent-Reach 可以在中间层完成意图判断、路由、结果汇总。第二类是内部知识库问答。不同知识域由不同 Agent 负责比如 HR 政策和报销流程是两套。通过 Agent-Reach 做“先路由后回答”明显比塞进一个大 Prompt 更可控。第三类是 AI 自动化工作流需要多个 Agent 协作完成复杂任务比如市场调研、竞品分析、合同初审。每个 Agent 只干自己擅长的事由 Agent-Reach 负责编排调用顺序和上下文传递。但也有不适合硬套的场景。如果你的系统里只有一个 Agent或者所有逻辑都写在同一个 Prompt 里那么引入独立的协作层就是过度设计。另外如果任务链路极其固定用传统工作流引擎比如 Airflow、Temporal 可能比 Agent 编排更稳定。Agent 的价值在于有自主性如果完全不需要自主性就别硬加。1.3 “Reach”到底在说什么“Reach”在英文里有“触达、延伸”的意思。我当时起这个名字是想强调一个理念单个 Agent 的能力边界是有限的但通过协作层它可以把能力触达给系统里的其他 Agent也可以被其他 Agent 触达。这种双向触达才是多智能体系统真正难做的地方。很多团队一开始都以为难点在单个 Agent 的推理能力上。调着调着就发现真正的瓶颈往往是 Agent 之间的“可触达性”。上下文传递丢字段路由规则互相覆盖某个 Agent 挂了其他 Agent 还在傻乎乎地等结果。而 Agent-Reach 这类中间层就是把“谁触达了谁、在什么条件下触达、触达的结果是否可靠”变成可管理、可观测的基础设施。2. 设计 Agent-Reach 时的几个关键取舍2.1 我为什么选了中心路由、星型拓扑刚开始我考虑过完全去中心化的方案每个 Agent 都维护一张“同伴目录”自己决定把消息发给谁。听起来很自由但我很快就否掉了。因为去中心化拓扑调试起来太痛苦请求在 Agent A 和 Agent C 之间绕了三个弯日志分散在各处你根本不知道哪一跳出了问题。Agent-Reach 采用的是星型拓扑所有 Agent 都注册到中心路由器Agent 之间的消息不直接互发而是都经过路由器转发。这样做有三个明显好处。一是集中的路由策略。你要调整“哪些问题应该发给数据分析 Agent”只需要改一处规则配置不用去改每个 Agent 的 Prompt 或者代码。二是集中的可观测性。所有消息都会经过路由器所以 trace_id 可以贯穿一整条调用链。出了问题时我可以直接在路由器这一层拉出整条链路的日志而不是去十台机器上翻。三是生命周期管理方便。中心路由器可以定时做健康检查发现某个 Agent 连续心跳失败就把它标记为不可用后续请求自动绕开它。当然星型拓扑的缺点是中心节点会成为瓶颈也会成为单点故障。为了解决这个问题我做了两个设计。第一路由器本身不存业务状态所有状态放到 Redis 和 PostgreSQL 里这样路由器实例可以随时水平扩容挂了直接拉起新实例。第二健康检查和路由规则缓存都做了本地缓存短暂断网不影响路由决策只是转发可能收到“目标不可达”的错误。2.2 统一消息信封所有智能体只认一套协议多智能体系统最容易踩的坑就是消息格式不统一。有的模型返回一个字符串有的返回一个 JSON 对象有的返回一个包含 metadata 的复杂结构。Agent-Reach 在协议层做了一个强制约束所有内部流转的消息都必须符合一个标准信封。这个信封分为三部分。头部是元信息包括消息 ID、发送者、接收者、消息类型、trace_id、时间戳。负载是真正的业务数据可以是字符串也可以是 JSON 对象但必须有明确的字段名。额外部分放权限和优先级标记比如是否允许该消息触发工具调用、该消息的 priority 是 high 还是 normal。为什么一定要统一信封因为一旦多个 Agent 之间开始交换信息你很快会遇到字段冲突。比如 Agent A 返回的结果里有个字段叫 statusAgent B 里 status 的含义完全不同。如果没有统一的字段语义后面根本没法维护。Agent-Reach 信封里专门设计了 context 区块用来存放 Agent 之间需要共享的语义化数据同时保留 raw_payload 区块存放原始输出。这样既保留了原始信息又提供了统一访问路径。在实际使用时Agent 往往只关心信封里的某个区块。比如路由只读头部和 status最终答案拼接只读 payload。这样设计还有一个附带好处每个 Agent 不需要了解其他 Agent 的内部细节它只需要知道“我把消息交给 Agent-ReachAgent-Reach 会给我返回一个标准信封”。2.3 注册中心的健康状态与生命周期管理Agent 不像普通服务那样可以随时拉起。一个 Agent 在启动时往往要做模型加载、工具初始化、Prompt 模板加载这些事情比较耗时。所以 Agent-Reach 把每个 Agent 的生命周期分成了四个状态注册中、在线、离线、故障。Agent 启动后先向中心发送注册请求携带自身 ID、能力描述、接口地址、支持的输入输出格式。中心收到后返回一个注册确认同时给该 Agent 分配一个健康检查令牌。之后 Agent 要周期性发送心跳默认是每 15 秒一次。连续三次心跳失败Agent 就被置为离线如果恢复心跳则重新进入在线状态。这里有一个值得注意的细节Agent 的“能力描述”不只是给人看的自然语言它是要参与路由匹配的。我建议把能力描述写成一组结构化的标签比如 capability: order_queryscope: ecommerceformat: json。这样路由规则可以直接做标签匹配而不需要每次都去做语义解析。你要是把能力描述写成一段长作文路由的时候就麻烦了。故障状态和离线状态的区别是离线只是心跳丢失Agent 可能还活着只是网络抖动故障状态则是 Agent 自己上报的错误比如模型调用失败、内部状态异常。故障状态的 Agent 会被系统优先隔离避免它继续接收新请求。这套生命周期管理虽然简单但在实际运行中帮我少处理了大量“幽灵请求”。后来我还加了主动探测机制每隔一段时间中心会向 Agent 发送一次 ping 级别的请求以防 Agent 心跳正常但对实际请求毫无反应。3. 从零到可用的实操记录3.1 技术栈和最小依赖Agent-Reach 对技术栈没有严格要求但我自己实现的时候用了 Python 和 FastAPI因为团队里大部分人写 Python而且和 AI 生态的集成最方便。核心依赖只有三个FastAPI 用来提供 HTTP 接口Redis 用来做状态存储和分布式锁PostgreSQL 用来持久化注册信息和路由规则。如果你不需要分布式能力最小版本其实只需要 FastAPI 就够了。Redis 可以先用内存变量替代PostgreSQL 可以用 SQLite 替代。我建议新手先用最简单的方式跑通再逐步加入状态存储和持久化。一上来就上全套分布式反而容易迷失。推荐的目录结构大概是这样的agent_reach/ ├── core/ # 路由、注册、状态管理 ├── transports/ # HTTP、WebSocket、消息队列接入 ├── plugins/ # 针对不同Agent框架的适配插件 ├── config/ # YAML 配置文件 ├── agents/ # 示例Agent └── main.py # 启动入口Agent-Reach 本身不限制接入方式。只要你的 Agent 能通过 HTTP 或 WebSocket 收发消息就可以接入。如果你用的是现成的 Agent 框架通常只需要写一个很薄的适配层把框架的输入输出转换成标准信封。3.2 核心配置项解读我在 config.yaml 里保留了最常用的一组配置。初次使用的人可以完全按下面这个模板来改server: port: 8000 host: 0.0.0.0 registry: heartbeat_interval: 15 heartbeat_timeout: 45 storage_backend: redis router: strategy: intent_priority default_timeout: 30 max_depth: 10 rules: - name: order_query match: labels: capability: order_query target: order_agent priority: 10 - name: data_analysis_fallback match: labels: scope: ecommerce target: data_agent priority: 1 agents: - id: order_agent endpoint: http://127.0.0.1:9001 capabilities: - order_query - refund_process format: json - id: data_agent endpoint: http://127.0.0.1:9002 capabilities: - data_analysis - report_generation format: json这里我重点说两个参数。一个是 router.max_depth。这个参数用来限制一次请求最多经过多少个 Agent。默认值是 10但实际我建议设置成 3 到 5。因为智能体协作的深度一旦超过三层每一层都会丢失一部分上下文信息最后生成的结果质量会明显下降。这个参数不是为了防死循环而是防止你无意识地让多个 Agent 互相“接力”绕来绕去把用户需求绕没了。另一个是 router.strategy。这里我写的是 intent_priority意思是先按语义意图匹配再按优先级选择。实际实现时可以支持多种策略比如随机、负载均衡、加权轮询。我通常在调试阶段用随机策略因为每个 Agent 都能被均匀地调用到生产环境再用 intint_priority 这类确定性策略保证核心业务路由稳定。3.3 把第一个 Agent 接进 Agent-Reach接入 Agent 的核心工作是写一个适配层。下面我以一个最简单的“订单查询 Agent”为例。这个 Agent 本身逻辑很简单收到一个问题判断是否包含订单号然后返回一个标准信封。为了演示我直接写成一个 FastAPI 服务from fastapi import FastAPI, Request from pydantic import BaseModel app FastAPI() class Envelope(BaseModel): sender: str receiver: str msg_type: str trace_id: str payload: dict priority: str normal app.post(/agent/order) async def order_agent(req: Request): data await req.json() # 这里简化处理实际应解析 envelope payload data.get(payload, {}) question payload.get(question, ) order_id extract_order_id(question) if not order_id: return { sender: order_agent, receiver: data.get(sender, router), msg_type: response, trace_id: data.get(trace_id, ), payload: {answer: 没有找到订单号请提供。, status: missing_param}, } # 模拟查询订单 result fake_query_order(order_id) return { sender: order_agent, receiver: data.get(sender, router), msg_type: response, trace_id: data.get(trace_id, ), payload: { answer: f订单 {order_id} 的状态是 {result[status]}, order_id: order_id, status: ok, }, }接入 Agent-Reach 的时候你需要写一个注册脚本让 Agent 在启动后主动向中心注册自己的能力。下面的代码演示了注册流程import requests REGISTER_URL http://127.0.0.1:8000/register def register_agent(agent_id, endpoint, capabilities, formatjson): body { id: agent_id, endpoint: endpoint, capabilities: capabilities, format: format, } resp requests.post(REGISTER_URL, jsonbody) resp.raise_for_status() print(f{agent_id} 注册成功) register_agent( order_agent, http://127.0.0.1:9001/agent/order, [order_query, refund_process], )跑起来之后你可以通过 Agent-Reach 的路由接口发送一个请求看看路由器会不会把问题分给 order_agent。这一步是验证整个链路是否通的关键。我当时第一次跑通的时候特别激动因为终于不用再手工指定“把这条消息发给谁”了。3.4 自定义路由规则和动态策略配置路由规则是整个系统里最容易被玩坏的地方。因为业务方今天加一个关键词明天加一个优先级后天又提出“这个用户应该是 VIP 优先”如果全写死在代码里改一次发一次版你会很崩溃。Agent-Reach 的做法是把路由规则拆成“匹配条件”和“目标动作”两部分存放在数据库里运行时动态加载。匹配条件支持字段级匹配包括 labels、sender、msg_type、payload 里的任意字段。目标动作可以是转发给某个 Agent也可以是先调用一个函数做预处理再进入下一步匹配。举个例子我设置了一条规则凡是 payload 里包含 refund 关键词的消息优先转发到售后 Agent并且标记 priority 为 high。{ name: refund_routing, match: { payload.keywords: [refund, 退款, 退货] }, action: { type: forward, target: after_sale_agent }, priority: 20 }这里的关键是匹配条件里的 payload.keywords 是一个数组。实际匹配时只要任意一个关键词命中就属于匹配成功。优先级数字越大越先被检查所以到达 priority 20 的 refund_routing 会优先于其他规则执行。我之前踩过一个坑把关键词匹配和语义匹配放在同一个优先级里。结果有些用户消息同时触发了两条规则路由器不知道该选哪个白白增加延迟。后来我把规则设计成两阶段先做硬匹配关键词、标签、字段值如果硬匹配没有结果再触发语义匹配调用一个轻量级分类模型。这样既快又准。如果你用的是 LLM 做意图识别我建议把意图识别结果放在 envelope 的 metadata 里而不是每次路由都现场调用模型。因为一次对话可能会有多次路由每次都调一次模型成本会指数级上涨。最好是入口处做一次全局意图识别把结果缓存到 metadata后续路由直接读缓存。3.5 超时、重试与链路追踪怎么落地多智能体系统一旦复杂起来超时和重试就是双刃剑。超时设短了稍微慢一点的模型调用就会误杀超时设长了用户等半天等不到结果。重试也一样盲目重试可能把下游 Agent 打到崩溃。Agent-Reach 里我对超时做了分层处理。第一层是 HTTP 连接超时默认 5 秒第二层是业务处理超时默认 30 秒第三层是整条链路的累计超时默认 90 秒。每一层超时触发后都会向调用方返回一个明确错误码而不是笼统地抛一个异常。这样可以快速区分是网络问题、Agent 处理慢还是多个环节叠加导致超时。重试策略我建议只在以下三种情况启用连接失败、响应超时、下游返回明确的“临时不可用”状态。对于业务逻辑错误的响应不要重试因为重试大概率还是得到同样的错误。Agent-Reach 里我做了指数退避重试初始延迟 1 秒退避因子 2最大重试次数 3。这个参数组合是我在压测之后定下来的既能缓解瞬时抖动又不会让系统雪上加霜。链路追踪方面每个请求在进入 Agent-Reach 时都会生成一个 trace_id随后所有子调用都带着这个 trace_id。我在日志里统一打印三样东西trace_id、当前 Agent 名称、耗时。这样排查问题的时候就变成了“按 trace_id 搜日志看哪个环节耗时异常”。如果你想更精细还可以给每个 Agent 调用增加 span_id但实际用下来trace_id 加 Agent 名称这个粒度已经够用。curl -X POST http://127.0.0.1:8000/route -H Content-Type: application/json \ -d {trace_id:abc123,msg_type:query,payload:{question:我的订单什么时候到}}后端日志里会看到类似这样的输出[abc123] router received: msg_typequery [abc123] intent_match - order_query [abc123] route matched: order_agent [abc123] order_agent response: statusok, latency812ms [abc123] final response sent这一行行日志比任何调试器都好用。4. 跑了一段时间之后踩过的坑和排查方法4.1 智能体之间出现循环调用多智能体系统最经典的故障就是两个 Agent 互相调用最后谁也不干活光在那传递消息了。我真实遇到过售后 Agent 把用户的问题转给客服主管 Agent客服主管 Agent 又觉得这个问题涉及订单转给订单 Agent订单 Agent 判断这不是订单问题又转回售后 Agent。整个过程循环了十几次直到超时。后来我在 Agent-Reach 里加了两个防线。一个是 max_depth 限制转发深度到达设置的上限后直接终止并返回错误。这个方法简单粗暴能防止最坏情况。另一个是环路检测路由器会记录当前 trace_id 下已经访问过的 Agent 列表如果目标 Agent 已经出现在访问列表里就不再转发而是返回一个“路由环路”错误。实操中我发现环路发生的原因往往不是路由规则配错了而是某个 Agent 的职责边界太模糊。比如售后 Agent 的标签里既写了 order_query又写了个 escalate结果它不知道该自己处理还是转给别人。后来我要求每个 Agent 注册的时候必须填写 main_capability每个 Agent 只能有一个核心能力标签。这个约束从源头上减少了循环调用。4.2 上下文越长回答越差另一个让我头疼的问题是多个 Agent 协作时每个 Agent 都会往共享上下文里塞一段内容。几轮过后上下文变得非常长RAM 倒是没爆但模型输出的质量明显下降有时候还会把很早之前某个 Agent 的错误信息当成最新结果引用出来。这个问题本质上是上下文污染。解决方法不是无限扩大上下文而是做上下文治理。Agent-Reach 里我给每个 Agent 的上下文设置了配额正常情况下单个 Agent 最多携带 4000 字数的上下文进入模型超过的部分必须做摘要压缩。摘要可以在 Agent 内部完成也可以在路由器层通过一个轻量摘要 Agent 完成。我个人的建议是路由器层只负责传递“结论性摘要”和“关键原始字段”不要传递完整对话历史。比如订单 Agent 只需要知道“用户问的是退款”不需要知道用户前面和客服聊了 20 句废话。这就像公司里开跨部门会议每个人只需要带自己部门的结论来不必把 200 页会议记录全背过来。4.3 路由规则越配越乱怎么办路由规则一开始只有 3 条后来变成 30 条再后来连写规则的人都忘了某些规则为什么存在。我踩过这个坑之后给 Agent-Reach 增加了一个规则可视化逻辑把每条规则的匹配次数、命中率、平均耗时都记录下来定期输出报表。通过报表我发现了大量“死规则”有一些规则从来没有任何请求命中过完全是因为某次临时排查问题加的后来忘了删。还有一些规则之间存在重复匹配两条规则条件相似优先级高的那条把低的那条完全覆盖了。最后我定了一条规矩每两个月集中清理一次路由规则匹配率连续 10 天低于 0.1% 的规则自动置为停用状态。另外一个治理技巧是路由规则的名字必须能看懂。不要用 rule1、rule2 这种名字而要用 order_query_v2、refund_with_vip_priority 这种带语义的名字。不然三个月后你自己都会怀疑人生。4.4 并发一高状态存储开始拖后腿最开始我用 PostgreSQL 存注册信息和路由规则用 Redis 存心跳和会话状态。后来发现高并发时 PostgreSQL 的连接数先扛不住了。排查之后发现问题不在于 PostgreSQL 查询多慢而在于我的代码在每次路由时都查了一次数据库加载全部规则相当于把数据库当缓存用。修复方案很简单路由规则在启动时加载进本地内存每隔 30 秒和数据库做一次同步。Redis 里的心跳数据做异步写入只在状态变化时通知主库。改完之后同样的并发量下 PostgreSQL 的负载降到了原来的十分之一。这个教训让我明白了一个道理协调层本身必须是高性能的它不能成为业务链路上的新瓶颈。4.5 常见问题排查速查表我把平时最容易遇到的几类问题整理成了一个表供你快速定位症状可能原因排查动作Agent 一直返回超时下游模型调用过慢查看该 Agent 的耗时统计考虑加大超时或换更快模型两个 Agent 互相转发路由规则重叠或职责模糊检查命中规则列表限制每个 Agent 的 main_capability回答内容张冠李戴上下文被污染开启上下文摘要开关只传关键结论路由规则不生效规则缓存未刷新手动触发规则同步检查新规则是否真正写入库Agent 状态一直离线心跳丢失或健康检查失败检查 Agent 进程是否存活网络是否稳定端点是否可访问请求量一高就慢中心节点单点负载过高水平扩容路由器检查数据库连接池这个表我直接写进了项目 README后来团队新人接手后不需要问我也能自己排查大部分问题。多智能体系统虽然听起来很酷但维护起来跟传统分布式系统一样拼的都是基本功日志、超时、状态管理、规则治理。Agent-Reach 把这些基本功收拢到一个可观测的中间层里让智能体之间既能各显神通又能协作得井井有条。最后再分享一点实际体会。不要指望一个框架能解决所有多 Agent 协作问题Agent-Reach 能保证消息不乱跑、状态不丢、问题可查但它不会替你想清楚每个 Agent 的职责边界。边界这种东西需要你在跑业务的过程中一点点修正。所以我的建议是先把一个最核心的场景跑通再加第二个、第三个不要一上来就接十个 Agent。现在这套框架内部还保留着当时那条最原始的路由规则如今已经长成了包含几十条规则、十几个服务的协作网络。以后我大概率还会往里加更多东西比如基于成本的路由、按模型速度自适应调度的策略但核心的思路不会变——让每个 Agent 都能被找到、被路由、被信任这才是多智能体系统能走远的地基。