Agent-Reach 是我前段时间在整理旧项目时重新翻出来打磨的一个小工具。名字拆开看就是 Agent智能体和 Reach触达当手里的 Agent 越来越多怎么让业务方一条请求就能“触达”到正确的那个 Agent而不是让调用方去记十几个不同地址、不同参数格式的接口。这个项目最初是因为团队里各小组各自训练和部署了几个垂直场景的 Agent有客服意图识别、工单信息抽取、报表自动生成结果业务方想接的时候发现每个 Agent 的调用方式都不一样有的走 HTTP有的走 Webhook有的甚至要自己拼 Prompt 再调底层模型。Agent-Reach 就是为了解决这个混乱状态而做的统一注册、统一调度、统一观测的中间层。如果你正在做 Agent 相关应用或者你的团队已经有三五个在线的 Agent 服务但每次联调都要看各自的文档那这篇文章应该能给你一个可以直接抄作业的落地思路。我会从整体设计、核心模块、代码实现到排查技巧全部过一遍重点讲清楚每一步为什么这么做。1. 项目定位为什么需要 Agent-Reach 这一层1.1 Agent 多起来之后的真实痛点很多人一开始做 Agent 的时候都是单机单服务一个 Agent 搞定所有事情。但真实业务里不是这样客服组做了一个意图分类 Agent数据组做了一个 SQL 转查询 Agent运维组做了一个日志异常分析 Agent这些 Agent 是不同人、不同时间、不同技术栈下做的。于是问题就来了业务方要同时对接三个 Agent就要处理三套认证方式、三套超时策略、三套错误码而且某个 Agent 升级了接口还得重新联调。我一开始也被这个问题折腾得够呛。有一次业务方反馈说“客服 Agent 经常超时”后来排查发现不是客服 Agent 慢而是调用方用了默认的 3 秒超时但客服 Agent 因为要调大模型正常响应就要 4 到 5 秒。这种问题本质上是缺少一个统一的“触达协议”调用方不知道该用多长超时、该传什么格式、出错该找谁。Agent-Reach 的核心思路不是重新造 Agent而是做一层薄薄的“调度网关”所有 Agent 都往这里注册自己的信息所有调用方都只跟网关打交道。网关负责把请求翻译成每个 Agent 能理解的格式负责设置合理的超时和重试负责把错误信息统一收敛。这个定位很像快递中转站寄件人不需要知道包裹最终走哪条航线只要把东西交到中转站后面的事中转站来协调。1.2 设计目标注册、调度、适配、观测四件事我给自己定的范围非常明确就做四件事Agent 注册与发现Agent 上线的时候把自己登记到注册表里包括名称、地址、协议类型、当前状态。这样调用方不再需要硬编码地址。统一调度入口所有外部调用统一走一个/v1/reach接口网关根据请求里声明的目标能力去路由到具体的 Agent。协议适配每个 Agent 内部可以保持自己的协议不变网关通过适配器做翻译。能不改业务方代码就不改。可观测每次调用都生成一个 trace_id记录从网关到 Agent 的完整调用链、耗时、返回结果。没有这个出了问题只能靠猜。这四个能力做成一个独立服务不侵入 Agent 内部逻辑。这个边界非常重要因为一旦你想给每个 Agent 装 SDK 做深度集成推广成本就上去了而且老 Agent 未必愿意配合改造。协议适配器可以单独部署Agent 侧只需要支持最普通的 HTTP 回调就能接入。1.3 适合谁来参考这套方案如果符合下面任意一条这篇文章的方案可以直接复用到你的场景你手上有多个独立部署的 Agent 服务且它们彼此之间协议不统一调用方维护成本高。你想给 Agent 服务加一层统一鉴权和访问控制但不想在每个 Agent 里重复实现。你想对 Agent 调用做全链路追踪比如记录某个任务到底被路由到了哪个 Agent耗时多少。你做的是中小型团队基建不想上 Service Mesh 这么重的方案只想用一个轻量服务解决问题。如果只有一个 Agent、一个调用方那没必要搞这套直接改 Agent 接口更省事。Agent-Reach 的收益一定是随着 Agent 数量和调用方数量增多才逐步放大的。2. 核心模块设计与注册表细节2.1 注册表Agent 的“身份档案”Agent-Reach 最核心的数据结构就是注册表里的 Agent 记录。我用 Pydantic 定义了一个AgentInfo模型字段不多但每一个都是踩过坑之后留下来的from typing import List, Optional from pydantic import BaseModel, Field from enum import Enum class AgentStatus(str, Enum): ONLINE online OFFLINE offline DEGRADED degraded class AgentProtocol(str, Enum): HTTP http WEBHOOK webhook class AgentInfo(BaseModel): agent_id: str Field(..., description全局唯一标识建议用 agent_ 前缀) name: str Field(..., description人类可读名称如 客服意图识别) endpoint: str Field(..., descriptionAgent 的调用地址) protocol: AgentProtocol AgentProtocol.HTTP capabilities: List[str] Field(..., description能力标签列表) status: AgentStatus AgentStatus.OFFLINE owner: str Field(, description负责人用于告警通知) version: str Field(1.0.0, descriptionAgent 版本号) max_timeout: int Field(5, description建议超时时间单位秒) created_at: Optional[str] None这里有几个值得注意的设计点。agent_id我建议用前缀加用途的方式比如agent_intent_cls、agent_sql_gen、agent_log_analysis。不要用自增数字因为后续做日志检索、权限配置的时候见名知意比查数字有意义得多。capabilities是一个字符串列表这是整个路由逻辑的“关键词索引”。比如客服意图识别 Agent 注册的是[intent, customer_service]当外部请求声明target_capability: intent时网关就能快速筛出所有带intent标签的 Agent。一开始我试图用结构化 Schema 描述能力比如“输入是什么类型、输出是什么类型”后来发现过度设计因为大家连自己 Agent 的输入输出规范都没理清楚强行上 Schema 只会增加接入门槛。先用标签等真的有约束需求再升级。max_timeout这个字段非常实用。不同 Agent 的响应速度差异很大日志分析 Agent 可能要 8 到 10 秒意图分类 Agent 可能 200 毫秒就回来了。如果网关对所有 Agent 用同一个超时时间必然有一边不舒服。注册的时候让 Agent 方声明自己的合理超时上限网关再在这个上限基础上做控制是成本最低的适配方式。2.2 统一请求与响应封装把差异关在笼子里有了注册表下一步是定义统一的消息格式。我参考了普通网关的做法设计了一组很朴素的协议{ trace_id: trc_20250101_ab12cd34ef56, task_id: task_20250101_0001, target_capability: intent, target_agent: agent_intent_cls, payload: { text: 我要退订这个月的流量包 } }trace_id是整个链路的唯一标识网关生成后传给 AgentAgent 在日志里带上这个 ID两边日志一拼就能看到完整链路。task_id是业务侧的任务标识适合异步场景里回查。target_capability和target_agent二选一如果调用方明确知道要找谁直接传target_agent如果不知道找谁只知道自己想干什么就传target_capability让网关来路由。这个设计很土但很好用。响应格式也做了统一{ trace_id: trc_20250101_ab12cd34ef56, agent_id: agent_intent_cls, status: success, data: { intent: unsubscribe, confidence: 0.97 } }一旦格式统一调用方就可以写一套通用的解析逻辑而不是每个 Agent 一套解析。这里我犯过一个错最开始把“响应统一”理解成“字段统一”试图让所有 Agent 返回完全一样的字段结构结果有的 Agent 就是没有置信度、没有意图层级硬填反而造假数据。后来想通了统一的是外壳状态、链路 ID、Agent 标识业务数据放data里保持 Agent 自己的结构。外壳统一保证可运维内核保留差异保证灵活。2.3 认证与鉴权API Key 的签发和存储Agent 服务暴露在公司内网不等于可以裸奔。Agent-Reach 做了一个简单的 API Key 管理Agent 注册时自动生成一对 Key 和 Secret调用方访问/v1/reach时必须在 Header 里带上X-API-Key。存储方面不要用明文存 Secret。我在数据库里存的是加盐哈希值盐是每个 Key 独立生成的随机串。校验的时候用hmac.compare_digest做常数时间比较避免时序侧信道攻击。签发逻辑很简单import secrets import hashlib import hmac def create_api_key(agent_id: str) - tuple[str, str]: key fak_{secrets.token_hex(16)} secret secrets.token_hex(32) salt secrets.token_hex(16) hashed_secret hashlib.sha256(f{salt}:{secret}.encode()).hexdigest() # 保存 key, salt, hashed_secret 到数据库 return key, secret调用方拿到的是key和secret但secret只在签发时显示一次之后就只剩哈希。这样即使数据库泄漏攻击者也拿不到可用凭据。很多人问内部服务之间搞这么严格有没有必要。我的看法是Agent 往往能触发比较重的操作比如自动发工单、自动查库哪怕只是内网有一层鉴权也能挡住很多误操作和误调用。3. 实操过程从零实现 Agent-Reach 调度网关3.1 技术选型和工程结构Agent-Reach 我用了 Python FastAPI Redis SQLite。选 FastAPI 是因为它原生支持异步适合做并行调度的 IO 密集场景。Redis 用来存注册表和健康状态主要利用它的过期时间特性做心跳天然回收。SQLite 用来存调用日志轻量到不需要单独部署数据库服务。如果调用量真的大了SQLite 换 PostgreSQL 也只需要改连接串和小部分 SQL。工程结构我分得很清晰agent_reach/ ├── main.py # FastAPI 入口 ├── models.py # Pydantic 模型 ├── registry.py # 注册与心跳逻辑 ├── gateway.py # 统一调度逻辑 ├── adapters.py # 协议适配器 ├── circuit_breaker.py # 熔断器 ├── storage.py # SQLite 读写 └── config.py # 配置管理这样拆的好处是每个文件职责单一出错的时候能快速定位。比如心跳异常直接看registry.py调用超时看gateway.py和adapters.py。3.2 注册与心跳Agent 怎么告诉网关“我还活着”Agent 上线后调用注册接口把AgentInfo提交上来网关把信息写入 Redis并启动一个心跳计时。心跳机制是最简单也最实用的存活检测方式from fastapi import APIRouter, HTTPException import redis.asyncio as aioredis router APIRouter() redis_client aioredis.from_url(redis://localhost:6379/0, decode_responsesTrue) router.post(/v1/agents/register) async def register_agent(info: AgentInfo): key fagent:{info.agent_id} exists await redis_client.exists(key) if exists: # 已有的 Agent 需要先下线再重新注册或者用版本号控制覆盖 raise HTTPException(status_code409, detailagent already exists) await redis_client.hset(key, mappinginfo.model_dump()) await redis_client.expire(key, 90) return {status: registered, agent_id: info.agent_id}心跳接口更简单Agent 每隔 30 秒来打一次点router.post(/v1/agents/{agent_id}/heartbeat) async def heartbeat(agent_id: str): key fagent:{agent_id} if not await redis_client.exists(key): raise HTTPException(status_code404, detailagent not found) await redis_client.expire(key, 90) await redis_client.hset(key, status, online) return {status: ok}Redis 的expire在这里充当了“自杀倒计时”Agent 每 30 秒续命一次如果超过 90 秒没有续命Key 自动消失下次调用就不会路由到它。这个设计比“维护一张状态表再定时扫描”轻太多完全靠 Redis 的 TTL 机制兜底不会漏。90 秒这个值是 30 秒心跳间隔的 3 倍给网络抖动留了足够余量。实操中踩过一个坑Agent 用同一个 Redis Key 存所有字段用hset更新单字段时如果不小心传入空对象会把整个 Key 覆盖掉。所以注册和心跳建议用不同的方法注册用完整写入心跳只更新status和expire。3.3 统一网关与并行调度一条请求分发到多个 Agent网关是整个项目的门面。外部调用方只调一个接口网关内部做能力匹配、目标筛选、批量调度。这里我实现了两个核心场景单目标调用请求里明确指定了target_agent。按能力批量调用请求里声明target_capability网关找出所有具备该能力的在线 Agent并行调用收集全部结果返回。批量调用是 Agent-Reach 最出彩的场景。比如调用方想知道一段文本的意图归属但团队里有两个意图识别 Agent用不同模型训练的你想让他们投票或者对比结果。这时网关并行发给两个 Agent然后汇总。调度核心代码import asyncio import time import uuid from fastapi import APIRouter, Request from adapters import get_adapter from registry import get_online_agents_by_capability from circuit_breaker import CircuitBreakerRegistry from storage import save_log router APIRouter() breaker_registry CircuitBreakerRegistry() def gen_trace_id() - str: return ftrc_{int(time.time())}_{uuid.uuid4().hex[:12]} router.post(/v1/reach) async def reach(request: Request): body await request.json() trace_id gen_trace_id() target_capability body.get(target_capability) target_agent body.get(target_agent) payload body.get(payload, {}) if target_agent: agents [get_agent(target_agent)] else: agents await get_online_agents_by_capability(target_capability) if not agents: return {trace_id: trace_id, status: error, message: no available agent} semaphore asyncio.Semaphore(10) # 控制并发数防止 Agent 被打爆 results [] async def call_one(agent): async with semaphore: breaker breaker_registry.get(agent.agent_id) if breaker.is_open(): return {agent_id: agent.agent_id, status: rejected, reason: circuit_open} try: adapter get_adapter(agent.protocol) result await adapter.invoke(agent, payload, trace_id, timeoutagent.max_timeout) breaker.record_success() return {agent_id: agent.agent_id, status: success, data: result} except Exception as e: breaker.record_failure() return {agent_id: agent.agent_id, status: error, message: str(e)} results await asyncio.gather(*[call_one(a) for a in agents]) await save_log(trace_id, body, results) return {trace_id: trace_id, results: results}有几个细节我特别想强调。asyncio.Semaphore(10)是给 Agent 的保护罩。如果你的场景是调用方并发 500 个请求全部打到网关网关如果同时把这 500 个请求转发给同一个 AgentAgent 大概率直接服务雪崩。信号量限制的是“对后端 Agent 的最大并发”不是“网关接收请求的并发”。这个数值要根据 Agent 本身的吞吐能力来调我之前默认 32结果把一个只能抗住 10 并发的老 Agent 打崩了后来缩到 10 就好了。asyncio.gather默认的行为是其中一个协程抛异常就整体取消但我在call_one内部已经 try 了一遍异常被转成结果对象所以gather不会因为单个 Agent 挂掉而整体失败。这一点当初也吃了亏第一版我没在call_one里捕获异常结果一个 Agent 超时整个 batch 的其它 Agent 结果全部丢失。3.4 协议适配器让旧 Agent 不改造也能接入最头疼的接入场景是那种“不能改代码”的 Agent。有些老系统是别的组维护的你不可能让对方为了配合你的网关改接口。所以 Agent-Reach 把协议适配逻辑放在网关注侧。适配器模式很好用。我定义了一个基础接口然后用不同的实现类处理不同协议from abc import ABC, abstractmethod import httpx class AgentAdapter(ABC): abstractmethod async def invoke(self, agent, payload, trace_id, timeout) - dict: pass class HttpJsonAdapter(AgentAdapter): async def invoke(self, agent, payload, trace_id, timeout) - dict: async with httpx.AsyncClient(timeouttimeout) as client: resp await client.post( agent.endpoint, json{trace_id: trace_id, payload: payload}, headers{X-Agent-Trace-ID: trace_id} ) resp.raise_for_status() return resp.json() class WebhookAdapter(AgentAdapter): async def invoke(self, agent, payload, trace_id, timeout) - dict: # 针对回调式 Agent先推任务再登记回调地址 callback_url fhttp://agent-reach.internal/callback/{trace_id} async with httpx.AsyncClient(timeouttimeout) as client: resp await client.post( agent.endpoint, json{payload: payload, callback_url: callback_url} ) resp.raise_for_status() # Webhook 场景立即返回受理结果业务状态走回调 return {submitted: True, task_id: resp.json().get(task_id)}HttpJsonAdapter 处理普通的 HTTP JSON 接口WebhookAdapter 处理回调型 Agent。接入新 Agent 的时候往往只需要新增一个适配器类不动网关核心代码。实际接入中我发现自己写的适配器有个问题每个 Agent 对 payload 字段的命名习惯不同有的喜欢{ text: ... }有的喜欢{ content: ... }。这种差异没法全自动消除只能靠注册表里增加一个可选的payload_mapping字段在适配器 invoke 之前做一层映射。这也是一个真实落地的妥协别想着 AI 自动解决字段映射先手动配置满足大部分情况就好。3.5 超时、重试与熔断参数怎么定才合理这块是我踩坑最多的部分值得单独讲。超时策略我采用“注册表声明 网关兜底”双层策略。Agent 注册的时候声明自己的max_timeout网关调用时用这个值作为网络请求超时。但为了防止某些 Agent 乱填网关设了一个硬顶默认 15 秒超过就拒绝注册或强制用 15 秒。 这里有个至理名言对 Agent 的超时不要设太短因为大模型推理不是传统的数据库查询生成式响应要 2 到 10 秒都很正常设 3 秒只会得到一片超时错误然后大家还以为是 Agent 坏了。重试策略对于非幂等操作重试要非常谨慎。Agent-Reach 默认只对GET类或调用方明确标记idempotent: true的请求做一次重试其余请求不自动重试。因为我没办法保证所有 Agent 都做了幂等处理重试导致的重复工单比“调用失败”更可怕。 正常模式下我会对返回 502、503、504 这些明确的网关错误做一次重试对应用层返回 500 直接放给调用方自行决定。熔断策略我用了一个简化的滑动窗口熔断器按 Agent 维度独立统计最近 1 分钟内的调用成功率。如果错误率超过 50% 且调用量大于 10 次熔断器打开后续请求直接短路返回不再转发。这个策略在保障线程资源上效果很明显避免了“一个 Agent 已经死了但调用方还在疯狂请求把网关连接池耗尽”的情况。import time class CircuitBreaker: def __init__(self, threshold_rate: float 0.5, min_requests: int 10): self.threshold_rate threshold_rate self.min_requests min_requests self.window [] self.open False def record_success(self): self.window.append((time.time(), True)) def record_failure(self): self.window.append((time.time(), False)) def is_open(self) - bool: if self.open: return True # 清理窗口外数据 now time.time() self.window [(t, s) for t, s in self.window if now - t 60] if len(self.window) self.min_requests: return False failure_count sum(1 for _, s in self.window if not s) failure_rate failure_count / len(self.window) if failure_rate self.threshold_rate: self.open True # 模拟半开恢复若干秒后自动关闭 self.open_until now 30 return self.open这版熔断器很朴素没有用三方库核心逻辑就是窗口统计加阈值判断。实际部署中我做了个定时任务熔断器打开 30 秒后自动置为关闭让流量逐渐恢复如果 Agent 还没恢复就会再次触发熔断形成“试探-失败-熔断-恢复”的循环。半开状态的试探流量控制我还没做精细但目前的自动恢复已经够用。4. 常见问题与排查技巧实录4.1 排查问题速查表下面是 Agent-Reach 实际跑起来之后我在运维排障中遇到的高频问题整理成速查表新手可以直接按表格定位故障现象可能原因排查步骤解决方案注册 Agent 后调用返回 404路由没有匹配到 agent_idNamespace 隔离问题查 Redis 中agent:{id}是否存在查调用方传的 ID 是否带空格注册时对 id 做 strip调用时统一小写Agent 心跳正常但频繁被标记离线心跳更新了status字段但没续期 TTL用redis-cli ttl agent:{id}查看剩余时间心跳逻辑里必须同时expire续期批量调用时一个 Agent 超时拖慢全部asyncio.gather没有为每个子请求单独兜底看日志中 gather 是否被整体取消在子协程内捕获异常超时分线程控制Agent 返回 200 但业务数据是错误内容适配器没有做响应 schema 校验查看 Agent 原始响应与网关透传响应差异增加可选response_validator钩子调用量平稳但 Agent 偶尔连接被拒网关并发超过 Agent 承受上限查看 Agent 端口连接数是否满调低 Semaphore 值或给 Agent 扩容熔断器一开就永久关闭所有流量熔断器没有自动恢复机制看is_open是否有半开逻辑增加 30 秒自动关闭逻辑同一 task_id 被重复执行调用方超时后重试了非幂等请求查看网关日志中同一 task_id 出现次数对 task_id 做去重缓存相同 ID 直接返回上次结果日志表增长过快每次调用记录全部 payload 和响应查询 SQLite 表大小生产环境只保留 trace_id、耗时和状态码payload 放冷存4.2 三个印象深刻的排障案例第一个案例是误判 Agent 离线。现象是有个数据分析 Agent 每天凌晨都会出现“短暂离线”业务方经常在凌晨收到告警。查了很久发现Agent 的心跳是进程内一个定时任务发的但该 Agent 凌晨会做模型热加载热加载期间 GIL 被占死心跳线程发不出去。这不是网络问题是 Agent 自己的执行节奏问题。解决方案是让心跳逻辑跑在独立线程或者干脆用外部探活脚本不要依赖 Agent 进程内的线程。这也提醒我一个原则心跳必须跟核心业务逻辑隔离。第二个案例是并发信号量设置不当。当时网关刚上线我把 Semaphore 设成了 64想着并发越高吞吐越大。结果接进来一个老 Agent用的还是 Flask 单进程部署并发一高直接在 accept 队列堆积调用方体验极差。后面把信号量降到 8再配合这个 Agent 的max_timeout声明虽然单个请求排队长一点但至少没有把老 Agent 打崩。这个教训告诉我网关的并发上限取决于最弱那个 Agent 的承受能力而不取决于网关自身的理论吞吐。第三个案例是 API Key 泄露问题。有一次我们排查一个奇怪的调用来源发现有人在内部论坛贴了一张截图里面无意间暴露了完整的 Key 和 Secret。因为没有轮换机制只能紧急线下禁用一个 Key 并重新签发。后来我在管理后台加了“密钥最后使用时间”和“一键轮换”功能同时把新 Secret 的显示改成一次性展示。内部服务也一样会出安全事件提前做好轮换工具能省很多事。4.3 排查链路问题的一个小习惯Agent-Reach 的日志我设计得比较简单每一条调用先写一行start日志包含 trace_id、target_capability、payload 前 100 字符结束再写一行end日志包含 trace_id、耗时、状态码。排查问题的时候用 trace_id 把所有日志 grep 出来按时间顺序看一遍基本就能定位问题出在网关还是 Agent。这个小习惯帮我省了大量盲查时间比任何复杂监控都好使。后来我还加了简单的耗时告警当单个请求在网关侧的耗时超过max_timeout的 80% 时记一条慢调用日志方便提前识别潜在劣化。5. 部署形态与后续演进的一些经验5.1 轻量部署一台机器加 systemd 就够了很多团队问 Agent-Reach 要不要上 K8s我的建议是前期完全没必要。这个服务本身很轻主要消耗在连接和转发上CPU 占用不高。我用一台 2C4G 的小机器加上 systemd 管理进程配合 Nginx 做入口代理就跑得非常稳。systemd 配置也很简单核心就三行[Unit] DescriptionAgent-Reach Gateway Afternetwork.target redis.service [Service] ExecStart/usr/local/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2 Restartalways RestartSec5 EnvironmentAGENT_REACH_CONFIG/etc/agent_reach/config.yaml [Install] WantedBymulti-user.targetworkers2是专门调的参数。这个服务虽然有异步特性但多进程能利用多核 CPU同时也不会因为一个进程崩溃导致全挂。注意这里不能用不够成熟的--reload模式上生产那个只适合开发。Redis 我用的是独立实例专门给 Agent-Reach 用没跟公司其他 Redis 混用。原因很简单注册表的 Key 如果被别的不小心清掉服务会静默失效。独立实例虽然浪费一点内存但隔离了事故域心里踏实。5.2 注册表的持久化问题Redis 虽然好用但有一个隐患重启后 Key 会丢。如果网关重启所有 Agent 的状态会全部变成未注册Agent 要等下一次心跳才能恢复身份。短期可以接受因为心跳间隔最多 90 秒服务能自愈。但如果你希望重启后立刻就有完整的注册表可以在 Agent 注册时同步落一份到 SQLite启动时先加载 SQLite 再叠加 Redis 的实时状态。我的做法是加一个启动时的“重建注册表”流程从 SQLite 读出所有 Agent 的静态信息然后逐个到 Redis 里查 TTL如果不存在就标记为 offline等 Agent 下次心跳再转 online。这样既不丢元数据又保留了 Redis 的时效性。5.3 两个可以继续扩展的方向Agent-Reach 再往下走有两个方向我觉得价值很高。一个是能力路由升级。目前是标签匹配简单但粗糙。后续可以引入调用方声明的“输入输出契约”网关根据契约做更精确的参数映射和结果校验。比如某个 Agent 需要{text}作为输入但调用方传的是{content}映射规则可以在注册表里配清楚。另一个是Agent 之间的协同编排。现在网关是“一次请求-多个 Agent 并行执行-汇总结果”但有些任务天然是分步骤的先意图识别再抽取实体最后查数据库生成回答。这需要引入一个简单的 DAG 编排引擎让网关支持多步骤流水线调用。我目前是把每一步写成一个 Python 逻辑硬编码在网关里还没做通用化但这是 Agent-Reach 最自然的演进方向。最后想分享一个实际心得。做 Agent-Reach 这类中间层最忌讳的不是技术实现不够优雅而是定位跑偏。我第一版的时候忍不住加了“动态扩容 Agent”“自动模型选择”“Prompt 自动优化”这些看起来很 AI 的功能结果发现根本没法在真实业务里落地因为每个团队对自己的 Agent 都有不可言说的定制逻辑通用优化根本推不动。后来老老实实回归到“注册、调度、适配、观测”这四个朴素的点上项目才真正跑起来。如果你想做一个有实际价值的 Agent 基建项目先管好触达这件事别被“智能”两个字带偏。