企业官网建设流程全解析

1. 项目缘起：Agent-Reach 是什么，它盯上了哪个痛点

做了快两年多智能体落地，我一直被一个问题卡着：单个 Agent 越做越聪明，但一旦到了“几个 Agent 一起干活”的阶段，问题就全冒出来了。

A 模型生成的结论，B 模型接收时格式对不上；C 任务跑到一半需要人确认，消息发不出去；D 工具调用失败之后没有重试机制，整条流程直接断在那儿。最后呈现的效果就是：明明每个智能体单独跑都挺正常，一拼在一起就变成一盘散沙。

Agent-Reach 就是我为了领域这个问题做的智能体触达与通信基础设施，定位一句话总结：让多个 Agent 之间、Agent 与工具之间、Agent 与人工之间，用一种约定好的、可追踪的方式互通消息。它不是一个具体业务功能的智能体，而是一个总线层，干的事类似公司前台：所有外部消息进来，前台先登记备案，再按标签路由到对应的人。

这个项目适合谁参考？两种人。一种是正在做多智能体协作方案，被“消息乱飞、流程断档、没人知道现在跑到哪个环节”折磨得睡不好觉的工程师；另一种是想给自己的智能体应用加一层可控、可审计、可人工介入能力的独立开发者和产品负责人。你要接的未必是那种万人级别的并发系统，Agent-Reach 这种轻量级消息中枢反而更贴合中小规模的落地场景。

别听到“基础设施”就觉得重，这套东西不是要取代什么大型消息队列平台，它就是想解决多智能体协作里“触达”这个环节的标准问题：谁发起、谁能接收、消息怎么走、失败了怎么补、人在哪里介入。说白了，它给 Agent 系统装了一根标准的“电话线”，而不是帮每个 Agent 重新发明一个电话交换机。

2. 整体架构：三个核心模块和一个消息包

2.1 把消息当作信封来理解

Agent-Reach 的核心抽象就一个，我觉得特别适合新手理解：消息是一封标准格式的信封。

信封上固定写着三块内容：

• 元数据（metadata）：谁发的、发给谁、什么类型的请求、唯一的编号。
• 载荷（payload）：真正要传递的业务内容，比如一张工单的标题和描述、一段待审核的话术、一批需要调用的工具参数。
• 附加信息（context）：关联的上文信息、来源、优先级、过期时间等。

有了统一的信封结构，ABC 三个 Agent 就不再各自定义消息格式。以前 A 用 JSON 的 message 字段，B 用 msgText，C 用 content，对接起来全是字段映射的脏活。现在大家只认 Agent-Reach 的消息包结构，只要按信封格式拆装，剩下的事情就全交给了基础设施。

2.2 三大核心模块：路由、队列、状态跟踪

为什么我把 Agent-Reach 拆成三个模块而不是揉成一个服务？因为我踩过揉一团的坑。早期版本我把路由、存储、任务处理全写在一个进程里，刚开始消息量少没什么感觉，一旦任务多了，日志里全是互相穿插的调用记录，排查一个问题得翻半天。

现在的架构是三个模块各司其职：

路由模块（Reach-Router）：负责决定消息到哪里去。它维护着一张内存态的路由表，接收端 Agent 启动时会向它注册自己负责哪些主题、哪些类型的消息。路由支持三种方式：精确匹配（发给指定 agent）、主题匹配（发给所有订阅了某主题的 agent）、语义匹配（按意图关键词路由）。

触达模块（Reach-Broker）：负责消息的可靠投递。Broker 把消息写入待发队列，尝试连接接收方。接收方确认收到才从队列移除，没有确认就按照重试策略等待后重发。它的存在保证了消息不会因为接收方瞬时抖动而静默丢包。

状态跟踪模块（Reach-Watch）：负责记录每一条消息的生命周期。从投递到确认，从处理开始到结束，从超时未处理到转入人工，每一步都有迹可循。这三个模块共享一个数据库，但通过不同的端口和进程独立部署。

模块独立还有一个附带好处：调试时可以单独重启路由而不影响正在处理的任务。我现在经常收到私信问“为什么我的消息积压在队列里”，查到最后十有八九是路由表挂了而 Broker 还在傻等。拆开部署以后，这种问题一眼就能定位。

2.3 消息格式的设计细节

Agent-Reach 定义的消息包用 JSON 传输，大致长这样：

{ "id": "msg_20250302105123_8f3a1c", "from": "agent.customer_service", "to": "agent.billing", "type": "request", "subtype": "refund_check", "priority": "high", "timestamp": "2025-03-02T10:51:23Z", "expire_at": "2025-03-02T11:21:23Z", "trace_id": "trace_9e2b7d9a1c", "idempotency_key": "refund_check_88210", "payload": { "order_id": "88210", "refund_amount": 129.90, "reason": "duplicate_charge" }, "context": { "source": "customer_service_web", "agent_version": "cs-agent-b2b-v1.4" } }

我特意设计了两个字段，trace_id和idempotency_key，这俩是我被现实教育出来的。

没有trace_id的时候，一条消息从客服 Agent 发给账单 Agent，账单 Agent 又调用支付工具，中间任何一个环节出问题，你在日志里看到的只是孤立的一条条记录，根本拼不出一条完整链路。有了trace_id，整条链路的日志串起来，排查问题的时间能缩短一半。

idempotency_key治理的是重复投递问题。重试机制虽然保证了不丢消息，但也带来了副作用：可能一条消息被投递了两次。有一次账单 Agent 收到两次退款请求，直接退了两遍钱，还是用户打电话来才发现的。从此我加了这个幂等键，接收方处理之前先查这个键，处理过就直接忽略。

3. 关键设计决策：为什么不能直接调 API

3.1 点对点调用的三大噩梦

提到多智能体协作，很多人第一反应是：直接让 Agent A 调用 Agent B 的 API 不就行了？

确实，两三个 Agent 的时候，点对点调用跑起来很爽。但一旦规模上来，就进入噩梦模式。

第一个噩梦是接口爆炸。3 个 Agent 互相连通需要 3 组接口，5 个需要 10 组，10 个就是 45 组。每一组的鉴权方式、参数格式、错误码都各搞一套，光是维护对接文档就能让人崩溃。

第二个噩梦是调用链不可控。A 调 B，B 调 C，C 挂了，A 只知道 B 没返回，根本不知道是 C 出了问题还是网络抖动。失败的定位需要穿透好几层调用，每次排查都是一场考古。

第三个噩梦是人为介入没有抓手。API 请求发出去就像泼出去的水，如果业务规则规定某类请求必须人工审核，API 模式根本接不住这个流程。

3.2 Agent-Reach 的解法：总线加可控投递

Agent-Reach 把“调用”降级为“触达”。调用是同步的、强耦合的、点对点的；触达是异步的、松耦合的、可路由的。

这个设计的直接效果，是新增一个 Agent 的成本大幅下降。新 Agent 启动后，只需要往路由模块注册自己负责的主题和消息类型，就能收到对应消息，其他 Agent 根本不需要改一行代码。

另一个效果是人为介入终于有了标准抓手。我把人工介入设计成了状态跟踪模块里的一种“特殊接收方”。消息被标记为 pending_human_review 后，不会推给任何 Agent 处理，而是在等待区等待人工拉取。这里有一个细节，人工审核消息的超时时间我会设得比 Agent 超时更长，例如 Agent 消息 30 秒内不确认就重试，人工审核我给了 24 小时。为什么？因为指望一个正在开会的人马上响应系统通知，是不现实的，设置过短的超时只会导致一堆误报。

还在 Agent 之间设了两级重试机制。接收方没确认，Broker 先等待 5 秒重试一次；连续三次失败，消息进入死信队列，由状态跟踪模块通知管理员。整个过程不需要调用方参与，真正的“发完即忘”。

3.3 异步带来的另外一个好处：并行与恢复

同步调用链上，如果 A 要等 B 回来才能做下一步，A 的计算资源就这么白等着。现在消息扔给了 Broker，A 可以立刻去处理下一个任务，B 完成后把结果作为新消息发回来，A 再接住。这本质上是消息驱动架构。

还有一个恢复场景值得说。有一次我把整个 Broker 进程重启，心里七上八下怕消息丢了。重启完成之后，Broker 从数据库里把还没确认的消息重新捞进队列继续投递。当时我心里那叫一个踏实。如果用的是点对点同步调用，这个时间窗口内所有请求都只能 502。

4. 实操全流程：从在 Agent 里注册到实际路由

4.1 基础设施准备

Agent-Reach 在实际部署时依赖两个基础组件：一个消息队列和一个状态数据库。我推荐轻量方案：

• 消息队列用 Redis 的 Stream，理由是小规模场景下 Redis 已经够用，省去单独部署一套完整消息队列组件的成本。
• 数据库用 MongoDB，存放状态快照。如果你不想额外引入 MongoDB，SQLite 也能跑，只是少了点并发能力。

依赖清单里主要是几个 Python 库：redis、pymongo、tornado，用来提供 HTTP 和 WebSocket 接口。

4.2 配置路由模块

Agent-Reach 的路由模块配置写在 YAML 里，我拿一个典型的客服场景举例：

router: listen_port: 9030 agents: - name: "agent.customer_service" topics: ["customer.inbound", "customer.feedback"] endpoint: "ws://localhost:9101/agent/ws" heartbeat: 30 - name: "agent.billing" topics: ["billing.inquiry", "billing.refund"] endpoint: "ws://localhost:9102/agent/ws" heartbeat: 30 - name: "agent.inventory" topics: ["inventory.check"] endpoint: "ws://localhost:9103/agent/ws" heartbeat: 30 routing_rules: - rule_id: "billing_mention" condition: "internal_ticket includes 账单" target: "agent.billing" - rule_id: "inventory_stock" condition: "internal_ticket includes 库存" target: "agent.inventory" - rule_id: "fallback_default" condition: "*" target: "agent.customer_service"

每个 Agent 会跟路由模块保持一个 WebSocket 长连接，每 30 秒发一次心跳。心跳的目的是让路由模块知道这个 Agent 还活着、还能接消息。如果你发现某些消息始终被路由到一个已经宕机的 Agent，先检查是不是心跳字段没有配好，或者 Agent 进程已经死掉但连接还没断开。

4.3 编写一个 Agent 接入

接入过程本身不复杂。拿一个简化版的企业内部服务 Agent 示例来展示：

import asyncio from agent_reach import AgentClient async def handle_message(msg): print(f"收到消息: {msg['type']}，携带参数 {msg['payload']}") result = {"status": "processed", "data": {"received_id": msg["id"]}} # 处理完成后回执一条消息给请求方 await client.send_to(msg["from"], "response", result, trace_id=msg["trace_id"]) async def main(): global client client = AgentClient( agent_name="agent.billing", router_url="ws://localhost:9030/router", topics=["billing.inquiry", "billing.refund"] ) await client.start(handler=handle_message) asyncio.run(main())

这段代码最核心的两处是：启动时注册自己的名字和主题，收到消息后通过同一个客户端发送回执。回执必须带上同一个trace_id，这样链路才能串起来。

4.4 业务系统发送消息

业务系统侧发送一条消息也很简单。举个例子，客服网页上用户提交了一张带“账单”关键词的工单：

import requests resp = requests.post("http://localhost:9030/v2/messages", json={ "to_type": "topic", "topic": "billing.refund", "from": "web.frontend", "priority": "normal", "type": "request", "subtype": "refund_check", "trace_id": "trace_billing_1029384756", "idempotency_key": "refund_check_88210", "payload": { "order_id": "88210", "refund_amount": 129.90, "reason": "duplicate_charge" } }) print(resp.status_code, resp.json())

注意to_type这个字段。它告诉路由模块这条消息是发给具体某个 Agent，还是发给一个主题的订阅者们。上面的例子走的是主题订阅模式，agent.billing注册了billing.refund主题，自然会收到。

4.5 人工接力与超时处理

Agent-Reach 的人工接力很有用，而且实现逻辑非常直接。你只需要在消息类型里指定一个特殊字段：

{ "type": "request", "subtype": "refund_over_threshold", "to": "human.review", "payload": {"order_id": "10010", "amount": 5200.00} }

这个消息送到human.review主题之后，不会触发任何 Agent 自动处理，而是进入人工工作队列。状态跟踪模块会把它标记为“等待人工处理”，24 小时没处理就向管理员发送提醒。

这个功能最适合的落地场景是金融、医疗等强合规行业。AI 可以给出初步判断，但最终的决策权必须留给人。你用 Agent-Reach 不需要为“人机协作”单独写一套复杂的协调逻辑，只是把消息发到一个特殊主题而已。

5. 实际踩过的坑：消息丢失、循环风暴和超时策略

5.1 消息“无声”丢失

第一次上线测试的时候，我发现一个问题：有些消息发送成功了，但接收方没有收到。查了 Broker 的日志，发现它已经把消息放到了 Redis Stream 里。但接收方消费队列后，会把消息从 Stream 里移除。

问题出在“消费确认”这个环节。接收方从队列读到消息之后立即返回 ACK，但实际上它还没处理完业务逻辑。如果这个瞬间进程崩了，消息已经从队列移除，但其实并未真正处理成功，这条消息就彻底丢了。

解法：把 ACK 时机后延。接收方必须执行完业务逻辑、把结果写入自己的持久化存储之后，才返回确认。如果中途崩溃，Broker 会因为超时未收到确认而重新投递。调整 ACK 时机之后，消息丢失率从百分之几降到了几乎为零。

5.2 两个 Agent 互相“踢皮球”的循环风暴

我的测试环境里出现过一次诡异的卡死：CPU 飙升，消息队列爆炸，状态数据库中充斥着大量消息记录。

顺着 trace_id 追踪后发现，agent.customer_service收到一条消息，因为识别不了内容，转给了agent.human_support；agent.human_support发现内容不需要人工处理，又转回agent.customer_service。两个 Agent 各转一次，形成无限循环。

这个问题的根源是：Agent 不具备“这条消息我应该拒收”的能力。后来我加了两个防御措施：

• 每个消息包增加hops字段，每转发一次加一，超过 5 次直接打入死信队列。
• 每个 Agent 在把消息转出之前，先检查目标是否跟来源相同，相同就停止转发。

这两个小改动在工程上成本极低，但彻底杜绝了循环风暴。

5.3 超时参数的“人性化”调整

最早期，我把所有超时都设为 30 秒。运行一段时间后发现，Agent 与 Agent 之间的消息确实能在几秒内处理完，但涉及人工审核的消息，30 秒根本不可能完成，导致大量消息被误判为超时。

后来我把超时参数改成按类型区分：

类型化超时之后，误报率显著下降，人机协作的体验好了很多。这个优化告诉我一个道理：基础设施的参数不能一刀切，要贴合实际业务的响应模式。

5.4 单元测试比最后联调便宜十倍

我在做这套系统的过程中最大的教训是：多智能体系统的单元测试，直接决定你的开发效率。

早期我只做整体联调，没有为路由规则写单元测试。每次改一次路由逻辑，都要手动启动三个 Agent 模拟一遍业务流程，一个小时就没了。

后来我为路由模块单独写了一套测试用例，把“收到请求 → 匹配规则 → 返回目标 Agent”这个过程抽象成纯函数测试，每次改规则跑一遍只需要半分钟。同理，Broker 重试逻辑和 Watch 跟踪逻辑也都做了单元测试。

Agent-Reach 自身的定位是一个连接系统，它不直接决定业务逻辑，但一个连接系统的故障影响面极大。没有基础测试护着，随便改一处路由规则，线上直接乱套。

6. 适用范围：什么时候值得用 Agent-Reach

这个项目不是银弹，它的适用场景比较明确。如果你遇到以下情况，可以考虑：

• 已经有多个 Agent 在各自运行，但它们之间缺少统一的消息协作规范。
• 需要人与 Agent 之间具备清晰、可审计的交接节点，尤其处于强合规行业。
• 需要追踪跨 Agent 的全链路日志，排查问题的时候不想再靠肉眼对时间戳。
• 希望新增 Agent 时不用去改已有 Agent 的代码，实现插拔式扩展。

反过来，以下场景你不应该用它：

• 只有单个 Agent、没有协作需求，引入消息总线纯属多余。
• 需要毫秒级同步响应，业务强依赖实时返回结果，异步消息架构不满足要求。
• 当前只有两三个 Agent 且不打算继续扩展，直接写几行调用代码更高效。

至于大规模高并发的场景，我个人不建议在初期就过度设计。Agent-Reach 设计时优先保证语义清晰和可控性，真要扛千万级消息，可以直接把底层的 Redis Stream 和 MongoDB 替换为更适合大规模压力的消息中间件，上层的信封格式和路由逻辑都不用动。

这套设计，本质上做的是一种标准化和减法。它让每个 Agent 不再需要关心“消息在到达我之前经历过什么、之后还要去哪里”，只需要关心“我收到了一封标准信封，处理完把回执放回信轨上”。

在我自己跑几个 Demo 项目的过程中，最直接的体感改善是：过去与人协作的 Agent 功能上线，每次都要在业务代码里手写审批流和消息转发，现在只需要往某个主题发送消息，剩下的一切都有基础设施接管。如果让我再选一次，我会把通信层的这套设计逻辑更早地定下来，因为它才是让多个智能体真正变成一个团队的前提。