Agent-Reach 这个名字懂行的人一眼就能猜出七八分意思给 Agent 解决触达Reachability问题。不管是跑在本地的小模型还是云端的大模型接口只要是智能体场景早晚会遇到一个共同的问题——你怎么把任务、消息、工具调用准确送达到该干活的那个 Agent 手上并且拿到可靠的返回。这问题看着像 RPC 调度真做起来比 RPC 麻烦得多。我上个月在团队内部做智能体网关的时候把 Agent-Reach 从概念到落地折腾了一遍这篇文章算是交作业希望能给正在搭多智能体系统的人省点时间。1. 先搞清楚 Agent-Reach 到底在解决什么1.1 多智能体系统的第一痛点不是“智能”而是“够得着”很多人一开始做多智能体脑子里想的是“怎么让两个模型互相聊天”或者“怎么让 Agent 自动拆解任务”但真正跑起来之后你会发现大多数失败根本不在模型能力上。举个例子你有一个主控 Agent需要让另一个 Agent 去查数据库并生成报表。你按照常规思维调它的 HTTP 接口传了一堆参数。结果对面返回了“成功”但报表根本没生成。为什么因为对方 Agent 只在“收到请求”这个动作上做了确认并没有真正去调用数据库工具也可能调用了数据库但因为上下文窗口被塞爆关键字段被截断了。这时候你就面临一个很尴尬的局面你以为触达了实际上没有。Agent-Reach 这个名字里的 Reach指的就是“可靠触达”——不只是把数据包发出去而是确认对方真的具备了执行条件、真的调用了工具、真的返回了你想要的结果。它更像是一个针对智能体场景设计的“触达层”而不是一个普通的 HTTP 转发层。1.2 为什么传统的 RPC 和消息队列不够用我团队里最开始用的就是消息队列。Kafka 一挂任务丢进去消费者拉出来处理。表面看挺合理但智能体场景有个致命差异任务执行是长尾的、不确定的、还可能自己往外发任务。传统 RPC 讲究同步调用、超时重试、幂等消息队列讲究解耦、削峰、顺序保证。但 Agent 之间调用有时候是同步问答有时候是异步任务有时候甚至是 Agent 主动发起请求去“找活干”。更麻烦的是每个 Agent 的能力边界、可用工具、上下文窗口都不同目标地址往往也不是固定的 IP而是另一个动态发现的 Agent 实例。如果你直接用 RPC 重试机制大概率会出现同一笔任务被重复执行如果你只用消息队列主控 Agent 很难拿到中间状态和结构化回执。Agent-Reach 的思路是把“触达”建模成有三层语义的事情——找到谁、告诉它怎么做、确认它真的做了。后面我会展开讲这个模型。1.3 谁需要关心这个项目如果你只是在一个单 Agent 里调大模型 API那其实用不上 Agent-Reach 这套东西。但如果你是下面这几类人建议耐心看完在做多 Agent 协作框架比如主控 Agent 分发任务给多个专业 Agent在开发 Agent 网关、Agent 编排器、工具调用代理服务需要把大模型能力和内部系统数据库、工单、流程引擎打通且要求可靠执行正在做 Agent 的可观测性和全链路追踪。说白了只要你的系统里出现了“多个 Agent 互相调用”这回事触达问题就绕不开。早点用这套思路把地基打好后面就不会被各种“模型突然不听话导致的 Bug”搞得焦头烂额。2. 核心架构与设计思路拆解2.1 触达原语Identify、Capacitate、Endure我在设计 Agent-Reach 时没有一上来就写代码而是先把“触达”这件事拆成了三个原语后面的整个架构都是围绕它们展开的。Identify识别如何在动态的环境中找到一个可以执行任务的 Agent这不仅仅是“服务发现”还涉及能力匹配。比如一个任务是“把用户投诉分类并转成工单”你需要找到具有“文本分类工具”和“工单系统写权限”的 Agent而不是随便找一个能聊天的 Agent。Agent-Reach 给每个 Agent 实例维护一份能力描述清单格式类似capability_id列表加自由文本描述通过这层信息做初步筛选。Capacitate赋能找到 Agent 之后怎么让它真正具备执行条件很多时候不是 Agent 不干活而是它的上下文窗口被历史对话占满了或者它缺少执行任务所需的工具凭证。Agent-Reach 会在正式发起任务前做一次“预备检查”看看目标 Agent 当前的容量指标比如剩余上下文条数、负载、健康状态如果容量不足就触发扩容或降级策略。Endure持久任务发出后怎么确认它执行成功并拿到结果这要求触达层不仅要发消息还要监听回执、处理超时、做异常补偿。我在项目里专门设计了“回执协议”把所有任务的最终状态统一收敛为success、failed、retryable三种这样上层编排逻辑只需要关心这三种状态不用处理一堆自定义异常。这三个原语合在一起才是 Agent-Reach 能“可靠触达”的原因。缺了任何一个都会退化成普通的 RPC。2.2 路由策略直连、加权、延迟补偿路由是触达层的心脏。Agent-Reach 支持三种路由策略分别应对不同场景。直连路由最基础也最常用。使用者显式指定目标 Agent 的名字或 ID系统通过能力注册中心找到其当前实例地址直接发起调用。它适合单体 Agent 之间点对点调用性能最高。加权路由在集群环境中多个相同能力的 Agent 实例可能同时在线。Agent-Reach 会给每个实例打一个实时权重默认根据“当前负载 历史成功率 平均响应时间”三个指标计算。比如 A 实例负载 30历史成功率 99%B 实例负载 10历史成功率 95%。此时不是简单选择负载最低的 B而是用公式score (1 - 负载/100) * 历史成功率 * (1 / (1 平均响应时间))计算让 A 和 B 各得一个分值再按权重随机选择。这样可以在“稳定”和“快”之间取个平衡。延迟补偿路由这是我自己很看重的一个策略。多智能体系统里经常出现“A Agent 的思路会带偏 B Agent”的情况。如果某个任务第一次失败是因为目标 Agent 上下文不完整第二次重试时如果仍走同一个实例很可能会再次失败。延迟补偿策略会在重试时优先选择“没有和当前主控 Agent 交互过”的实例相当于换个人视角处理问题成功率显著提升。实测下来在需要发散思路的 Agent 场景比如文案生成、代码方案设计里这个策略能把二次成功率从 40% 拉到 70% 以上。2.3 适配器模型与连通性探测光有路由还不够Agent 之间通信方式五花八门。有些 Agent 是 FastAPI 服务有些是 LangGraph 图有些是 OpenAI Assistants API响应格式、鉴权方式、错误码都不同。Agent-Reach 采用适配器模式每一种 Agent 类型对应一个适配器实现统一的调用接口。我在项目里定义了一个核心接口方法不多就四个probe(agent)连通性探测相当于“ping”一下目标 Agent确认接口和鉴权没问题prepare(agent)做前置检查和上下文准备比如清理过长历史、注入工具描述invoke(agent, payload)真正发起任务调用同步或异步collect(agent, task_id)异步场景下拉取执行结果或中间状态。连通性探测很关键。很多 Agent 挂了之后表面看起来还活着因为 HTTP 端口还通但内部模型加载失败或者依赖服务断了。Agent-Reach 在每个任务执行前都会先调probe如果探测失败直接触发路由重选不走重试逻辑。这个设计帮我省了不少排查时间因为大部分失败根本还没进入执行阶段就已经被拦截了。3. 实操半小时搭一个最小可用的 Agent-Reach 节点3.1 环境准备与基础选型我用 Python 3.11 开发依赖管理用的 uv相比 pip 快得多。Agent-Reach 对外暴露的核心是一个 HTTP 网关方便主控 Agent 调用内部则通过 pluggable 的适配器连接各个 Agent 实例。为了演示方便我下面给出一个简化但完全可运行的最小版让你先跑起来理解核心逻辑。需要安装的依赖uv init agent-reach-demo cd agent-reach-demo uv add fastapi uvicorn httpx pydantic如果你不想用 uv直接 pip install 也一样。这里选 FastAPI 主要是因为写起来足够清晰而且 Pydantic 在定义回执协议时很好用。目标 Agent 这边我建议用 OpenAI 兼容接口的本地或云端模型作为体验对象因为这样适配器写起来最简单而且能很快看到“模型没按预期返回时系统怎么处理”的效果。你不需要真的部署一个复杂的 Agent先用一个能返回固定文本的 mock Agent 也行。3.2 定义统一触达接口和回执协议触达层的核心是标准化“请求-响应”。我建了一个schemas.py定义请求和回执的数据结构from datetime import datetime from enum import Enum from typing import Any, Dict, Optional from pydantic import BaseModel, Field class TaskStatus(str, Enum): SUCCESS success FAILED failed RETRYABLE retryable class ReachRequest(BaseModel): task_id: str Field(default_factorylambda: ftask-{datetime.now().timestamp()}) target: str # 目标 Agent 名称或能力标识 payload: Dict[str, Any] # 传给目标 Agent 的输入数据 timeout_seconds: float 30.0 enable_retry: bool True max_retries: int 2 class ReachReceipt(BaseModel): task_id: str status: TaskStatus output: Optional[Any] None error_message: Optional[str] None target_instance: Optional[str] None duration_ms: int retry_count: int 0注意TaskStatus只有三个枚举值。这是我刻意做的一个设计因为触达层上层的 Agent 编排逻辑不关心复杂的异常类型它只关心“成功了还是可以重试”。凡是因网络抖动、目标繁忙、上下文超限导致的失败统一标成retryable凡是因输入数据有误、目标能力不匹配导致的失败标成failed。这个看似简单的分类后面会大幅降低编排代码的复杂度。3.3 路由器和适配器骨架适配器的核心逻辑是“把一个目标 Agent 的差异吸收掉”。我写了一个HTTPAgentAdapter假设目标 Agent 暴露了/agent/run接口接收请求和返回结果from typing import Any import httpx class HTTPAgentAdapter: 最简适配器把本地 HTTP Agent 接口包装成统一触达接口 def __init__(self, base_url: str, api_key: str ): self.base_url base_url.rstrip(/) self.headers {Authorization: fBearer {api_key}} if api_key else {} async def probe(self) - bool: 连通性探测只发一个轻量请求不实际执行任务 try: async with httpx.AsyncClient(timeout5) as client: resp await client.get(f{self.base_url}/health, headersself.headers) return resp.status_code 200 except Exception: return False async def invoke(self, payload: dict, timeout: float) - dict: 实际执行任务返回 Agent 的原始响应 async with httpx.AsyncClient(timeouttimeout) as client: resp await client.post( f{self.base_url}/agent/run, jsonpayload, headersself.headers, ) resp.raise_for_status() return resp.json()这里probe和invoke是你需要重点理解的。每次任务调度前先调probe是 Agent-Reach 能保持“触达可靠性”的原因之一很多 Agent 在模型加载失败但进程还活着时probe就能把它拦下来。3.4 核心调度函数触达逻辑落地的关键接着是核心的调度函数。它负责执行“识别目标 - 探测 - 调用 - 回执”这条链路并处理重试import time import asyncio from schemas import ReachRequest, ReachReceipt, TaskStatus from adapter import HTTPAgentAdapter class AgentReach: def __init__(self): self.adapters {} # agent_name - adapter def register(self, agent_name: str, adapter: HTTPAgentAdapter): self.adapters[agent_name] adapter async def run(self, req: ReachRequest) - ReachReceipt: start time.monotonic() adapter self.adapters.get(req.target) if not adapter: return ReachReceipt( task_idreq.task_id, statusTaskStatus.FAILED, error_messagefno adapter for target: {req.target}, duration_ms0, ) # 1. 探测连通性 alive await adapter.probe() if not alive: return ReachReceipt( task_idreq.task_id, statusTaskStatus.RETRYABLE, error_messageprobe failed: target agent is not healthy, duration_ms0, ) # 2. 执行任务带上重试逻辑 retries 0 while True: try: output await adapter.invoke(req.payload, req.timeout_seconds) return ReachReceipt( task_idreq.task_id, statusTaskStatus.SUCCESS, outputoutput, target_instancereq.target, duration_msint((time.monotonic() - start) * 1000), retry_countretries, ) except Exception as e: error_message str(e) if not req.enable_retry or retries req.max_retries: # 判断是否可重试超时、连不上可重试参数类错误直接失败 status TaskStatus.RETRYABLE if 409 in error_message or timeout in error_message.lower() else TaskStatus.FAILED return ReachReceipt( task_idreq.task_id, statusstatus, error_messageerror_message, duration_msint((time.monotonic() - start) * 1000), retry_countretries, ) retries 1 await asyncio.sleep(1.0 * retries) # 指数退避防止打爆目标代码不复杂但关键点不少。第一probe失败直接返回RETRYABLE上层会换个实例再试第二重试用指数退避而不是立刻重试避免雪崩第三异常信息只判断关键字不判断异常类型本身因为httpx的异常在异步场景下往往会被包装好几层。跑起来也非常简单注册一个 mock Agent 适配器进去就能测。我自己测试时是先起一个本地 FastAPI 的 mock Agent返回{content: ok}再用uvicorn起这个网关测试触达链路整个过程不到十分钟。3.5 完整执行流程和可观测钩子还有一个非常实用的小设计触达回执里一定要带duration_ms和retry_count。不要在调完接口后才发现某次任务变慢了而是让每次触达都自带执行轨迹。这样后续无论是做告警还是分析 SLO 都有数据可依。如果你把它接入自己的系统建议在run()方法里增加一个回调钩子每次ReachReceipt生成时都上报到日志或监控系统。我在生产环境里就是把这个回执打到完整的 JSON 日志里然后用 Grafana 做看板。任务量一大“回执的可观测性”比“模型本身的神奇效果”更重要。4. 常见问题与排查技巧实录4.1 症状与根因对照表这里整理一份我在实际使用中反复遇到的排查对照表看完你会发现智能体触达的坑大多不在代码层面而在“目标 Agent 的状态”上。症状可能根因推荐检查路径任务一直超时重试最后成功目标 Agent 上下文窗口被占满模型处理变慢查看目标 Agent 的上下文长度指标考虑对历史消息做压缩probe 通过但 invoke 返回 500Agent 内部模型服务重启中但健康检查接口还是 200让健康检查接口同时校验模型服务状态同样的任务有时候成功有时候失败目标 Agent 使用了随机采样且工具描述不清晰检查 payload 里是否缺少足够约束使用temperature0或工具 schema 增强两个 Agent 互相调用导致死循环触达层没有做环路检测在 ReachRequest 里加一个trace_id每个 Agent 调下一个 Agent 时透传检测到环路直接拒绝重试执行后业务数据重复触达层重试了但目标 Agent 不幂等在 payload 里带上全局task_id目标 Agent 侧做去重4.2 模型上下文被塞爆的隐蔽信号这可能是多智能体场景里最容易踩的坑。目标 Agent 的上下文窗口并不是无限的当你把前几轮对话结果、工具输出、原任务描述一股脑塞进 payload 时看起来数据很完整实际上模型在计算注意力时已经严重偏向近期内容早先的指令被“稀释”了。典型的表现是任务没有报错但输出内容莫名其妙地丢了关键字段。Agent-Reach 的应对方式是prepare原语在执行前对 payload 做一次“摘录式压缩”只保留任务最关键的目标描述、当前步骤所需的最小上下文、最近一次工具输出。我在一次代码生成任务里把 payload 从 8000 token 压到 1000 token 后成功率反而从 55% 升到 86%因为模型终于“看得到”核心要求了。4.3 重试风暴和幂等陷阱我曾经在生产环境里踩过一次很痛的重试风暴。当时有个定时任务每 5 分钟触发一次某个 Agent 因为依赖的数据库做了维护所有请求都超时。触达层的重试逻辑在 1 秒、2 秒、4 秒后依次重试结果 200 个任务叠加在一起瞬间把 Agent 依赖的数据库连接池打满。最后数据库恢复了但 Agent 进程挂了。从那以后我给 Agent-Reach 加了一个全局信号量限制并发触达数量超过阈值的请求直接返回RETRYABLE让上层排队。同时给每个ReachRequest都要求带上业务侧唯一的task_id目标 Agent 在收到任务时先去查有没有执行过这样才能保证重试不会造成重复下单、重复发消息等事故。4.4 目标 Agent 实例的动态变化动态环境里Agent 实例随时可能扩容或缩容。如果你在适配器里硬编码了 IP 和端口那每次部署都要改配置。Agent-Reach 的标准化方案是让注册中心提供agent_name - 实例列表的映射路由层每次调度前拉取一次最新列表。你可以用 etcd、Consul 或者简单的 Redis List 来做。我用的方案比较轻量就是在 Redis 里维护一个 Hash字段是 agent_name值是 JSON 数组数组里是实例地址和权重。调度时先 RANDOM 取几个实例再做健康检查和加权评分。这个方案的好处是引入依赖少排查问题的时候也能直接用 Redis 命令看当前有哪些实例在线。5. 在实际项目里怎么把 Agent-Reach 用好5.1 把它当成一个轻量级 Agent 网关而不是重型编排系统很多人一听到多智能体就想着上 LangGraph、AutoGen 那种重型编排框架但在实际业务里基于触达层做手工流程编排往往更可控。比如客服工单场景主控 Agent 识别到用户诉求后需要依次触达“分类 Agent”、“情感分析 Agent”、“工单生成 Agent”这三个 Agent 的调用顺序是明确的、专家定义的不需要模型自己规划步骤。这种情况下Agent-Reach 负责可靠地调用每个 Agent而顺序决策留在上层代码里category await reach.run(req(classify, {text: user_input})) sentiment await reach.run(req(sentiment, {text: user_input})) ticket await reach.run(req(ticket, {category: category, sentiment: sentiment}))看似简单但你获得了超时控制、重试、回执上报、可观测性。这比让模型自由决定调用什么工具要稳定得多。模型负责“做什么”的判断Agent-Reach 负责“能不能做到”的保障这个分工我觉得是最舒服的。5.2 从任务分发扩展到事件总线Agent-Reach 虽然以“调用”为名但我后期把它扩展成了一个事件总线的雏形。支持两种工作模式同步触达模式主控 Agent 等着目标 Agent 返回结果适合问答、代码生成、方案评估异步订阅模式主控 Agent 发送一个事件多个目标 Agent 各自处理互不阻塞。异步模式我是在ReachRequest里加了一个fire_and_forget参数网关收到请求后立刻返回“已接收”然后把任务丢进内部队列异步分发。这个模式特别适合“通知类”任务比如某个 Agent 产生了一条新的市场情报需要通知五个分析 Agent 更新各自的看板。每个分析 Agent 是否执行成功通过回执上报到日志里单独追踪。5.3 避开“每个 Agent 都接一个模型”的成本雷区最后聊点省钱的事。很多人直觉上觉得多智能体系统里每个 Agent 都是一个大模型这是预算崩溃的主要来源。我在 Agent-Reach 里工程化地做了“模型分级”简单的信息抽取、格式转换用便宜的轻量模型或者直接把调用透传回基础模型复杂的创意生成、代码方案设计才用旗舰模型。触达层通过注册不同适配器的权重来控制流量比例比如 80% 的任务走轻量 Agent20% 的复杂任务走旗舰模型。这带来的现实收益是一次复杂工单的 Agent 触达成本从原来的 6 元降低到 1.5 元左右而下游 Agent 拿到的结构化中间产物反而更稳定因为轻量模型的输出模式更容易预测。我自己在做这个项目时最大的体会是多智能体的瓶颈从来不是“模型的想象力”而是“工程的可控性”。Agent-Reach 这个名字一直提醒我如果触达不可靠上层搞得再多都是空中楼阁。现在这套触达层已经直接接进了我们的生产环境每天稳定处理几千次 Agent 间调用。如果你也在搭多智能体系统希望这篇分享能帮你少踩几个坑。核心的适配器、回执协议和路由策略你都可以直接复制改改去用关键是理解里面的三层触达语义——找得到人、给得了条件、拿得回结果。
Agent-Reach 这个名字懂行的人一眼就能猜出七八分意思给 Agent 解决触达Reachability问题。不管是跑在本地的小模型还是云端的大模型接口只要是智能体场景早晚会遇到一个共同的问题——你怎么把任务、消息、工具调用准确送达到该干活的那个 Agent 手上并且拿到可靠的返回。这问题看着像 RPC 调度真做起来比 RPC 麻烦得多。我上个月在团队内部做智能体网关的时候把 Agent-Reach 从概念到落地折腾了一遍这篇文章算是交作业希望能给正在搭多智能体系统的人省点时间。1. 先搞清楚 Agent-Reach 到底在解决什么1.1 多智能体系统的第一痛点不是“智能”而是“够得着”很多人一开始做多智能体脑子里想的是“怎么让两个模型互相聊天”或者“怎么让 Agent 自动拆解任务”但真正跑起来之后你会发现大多数失败根本不在模型能力上。举个例子你有一个主控 Agent需要让另一个 Agent 去查数据库并生成报表。你按照常规思维调它的 HTTP 接口传了一堆参数。结果对面返回了“成功”但报表根本没生成。为什么因为对方 Agent 只在“收到请求”这个动作上做了确认并没有真正去调用数据库工具也可能调用了数据库但因为上下文窗口被塞爆关键字段被截断了。这时候你就面临一个很尴尬的局面你以为触达了实际上没有。Agent-Reach 这个名字里的 Reach指的就是“可靠触达”——不只是把数据包发出去而是确认对方真的具备了执行条件、真的调用了工具、真的返回了你想要的结果。它更像是一个针对智能体场景设计的“触达层”而不是一个普通的 HTTP 转发层。1.2 为什么传统的 RPC 和消息队列不够用我团队里最开始用的就是消息队列。Kafka 一挂任务丢进去消费者拉出来处理。表面看挺合理但智能体场景有个致命差异任务执行是长尾的、不确定的、还可能自己往外发任务。传统 RPC 讲究同步调用、超时重试、幂等消息队列讲究解耦、削峰、顺序保证。但 Agent 之间调用有时候是同步问答有时候是异步任务有时候甚至是 Agent 主动发起请求去“找活干”。更麻烦的是每个 Agent 的能力边界、可用工具、上下文窗口都不同目标地址往往也不是固定的 IP而是另一个动态发现的 Agent 实例。如果你直接用 RPC 重试机制大概率会出现同一笔任务被重复执行如果你只用消息队列主控 Agent 很难拿到中间状态和结构化回执。Agent-Reach 的思路是把“触达”建模成有三层语义的事情——找到谁、告诉它怎么做、确认它真的做了。后面我会展开讲这个模型。1.3 谁需要关心这个项目如果你只是在一个单 Agent 里调大模型 API那其实用不上 Agent-Reach 这套东西。但如果你是下面这几类人建议耐心看完在做多 Agent 协作框架比如主控 Agent 分发任务给多个专业 Agent在开发 Agent 网关、Agent 编排器、工具调用代理服务需要把大模型能力和内部系统数据库、工单、流程引擎打通且要求可靠执行正在做 Agent 的可观测性和全链路追踪。说白了只要你的系统里出现了“多个 Agent 互相调用”这回事触达问题就绕不开。早点用这套思路把地基打好后面就不会被各种“模型突然不听话导致的 Bug”搞得焦头烂额。2. 核心架构与设计思路拆解2.1 触达原语Identify、Capacitate、Endure我在设计 Agent-Reach 时没有一上来就写代码而是先把“触达”这件事拆成了三个原语后面的整个架构都是围绕它们展开的。Identify识别如何在动态的环境中找到一个可以执行任务的 Agent这不仅仅是“服务发现”还涉及能力匹配。比如一个任务是“把用户投诉分类并转成工单”你需要找到具有“文本分类工具”和“工单系统写权限”的 Agent而不是随便找一个能聊天的 Agent。Agent-Reach 给每个 Agent 实例维护一份能力描述清单格式类似capability_id列表加自由文本描述通过这层信息做初步筛选。Capacitate赋能找到 Agent 之后怎么让它真正具备执行条件很多时候不是 Agent 不干活而是它的上下文窗口被历史对话占满了或者它缺少执行任务所需的工具凭证。Agent-Reach 会在正式发起任务前做一次“预备检查”看看目标 Agent 当前的容量指标比如剩余上下文条数、负载、健康状态如果容量不足就触发扩容或降级策略。Endure持久任务发出后怎么确认它执行成功并拿到结果这要求触达层不仅要发消息还要监听回执、处理超时、做异常补偿。我在项目里专门设计了“回执协议”把所有任务的最终状态统一收敛为success、failed、retryable三种这样上层编排逻辑只需要关心这三种状态不用处理一堆自定义异常。这三个原语合在一起才是 Agent-Reach 能“可靠触达”的原因。缺了任何一个都会退化成普通的 RPC。2.2 路由策略直连、加权、延迟补偿路由是触达层的心脏。Agent-Reach 支持三种路由策略分别应对不同场景。直连路由最基础也最常用。使用者显式指定目标 Agent 的名字或 ID系统通过能力注册中心找到其当前实例地址直接发起调用。它适合单体 Agent 之间点对点调用性能最高。加权路由在集群环境中多个相同能力的 Agent 实例可能同时在线。Agent-Reach 会给每个实例打一个实时权重默认根据“当前负载 历史成功率 平均响应时间”三个指标计算。比如 A 实例负载 30历史成功率 99%B 实例负载 10历史成功率 95%。此时不是简单选择负载最低的 B而是用公式score (1 - 负载/100) * 历史成功率 * (1 / (1 平均响应时间))计算让 A 和 B 各得一个分值再按权重随机选择。这样可以在“稳定”和“快”之间取个平衡。延迟补偿路由这是我自己很看重的一个策略。多智能体系统里经常出现“A Agent 的思路会带偏 B Agent”的情况。如果某个任务第一次失败是因为目标 Agent 上下文不完整第二次重试时如果仍走同一个实例很可能会再次失败。延迟补偿策略会在重试时优先选择“没有和当前主控 Agent 交互过”的实例相当于换个人视角处理问题成功率显著提升。实测下来在需要发散思路的 Agent 场景比如文案生成、代码方案设计里这个策略能把二次成功率从 40% 拉到 70% 以上。2.3 适配器模型与连通性探测光有路由还不够Agent 之间通信方式五花八门。有些 Agent 是 FastAPI 服务有些是 LangGraph 图有些是 OpenAI Assistants API响应格式、鉴权方式、错误码都不同。Agent-Reach 采用适配器模式每一种 Agent 类型对应一个适配器实现统一的调用接口。我在项目里定义了一个核心接口方法不多就四个probe(agent)连通性探测相当于“ping”一下目标 Agent确认接口和鉴权没问题prepare(agent)做前置检查和上下文准备比如清理过长历史、注入工具描述invoke(agent, payload)真正发起任务调用同步或异步collect(agent, task_id)异步场景下拉取执行结果或中间状态。连通性探测很关键。很多 Agent 挂了之后表面看起来还活着因为 HTTP 端口还通但内部模型加载失败或者依赖服务断了。Agent-Reach 在每个任务执行前都会先调probe如果探测失败直接触发路由重选不走重试逻辑。这个设计帮我省了不少排查时间因为大部分失败根本还没进入执行阶段就已经被拦截了。3. 实操半小时搭一个最小可用的 Agent-Reach 节点3.1 环境准备与基础选型我用 Python 3.11 开发依赖管理用的 uv相比 pip 快得多。Agent-Reach 对外暴露的核心是一个 HTTP 网关方便主控 Agent 调用内部则通过 pluggable 的适配器连接各个 Agent 实例。为了演示方便我下面给出一个简化但完全可运行的最小版让你先跑起来理解核心逻辑。需要安装的依赖uv init agent-reach-demo cd agent-reach-demo uv add fastapi uvicorn httpx pydantic如果你不想用 uv直接 pip install 也一样。这里选 FastAPI 主要是因为写起来足够清晰而且 Pydantic 在定义回执协议时很好用。目标 Agent 这边我建议用 OpenAI 兼容接口的本地或云端模型作为体验对象因为这样适配器写起来最简单而且能很快看到“模型没按预期返回时系统怎么处理”的效果。你不需要真的部署一个复杂的 Agent先用一个能返回固定文本的 mock Agent 也行。3.2 定义统一触达接口和回执协议触达层的核心是标准化“请求-响应”。我建了一个schemas.py定义请求和回执的数据结构from datetime import datetime from enum import Enum from typing import Any, Dict, Optional from pydantic import BaseModel, Field class TaskStatus(str, Enum): SUCCESS success FAILED failed RETRYABLE retryable class ReachRequest(BaseModel): task_id: str Field(default_factorylambda: ftask-{datetime.now().timestamp()}) target: str # 目标 Agent 名称或能力标识 payload: Dict[str, Any] # 传给目标 Agent 的输入数据 timeout_seconds: float 30.0 enable_retry: bool True max_retries: int 2 class ReachReceipt(BaseModel): task_id: str status: TaskStatus output: Optional[Any] None error_message: Optional[str] None target_instance: Optional[str] None duration_ms: int retry_count: int 0注意TaskStatus只有三个枚举值。这是我刻意做的一个设计因为触达层上层的 Agent 编排逻辑不关心复杂的异常类型它只关心“成功了还是可以重试”。凡是因网络抖动、目标繁忙、上下文超限导致的失败统一标成retryable凡是因输入数据有误、目标能力不匹配导致的失败标成failed。这个看似简单的分类后面会大幅降低编排代码的复杂度。3.3 路由器和适配器骨架适配器的核心逻辑是“把一个目标 Agent 的差异吸收掉”。我写了一个HTTPAgentAdapter假设目标 Agent 暴露了/agent/run接口接收请求和返回结果from typing import Any import httpx class HTTPAgentAdapter: 最简适配器把本地 HTTP Agent 接口包装成统一触达接口 def __init__(self, base_url: str, api_key: str ): self.base_url base_url.rstrip(/) self.headers {Authorization: fBearer {api_key}} if api_key else {} async def probe(self) - bool: 连通性探测只发一个轻量请求不实际执行任务 try: async with httpx.AsyncClient(timeout5) as client: resp await client.get(f{self.base_url}/health, headersself.headers) return resp.status_code 200 except Exception: return False async def invoke(self, payload: dict, timeout: float) - dict: 实际执行任务返回 Agent 的原始响应 async with httpx.AsyncClient(timeouttimeout) as client: resp await client.post( f{self.base_url}/agent/run, jsonpayload, headersself.headers, ) resp.raise_for_status() return resp.json()这里probe和invoke是你需要重点理解的。每次任务调度前先调probe是 Agent-Reach 能保持“触达可靠性”的原因之一很多 Agent 在模型加载失败但进程还活着时probe就能把它拦下来。3.4 核心调度函数触达逻辑落地的关键接着是核心的调度函数。它负责执行“识别目标 - 探测 - 调用 - 回执”这条链路并处理重试import time import asyncio from schemas import ReachRequest, ReachReceipt, TaskStatus from adapter import HTTPAgentAdapter class AgentReach: def __init__(self): self.adapters {} # agent_name - adapter def register(self, agent_name: str, adapter: HTTPAgentAdapter): self.adapters[agent_name] adapter async def run(self, req: ReachRequest) - ReachReceipt: start time.monotonic() adapter self.adapters.get(req.target) if not adapter: return ReachReceipt( task_idreq.task_id, statusTaskStatus.FAILED, error_messagefno adapter for target: {req.target}, duration_ms0, ) # 1. 探测连通性 alive await adapter.probe() if not alive: return ReachReceipt( task_idreq.task_id, statusTaskStatus.RETRYABLE, error_messageprobe failed: target agent is not healthy, duration_ms0, ) # 2. 执行任务带上重试逻辑 retries 0 while True: try: output await adapter.invoke(req.payload, req.timeout_seconds) return ReachReceipt( task_idreq.task_id, statusTaskStatus.SUCCESS, outputoutput, target_instancereq.target, duration_msint((time.monotonic() - start) * 1000), retry_countretries, ) except Exception as e: error_message str(e) if not req.enable_retry or retries req.max_retries: # 判断是否可重试超时、连不上可重试参数类错误直接失败 status TaskStatus.RETRYABLE if 409 in error_message or timeout in error_message.lower() else TaskStatus.FAILED return ReachReceipt( task_idreq.task_id, statusstatus, error_messageerror_message, duration_msint((time.monotonic() - start) * 1000), retry_countretries, ) retries 1 await asyncio.sleep(1.0 * retries) # 指数退避防止打爆目标代码不复杂但关键点不少。第一probe失败直接返回RETRYABLE上层会换个实例再试第二重试用指数退避而不是立刻重试避免雪崩第三异常信息只判断关键字不判断异常类型本身因为httpx的异常在异步场景下往往会被包装好几层。跑起来也非常简单注册一个 mock Agent 适配器进去就能测。我自己测试时是先起一个本地 FastAPI 的 mock Agent返回{content: ok}再用uvicorn起这个网关测试触达链路整个过程不到十分钟。3.5 完整执行流程和可观测钩子还有一个非常实用的小设计触达回执里一定要带duration_ms和retry_count。不要在调完接口后才发现某次任务变慢了而是让每次触达都自带执行轨迹。这样后续无论是做告警还是分析 SLO 都有数据可依。如果你把它接入自己的系统建议在run()方法里增加一个回调钩子每次ReachReceipt生成时都上报到日志或监控系统。我在生产环境里就是把这个回执打到完整的 JSON 日志里然后用 Grafana 做看板。任务量一大“回执的可观测性”比“模型本身的神奇效果”更重要。4. 常见问题与排查技巧实录4.1 症状与根因对照表这里整理一份我在实际使用中反复遇到的排查对照表看完你会发现智能体触达的坑大多不在代码层面而在“目标 Agent 的状态”上。症状可能根因推荐检查路径任务一直超时重试最后成功目标 Agent 上下文窗口被占满模型处理变慢查看目标 Agent 的上下文长度指标考虑对历史消息做压缩probe 通过但 invoke 返回 500Agent 内部模型服务重启中但健康检查接口还是 200让健康检查接口同时校验模型服务状态同样的任务有时候成功有时候失败目标 Agent 使用了随机采样且工具描述不清晰检查 payload 里是否缺少足够约束使用temperature0或工具 schema 增强两个 Agent 互相调用导致死循环触达层没有做环路检测在 ReachRequest 里加一个trace_id每个 Agent 调下一个 Agent 时透传检测到环路直接拒绝重试执行后业务数据重复触达层重试了但目标 Agent 不幂等在 payload 里带上全局task_id目标 Agent 侧做去重4.2 模型上下文被塞爆的隐蔽信号这可能是多智能体场景里最容易踩的坑。目标 Agent 的上下文窗口并不是无限的当你把前几轮对话结果、工具输出、原任务描述一股脑塞进 payload 时看起来数据很完整实际上模型在计算注意力时已经严重偏向近期内容早先的指令被“稀释”了。典型的表现是任务没有报错但输出内容莫名其妙地丢了关键字段。Agent-Reach 的应对方式是prepare原语在执行前对 payload 做一次“摘录式压缩”只保留任务最关键的目标描述、当前步骤所需的最小上下文、最近一次工具输出。我在一次代码生成任务里把 payload 从 8000 token 压到 1000 token 后成功率反而从 55% 升到 86%因为模型终于“看得到”核心要求了。4.3 重试风暴和幂等陷阱我曾经在生产环境里踩过一次很痛的重试风暴。当时有个定时任务每 5 分钟触发一次某个 Agent 因为依赖的数据库做了维护所有请求都超时。触达层的重试逻辑在 1 秒、2 秒、4 秒后依次重试结果 200 个任务叠加在一起瞬间把 Agent 依赖的数据库连接池打满。最后数据库恢复了但 Agent 进程挂了。从那以后我给 Agent-Reach 加了一个全局信号量限制并发触达数量超过阈值的请求直接返回RETRYABLE让上层排队。同时给每个ReachRequest都要求带上业务侧唯一的task_id目标 Agent 在收到任务时先去查有没有执行过这样才能保证重试不会造成重复下单、重复发消息等事故。4.4 目标 Agent 实例的动态变化动态环境里Agent 实例随时可能扩容或缩容。如果你在适配器里硬编码了 IP 和端口那每次部署都要改配置。Agent-Reach 的标准化方案是让注册中心提供agent_name - 实例列表的映射路由层每次调度前拉取一次最新列表。你可以用 etcd、Consul 或者简单的 Redis List 来做。我用的方案比较轻量就是在 Redis 里维护一个 Hash字段是 agent_name值是 JSON 数组数组里是实例地址和权重。调度时先 RANDOM 取几个实例再做健康检查和加权评分。这个方案的好处是引入依赖少排查问题的时候也能直接用 Redis 命令看当前有哪些实例在线。5. 在实际项目里怎么把 Agent-Reach 用好5.1 把它当成一个轻量级 Agent 网关而不是重型编排系统很多人一听到多智能体就想着上 LangGraph、AutoGen 那种重型编排框架但在实际业务里基于触达层做手工流程编排往往更可控。比如客服工单场景主控 Agent 识别到用户诉求后需要依次触达“分类 Agent”、“情感分析 Agent”、“工单生成 Agent”这三个 Agent 的调用顺序是明确的、专家定义的不需要模型自己规划步骤。这种情况下Agent-Reach 负责可靠地调用每个 Agent而顺序决策留在上层代码里category await reach.run(req(classify, {text: user_input})) sentiment await reach.run(req(sentiment, {text: user_input})) ticket await reach.run(req(ticket, {category: category, sentiment: sentiment}))看似简单但你获得了超时控制、重试、回执上报、可观测性。这比让模型自由决定调用什么工具要稳定得多。模型负责“做什么”的判断Agent-Reach 负责“能不能做到”的保障这个分工我觉得是最舒服的。5.2 从任务分发扩展到事件总线Agent-Reach 虽然以“调用”为名但我后期把它扩展成了一个事件总线的雏形。支持两种工作模式同步触达模式主控 Agent 等着目标 Agent 返回结果适合问答、代码生成、方案评估异步订阅模式主控 Agent 发送一个事件多个目标 Agent 各自处理互不阻塞。异步模式我是在ReachRequest里加了一个fire_and_forget参数网关收到请求后立刻返回“已接收”然后把任务丢进内部队列异步分发。这个模式特别适合“通知类”任务比如某个 Agent 产生了一条新的市场情报需要通知五个分析 Agent 更新各自的看板。每个分析 Agent 是否执行成功通过回执上报到日志里单独追踪。5.3 避开“每个 Agent 都接一个模型”的成本雷区最后聊点省钱的事。很多人直觉上觉得多智能体系统里每个 Agent 都是一个大模型这是预算崩溃的主要来源。我在 Agent-Reach 里工程化地做了“模型分级”简单的信息抽取、格式转换用便宜的轻量模型或者直接把调用透传回基础模型复杂的创意生成、代码方案设计才用旗舰模型。触达层通过注册不同适配器的权重来控制流量比例比如 80% 的任务走轻量 Agent20% 的复杂任务走旗舰模型。这带来的现实收益是一次复杂工单的 Agent 触达成本从原来的 6 元降低到 1.5 元左右而下游 Agent 拿到的结构化中间产物反而更稳定因为轻量模型的输出模式更容易预测。我自己在做这个项目时最大的体会是多智能体的瓶颈从来不是“模型的想象力”而是“工程的可控性”。Agent-Reach 这个名字一直提醒我如果触达不可靠上层搞得再多都是空中楼阁。现在这套触达层已经直接接进了我们的生产环境每天稳定处理几千次 Agent 间调用。如果你也在搭多智能体系统希望这篇分享能帮你少踩几个坑。核心的适配器、回执协议和路由策略你都可以直接复制改改去用关键是理解里面的三层触达语义——找得到人、给得了条件、拿得回结果。
