☰
Agent-Reach:让多Agent协作不再乱套的轻量路由交接框架
2026/10/27 16:10:27 网站建设 项目流程

做 Agent-Reach 这个项目,起因特别朴素:我自己手底下一堆 AI Agent 单个跑都没问题,凑在一起就乱套。客服机器人只会答订单,内容总结器只认长文本,数据分析助手只吃表格,可用户不会按我的系统划分来提问题——他问一句"我上个订单的退款到底卡在哪里",按道理应该先找客服、再查订单系统、再拉支付数据,但当时没有任何一个 Agent 会主动把球传出去。

那段时间我试过几种现成的编排方案,要么太重(光配置中心就几百个字段),要么太黑盒(Agent 互相调来调去,出了问题根本不知道谁先甩的锅)。后来我干脆自己写了一个轻量路由与交接框架,取名 Agent-Reach。Reach 在这里有两层意思:一是让请求能精准触达对口的 Agent,二是让每个 Agent 的能力边界可以被显式地发现和触达。这篇文章就把这个项目的设计思路、最小实现、完整链路和踩坑记录摊开来讲,适合正在做多 Agent 协作、或者被"Agent 会了但不会配合"折磨过的朋友参考。

1. 项目缘起:多 Agent 场景里最缺的不是模型,是"总机"

1.1 没有路由的多 Agent 协作现场

你先想象一个场景:三个 Agent 各自都有独立的 HTTP 服务,客服 Agent 挂在 8001 端口,技术顾问 Agent 挂在 8002 端口,工单处理 Agent 挂在 8003 端口。用户问"我的退款申请提交三天了还没到账,帮我查一下",请求打进来之后,系统怎么决定该把这句话送给哪个 Agent?

我最早的做法很直接:用关键词硬匹配。"退款"命中客服,"查一下"命中工单,结果一句话被同时发给了两个 Agent,两边各自回复了一段,用户收到的答案互相矛盾。更麻烦的是,客服 Agent 查完订单状态后,发现需要进一步看支付渠道的回执,它根本没有能力去调用技术顾问 Agent——因为它们之间没有任何约定的通信协议。

这个问题说白了就是:单 Agent 的能力被模型和 Prompt 定义,多 Agent 系统的能力却被"谁能找到谁"定义。模型再聪明,如果请求到了错误的 Agent 门口,或者正确的 Agent 根本接不到请求,整个系统就是散的。当时的系统就像一栋没有总机的办公楼,每个房间里都坐着专家,但电话线没接起来。

1.2 "能答"和"能触达"是两回事

我后来跟朋友聊这个项目时打了个比方:一个 Agent 就像一家餐馆的后厨,菜单写得好、厨师手艺高,只是"能答";但客人走进来,得有服务员把菜单递过去、把菜品端出来,这是"触达"。大多数 AI 应用开发者的精力都花在改进后厨——换更大的模型、写更细的 Prompt——却忽略了服务员这一层。

Agent-Reach 的定位就是那个服务员:它维护一张"哪个 Agent 擅长什么"的注册表,接收任意入站请求,根据请求内容和每个 Agent 的能力描述做打分匹配,必要时还会在多个 Agent 之间做一次或多次交接,把上一站的结论作为上下文传给下一站。整个过程对用户透明,用户只看到一个入口,背后是几个 Agent 在接力干活。

这个项目我前后迭代了大概三周,从最开始 200 行的 FastAPI 脚本,到后来加入了心跳检测、语义路由、交接上下文令牌、超时熔断,代码量涨到了接近 1500 行。下面我按设计决策的顺序来讲,这样比直接贴完整代码更清楚。

2. 核心理念:把"路由"和"交接"当成一等公民来设计

2.1 能力注册表:先让系统知道"谁在、谁能干什么"

做多 Agent 系统,我踩过的第一个认知坑就是:过度依赖模型去理解 Agent 列表。很多编排框架的做法是,把 Agent 的描述拼进一个大 Prompt,让 LLM 自己决定调用谁。这在 Agent 只有两三个的时候挺好用,一旦超过五个,模型就会出现选择漂移——同一个问题,今天选 A,明天选 B,而且你很难解释它为什么这么选。

Agent-Reach 反其道而行之:每个 Agent 启动时必须向注册中心上报一份结构化的能力声明,就像是给总机交了一份"分机号码 + 擅长领域 + 受理格式"的登记表。我用的数据结构很简单:

@dataclass class AgentSpec: name: str # Agent 唯一标识,比如 "customer_service" description: str # 给路由器看的自然语言能力描述 input_schema: dict # 期望收到的消息结构,路由后会用 JSON Schema 校验 endpoint: str # 该 Agent 的 HTTP 地址 priority: int = 0 # 同分时谁优先 ttl: int = 60 # 心跳超时秒数,超过视为下线

描述字段我要求写得非常具体,不要写"处理各种问题"这种废话,要写"处理订单查询、退款进度、优惠券使用问题;需要订单号或用户 ID;可以回答支付渠道相关疑问"。为什么必须具体?因为语义路由的质量上限,完全取决于 description 和真实能力之间的匹配度。描述越抽象,向量检索的召回就越差。

2.2 两类路由策略:语义路由与声明式路由

注册表有了之后,接下来就是路由本身。Agent-Reach 里我实现了两种策略,可以混用。

第一类是语义路由:把用户查询和每个 Agent 的 description 都转成向量,算余弦相似度,取最高分且超过阈值的那个。这种适合开放式的用户表达,比如"我东西还没到"这种话,你很难靠关键词猜到它在说物流,但向量能把它和"物流查询 Agent"的 description 拉近。我的实现片段:

def semantic_route(query: str, candidates: list[AgentSpec], threshold: float = 0.72): q_vec = embed(query) # 复用同一个 embedding 服务,避免引入额外模型 scored = [] for spec in candidates: d = cos_sim(q_vec, embed(spec.description)) scored.append((d, spec)) scored.sort(key=lambda x: x[0], reverse=True) return scored[0][1] if scored[0][0] >= threshold else None

第二类是声明式路由:针对意图特别明确、规则特别稳定的场景。比如请求头里带着intent=refund,就直接走客服 Agent,完全不过模型。这类规则我放在一个 YAML 里维护,不写死在代码中。为什么保留这种相对"笨"的方式?因为语义路由有它的软肋——对专有名词、编号、短查询不敏感,比如用户发一个订单号"PO-2024-0831",你跟它谈语义是谈不出来的,必须靠显式规则硬路由。

实际项目中,我强烈建议两条腿走路:规则兜底保证确定性,语义路由覆盖开放性表达,两者权重可以配。Agent-Reach 的默认逻辑是:先看有没有命中声明式规则,命中就走规则;没命中再走语义路由;语义路由低于阈值就返回兜底话术,而不是硬塞给一个不匹配的 Agent。

2.3 交接协议:上下文怎么传才不会丢

路由只解决了"第一棒交给谁",多 Agent 系统真正难的是"第二棒、第三棒怎么接"。我最早的做法非常粗暴:Agent A 的处理结果直接拼进 Agent B 的 Prompt。结果 B 经常被 A 的一大段废话带偏,而且 A 如果没把原始问题原样保留,B 就根本不知道用户在问什么。

所以我设计了一个交接信封(Handoff Envelope),每一次请求从入口到最终返回,都带着同一个信封走:

class HandoffEnvelope(BaseModel): trace_id: str # 整条链路的追踪 ID,排查问题全靠它 session_id: str # 会话 ID,用于多轮上下文 step: int # 当前已经经过了几次交接 max_steps: int = 5 # 最多跳数,防止路由死循环 history: list[dict] # 前面 Agent 的结论摘要,按时间顺序 payload: dict # 原始诉求 + 当前 Agent 需要的字段,永不丢失

这里有一个关键设计决策:原始 payload 始终保留,history 只增不减。每个 Agent 返回时,只能往 history 里 append 自己的结论摘要,不能修改 payload。为什么?因为后续 Agent 经常需要回头查看原始问题。客服 Agent 把用户的话压缩成"用户想退款"再传给技术 Agent,技术 Agent 想核对支付流水号时发现原文丢了,这种事故我在早期版本里出过不止一次。保留原文,只是多存几个 token 的事,但能省掉整条链路的返工成本。

另外,交接不是无条件的。Agent 可以在结果里标注need_handoff=true并附带suggest_target字段,路由器才会继续找下一棒;否则就直接把当前结果返回给用户。这么做的理由是:让"是否交接"这个决策尽量靠近业务现场,而不是由全局路由器猜。

3. 最小可用的实现:Python + FastAPI 搭出的骨架能跑多远

3.1 目录结构与核心类

如果你只想快速复刻一个 Agent-Reach 雏形,我给的目录结构是这样的:

agent-reach/ ├── main.py # FastAPI 入口,路由服务本身 ├── registry.py # 注册表与心跳管理 ├── router.py # 规则路由 + 语义路由 ├── dispatcher.py # 交接调度,负责跳数控制和超时 ├── agents/ │ ├── customer_service.py │ ├── tech_consult.py │ └── ticket_writer.py └── config.yaml # 声明式路由规则

路由服务本身不执行业务,只做三件事:维护注册表、计算路由、调度交接。每个真正的 Agent 是独立部署的 HTTP 服务,通过统一接口被调用。这种解耦带来的好处是,你可以随时上线一个新 Agent,只要它注册一下就能被路由到,不需要改主服务的代码。

核心类就两个。一个是AgentRegistry,负责存注册信息和心跳时间;另一个是Dispatcher,负责执行交接链路。我把 Dispatcher 的逻辑单独抽出来,是因为踩过"路由器和调度器混在一起"的坑——混在一起时,你想加一个超时都分不清是路由超时还是 Agent 调用超时。

3.2 注册与心跳:先让 Agent 报个到

注册接口很直接,Agent 启动后调用POST /register,带上自己的 AgentSpec。为了应对 Agent 崩溃重启的情况,我加了一个心跳机制:每个 Agent 每 30 秒调一次POST /heartbeat,注册中心记录last_heartbeat。路由时,只有心跳在 TTL 内的 Agent 才进入候选列表。

@app.post("/register") async def register(spec: AgentSpec): registry[spec.name] = { "spec": spec, "last_heartbeat": time.time(), "status": "healthy", } return {"ok": True, "name": spec.name} @app.post("/heartbeat") async def heartbeat(name: str): if name in registry: registry[name]["last_heartbeat"] = time.time() registry[name]["status"] = "healthy" return {"ok": True}

这个机制的初衷是:多 Agent 系统里,单个 Agent 挂掉不该拖垮整条链路。没有心跳,你可能会把请求路由到一个已经死掉的 Agent,然后等一个永远不会回来的响应,最后只能靠外层超时兜底,体验很差。有了心跳,路由阶段就能提前把不健康节点摘掉。实际部署时我还加了一个/health轮询作为心跳的补充,用来检测那种"进程还活着但业务已经不正常"的僵尸状态。

3.3 路由器的匹配逻辑:关键词、规则、向量三级递进

router.py 里我实现了一个三级匹配流程。第一级,查 config.yaml 里的规则表,命中intent或特定字段的直接返回;第二级,对入站文本做轻量关键词打分,作为语义路由的辅助特征;第三级才是向量相似度。

rules: - match: query_type: exact field: intent value: refund target: customer_service - match: query_type: contains field: text value: "PO-" target: order_query

为什么保留第二级关键词打分?因为纯向量方案在某些短文本上得分虚高。我实测过一个案例:用户问"发票",技术 Agent 的 description 里有"接口文档",语义相似度打到 0.68,差点被路由过去。加上关键词权重后,系统会给"发票"这个词叠加一个偏向财务 Agent 的分数,明显更稳。所以我的建议是:语义路由负责召回,关键词和规则负责纠偏,两者是配合关系,不是替代关系。

4. 一次完整的链路演示:客服分流、技术查询、工单处理三个 Agent 接力

4.1 三个 Agent 的角色配置

我把一个最常见的客服 + 技术支持场景搬上来做演示。三个 Agent 的注册信息如下:

Agent 名能力描述(简写)期望输入典型响应耗时
customer_service订单查询、退款进度、优惠券,能读订单号{query, user_id}1.2s
tech_consultAPI 文档、SDK 集成、报错日志分析{query, logs}2.8s
ticket_writer把结构化问题写入工单系统,返回工单号{summary, priority}0.3s

用户的实际提问是:"调用你们的支付接口一直报 500,退款是不是也受影响?"这句话同时涉及两个领域:技术报错和退款业务。

4.2 从请求进入到结果返回的时序拆解

整个流程我拆成了六步。

第一步,请求进入 Agent-Reach 主服务,路由器先跑规则表,没命中任何规则。第二步,语义路由打分,tech_consult得分 0.81 最高,customer_service0.75 次之,系统选择 tech_consult 作为第一棒。

第三步,Dispatcher 调用 tech_consult,传入原始 payload。技术 Agent 分析日志后得出"支付接口返回 500 是因为签名过期,和退款无关",但它发现用户还惦记退款,于是在结果里标记need_handoff=true,suggest_target="customer_service"。

第四步,Dispatcher 检查 step 还没到上限,就带着信封继续找 customer_service。这里要注意,信封里历史已经有技术 Agent 的结论,但原始 payload 原封不动。第五步,客服 Agent 读取历史和技术结论,知道技术问题已解决,于是专注处理退款部分,返回"退款不受影响,预计两个工作日到账"。

第六步,按业务规则,凡涉及"接口报错 + 用户疑虑"的对话应记录工单,ticket_writer 被第三次调用,把整个对话摘要和两个 Agent 的结论写入工单系统,返回工单号。最终用户收到一条综合答复和工单号。

第 4.2 节值得留意的是:每一次交接都意味着一次额外的 LLM 调用,所以我在 Dispatcher 里做了一层短路判断——如果第一棒 Agent 返回结果已经很完整,且没有need_handoff标记,就绝不强行交接。Agent 之间要交接,必须自己给出理由。

4.3 延迟与成本实测数据

这个链路我压测过一轮,用 200 条真实历史对话回放,数据如下:

链路类型平均延迟P95 延迟单条平均 token 消耗
单 Agent 直答1.1s2.0s0.8k
一次路由 + 一次交接1.9s3.1s1.4k
两次交接(技术 -> 客服 -> 工单)4.3s5.8s2.6k

看到延迟后我先是不太满意,后来冷静想了想,多 Agent 接力本来就不是免费的——它换来的是"每个 Agent 只需在自己专业范围内做判断",单步质量更高,整体幻觉率下降。实测幻觉率从单 Agent 直答的 11.5% 降到了交接链路的 6.2%,这个收益对我来说是值得的。

不过这个数字也提醒了我:如果链路里任何一个 Agent 的调用超过 3 秒,整体体验就会明显变差,所以超时控制必须做在 Dispatcher 层,而不是依赖每个 Agent 自己自觉,这一点我在下一节专门展开。

5. 接踵而至的坑:调试 Agent-Reach 时最折磨人的几个问题

5.1 Agent 循环甩锅:路由回路与最大跳数

第一个坑来得特别快。上线第二天,用户问了一句"我想改一下收货地址",结果系统在客服 Agent 和订单 Agent 之间来回交接了四次,最后超时报错。我查 trace 日志发现,客服 Agent 认为改地址属于订单模块,订单 Agent 认为改地址需要先确认用户身份所以又传回客服,两个 Agent 互相客气,谁也不肯收尾。

根因是:我没有给路由加跳数上限,也没有阻止同一 Agent 被重复访问。修复方案有两层。第一层是硬性的,在 Dispatcher 里维护一个visited集合,同一个 Agent 在一次请求链路中最多被访问一次;同时把max_steps从 5 降到 3,超过就触发兜底话术。第二层是软性的,我在交接信封里加了一个previous_conclusions字段,让 Agent 在决定交接前能看到"上一棒已经说过什么",避免重复做无意义判断。

如果你也在做类似系统,我强烈建议把 trace_id 和 step 信息打到日志里——排查这种循环问题,没有 trace 你只能靠猜。

5.2 上下文在交接中悄悄缩水:Trace 追踪还原真相

第二个坑更隐蔽。某天用户反馈,经过两次交接之后,技术 Agent 的回答明显"变傻"了,给出的建议和原始报错日志完全不沾边。我一开始以为是模型问题,后来打开 trace 才发现:客服 Agent 在生成 history 摘要时,为了省 token,把原始报错信息里的堆栈关键行给截掉了,技术 Agent 拿到的 history 里只有一句"用户遇到支付报错",自然只能给通用建议。

这事的教训是:摘要必须标注"可能不完整",而且原始 payload 中的关键结构化字段要单独保留,不能只依赖自然语言摘要。我后来在交接协议里加了一条约定:凡是input_schema中标记为preserve=true的字段,Agent 在生成 history 时必须原样拷贝,不得压缩。这条约定从协议层面杜绝了上下文缩水。

防患于未然比事后补漏重要得多。我在 Dispatcher 里加了一个校验逻辑:每次交接前,比对 payload 中 preserve 字段在 history 里是否存在,缺失就直接拒绝交接并报警。

5.3 超时与并发:别让一个慢 Agent 拖垮整条链

第三个坑出现在并发压力下。某次推广活动带来流量高峰,路由服务吞吐骤降。排查后发现,罪魁祸首是技术顾问 Agent 的一次 LLM 调用耗时 28 秒,而 Dispatcher 按同步方式等待它,导致后续 40 个请求全部排队。

修复做了三件事。第一,给每个 Agent 调用加独立超时,默认 15 秒,超时直接降级为该 Agent 的兜底回答,不再等待。第二,引入信号量控制并发上限,同一个 Agent 的并发调用不超过 8 个,超出部分走排队队列。第三,增加熔断:如果某个 Agent 在 2 分钟内超时率达到 50%,路由器自动把它标记为降级状态,请求直接路由到备选 Agent 或兜底话术。

async def call_with_timeout(agent: AgentSpec, envelope: HandoffEnvelope, timeout: float = 15.0): try: return await asyncio.wait_for(call_agent(agent, envelope), timeout=timeout) except asyncio.TimeoutError: log_warning(f"agent timeout: {agent.name}, trace_id={envelope.trace_id}") return {"type": "timeout", "fallback": "当前服务繁忙,请稍后重试"}

这个坑其实暴露了一个更本质的问题:多 Agent 系统的瓶颈往往不在模型,而在调度策略。模型再快,如果调度层同步等待、没有超时、没有并发控制,整个系统就是一个巨大的串行队列。把这三个机制加上之后,高峰期的 P95 延迟从 9.3 秒降到了 4.1 秒,效果立竿见影。

6. 我用下来的结论与后续打算

6.1 什么场景下 Agent-Reach 最值钱

几个月用下来,我总结出这个框架真正有价值的三个场景。第一,业务交互链路过长的场景,典型如客服、售前咨询,一个请求需要查订单、查物流、查优惠券,多 Agent 分工天然合理。第二,Agent 领域边界清晰的场景,比如财务、技术、运营各管一摊,互相不抢活,路由准确率高。第三,需要审计与追溯的场景,每个 Agent 的结论都在 history 里留痕,出了问题能追到具体是哪一环判断失误。

反过来,如果只是单个 Agent 就能覆盖 80% 的请求,或者两个 Agent 之间几乎没有交接需求,那 Agent-Reach 就属于过度设计,老老实实用一个模型加一个 Prompt 就行。

6.2 不要过度设计:什么时候该拆掉它

我做这个项目最大的体会,是学会克制。早期版本我加了一堆花哨功能:自动重试、动态 Prompt 拼接、Agent 之间的图拓扑配置。后来发现 90% 的情况下用不到,反而增加了排查成本。现在我砍到只剩四个核心能力:注册、路由、交接、超时熔断。够用,且每个组件都能在一分钟内讲清楚。

如果你准备借鉴这个方案,我的建议是先把单一 Agent 的服务质量打磨到位,再引入路由层。路由是放大器,不是救火队。Agent 本身答不好,路由再准也没用。

6.3 后续计划:观测、回滚与社区化

接下来我打算做两件事。一是完善观测面板,把每条链路的 trace_id、step、各 Agent 耗时、token 消耗可视化出来,正在考虑是否把数据接到现有的监控系统里。二是做一个简单的回滚机制——当路由准确率在某段时间内下降明显时,能把路由策略自动回退到上一版本,而不是干等人工介入。Agent-Reach 的代码量不大,我准备整理干净后开源,把注册表协议和交接信封这两个核心抽象提炼成一版规范,让不同语言实现的 Agent 也能互相协作。

最后说句掏心窝的话:做多 Agent 系统的这几个月,我最大的体会是,AI 应用开发的复杂度正在从"训练模型"转移到"编排一堆会各自说话的组件"。Agent-Reach 只是我在这个方向上的一小块拼图,它解决的是"谁能找到谁"的问题,而这个问题,恰恰是大模型应用从 Demo 走向生产时最容易忽略、也最致命的一环。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询