做 Agent 项目做到第三个年头我发现一个特别容易被忽略的问题Agent 本身不难跑通难的是让 Agent 真正“Reach”——触达用户、触达系统、触达另一批 Agent。这也是我为什么在搭 Agent-Reach 时把名字钉死在“触达”这两个字上。Agent-Reach 在我这里不是一个聊天机器人也不是某个单独模型的封装而是一层统一触达与治理网关所有智能体以标准端点注册所有调用方通过统一入口完成发现、路由、编排、限流与审计。它能解决什么问题一句话避免你的 Agent 集群变成一堆谁也找不到、谁也不敢用的“模型岛”。适合正在搭多 Agent 平台、或者准备把 Agent 能力开放给内部业务方的团队参考。1. 为什么需要 Agent-ReachAgent 不是模型而是业务单元1.1 智能体集群的真正瓶颈在“送达”单体 Bot 时代一个入口对应一个 Prompt链路不复杂。但当 Agent 数量超过十个你会发现最大的问题不再是单个 Agent 效果好不好而是“谁能提供什么能力、谁可以调用谁”这件事根本不清晰。多个 Agent 要接入同一个 IM、同一个工单系统、同一个数据平台每个 Agent 各自实现一套鉴权、限流、日志结果就是重复建设而且行为不一致。有的 Agent 对外裸奔有的 Agent 谁也调不动最后都变成演示 Demo。Agent-Reach 的核心是把“送达”变成平台能力而不是每个 Agent 自己的家务事。打个比方Agent 是手机 AppAgent-Reach 是应用商店。开发者不需要自己满世界找用户用户也不需要记一堆 URL大家只需要通过商店这个统一入口完成检索、下载、更新、安全审核。落到工程上就是注册、发现、路由、编排、限流、审计这一套机制。以前这些事散落在每个 Agent 里现在统一收口才可能让 Agent 集群真正跑起来。我见过不少团队在模型效果上卷得很厉害Prompt 写了几十版工具调用调得也很顺但一到“上线给业务方用”就卡住业务方不知道入口在哪不知道传什么参数更不知道这个 Agent 能不能处理他的问题。Agent-Reach 解决的就是这个“最后一公里”问题。它不负责提升某个 Agent 的智商但负责让每个 Agent 的能力可以被准确、安全、可控地触达到需要它的地方。1.2 Agent-Reach 要解决的三类问题第一个问题是“调用方不知道用哪个 Agent”。业务方想发起一个营销活动但他不知道系统里已经有活动策划 Agent、素材生成 Agent、风险审核 Agent于是自己写死一套逻辑绕过了平台。第二个问题是“Agent 之间互相找不到”。编排同事做多 Agent 协作时把子 Agent 地址硬编码在代码里子 Agent 接口一升级整个链路直接断掉。第三个问题是“上线之后没有审计”。输入、输出、模型调用、费用全部散落出了问题没人能说清是哪一环造成的。这三个问题叠加起来表现就是 Agent 越多系统越乱。Agent-Reach 的解法很简单所有 Agent 按照统一元数据注册所有调用通过统一网关分发所有链路都保留 trace_id。有了注册信息调用方可以搜索有了统一网关路由和限流可以集中控制有了 trace_id问题可以回溯。这不是什么高深技术但非常管用。问题常规表现Agent-Reach 的解法调用方不知道用哪个 Agent业务方重复开发入口五花八门统一注册中心 语义检索Agent 之间互相找不到子 Agent 被硬编码升级即断服务发现 动态路由上线后没有审计无法定位故障、费用、合规问题全链路 trace_id 调用日志1.3 这项能力的影响范围远比聊天机器人广很多人一听到 Agent 就想到客服问答。真做起来你会发现Agent-Reach 的价值更多体现在企业内部工具和自动化流程里。比如运维团队做一个告警分析 Agent它收到告警后需要调用链路追踪 Agent 查调用链再调用日志分析 Agent 查错误堆栈最后把结论通知值班人。这里的核心不是某一个 Agent 的推理能力而是多个 Agent 能不能被可靠地编排起来、互相触达。再比如数据分析场景业务问一句“上周转化率为什么下降”数据 Agent 要拆解成看板查询、异常检测、归因分析、报告生成四个子任务每个子任务由不同 Agent 完成。没有 Agent-Reach 这层编排硬要让一个 Agent 把所有事都干了Prompt 会越来越长成本会越来越高出错时更是没法定位。所以我认为Agent-Reach 的适用范围是所有“多个智能体需要协作”的场景它决定了你的 Agent 网络能长多大、能承担多少真实业务。2. Agent-Reach 的整体设计把 Agent 变成可触达的“能力端点”2.1 核心抽象每个 Agent 都是标准端点Agent-Reach 里最重要的一条设计原则是“把 Agent 当端点而不是当进程”。每个 Agent 无论内部是什么模型、什么框架对外只暴露统一的元数据。一个标准的 Agent 元数据大概长这样{ agent_id: campaign_writer, name: 活动文案生成, description: 根据活动主题和目标渠道生成推广文案与话题建议适合市场部做 campaign 时使用, version: 2.1.0, tags: [marketing, copywriting], owner: team-content, input_schema: { type: object, properties: { topic: {type: string}, channel: {type: string, enum: [wechat, xiaohongshu, douyin]} }, required: [topic] }, output_schema: { type: object, properties: { copy: {type: string}, suggestions: {type: array, items: {type: string}} } }, auth: [oauth2], rate_limit: {rpm: 60}, timeout: 30 }这段 JSON 看起来很普通但它解决了一个关键问题模型的槽位是自然语言而系统之间需要确定性。用户可以说“帮我写一个活动文案”但系统不能靠猜它需要知道这个 Agent 接受什么参数、返回什么结构。input_schema 和 output_schema 就是系统侧的“接口契约”description 则是给路由模型看的“说明文档”。这里有个经验description 一定不要只写“能做什么”还要写“什么时候不能用”。比如“本 Agent 仅用于文案生成不负责图片生成也不负责投放策略”这句话能显著降低模型误调用概率。后面我会专门讲 Agent 选错的坑这里先提一句description 写得好路由就成功了一半。2.2 注册与发现让 Agent 可以被检索所有 Agent 启动时先调用 Agent-Reach 的注册接口把自己的元数据上报上来。注册中心会把注册信息落库同时启动心跳机制。注册接口我用得比较多的是下面这种形态app.post(/v1/agents/register) async def register_agent(meta: AgentMeta): # 校验 agent_id 唯一性 existing await redis.get(fagent:{meta.agent_id}) if existing and existing ! meta.version: # 触发灰度升级流程而不是直接覆盖 await audit.record(agent_updated, meta) # 写入注册表 await redis.hset(agent:registry, meta.agent_id, meta.json()) # 记录心跳时间用于 TTL 清理 await redis.zadd(agent:heartbeat, {meta.agent_id: time.time()}) # 生成访问凭证 token secrets.token_urlsafe(32) await redis.set(fagent:token:{meta.agent_id}, token) return {agent_id: meta.agent_id, token: token}发现机制我一般提供三种精确匹配、标签过滤、语义检索。精确匹配就是调用方明确知道 agent_id直接路由标签过滤适合“我要找一个能做图片生成的 Agent”这种半结构化查询语义检索适合“用户说了一句自然语言系统需要判断交给哪个 Agent”的场景。语义检索可以基于向量数据库做但这里有个容易被忽略的点向量检索只能给候选排序不能直接决定调用哪个 Agent。最终决策一定要过一层“置信度阈值”低于阈值就转人工或返回“我不确定该用哪个能力”。2.3 路由与护栏让每次触达都可控路由策略我总结为三种直连路由、语义路由、编排路由。直连路由适合调用方已经确定目标 Agent语义路由适合把用户请求自动分发给合适 Agent编排路由适合一个任务需要多个 Agent 协作的场景。实际系统里这三种会同时存在Agent-Reach 的角色是把它们统一收口。只把请求转发出去还不够每次触达都要过护栏。护栏包括超时控制、并发限制、额度限制、参数校验、内容安全校验、人工审批。内容安全这块我建议做成“前置校验 后置校验”两道。前置校验在请求进入网关时做拦截明显不合规的输入后置校验在 Agent 返回结果后做防止模型输出绕过规则。两道校验都通过才允许最终响应返回给调用方。这里的一个实战心得是护栏尽量放在网关上不要放在 Agent 里。因为一旦 Agent 数量多了你不可能逐个去改它们的代码。网关上集中配置规则新增 Agent 时自动继承才具备可扩展性。Agent-Reach 本质上做的事情就是把这些本来要重复一万遍的“接线和安全工作”标准化。3. 最小 Agent-Reach 网关实现用 FastAPI 半天跑通3.1 基础组件选型如果从零开始搭一个最小可用版本我推荐 FastAPI 做网关Redis 做注册中心和心跳存储PostgreSQL pgvector 做元数据和向量检索。这套组合的好处是FastAPI 写起来快天然支持异步Redis 适合做 TTL 和排行榜pgvector 可以在不引入额外搜索服务的情况下完成语义检索。当然如果是纯 Demo也可以先用内存字典模拟注册中心。我之前带新人时会给一个“玩具版”思路用 Python dict 存 Agent 元数据用列表存心跳用关键词匹配代替向量检索。这个版本跑通之后再平滑迁移到 Redis 和 pgvector。直接把架构堆得很重反而会让初学者迷失在基础设施里。3.2 注册与心跳的实现细节除了注册接口心跳机制一定要做。它决定了 Agent-Reach 能不能自动发现“某个 Agent 已经不可用”。我通常要求每个 Agent 暴露 /healthz 和 /readyz 两个接口healthz 表示进程还活着readyz 表示依赖都可用。网关每隔 10 秒拉一次 readyz连续失败三次就把该 Agent 标记为不可触达路由时自动跳过。心跳数据放在 Redis 的 zset 里score 存最新心跳时间戳。定期清理超过 90 秒没有心跳的注册项避免僵尸 Agent 占着坑。这个机制在本地开发时容易踩坑开发同学本地起了一个 Agent联调完直接关电脑但注册表里还留着旧地址其他环境跑到这里就开始报连接失败。有了 TTL 清理这类问题会少很多。3.3 统一调用入口把 Agent 变成 APIAgent-Reach 对调用方暴露的入口非常少核心就一个POST /v1/agents/invoke。调用方不用关心目标 Agent 部署在哪、内部是什么协议只需要传 prompt 或结构化参数网关负责找 Agent、调用 Agent、返回统一结果。app.post(/v1/agents/invoke) async def invoke(request: InvokeRequest): trace_id request.trace_id or uuid4().hex # 1. 找到候选 Agent candidates await find_candidates(request.query, request.agent_id) if not candidates: return error_response(350404, no available agent, trace_id) # 2. 按策略挑选最终 Agent agent select_agent(candidates, request) if agent is None: return error_response(350404, agent confidence too low, trace_id) # 3. 前置护栏参数校验、内容安全、限流 await guard.pre_check(request, agent, trace_id) # 4. 调用 Agent try: result await call_agent(agent, request.payload, trace_id, timeoutagent.timeout) except TimeoutError: return error_response(350408, agent timeout, trace_id) except AgentUnavailable: return error_response(350503, agent unavailable, trace_id) # 5. 后置校验输出结构、内容安全 await guard.post_check(result, agent, trace_id) # 6. 记录审计日志 await audit.log(trace_id, agent.agent_id, request.payload, result) return success_response(result, trace_id)这段代码把前面说的路径都串起来了。有一点需要留意find_candidates如果只靠向量相似度可能会把不相关的 Agent 也捞出来所以我会同时结合标签匹配和 description 关键词匹配做过滤最后再用向量排序。这种方式在冷启动阶段比单靠向量检索稳定得多。3.4 统一返回与错误码约定调用方最讨厌的一件事就是每个 Agent 返回格式都不一样。Agent-Reach 内部强制要求所有结果都包装成统一结构{ code: 0, message: ok, data: {}, trace_id: a1b2c3d4 }code 为 0 表示成功非 0 表示失败。我给自己定了一套错误码code含义调用方建议350404Agent 不存在或不可达检查 agent_id或换一个 Agent350408Agent 调用超时稍后重试或走降级通道350429触发限流排队等待或退避重试350502Agent 返回内容校验失败查网关后置校验日志350503Agent 服务不可用等待恢复或走降级 Agent统一错误码这件事看起来简单但在多 Agent 体系里非常重要。因为一个编排任务可能会有几十次子调用错误码不统一上层根本没法做重试和降级策略。4. 用 Agent-Reach 编排多 Agent让子 Agent 之间真正“互相触达”4.1 场景拆解一个活动策划任务用一个最简单的例子说明编排某个市场部同学在系统里发起“为新产品做一个上市推广策划”。这个任务不是单个 Agent 能搞定的通常需要四步舆情洞察 Agent 先分析目前市场上同类产品的讨论热点和用户痛点的数据。活动策划 Agent 根据洞察结果产出整体推广策略包括目标人群、渠道选择。内容生成 Agent 根据策略批量生成渠道文案、海报标题等物料。合规审核 Agent 对生成的文案做风险检查有问题打回重写。没有 Agent-Reach 时这个流程得靠人手工把上一个 Agent 的输出复制给下一个 Agent。有了 Agent-Reach你可以把链路定义成显式的工作流每个环节对应一个 Agent由编排引擎自动调度。4.2 编排配置与执行引擎我在 Agent-Reach 里用 DAG 来表达编排关系。每个 task 节点包含 agent_id、输入映射、输出映射和依赖关系workflow: product_launch_campaign steps: - id: insight agent: market_insight_agent depends_on: [] - id: strategy agent: campaign_strategy_agent depends_on: [insight] - id: copywriting agent: content_generator_agent depends_on: [strategy] - id: compliance agent: compliance_review_agent depends_on: [copywriting]执行引擎从没有依赖的节点开始被依赖的节点完成后再触发下游节点。没有依赖关系的节点可以并行执行带依赖关系的节点则要等待上游结果。把编排显式写成 YAML比把编排逻辑藏在某个 Agent 的 Prompt 里要可控得多。Prompt 里的“你第一步调用 A第二步调用 B”很难调试而 YAML 配置每一步都能看到输入输出也方便做人工插入审批。编排引擎还要考虑“部分成功”的问题。比如文案生成有两个渠道其中一个渠道的 Agent 超时了整个流程要不要回滚我建议不要整体回滚而是把成功结果先返回失败节点标记为“可重试”并在最终结果里给出提示。因为业务方往往宁肯拿到部分结果也不愿意干等全部完成。4.3 上下文传递与状态管理编排过程中最容易出的问题就是上下文丢失。子 Agent 需要理解上游在问什么它的返回又要回传给下游。我一般用 task_id 和 parent_trace_id 来串联。每个节点接收的输入里除了显式参数还会带上上游摘要而不是整个对话历史。这个设计是防止上下文无限膨胀也是控制 token 成本的关键。状态管理我会在 Redis 里缓存每个节点的输入和输出键名类似ctx:{workflow_id}:{node_id}。执行到某个节点时节点从 context 里取上游产物处理完再把产物写回 context。如果中途宕机重跑时不需要从头开始已经完成的节点可以直接复用结果。这个能力在生产环境里非常实用因为大模型调用一多任务动辄几十秒全流程重跑代价太大。4.4 人工审批与回退机制不是所有 Agent 调用都应该全自动。Agent-Reach 里我会给任务增加“需要人工确认”的开关。比如合规审核 Agent 发现一个风险点它不应该擅自决定“再审一次”而是要把任务置为 pending通知相关人去人工确认。审批通过后流程继续走审批驳回则把结果返回到上一个节点重新生成。这个机制其实是在“自动化”和“可控性”之间找平衡。Agent 能力再强在高风险决策上依然需要人来拍板。Agent-Reach 把人工审批作为一等公民而不是事后补救。配置很简单在 workflow 的节点上增加approval_required: true执行引擎看到这个标记就自动等待审批回调。5. 实战中踩过的坑和排查记录5.1 超时、重试与幂等大模型 Agent 的延迟非常不稳定。有的请求 2 秒返回有的请求 45 秒还在思考。最开始我按普通 HTTP 服务的标准设置 5 秒超时结果线上全是超时告警。后来改为“连接超时 3 秒 读取超时 60 秒”并把超过 30 秒的调用标记为慢调用单独观测。重试也要特别小心。模型调用可能已经成功只是响应超时如果直接重试用户会收到两份结果还可能产生额外费用。我的做法是每次调用带上幂等键Idempotency-Key网关在 Redis 里缓存最近 5 分钟的响应内容重试时直接返回缓存结果。这个思路和支付系统里的幂等设计一脉相承非常值得引入。5.2 Agent 选错模型幻觉导致的误调用这是我在 Agent-Reach 项目里遇到最多的问题。用户说“帮我写个标题”系统却调用了“用户画像分析 Agent”。原因通常是两个一是语义检索把不相关 Agent 的向量排到了前面二是候选 Agent 的 description 写得含糊模型无法区分。解决手段有三个description 里写清楚“不适用的场景”相当于给路由模型画了一条禁止线。对 query 先做一次意图分类用一个小模型判断任务领域再缩小候选范围。语义检索的置信度低于阈值时不允许自动调用必须返回“需要你补充信息”或转人工。第三个手段最容易被忽略。很多人为了“智能感”强行让系统什么请求都直接执行结果就是错误被自动化放大了。在 Agent-Reach 里我宁愿让系统多问一次“你确定要生成图片吗”也不希望它调用一个错误的 Agent 输出垃圾结果。5.3 注册信息垃圾与版本漂移Agent 上线时间长了注册表会变得非常混乱。最典型的情况是Agent 逻辑已经升级但 description 还停留在旧版本或者某个 Agent 的 timeout 配置不合理但没人去调整再或者 Agent 下线之后注册信息没有清理路由偶尔命中一个已经不存在的内网地址。这些都是“蚂蚁搬家式”的问题单独看不大合在一起会让系统很不可信。我从第三个月开始强制执行 Agent 注册信息审查每次 Agent 版本更新必须同时更新 description、input_schema、output_schema、timeout。网关端每 24 小时跑一次体检找出描述太长、schema 不合法、心跳长期缺失的 Agent推到管理人那里处理。Agent-Reach 里注册中心不只是服务发现它也是 Agent 治理解析器——它有多规范整个 Agent 体系就有多规范。5.4 日志、链路追踪与数据脱敏最后是观测。Agent-Reach 需要记录五类数据调用方信息、目标 Agent 信息、输入参数、输出结果、模型调用链路。每条记录必须带上 trace_id让一次用户请求可以串起从网关到子 Agent 的完整链路。但这里有一个矛盾日志越完整数据泄漏风险越大。尤其用户的输入可能包含姓名、手机号、地址等敏感信息。我的做法是“原始日志脱敏 明细日志受限访问”。默认日志只记录字段名、字段类型和脱敏后的值比如phone: 138****1234只有在审计系统里通过权限申请后才能看到完整明文。别嫌麻烦这一步在 Agent 真正接入业务后会救你很多次。否则光一条合规检查就够你折腾一个月。最后再分享一个小技巧如果你准备自己搭 Agent-Reach我强烈建议给每个接入的 Agent 强制暴露/healthz和/readyz两个端点。很多团队只有/healthz进程活着就以为健康但依赖的模型 API 可能已经不通了缓存也可能连不上。我们在/readyz里会检查模型服务连通性、缓存状态、本地任务队列长度每 10 秒检查一次连续失败就把 Agent 从路由表里摘掉。这个改动看起来不起眼但它让 Agent-Reach 的可用性提升了一个数量级。智能体的世界很复杂先把“谁能被触达”这件事管好后面的一切才有意义。 
长河君 · 2026/10/26 22:11:26 · 阅读 1.2k Agent-Reach：让多Agent真正互相触达的统一网关设计
做 Agent 项目做到第三个年头我发现一个特别容易被忽略的问题Agent 本身不难跑通难的是让 Agent 真正“Reach”——触达用户、触达系统、触达另一批 Agent。这也是我为什么在搭 Agent-Reach 时把名字钉死在“触达”这两个字上。Agent-Reach 在我这里不是一个聊天机器人也不是某个单独模型的封装而是一层统一触达与治理网关所有智能体以标准端点注册所有调用方通过统一入口完成发现、路由、编排、限流与审计。它能解决什么问题一句话避免你的 Agent 集群变成一堆谁也找不到、谁也不敢用的“模型岛”。适合正在搭多 Agent 平台、或者准备把 Agent 能力开放给内部业务方的团队参考。1. 为什么需要 Agent-ReachAgent 不是模型而是业务单元1.1 智能体集群的真正瓶颈在“送达”单体 Bot 时代一个入口对应一个 Prompt链路不复杂。但当 Agent 数量超过十个你会发现最大的问题不再是单个 Agent 效果好不好而是“谁能提供什么能力、谁可以调用谁”这件事根本不清晰。多个 Agent 要接入同一个 IM、同一个工单系统、同一个数据平台每个 Agent 各自实现一套鉴权、限流、日志结果就是重复建设而且行为不一致。有的 Agent 对外裸奔有的 Agent 谁也调不动最后都变成演示 Demo。Agent-Reach 的核心是把“送达”变成平台能力而不是每个 Agent 自己的家务事。打个比方Agent 是手机 AppAgent-Reach 是应用商店。开发者不需要自己满世界找用户用户也不需要记一堆 URL大家只需要通过商店这个统一入口完成检索、下载、更新、安全审核。落到工程上就是注册、发现、路由、编排、限流、审计这一套机制。以前这些事散落在每个 Agent 里现在统一收口才可能让 Agent 集群真正跑起来。我见过不少团队在模型效果上卷得很厉害Prompt 写了几十版工具调用调得也很顺但一到“上线给业务方用”就卡住业务方不知道入口在哪不知道传什么参数更不知道这个 Agent 能不能处理他的问题。Agent-Reach 解决的就是这个“最后一公里”问题。它不负责提升某个 Agent 的智商但负责让每个 Agent 的能力可以被准确、安全、可控地触达到需要它的地方。1.2 Agent-Reach 要解决的三类问题第一个问题是“调用方不知道用哪个 Agent”。业务方想发起一个营销活动但他不知道系统里已经有活动策划 Agent、素材生成 Agent、风险审核 Agent于是自己写死一套逻辑绕过了平台。第二个问题是“Agent 之间互相找不到”。编排同事做多 Agent 协作时把子 Agent 地址硬编码在代码里子 Agent 接口一升级整个链路直接断掉。第三个问题是“上线之后没有审计”。输入、输出、模型调用、费用全部散落出了问题没人能说清是哪一环造成的。这三个问题叠加起来表现就是 Agent 越多系统越乱。Agent-Reach 的解法很简单所有 Agent 按照统一元数据注册所有调用通过统一网关分发所有链路都保留 trace_id。有了注册信息调用方可以搜索有了统一网关路由和限流可以集中控制有了 trace_id问题可以回溯。这不是什么高深技术但非常管用。问题常规表现Agent-Reach 的解法调用方不知道用哪个 Agent业务方重复开发入口五花八门统一注册中心 语义检索Agent 之间互相找不到子 Agent 被硬编码升级即断服务发现 动态路由上线后没有审计无法定位故障、费用、合规问题全链路 trace_id 调用日志1.3 这项能力的影响范围远比聊天机器人广很多人一听到 Agent 就想到客服问答。真做起来你会发现Agent-Reach 的价值更多体现在企业内部工具和自动化流程里。比如运维团队做一个告警分析 Agent它收到告警后需要调用链路追踪 Agent 查调用链再调用日志分析 Agent 查错误堆栈最后把结论通知值班人。这里的核心不是某一个 Agent 的推理能力而是多个 Agent 能不能被可靠地编排起来、互相触达。再比如数据分析场景业务问一句“上周转化率为什么下降”数据 Agent 要拆解成看板查询、异常检测、归因分析、报告生成四个子任务每个子任务由不同 Agent 完成。没有 Agent-Reach 这层编排硬要让一个 Agent 把所有事都干了Prompt 会越来越长成本会越来越高出错时更是没法定位。所以我认为Agent-Reach 的适用范围是所有“多个智能体需要协作”的场景它决定了你的 Agent 网络能长多大、能承担多少真实业务。2. Agent-Reach 的整体设计把 Agent 变成可触达的“能力端点”2.1 核心抽象每个 Agent 都是标准端点Agent-Reach 里最重要的一条设计原则是“把 Agent 当端点而不是当进程”。每个 Agent 无论内部是什么模型、什么框架对外只暴露统一的元数据。一个标准的 Agent 元数据大概长这样{ agent_id: campaign_writer, name: 活动文案生成, description: 根据活动主题和目标渠道生成推广文案与话题建议适合市场部做 campaign 时使用, version: 2.1.0, tags: [marketing, copywriting], owner: team-content, input_schema: { type: object, properties: { topic: {type: string}, channel: {type: string, enum: [wechat, xiaohongshu, douyin]} }, required: [topic] }, output_schema: { type: object, properties: { copy: {type: string}, suggestions: {type: array, items: {type: string}} } }, auth: [oauth2], rate_limit: {rpm: 60}, timeout: 30 }这段 JSON 看起来很普通但它解决了一个关键问题模型的槽位是自然语言而系统之间需要确定性。用户可以说“帮我写一个活动文案”但系统不能靠猜它需要知道这个 Agent 接受什么参数、返回什么结构。input_schema 和 output_schema 就是系统侧的“接口契约”description 则是给路由模型看的“说明文档”。这里有个经验description 一定不要只写“能做什么”还要写“什么时候不能用”。比如“本 Agent 仅用于文案生成不负责图片生成也不负责投放策略”这句话能显著降低模型误调用概率。后面我会专门讲 Agent 选错的坑这里先提一句description 写得好路由就成功了一半。2.2 注册与发现让 Agent 可以被检索所有 Agent 启动时先调用 Agent-Reach 的注册接口把自己的元数据上报上来。注册中心会把注册信息落库同时启动心跳机制。注册接口我用得比较多的是下面这种形态app.post(/v1/agents/register) async def register_agent(meta: AgentMeta): # 校验 agent_id 唯一性 existing await redis.get(fagent:{meta.agent_id}) if existing and existing ! meta.version: # 触发灰度升级流程而不是直接覆盖 await audit.record(agent_updated, meta) # 写入注册表 await redis.hset(agent:registry, meta.agent_id, meta.json()) # 记录心跳时间用于 TTL 清理 await redis.zadd(agent:heartbeat, {meta.agent_id: time.time()}) # 生成访问凭证 token secrets.token_urlsafe(32) await redis.set(fagent:token:{meta.agent_id}, token) return {agent_id: meta.agent_id, token: token}发现机制我一般提供三种精确匹配、标签过滤、语义检索。精确匹配就是调用方明确知道 agent_id直接路由标签过滤适合“我要找一个能做图片生成的 Agent”这种半结构化查询语义检索适合“用户说了一句自然语言系统需要判断交给哪个 Agent”的场景。语义检索可以基于向量数据库做但这里有个容易被忽略的点向量检索只能给候选排序不能直接决定调用哪个 Agent。最终决策一定要过一层“置信度阈值”低于阈值就转人工或返回“我不确定该用哪个能力”。2.3 路由与护栏让每次触达都可控路由策略我总结为三种直连路由、语义路由、编排路由。直连路由适合调用方已经确定目标 Agent语义路由适合把用户请求自动分发给合适 Agent编排路由适合一个任务需要多个 Agent 协作的场景。实际系统里这三种会同时存在Agent-Reach 的角色是把它们统一收口。只把请求转发出去还不够每次触达都要过护栏。护栏包括超时控制、并发限制、额度限制、参数校验、内容安全校验、人工审批。内容安全这块我建议做成“前置校验 后置校验”两道。前置校验在请求进入网关时做拦截明显不合规的输入后置校验在 Agent 返回结果后做防止模型输出绕过规则。两道校验都通过才允许最终响应返回给调用方。这里的一个实战心得是护栏尽量放在网关上不要放在 Agent 里。因为一旦 Agent 数量多了你不可能逐个去改它们的代码。网关上集中配置规则新增 Agent 时自动继承才具备可扩展性。Agent-Reach 本质上做的事情就是把这些本来要重复一万遍的“接线和安全工作”标准化。3. 最小 Agent-Reach 网关实现用 FastAPI 半天跑通3.1 基础组件选型如果从零开始搭一个最小可用版本我推荐 FastAPI 做网关Redis 做注册中心和心跳存储PostgreSQL pgvector 做元数据和向量检索。这套组合的好处是FastAPI 写起来快天然支持异步Redis 适合做 TTL 和排行榜pgvector 可以在不引入额外搜索服务的情况下完成语义检索。当然如果是纯 Demo也可以先用内存字典模拟注册中心。我之前带新人时会给一个“玩具版”思路用 Python dict 存 Agent 元数据用列表存心跳用关键词匹配代替向量检索。这个版本跑通之后再平滑迁移到 Redis 和 pgvector。直接把架构堆得很重反而会让初学者迷失在基础设施里。3.2 注册与心跳的实现细节除了注册接口心跳机制一定要做。它决定了 Agent-Reach 能不能自动发现“某个 Agent 已经不可用”。我通常要求每个 Agent 暴露 /healthz 和 /readyz 两个接口healthz 表示进程还活着readyz 表示依赖都可用。网关每隔 10 秒拉一次 readyz连续失败三次就把该 Agent 标记为不可触达路由时自动跳过。心跳数据放在 Redis 的 zset 里score 存最新心跳时间戳。定期清理超过 90 秒没有心跳的注册项避免僵尸 Agent 占着坑。这个机制在本地开发时容易踩坑开发同学本地起了一个 Agent联调完直接关电脑但注册表里还留着旧地址其他环境跑到这里就开始报连接失败。有了 TTL 清理这类问题会少很多。3.3 统一调用入口把 Agent 变成 APIAgent-Reach 对调用方暴露的入口非常少核心就一个POST /v1/agents/invoke。调用方不用关心目标 Agent 部署在哪、内部是什么协议只需要传 prompt 或结构化参数网关负责找 Agent、调用 Agent、返回统一结果。app.post(/v1/agents/invoke) async def invoke(request: InvokeRequest): trace_id request.trace_id or uuid4().hex # 1. 找到候选 Agent candidates await find_candidates(request.query, request.agent_id) if not candidates: return error_response(350404, no available agent, trace_id) # 2. 按策略挑选最终 Agent agent select_agent(candidates, request) if agent is None: return error_response(350404, agent confidence too low, trace_id) # 3. 前置护栏参数校验、内容安全、限流 await guard.pre_check(request, agent, trace_id) # 4. 调用 Agent try: result await call_agent(agent, request.payload, trace_id, timeoutagent.timeout) except TimeoutError: return error_response(350408, agent timeout, trace_id) except AgentUnavailable: return error_response(350503, agent unavailable, trace_id) # 5. 后置校验输出结构、内容安全 await guard.post_check(result, agent, trace_id) # 6. 记录审计日志 await audit.log(trace_id, agent.agent_id, request.payload, result) return success_response(result, trace_id)这段代码把前面说的路径都串起来了。有一点需要留意find_candidates如果只靠向量相似度可能会把不相关的 Agent 也捞出来所以我会同时结合标签匹配和 description 关键词匹配做过滤最后再用向量排序。这种方式在冷启动阶段比单靠向量检索稳定得多。3.4 统一返回与错误码约定调用方最讨厌的一件事就是每个 Agent 返回格式都不一样。Agent-Reach 内部强制要求所有结果都包装成统一结构{ code: 0, message: ok, data: {}, trace_id: a1b2c3d4 }code 为 0 表示成功非 0 表示失败。我给自己定了一套错误码code含义调用方建议350404Agent 不存在或不可达检查 agent_id或换一个 Agent350408Agent 调用超时稍后重试或走降级通道350429触发限流排队等待或退避重试350502Agent 返回内容校验失败查网关后置校验日志350503Agent 服务不可用等待恢复或走降级 Agent统一错误码这件事看起来简单但在多 Agent 体系里非常重要。因为一个编排任务可能会有几十次子调用错误码不统一上层根本没法做重试和降级策略。4. 用 Agent-Reach 编排多 Agent让子 Agent 之间真正“互相触达”4.1 场景拆解一个活动策划任务用一个最简单的例子说明编排某个市场部同学在系统里发起“为新产品做一个上市推广策划”。这个任务不是单个 Agent 能搞定的通常需要四步舆情洞察 Agent 先分析目前市场上同类产品的讨论热点和用户痛点的数据。活动策划 Agent 根据洞察结果产出整体推广策略包括目标人群、渠道选择。内容生成 Agent 根据策略批量生成渠道文案、海报标题等物料。合规审核 Agent 对生成的文案做风险检查有问题打回重写。没有 Agent-Reach 时这个流程得靠人手工把上一个 Agent 的输出复制给下一个 Agent。有了 Agent-Reach你可以把链路定义成显式的工作流每个环节对应一个 Agent由编排引擎自动调度。4.2 编排配置与执行引擎我在 Agent-Reach 里用 DAG 来表达编排关系。每个 task 节点包含 agent_id、输入映射、输出映射和依赖关系workflow: product_launch_campaign steps: - id: insight agent: market_insight_agent depends_on: [] - id: strategy agent: campaign_strategy_agent depends_on: [insight] - id: copywriting agent: content_generator_agent depends_on: [strategy] - id: compliance agent: compliance_review_agent depends_on: [copywriting]执行引擎从没有依赖的节点开始被依赖的节点完成后再触发下游节点。没有依赖关系的节点可以并行执行带依赖关系的节点则要等待上游结果。把编排显式写成 YAML比把编排逻辑藏在某个 Agent 的 Prompt 里要可控得多。Prompt 里的“你第一步调用 A第二步调用 B”很难调试而 YAML 配置每一步都能看到输入输出也方便做人工插入审批。编排引擎还要考虑“部分成功”的问题。比如文案生成有两个渠道其中一个渠道的 Agent 超时了整个流程要不要回滚我建议不要整体回滚而是把成功结果先返回失败节点标记为“可重试”并在最终结果里给出提示。因为业务方往往宁肯拿到部分结果也不愿意干等全部完成。4.3 上下文传递与状态管理编排过程中最容易出的问题就是上下文丢失。子 Agent 需要理解上游在问什么它的返回又要回传给下游。我一般用 task_id 和 parent_trace_id 来串联。每个节点接收的输入里除了显式参数还会带上上游摘要而不是整个对话历史。这个设计是防止上下文无限膨胀也是控制 token 成本的关键。状态管理我会在 Redis 里缓存每个节点的输入和输出键名类似ctx:{workflow_id}:{node_id}。执行到某个节点时节点从 context 里取上游产物处理完再把产物写回 context。如果中途宕机重跑时不需要从头开始已经完成的节点可以直接复用结果。这个能力在生产环境里非常实用因为大模型调用一多任务动辄几十秒全流程重跑代价太大。4.4 人工审批与回退机制不是所有 Agent 调用都应该全自动。Agent-Reach 里我会给任务增加“需要人工确认”的开关。比如合规审核 Agent 发现一个风险点它不应该擅自决定“再审一次”而是要把任务置为 pending通知相关人去人工确认。审批通过后流程继续走审批驳回则把结果返回到上一个节点重新生成。这个机制其实是在“自动化”和“可控性”之间找平衡。Agent 能力再强在高风险决策上依然需要人来拍板。Agent-Reach 把人工审批作为一等公民而不是事后补救。配置很简单在 workflow 的节点上增加approval_required: true执行引擎看到这个标记就自动等待审批回调。5. 实战中踩过的坑和排查记录5.1 超时、重试与幂等大模型 Agent 的延迟非常不稳定。有的请求 2 秒返回有的请求 45 秒还在思考。最开始我按普通 HTTP 服务的标准设置 5 秒超时结果线上全是超时告警。后来改为“连接超时 3 秒 读取超时 60 秒”并把超过 30 秒的调用标记为慢调用单独观测。重试也要特别小心。模型调用可能已经成功只是响应超时如果直接重试用户会收到两份结果还可能产生额外费用。我的做法是每次调用带上幂等键Idempotency-Key网关在 Redis 里缓存最近 5 分钟的响应内容重试时直接返回缓存结果。这个思路和支付系统里的幂等设计一脉相承非常值得引入。5.2 Agent 选错模型幻觉导致的误调用这是我在 Agent-Reach 项目里遇到最多的问题。用户说“帮我写个标题”系统却调用了“用户画像分析 Agent”。原因通常是两个一是语义检索把不相关 Agent 的向量排到了前面二是候选 Agent 的 description 写得含糊模型无法区分。解决手段有三个description 里写清楚“不适用的场景”相当于给路由模型画了一条禁止线。对 query 先做一次意图分类用一个小模型判断任务领域再缩小候选范围。语义检索的置信度低于阈值时不允许自动调用必须返回“需要你补充信息”或转人工。第三个手段最容易被忽略。很多人为了“智能感”强行让系统什么请求都直接执行结果就是错误被自动化放大了。在 Agent-Reach 里我宁愿让系统多问一次“你确定要生成图片吗”也不希望它调用一个错误的 Agent 输出垃圾结果。5.3 注册信息垃圾与版本漂移Agent 上线时间长了注册表会变得非常混乱。最典型的情况是Agent 逻辑已经升级但 description 还停留在旧版本或者某个 Agent 的 timeout 配置不合理但没人去调整再或者 Agent 下线之后注册信息没有清理路由偶尔命中一个已经不存在的内网地址。这些都是“蚂蚁搬家式”的问题单独看不大合在一起会让系统很不可信。我从第三个月开始强制执行 Agent 注册信息审查每次 Agent 版本更新必须同时更新 description、input_schema、output_schema、timeout。网关端每 24 小时跑一次体检找出描述太长、schema 不合法、心跳长期缺失的 Agent推到管理人那里处理。Agent-Reach 里注册中心不只是服务发现它也是 Agent 治理解析器——它有多规范整个 Agent 体系就有多规范。5.4 日志、链路追踪与数据脱敏最后是观测。Agent-Reach 需要记录五类数据调用方信息、目标 Agent 信息、输入参数、输出结果、模型调用链路。每条记录必须带上 trace_id让一次用户请求可以串起从网关到子 Agent 的完整链路。但这里有一个矛盾日志越完整数据泄漏风险越大。尤其用户的输入可能包含姓名、手机号、地址等敏感信息。我的做法是“原始日志脱敏 明细日志受限访问”。默认日志只记录字段名、字段类型和脱敏后的值比如phone: 138****1234只有在审计系统里通过权限申请后才能看到完整明文。别嫌麻烦这一步在 Agent 真正接入业务后会救你很多次。否则光一条合规检查就够你折腾一个月。最后再分享一个小技巧如果你准备自己搭 Agent-Reach我强烈建议给每个接入的 Agent 强制暴露/healthz和/readyz两个端点。很多团队只有/healthz进程活着就以为健康但依赖的模型 API 可能已经不通了缓存也可能连不上。我们在/readyz里会检查模型服务连通性、缓存状态、本地任务队列长度每 10 秒检查一次连续失败就把 Agent 从路由表里摘掉。这个改动看起来不起眼但它让 Agent-Reach 的可用性提升了一个数量级。智能体的世界很复杂先把“谁能被触达”这件事管好后面的一切才有意义。