1. 为什么我做了 Agent-Reach多智能体协作里最容易被低估的“触达”问题先说个真实场景。上个月我们内部搞了一次多智能体联调三个 Agent 分别负责查库存、生成报价、执行订单。单拿出来每一个都跑得好好的结果串起来之后第一个 Agent 在对话历史里翻到了三个月前一条已经失效的折扣信息第二个 Agent 拿着这个信息直接算错了价格第三个 Agent 还想当然地调了一个已经被下线的接口。排查到最后问题根本不在模型能力而是在 Agent 之间“怎么找到对方、怎么传上下文、怎么控制调用边界”这一整条链路。那段时间我每晚都在想同一个问题能不能把“触达”这件事做成一种可配置、可观测、可管控的通用能力这就是 Agent-Reach 的起点。Agent-Reach 不是一个模型也不是某个具体的 Agent 应用而是一套面向多智能体协作场景的触达编排方案。它要管的事情很具体当一个 Agent 需要调用另一个 Agent 的能力时消息该走哪条路由、上下文该带多少、能调用哪些工具、不能越过哪些边界、失败了怎么重试、整条调用链怎么追踪。简单说单 Agent 的能力决定它“聪明不聪明”Agent-Reach 决定一群 Agent 在一起时“找不找得到人、传不传得对话、控不控得住边界”。这篇文章主要聊三部分Agent-Reach 的设计思路、核心代码实现、我在实际部署和联调中踩过的坑。适合三类人看正在搭多 Agent 平台的架构师、做 AI 自动化流程的开发者以及因为 Agent 互相乱调而头疼的运维同学。如果你只是刚接触多智能体也没关系我会把里面的关键概念都用大白话拆开讲。1.1 一次差点让项目翻车的联调事故先把这个事故完整还原一下因为它是整个 Agent-Reach 的起源。我们当时的架构很简单用户输入先进主 AgentRouter主 Agent 根据语义把请求分发给子 Agent子 Agent 再调用各自的工具。看起来没什么问题实际跑起来全是洞。第一个洞是“调用风暴”。某个子 Agent 在处理退款请求时因为拿不到足够的用户信息连续向主 Agent 发起了三次回传请求主 Agent 又把它当成新任务重新分发结果同一笔退款在几分钟内被重复处理了两遍。第二个洞是“上下文越界”。子 Agent 之间传消息时默认把整段对话历史都带上一个负责写文案的 Agent 被塞进了大量数据库查询日志输出质量明显下降还白白浪费了几万 token。第三个洞是“幽灵接口调用”。有 Agent 基于训练语料中的旧接口格式生成调用请求但真实环境里那个接口早就换路径了失败后它不报错反而尝试了三种参数变体把日志刷得乱七八糟。这三个洞其实指向同一个根因我们没有对 Agent 之间的触达行为做任何约束。每个 Agent 都像一台可以随意踩油门的车但没有车道、没有红绿灯、没有限速牌。Agent-Reach 要补的就是基础设施这一层。1.2 Agent-Reach 到底解决什么四种“触达”我把多智能体协作里的核心问题归纳成四种触达Agent-Reach 的所有设计都是围绕这四种触达展开的。触达类型核心问题失控时的典型表现路由触达一个 Agent 怎么找到有能力处理该请求的另一个 Agent请求被错误转发、重复分发、找不到处理者上下文触达消息在传递时应该携带哪些有效信息上下文爆炸、无效信息污染、token 超限资源触达Agent 能调用哪些工具、接口、数据库调用已下线接口、越权访问、副作用不可控权限触达不同 Agent 之间允许什么粒度的互相调用子 Agent 反向触发高危操作、绕过审批链四种触达不是平级的。资源触达和权限触达是底座路由触达和上下文触达是表现层。Agent-Reach 的做法是把这四类规则统一描述、统一执行、统一观测而不是像以前那样散落在每个 Agent 的 prompt 里碰运气。1.3 什么场景需要它什么人适合读诚实说如果你的项目只有一个 Agent、只调几个外部 API那 Agent-Reach 是过度设计你直接写个工具函数就行。但一旦你的系统开始出现这些信号就该考虑它了主 Agent 的 prompt 越来越臃肿、子 Agent 之间互相调用没有规律可循、一次用户请求会触发十几跳内部消息、出了问题只能靠翻日志猜链路。具体来说Agent-Reach 比较适合这几类场景企业内部的多 Agent 自动化中台、客服和销售场景的 Agent 协作流、代码生成与执行沙箱需要有严格的资源边界、以及由多个垂直领域 Agent 组成的内容生产系统。读这篇文章的人我假设你至少有一个跑通的多 Agent demo或者正在设计这样的系统不然后面代码部分的体感会弱一些。2. Agent-Reach 的架构设计把触达控制做成平台能力Agent-Reach 在设计上坚持一个原则触达控制不应该是每个 Agent 自己的“自觉”而应该下沉成平台层的强制约束。好比一个公司的跨部门协作不能指望着每个员工都主动守规矩得有统一的 OA 流程、权限系统和审计日志。所以 Agent-Reach 没有做成 SDK 让你在每个 Agent 里手动埋点而是做成了一个代理层所有 Agent 之间的通信都从它这里经过。这样做有两个直接好处第一新接入一个 Agent 不需要改它的内部逻辑只需要声明它“能干什么、不能干什么”第二所有触达行为都会留下结构化日志排查问题的时候不用再靠猜。2.1 六个层级的分工Agent-Reach 的内部结构从下往上分成六层每一层只解决一种触达问题。接入层负责把不同类型的 AgentOpenAI 接口、自研模型、纯规则脚本统一成标准的消息格式屏蔽底层差异。编排层负责处理外部进来的用户请求决定要不要拆分成子任务、按什么顺序分发这是整个系统的入口。路由层根据路由表把消息送到目标 Agent路由表支持精确匹配、模糊匹配和兜底策略。触达协议层这是 Agent-Reach 比较核心的一层。它定义了消息中允许携带什么元信息、上下文的裁剪规则、同步还是异步、以及超时和重试预算。策略层负责加载权限和资源边界规则比如某个 Agent 不允许调用支付接口某些外部工具只能在特定时间段访问。可观测层把每一次触达包括成功、失败、被拦截记录为一条 reach trace包含链路 ID、跳数、耗时、调用来源和目标。这么分层的好处是你想改某一个维度的规则时不用动其他层。比如要调整上下文裁剪策略只需要改触达协议层的配置路由和权限完全不受影响。2.2 路由与覆盖范围白名单不是限制是安全网很多人刚接触 Agent-Reach 时对“白名单路由”这个设计有点抵触觉得把 Agent 之间的调用限制死了会影响灵活性。我的观点恰恰相反在多智能体系统里自由度越高出事故的概率越大白名单看起来是限制其实是安全网。Agent-Reach 路由层不搞“Agent 自己决定找谁”的分布式自主调用而是统一维护一张路由表。路由表里有三种条目精确路由task_type 完全匹配时走指定 Agent、模糊路由用关键词权重匹配比如“退款”命中客服 Agent 的概率设为 0.9、拒绝路由明确禁止某些 task_type 出现在某些 Agent 之间比如“删除数据库”这种任务禁止从低权限 Agent 转发。我见过很多失败的多智能体项目问题都出在“让 Agent 自己找伙伴”。模型在开放环境下做工具选择已经不够稳定再让它做跨 Agent 路由选择错误率会叠加。所以 Agent-Reach 把路由决策从模型手里拿回来交给确定性的规则引擎。模型负责理解语义、提取参数Agent-Reach 负责决定把参数交给谁。这和微服务架构里的 API 网关是一个道理如果每个服务都自己去服务发现、自己决定调用谁那系统离雪崩就不远了。2.3 触达半径上下文不是传得越多越好上下文触达是 Agent-Reach 里我个人认为最有价值的设计。大多数多智能体系统传上下文的方式是“整段复制”A 把用户原始输入 自己的思考 工具返回结果 历史消息全部塞给 B。结果就是 B 收到的信息里至少有一半是噪声模型要在噪声里找关键信息既慢又容易出错。Agent-Reach 引入了一个叫“触达半径”的概念。每个 Agent 在注册时声明自己的上下文偏好处理这个类型的任务最多需要多少 token、需要什么类型的上下文用户摘要、工具返回、历史决策、哪些信息明确不需要。当 A 要向 B 传消息时触达协议层会根据 B 的偏好对上下文做一次裁剪和摘要而不是直接转发原文。我举一个具体例子。用户要求“把上周的销售数据整理成周报并发送给管理层”主 Agent 手里有整整 5 万 token 的原始数据。如果直接传给写周报的 Agent成本高不说模型还容易抓错重点。Agent-Reach 的做法是先让一个轻量的聚合组件把数据加工成“核心结论 关键图表 异常标注”再把这份摘要传给写周报的 Agenttoken 直接降到 3000质量反而更稳定。后续第四节我会给出具体的实现代码。3. 核心代码实现路由策略、退避重试与上下文裁剪理论讲得再多不如直接看代码。Agent-Reach 的核心实现不算复杂它的聪明之处在于把策略与执行分离。我用 Python 写了最小的可运行版本大家可以直接照着搭。3.1 路由策略配置用 DSL 表达触达规则Agent-Reach 使用 YAML 作为策略描述语言原因很简单可读性好非技术人员也能维护。下面是系统里一份典型的配置routes: - name: refund_flow match: type: keyword_weight keywords: [退款, 退货, refund] min_score: 0.6 target: agent:refund_service fallback: agent:human_handoff allowed_hops: 3 - name: report_flow match: type: exact task_type: generate_weekly_report target: agent:report_writer required_context: template: summary max_tokens: 4096 include_fields: [conclusion, key_metrics, anomalies] reach_policy: default_timeout_ms: 5000 retry_budget: 3 backoff_base_ms: 200 max_parallel_calls: 8 blocklist: - from: agent:report_writer to: agent:payment_executor reason: 写周报的 Agent 不允许触发支付操作路由配置里有个关键的allowed_hops字段这是防循环触达的核心。每次消息经过一个 Agent跳数就加一超过上限直接丢弃并告警。required_context则声明了上下文触达的要求目标 Agent 不需要原始数据只需要summary模板、最多 4096 token、只要结论和异常字段。这部分的设计思路是把“该找谁”和“该带什么”都变成可声明的配置而不是写死在代码里。这样即便系统里有几十个 Agent运维同学也能在开会时打开 YAML 直接讨论某个流程该不该加白名单而不是去翻代码。3.2 路由执行器匹配、转发与退避重试配置是静态的真正干活的是路由执行器。核心代码如下import asyncio import random import yaml from dataclasses import dataclass from typing import Any, Optional dataclass class ReachContext: trace_id: str source: str target: str hops: int payload: dict policy: dict class ReachRouter: def __init__(self, config_path: str): with open(config_path, r, encodingutf-8) as f: self.config yaml.safe_load(f) self.routes {r[name]: r for r in self.config[routes]} self.policy self.config[reach_policy] def match_route(self, task_type: str, keywords: list[str]) - Optional[dict]: for route in self.routes.values(): matcher route[match] if matcher[type] exact: if task_type matcher[task_type]: return route elif matcher[type] keyword_weight: score sum(1 for kw in matcher[keywords] if kw in keywords) score score / len(matcher[keywords]) if score matcher.get(min_score, 0.5): return route return None async def forward(self, ctx: ReachContext, agent_client: Any) - dict: if ctx.hops ctx.policy[max_hops]: raise RuntimeError(freach hops exceeded: {ctx.trace_id}) max_retry ctx.policy[retry_budget] base_backoff ctx.policy[backoff_base_ms] for attempt in range(max_retry 1): try: result await agent_client.invoke(ctx.payload) ctx.payload[result] result return ctx.payload except Exception as e: if attempt max_retry: raise delay base_backoff * (2 ** attempt) random.uniform(0, 50) await asyncio.sleep(delay / 1000) return {}这里有两个值得说明的设计点。第一重试不是无限重试retry_budget等于 3 时总共最多尝试 4 次而且退避时间按指数增长200ms、400ms、800ms 再加一点随机抖动。随机抖动是为了避免多个 Agent 同时失败后在同一个时间点集体重试把系统打出一个新的峰值。第二重试只覆盖“调用 Agent 过程”的异常如果上一次调用已经成功执行但响应丢失重试可能会造成重复操作。这个问题我在第五节会专门讲怎么用幂等键处理。3.3 上下文裁剪器按触达半径压缩信息再来看上下文裁剪器的实现。很多刚接触 Agent-Reach 的读者想当然地以为“裁剪就是把文本截断”其实不是最有效的裁剪是按照接收方 Agent 的偏好做结构化抽取。class ContextTrimmer: def __init__(self, token_estimatorNone): if token_estimator is None: self.token_estimator lambda text: len(text) // 3 else: self.token_estimator token_estimator def trim(self, raw_context: dict, route: dict) - dict: required route.get(required_context, {}) template required.get(template, full) max_tokens required.get(max_tokens, 4096) include_fields required.get(include_fields, []) if template summary: # 只提取关键字段priority 字段在前 trimmed {} for field in include_fields: if field in raw_context: trimmed[field] raw_context[field] # 如果还是超出预算则对最长的字段做递归摘要 while self._estimate_tokens(trimmed) max_tokens: longest_key max(trimmed, keylambda k: self._estimate_tokens(trimmed[k])) trimmed[longest_key] self._summarize(trimmed[longest_key]) trimmed[_trim_note] auto-trimmed by Agent-Reach return trimmed # full 模板原样传出但会记录警告 if self._estimate_tokens(raw_context) max_tokens: return { **raw_context, _trim_warning: fcontext exceeds {max_tokens} tokens, } return raw_context def _estimate_tokens(self, data): if isinstance(data, str): return self.token_estimator(data) return sum(self._estimate_tokens(v) for v in data.values()) def _summarize(self, text: str) - str: # 生产环境可以接 LLM 或抽取式摘要 if len(text) 500: return text return text[:500] ...[truncated by Agent-Reach]_summarize里我故意留了简化的截断逻辑实际生产环境可以替换成 LLM 摘要调用。不过要注意摘要本身也是有成本的如果每一次消息传递都调一次 LLM 做摘要整体延迟和费用都会上去。建议的做法是对高频触达的 Agent预先把原始数据处理成多级摘要缓存按需取用不同精度的版本。3.4 可观测性reach trace 怎么记才有效最后是触达链路的可观测性。Agent-Reach 里每一条消息都会生成一个结构化的 trace 日志字段如下{ reach_trace_id: rt_8f2a91c, source_agent: main_router, target_agent: refund_service, route_name: refund_flow, hops: 2, context_tokens_before: 48200, context_tokens_after: 3800, decision: allowed, duration_ms: 1240, retry_count: 1 }这个日志的价值在于它能直接回答“一个问题卡在哪一跳”。比如你看到duration_ms在某条路由上突然飙升同时retry_count从 0 变成 3基本可以断定目标 Agent 出现了性能问题。又比如context_tokens_before和after的差距很小说明触达协议层没有正确裁剪需要检查路由配置里的required_context是否被目标 Agent 正确声明。4. 从零到一一次完整的 Agent-Reach 联调实录理论代码都有了接下来走一遍实际联调过程。我尽量按真实操作的顺序来包括配置、启动、压测、调参方便你照着复现。4.1 最小拓扑与角色划分这次联调我用了三个 Agent 组成的最小系统场景是“用户申请退款系统判断是否符合政策符合则退款不符合则转人工”。main_router主路由 Agent接收用户输入负责语义识别与任务分发。refund_checker退款校验 Agent查询订单状态、退款政策输出是否可退以及退款金额。payment_executor支付执行 Agent负责实际发起退款转账属于高风险 Agent必须严格保护。按照第五节的权限原则payment_executor只允许被refund_checker在特定条件下调用主路由不能直接触达它。这个限制写在路由配置里即便模型胡说八道想跨级调用Agent-Reach 也会在策略层拦下来。4.2 完整配置与启动流程routes: - name: parse_refund_request match: type: keyword_weight keywords: [退款, 退钱, refund] min_score: 0.7 target: agent:refund_checker allowed_hops: 3 - name: execute_refund match: type: exact task_type: refund_approved target: agent:payment_executor allowed_hops: 2 # 必须具备 refund_checker 的签名才允许执行 required_claims: - role: refund_checker action: approve_refund reach_policy: max_hops: 5 default_timeout_ms: 3000 retry_budget: 2 backoff_base_ms: 150 blocklist: - from: agent:main_router to: agent:payment_executor reason: 主路由不允许直接触达支付执行器启动流程很简单先启动三个 Agent 服务再启动 Agent-Reach 代理层然后把三个 Agent 的注册信息名称、能力描述、上下文偏好、回调地址写进 Agent-Reach 的服务目录。联调时所有的消息都不直接走 Agent 之间的点对点连接而是统一发给 Agent-Reach 的入口由它做路由分发。这样一个小时后整个系统的触达路径就全部处于可观测状态了。4.3 压测与参数调整记录我做了两轮压测第一轮用默认参数第二轮根据结果调参。测试场景是模拟 1000 个退款请求观察三类指标正确路由率、平均端到端耗时、越权拦截次数。参数版本正确路由率P95 耗时被拦截的越权调用上下文裁剪后平均 token默认超时 5s重试 392.4%4120ms374200调参后超时 3s重试 297.8%1860ms413850第一轮正确路由率不高的原因很有代表性refund_checker在处理部分请求时响应超过了 3 秒触发了重试但重试后的重复请求又把队列堵住形成连锁延迟。把超时从 5 秒降到 3 秒、重试预算从 3 降到 2 之后反而因为系统更早地放弃了“慢请求”整体吞吐反而上去了。越权拦截次数从 37 升到 41看起来变差了其实不是。调参前主路由有更多机会直接触达支付执行器部分请求因为路由混乱在早期就失败了压根轮不到策略层拦截。调参后路由更准确真正到达支付环节的请求更多被拦截的绝对次数自然上升。这里要看的指标是“拦截率”实际上从 3.7% 降到了 4.1% 的假象背后高风险操作总数从 1000 次降至 980 次拦截率实际是上升的。所以排查问题时一定要结合总量看比例不能只看绝对值。5. 线上踩过的坑常见问题与排查技巧说到底Agent-Reach 不是装了就能一劳永逸。上线两个月我这边遇到过不少奇怪的问题挑几个典型的说说顺带整理成速查表方便你遇到同类问题时快速定位。5.1 Agent 循环调用与幽灵消息这是多 Agent 系统里最经典的事故。现象是某条消息在几个 Agent 之间来回传递日志里全是“转发成功”但没有任何一个 Agent 真正在做有效输出直到allowed_hops耗尽才被拦截。我遇到的具体场景是主路由把一个退款请求转给refund_checkerrefund_checker觉得信息不足把消息回传给主路由补充请求主路由又没有把消息标记为“来自子 Agent 的回传”而是当成全新任务重新分发于是又转回refund_checker形成死循环。解决办法有两层第一层是严格规范消息类型回传消息必须带parent_trace_id和callback_typerequest_more_info路由层遇到这类消息只做信息合并不再走分发逻辑第二层是保留allowed_hops限制让系统在失控时能够兜底熔断。5.2 上下文越过触达半径模型开始“拼接幻觉”有一次我发现负责生成日报的 Agent 在输出里混进了一条不该出现的旧订单数据。追查 trace 后发现主路由把整段用户对话历史原样传给了日报 Agent其中包含三个月前的一次订单查询记录日报 Agent 把这个历史信息当成了本周数据直接写进了总结。这不是模型的问题是上下文触达失控。解决方式是严格执行required_context模板并且在模板里声明include_fields白名单没有出现在白名单里的字段一律丢弃。同时建议在消息里加一个_trim_note标记我在代码里就是这么做的让下游 Agent 知道自己收到的是裁剪后的摘要而不是完整原始数据。这个标记虽然不起眼但能显著减少模型把摘要当成完整数据的概率。5.3 权限边界失效的三种典型路径第一种是配置覆盖。后加的路由规则把payment_executor的required_claims漏掉了等于开了个后门。我的建议是每次配置变更都跑一遍静态校验脚本检查每个高风险 Agent 是否仍然有完整的权限声明。第二种是错误地在 Agent 的 prompt 里写“你可以调用任何工具”模型在生成函数调用时绕过了 Agent-Reach 的 API直接直连底层 HTTP 接口。这个问题得靠网络层封禁解决Agent-Reach 的策略层管不到绕过它的流量。第三种是回传消息里携带了额外的工具调用意图目标 Agent 在解析时把这个意图当成用户指令执行了。解决方案是解析消息时只能读取payload里白名单字段其他字段一律忽略。5.4 排查工具箱一页纸速查表症状可能原因排查步骤修复方案请求在多个 Agent 之间反复横跳回传消息被当成新任务分发查 reach trace 的 hops 和 route_name规范 callback_type分发表里增加回传拦截下游 Agent 输出包含无关历史信息上下文未按模板裁剪对比 context_tokens_before 和 after配置 required_context 白名单某条路由耗时突然变长目标 Agent 响应慢触发多次重试看 retry_count 与 duration_ms缩短超时时间降低重试预算高风险 Agent 被异常调用配置缺失或直连绕过查拦截日志和网络访问日志补 claims网络层加白名单 ACL重复执行业务操作首次调用成功但响应丢失重试导致重复检查业务侧对账日志引入幂等键Agent-Reach 重试时携带相同幂等 ID老实说Agent-Reach 这套方案并不惊艳它的价值在于把很多人默认“靠模型自觉”的事情变成了一套强制机制。我实际部署中体会最深的有三点第一路由决策一定要从模型手里拿回来模型理解语义就够了别让它做网络拓扑层面的选择第二上下文裁剪的收益比想象中还要大不仅省钱还直接提升了下游 Agent 的准确率第三可观测性是一切排障的前提没有 reach trace上面这些问题每一个都要靠瞎猜浪费大半天。另外还有一个非常实用的建议如果你刚开始搭不要一上来就追求全自动的 Agent 自主协作先把路由、权限、上下文模板这些确定性部分搞扎实再逐步放开自由度。这个顺序走下来系统稳定性会肉眼可见地提升。