企业官网建设流程全解析

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(..., description="Agent 的调用地址") protocol: AgentProtocol = AgentProtocol.HTTP capabilities: List[str] = Field(..., description="能力标签列表") status: AgentStatus = AgentStatus.OFFLINE owner: str = Field("", description="负责人，用于告警通知") version: str = Field("1.0.0", description="Agent 版本号") 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是整个链路的唯一标识，网关生成后传给 Agent，Agent 在日志里带上这个 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 = f"ak_{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_responses=True) @router.post("/v1/agents/register") async def register_agent(info: AgentInfo): key = f"agent:{info.agent_id}" exists = await redis_client.exists(key) if exists: # 已有的 Agent 需要先下线再重新注册，或者用版本号控制覆盖 raise HTTPException(status_code=409, detail="agent already exists") await redis_client.hset(key, mapping=info.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 = f"agent:{agent_id}" if not await redis_client.exists(key): raise HTTPException(status_code=404, detail="agent 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 f"trc_{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, timeout=agent.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 个请求转发给同一个 Agent，Agent 大概率直接服务雪崩。信号量限制的是“对后端 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(timeout=timeout) 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 = f"http://agent-reach.internal/callback/{trace_id}" async with httpx.AsyncClient(timeout=timeout) 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 实际跑起来之后我在运维排障中遇到的高频问题，整理成速查表，新手可以直接按表格定位：

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] Description=Agent-Reach Gateway After=network.target redis.service [Service] ExecStart=/usr/local/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2 Restart=always RestartSec=5 Environment=AGENT_REACH_CONFIG=/etc/agent_reach/config.yaml [Install] WantedBy=multi-user.target

workers=2是专门调的参数。这个服务虽然有异步特性，但多进程能利用多核 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 基建项目，先管好触达这件事，别被“智能”两个字带偏。