企业官网建设流程全解析

Agent-Reach 这个项目名，是我今年折腾多智能体协作时被现实逼出来的。

一开始我以为难点在模型推理能力，后来才发现真正乱的是 Agent 之间的触达关系：谁可以找谁干活、按什么条件找、边界划在哪里。当你同时跑十几个带工具的 AI Agent 时，这些问题的优先级会迅速超过模型本身的智能，毕竟一个模型再聪明，也不知道自己该找哪个 Agent 拿数据、该把结果交给谁。Agent-Reach 的思路很简单：把“谁和谁可以触达、以什么规则触达”从业务代码里抽出来，做成一个独立的触达层。

这篇文章我会从项目缘起、架构设计、最小实现、排查实录到适用边界，完整记录这套东西的设计过程和踩过的坑，适合正在做多 Agent 编排、AI 自动化流水线，或者已经在用框架但控制不住 Agent 之间乱串门的工程师。

1. 项目缘起：Agent 一多，先乱的是通信，不是推理

1.1 三个真实场景让我决定动手

第一个场景是客服机器人集群。售前、售后、退款、物流四个 Agent 各管一段，但用户不会按后台的服务目录说话，他只会说“我要退昨天那单”。这句话同时命中售后 Agent 的“退货咨询”能力和退款 Agent 的“退单处理”能力。如果两个 Agent 都去抢这个请求，用户会被两套话术同时轰炸；如果都让，就会没人处理。传统做法是在入口写一堆 if-else 规则，但业务一变，规则就废。

第二个场景是内容生产流水线。我用 Agent A 写初稿，Agent B 审核，Agent C 发布。某天 C 发现稿子有敏感词，需要回退给 B 修改。按理说调用链是清晰的，但 A、B、C 各自有自己的上下文和工具集，C 怎么知道 B 现在能不能接、接了之后 B 有没有权限改这部分内容？在代码里把调用关系写死，跟上一条一样，改动成本很高。

第三个场景是 IoT 指令控制。我做过一个实验环境，里面有 20 多个可以操作设备的 Agent，每个 Agent 管一组设备。用户来一句“把客厅灯关掉”，这句话理论上能匹配所有带“灯”能力的 Agent，因为它们的 capability 标签里都有 lighting。如果按匹配度分发，会有好几个 Agent 同时对真实设备下单。那时候我就意识到，Agent 之间的触达必须设计成一层专门的问题，不能当作普通函数调用来处理。

1.2 Agent-Reach 到底解决哪几类问题

从上面几个案例逆推，Agent-Reach 要回答的是三个问题。

第一是发现问题。新 Agent 上线，老 Agent 不应该改一行配置就能找到它。就像微服务里的服务注册发现，但比那个难一点，因为 Agent 的能力是动态的，它可能今天多挂了一个“图片识别”的工具，明天又下掉了。

第二是路由问题。一个请求来了，应该触达哪一个 Agent、按什么权重触达、要不要灰度、目标 Agent 当前是否过载。这层逻辑如果散落在各个 Agent 的 prompt 里，几乎没法维护。

第三是边界问题。哪些 Agent 可以互访，哪些必须隔离，上下文里哪些字段能跨 Agent 传递，哪些只能留在本地。这个比权限系统更细致，因为 Agent 的“越权”往往不是安全意义上的漏洞，而是业务语义的错乱——退款 Agent 改了订单价格，谁该负责？

我把这三类问题统称为“触达问题”，Agent-Reach 这个名字里的 Reach，就是想说清楚一个 Agent 能触及的范围到底有多大。

2. 架构设计：把“触达”变成一层专门的基础设施

2.1 五个核心模块如何在一次调用里配合

Agent-Reach 不是又一个 Agent 框架，它更像一个夹在规划器和执行 Agent 之间的基础设施层。你可以把它理解成“注册中心 + 网关 + 消息总线的三合一”，只是所有接口的语义都面向 Agent。

整个系统由五个模块组成。

Agent Registry 负责登记每个 Agent 的能力快照、当前状态、负载情况和工作模式。它解决的是“有哪些 Agent 在场”的问题。

Reach Topology 维护一张有向图，图中每个节点是一个 Agent，每条边表示“从这个 Agent 出发，能触达那个 Agent”。这张图是动态更新的，不再是一张静态架构图。

Message Bus 负责异步投递调用请求。之所以不用标准的同步 REST，是因为 Agent 执行任务通常耗时较长，从几秒到几分钟不等，同步等待会让调用方长期占用连接，而且一旦目标 Agent 重启，整条链路就断了。

Policy Engine 是灵魂。它决定每条触达请求是否被允许、是否要在投递前做参数改写、要不要走降级路径。硬编码路由在 Agent 数量少的时候很顺手，但一旦多了，规则就变成了一堆无处安放的 if。

State Store 保存每一次触达的记录和状态，包括发起了谁、投给了谁、结果如何、失败原因是什么。没有这一层，整个系统就变成一个黑盒，Agent 之间相互推诿时你根本找不到证据。

在一次触达请求的生命周期里，流程是这样的：调用方把请求发给 Agent-Reach 的 API，Agent-Reach 先查 Registry 确认目标 Agent 在场，再查 Topology 确认这两个 Agent 之间可达，接着 Policy Engine 检查规则是否放行，最后把请求写进 Message Bus。目标 Agent 消费后执行，结果写回 State Store，调用方通过轮询或者回调拿到结果。

2.2 三个核心抽象：可达 Agent、调用意图、作用域与 TTL

我做了三个抽象来支撑上面这套流程，它们是我在实际代码里用的基础类。

第一个是可达 Agent。它不是简单地指 Agent 的 ID 和地址，而是一个完整的注册表条目，包含 agent_id、capabilities、access_policy、adapter 配置和当前状态。负载信息不是它自己报的，而是由 Message Bus 侧统计的投递窗口来生成，因为 Agent 说自己很闲往往不可信，真实的执行吞吐才是标准。

第二个是调用意图。它不是一次普通的函数调用，而是一个带有语义的对象，字段包括 source_agent、target_agent、intent 类型、playload、ttl、idempotency_key。之所以做成对象而非 HTTP 调用，是为了让策略引擎有机会在投递前做检查。你可以把调用意图理解成一份数字化的“协作文书”，函件到哪一步都要留痕，而不是一个没人管的数据包。

第三个是作用域与 TTL。作用域限制的是这条请求可以触达的数据范围，比如退款 Agent 只允许触达订单号前缀为 after_sale 的记录；TTL 限制的是这条请求的有效时间，超过时间就取消。这两个参数是我从分布式任务系统里借来的思路，但应用在 Agent 场景更严格，因为模型对上下文的处理是非确定的，你永远不知道它会不会把不应该带走的字段塞进结果里。

2.3 为什么不能把触达规则写死在代码里

有一个很常见的做法是在业务代码里直接写死“如果我处理的订单属于售后，就调用 refund_agent”，这在 10 个 Agent 以下能跑，但有两个隐患。

第一是变更成本。每加一个 Agent，就要重新梳理一遍所有可能调用它的地方。Agent 的职责经常微调，比如退款 Agent 增加了一个“拒单原因分析”能力，那入口处调用的判断条件可能要改，但没人记得住全部调用点，改漏了就出问题。

第二是灰度没法做。写死路由本质上是一种全有或全无的逻辑。你没法说“新上线的 Agent 先放 10% 流量”，即使想灰度，也只能临时改代码再部署。策略引擎把路由逻辑变成运行期可改的规则，灰度就是改一条配置的事，不用重新发版。

我对策略引擎的定位是：把“你能触达谁”和“你被允许触达谁”分开。Agent 在能力上可以看到很多候选目标，但在策略上只能触达规则放行的目标。规则的核心方向是 default-deny，也就是默认情况下任何 Agent 都不可触达目标，只有显式规则放行才允许。这套设计和零信任的思路在逻辑上是同构的，只不过 Agent-Reach 管的是业务边界，纯技术机制。

3. 实操：手写一个最小可用的 Agent-Reach

3.1 技术选型和目录结构

我选型的原则是：能少依赖就少依赖，能用现成组件就用现成组件，重点放在可替换的接口上。

语言选了 Python 3.11，因为这个系统要跟各种 Agent 集成，Python 生态的 Agent 库最多。API 层用 FastAPI，写 handler 轻量，文档也自动生成。注册表，我用 Redis Hash 来存 Agent 信息，利用 Redis 的 TTL 可以做 Agent 的自动过期。消息通道用 Redis Streams，而不是直接上 Kafka，理由是它用 XADD、XREADGROUP 就能提供足够可靠的消息队列语义，一个 Redis 实例同时兼了注册表和消息队列，部署成本低很多。等消息量大到 Redis 扛不住时，再替换成 Kafka 也很顺滑，因为 Redis Streams 的消费组模型本来就很接近 Kafka。

Adapters 做一层抽象，因为我发现不同 Agent 的通信方式差得很远。有些 Agent 暴露了 HTTP 接口，有些是本地函数，有些是通过 WebSocket 长连接。解决思路是给每个 Agent 配一个 adapter，Agent-Reach 只跟 adapter 交互，adapter 去跟真实 Agent 通信。

目录结构长这样：

agent_reach/ ├── core/ │ ├── registry.py # Agent 注册表 │ ├── topology.py # 触达拓扑 │ └── policy.py # 策略引擎 ├── bus/ │ ├── base.py # 消息总线抽象 │ └── redis_streams.py # Redis Streams 实现 ├── api/ │ └── main.py # FastAPI 入口 ├── adapters/ │ ├── http.py # HTTP 调用适配 │ └── local.py # 本地函数适配 └── models/ └── invocation.py # 触达请求模型

3.2 注册表与触达策略的最小实现

先用 Pydantic 定义 Agent 注册表的核心数据结构。这一层不需要多复杂，关键在于字段要留足。

from pydantic import BaseModel from typing import List, Optional class AgentRegistration(BaseModel): agent_id: str name: str capabilities: List[str] domain: str # 所属业务域，比如 "after_sale" status: str = "active" # active / paused / draining adapter: str # 用哪个 adapter，比如 "http" endpoint: Optional[str] = None

这个类对应的存储方式，是用 Redis Hash：

import redis, json r = redis.Redis.from_url("redis://localhost:6379") def register_agent(reg: AgentRegistration): key = f"agent:{reg.agent_id}" r.hset(key, mapping={ "name": reg.name, "capabilities": json.dumps(reg.capabilities), "domain": reg.domain, "status": reg.status, "adapter": reg.adapter, "endpoint": reg.endpoint or "", }) r.expire(key, 3600) # 心跳超时自动下线

注意 expire 那行。Agent 每次启动都要重新注册，并定期发心跳，如果超过一小时没更新，Registry 就认为它挂了，自动从拓扑里摘掉。这是从微服务健康检查里借鉴的手段，但比服务健康检查更严格，因为 Agent 可能只是休眠，不是崩溃。

策略引擎的最小实现，我用 JSON 规则块来做判断。这个方案的好处是规则方便存进 Redis，改的时候不用重新部署代码。

class PolicyRule(BaseModel): intent: str # 调用意图，比如 "refund_request" allow_from: List[str] # 允许的调用方列表，"*" 表示全部 allow_target_domain: str # 允许触达的目标域 scope: Optional[str] = None # 数据作用域，如 "order_id:after_sale_*" fallback: Optional[str] = None # 不允许时的降级目标 def check_policy(rule: PolicyRule, invocation: dict) -> bool: if invocation["intent"] != rule.intent: return False if "router" not in rule.allow_from and invocation["source_agent"] not in rule.allow_from: return False if invocation["target_domain"] != rule.allow_target_domain: return False if rule.scope: # 伪代码：用正则或者前缀匹配 if not invocation["payload"].get("order_id", "").startswith(rule.scope.rstrip("*")): return False return True

这个 check 函数看起来简单，但它在真实系统里承担了 90% 的边界控制。绝大多数非法触达不是被用户识别的，而是被这个函数拦下来的。

3.3 一次触达请求的完整链路

API 层暴露一个 POST /invoke 端点，接收调用方发来的调用意图。整个链路我拆成四步，每一步都有明确状态记录。

第一步，接收请求并生成 invocation ID，用 UUID 保证全局唯一。这个 ID 同时作为 idempotency key，用于防止重复投递。

第二步，执行三层检查。查 Registry，确认目标 Agent 存在且状态是 active；查 Topology，确认调用方和目标 Agent 之间有可达的边；查策略，确认这条调用意图和来源被放行。任何一层失败，就直接返回 403 或 404，并在 State Store 写一条 denied 记录。

第三步，投递到消息总线。我用的 Redis Streams 是这么操作的：

def dispatch(invocation: dict): stream_key = f"stream:{invocation['target_agent']}" r.xadd(stream_key, invocation, maxlen=10000)

每个目标 Agent 只消费自己的 stream，而不是所有 Agent 共用一个队列。这样天然的隔离，一个 Agent 消费慢了不会拖累其他 Agent，重试也只需重放单个 stream。

第四步，Agent 侧消费后，由 adapter 真正执行。执行完必须写回结果，写到 State Store。State Store 可以直接复用 Redis 的 String 或 Hash，但要注意键的设计，以 invocation_id 为主键，值为结果状态和耗时。

def finish_invocation(invocation_id: str, result: dict, status: str): r.hset(f"invocation:{invocation_id}", mapping={ "status": status, "result": json.dumps(result), "finished_at": time.time(), })

到这里，一次触达就闭环了。调用方如果需要等待结果，就轮询这个 invocation key，或者注册回调地址，由 Agent-Reach 在完成时做一次 HTTP 回调。

3.4 部署时我会注意的几件事

Agent 是分布式的，部署时这几件事特别容易踩雷。

第一，Redis 单独部署，不要和 Agent 进程混在一起。Agent 是长耗时任务，Redis 是高频读写，两者混跑会导致网络抖动和 CPU 竞争，一旦 Redis 慢，所有 Agent 的注册心跳都受影响。

第二，把 Agent 分成状态有状态和无状态两类。内容生成、审核这类 Agent 是无状态的，谁接都可以；但某些 Agent，比如正在操作一个长时间运行的订单流程，它内部维护了状态，就必须用粘性路由，同一个订单 ID 的后续调用要固定发给同一个实例。我在 registry 里加了一个 sticky_key 字段，路由的时候按照它做一致性哈希。

第三，异步消费不能丢数据。Redis Streams 投递之后，如果 Agent 消费了但没执行完就宕机，这条 invocation 会被 pending，需要单独的程序去扫描 pending 队列并处理。这个可以做成一个定时 task，每小时滚动检查一次。实测中这条 scanner 是排查僵尸任务的核心工具。

4. 试运行：四类典型故障和排查记录

4.1 静默失败：返回值丢了，但状态没写

第一个大坑是 Agent 静默失败。现象是调用方等不到结果，但目标 Agent 也没有任何报错。排查了半天，发现问题出在消费端：Agent 作为一个长期运行的 worker，如果业务代码抛了异常，而 worker 的进程没有强制退出，这个 invocation 就不会有任何后续记录。调用方不知道是没投达还是没处理，目标 Agent 也不知道它丢了一个任务。

解决方法是强制要求：一条 invocation 只要被消费，就必须在短时间内落一个状态，哪怕结果是 failed。我在 Message Bus 的 adapter 层，把整个执行过程包在 try-except 里，无论成功失败都调完 finish_invocation。这不只是技术规范，更是一种与 Agent 协作时的信任机制：系统可以容忍失败，但不能容忍无响应。

4.2 广播风暴：所有 Agent 都在响应，谁都没有责任

第二个坑更致命。初始版本的策略规则太宽松，我把所有路由都设计成了“capability 匹配即尝试”。结果有一次，一个“获取今日生产报表”的请求被丢给了 5 个有数据分析能力的 Agent，5 个 Agent 都执行了，写回了 5 个不同版本的报表。没有人出错，但结果是完全不可控的。

这暴露了一个关键问题：触达不是“谁有能力谁上”，而是“谁有职责谁上”。能力的语义太泛了，一个“报表生成”能力可能匹配到 AI 助手 Agent、传统 BI Agent、甚至数据清洗 Agent。Agent-Reach 后来把默认路由模式改成了“由拓扑图中入度最高的目标 Agent 执行”，也就是按职责显式声明，而不是按能力自动匹配。如果你真的需要多个 Agent 竞争一个任务，必须给 invocation 显式声明 competitive 模式，并配置冲突处理规则。

4.3 环形依赖：两个 Agent 互相等，死锁

第三个坑是环依赖。当时我有两个 Agent：一个做订单审核，一个做风险分析。审核 Agent 在处理某个订单时发现需要风险分析结果，于是调用了风险 Agent；而风险 Agent 的内部逻辑发现要审核订单的当前状态，又反过来调用审核 Agent。两边都在等对方的结果，所有 invocation 都卡在 pending 里，而且没有超时机制，因为我在最初设置 TTL 时留了太大的裕度，默认给到 600 秒，等于没设。

排查过程让我意识到，Agent 之间的依赖不仅存在于业务数据流里，还可能藏在上下文中。后来我加了两个自我保护：一是投递前做拓扑有向图的环检测，能够发现“查询路径”中的环；二是给所有 invocation 设置了强制的 max_ttl，最长不超过 120 秒，超过就直接失败并返回可读的错误。环检测不一定能防住所有情况，因为 Agent 是动态的，但有了 TTL 兜底，系统不会挂死，最多失败几条，人力再去处理。

4.4 重复执行与状态不一致：重试其实是双刃剑

第四个坑跟 Redis Streams 的消费机制有关。默认情况下，如果消费端处理了消息但是没有及时 ACK，Redis 会把这条消息认为是 Pending，下一次重试投递时，Agent 可能再执行一遍同一个任务。普通查询重复执行还好，但退款、创建工单这类写操作，重复执行是致命的。

真正负责的做法是给每个 invocation 配一个幂等键。我在模型里加了 idempotency_key，这个键不是 invocation ID，而是由调用方生成的业务幂等键，例如“用户-201-退款请求”。消费者在执行写操作前，先检查这个键是不是已经处理过，如果处理过就带上已处理结果直接返回。这个技巧，是我从支付网关的接口设计里搬过来的，成本低还非常管用。

4.5 一组测试环境参考数据

我的测试环境是 6 台虚拟机，跑 20 个 Agent，Intel 64 核的普通配置，Redis 单独一台。负载压到每分钟 2000 条 invocation 时：

• 路由层（从 API 接收到投递到 Stream）的 P99 耗时约 40ms，基本瓶颈在 FastAPI 和 Redis 之间的网络往返。
• 策略检查的平均耗时约 0.3ms，主要开销是反序列化规则块和匹配 scope 前缀。
• 增加策略规则后，性能下降不明显，因为规则数在几百条量级时足够轻量。
• Agent 消费端的耗时完全看业务复杂度，从几百毫秒到几十秒都有。
• 引入 Agent-Reach 之后，系统的异常率从 18% 降到 3% 左右。18% 里的很多异常，来自“该响应的 Agent 没响、不该响的 Agent 抢着响”。

这些数据不算是权威基准测试，只代表这套架构在我这种中小规模的负载下，性能开销是可接受的。如果你要做超大流量，建议优先替换消息层。

5. 什么时候该用 Agent-Reach，什么时候别用

5.1 适合的三种典型场景

第一种，你有超过 10 个职责不同的 Agent，并且它们共享同一批工具或上下文。这个数量级，人工维护调用关系开始吃力，Agent-Reach 的价值开始显现。

第二种，Agent 数量不固定，经常新增、下线能力。比如你每周都在给 Agent 添加新技能或工具，那静态的调用链就是噩梦。动态注册和策略引擎会帮你省掉大量改动。

第三种，需要做责任隔离或审计。比如你想做到：退款 Agent 永远不能访问库存数据，外部合作方的 Agent 只能触达公司对外暴露的接口。传统的权限系统控制的是“用户能否调用某 API”，Agent-Reach 控制的是“Agent 能否触达某 Agent”，这个隔离逻辑是同一套，但在语义上更贴近 Agent 协作场景。

5.2 不适合的场景，我也不会硬上

如果你只有两个 Agent，做简单的串行流水线，那用 Agent-Reach 是浪费。直接写一个函数调用，几十毫秒就完事，引入消息总线、注册表、策略引擎，延迟至少增加几百毫秒。这种场景下，系统的复杂度往往比 Agent 本身更祸。

如果你只有一个 Agent，配了一堆工具，那你需要的只是 function calling 或者工具路由，不是跨 Agent 触达。把 Agent-Reach 硬塞进来，属于杀鸡用牛刀。

我自己的判断标准很朴素：如果你不需要回答“这个请求该由哪个 Agent 处理”这个问题，就不需要它；只要你这套系统有多个 Agent、并且职责经常重叠，就值得认真考虑。

5.3 后续我想做的三个扩展方向

第一，自适应路由。现在路由规则是人工配置的，但我已经积累了历史 invocation 的执行结果。可以做一个简单学习者，定时扫描成功率和耗时，把表现稳定的 Agent 权重调高，失败的 Agent 自动降级。这个不难，但要注意不能让它学出怪路径，所以要在策略引擎之上约束。

第二，拓扑自动学习。目前 Topology 的边是手工建，新 Agent 上线时需要声明我允许谁触达它。其实可以通过分析历史 invocation 的调用模式来建议边的建立，然后由人确认。这能减少运维成本，但还不能全自动，因为自动生成的边可能带来环和安全隐患。

第三，跨实例的 Reach 协议。两个独立部署的 Agent-Reach 实例，可以像联盟一样开放一部分可达能力。比如一个团队的内容生产 Agent，可以调用另一个团队的图片生成 Agent，中间通过 Agent-Reach 的边界网关做鉴权和限流。这一块涉及跨部门的责任边界，但值得探索。

现在返回去谈个人体会。Agent-Reach 最让我受益的设计，不是路由快，也不是消息队列稳，而是那一行 default-deny 的默认策略。所有 Agent 默认不可触达，只有显式规则放行，这是我在整个项目里最坚持的一条。初期会觉得繁琐，每加一个 Agent 都要手动写规则，但后来每次出问题都能快速定位，靠的正是这条铁规矩。如果让我重新做一次，我会更早把它实现，而不是等广播风暴出现了再补。