企业官网建设流程全解析

前阵子我这边同时在跑三个Agent：一个负责收集线索和整理销售情报，一个负责根据素材自动生成推广内容，还有一个做客户回访和建联。单看每个Agent都挺能干，但真正串起来的时候，我发现事情开始变得不可控——它们之间互相喊话靠的是轮询数据库、互相监听临时文件、甚至有人直接让我手工复制产物再塞进下一个Agent的输入框。一时间我分不清自己是在做系统集成，还是在当人肉管道。后来我花了两周时间写了一个轻量级中间层，把它叫Agent-Reach。这个东西解决的核心问题就一句话：让Agent之间、Agent与外部渠道之间，有一个稳定、可控、可观测的触达方式。今天这篇文章就把它的设计思路、核心实现和踩坑记录完整捋一遍，给正在搭多Agent系统的朋友做一个参考。

先说清楚Agent-Reach适合谁。如果你只是单机调一个Agent，不需要这玩意儿；如果我上面说的三个Agent之间的协作场景你还没有遇到，那也暂时用不上。但如果你已经开始让不同Agent互相传递结果、想让Agent自动发飞书通知、想给Agent挂一个Webhook入口、想让多个Agent共享同一套消息路由能力，那Agent-Reach的模型对你应该是有用的。

1. 为什么在Agent协作里，最难的不是模型能力而是"触达"

1.1 我遇到的实际场景：三个Agent各说各话

我先交代一下具体背景。线索Agent是Python写的，每小时从公开渠道抓一批潜在客户信息，它需要一个地方把结果吐出来。内容Agent是Node写的，它能接收一段素材自动生成小红书文案，但它不关心素材从哪来。回访Agent更麻烦，它需要知道内容Agent生成的结果，再决定给哪些客户发消息，同时还要把回访记录回传给线索Agent闭环。

最原始的方案是什么？线索Agent把结果写进MySQL一张表，内容Agent定时扫表，回访Agent再从内容表里读。这个方案的问题不是跑不通，而是只要有一个Agent挂了、或者表结构变了、或者有人手动改了数据，整条链路就静默失败。排查的时候要从日志一层层翻，非常痛苦。我意识到，真正缺的不是Agent的推理能力，而是它们之间的一条标准化通路。

这个痛点在多Agent系统里其实有个名字，叫Agent-to-Agent通信问题。单个Agent内部有ReAct循环、有工具调用，很成熟；但多个Agent之间怎么串联、怎么定协议、怎么确认送达，基本上没有统一答案。市面上有消息队列、有HTTP Webhook、有RPC框架，但很少有专门以Agent为第一公民设计的触达中间件。

1.2 现有方案梳理：都不是不行，但都差一点意思

我mentally过了一遍可以复用的技术方案，列个表感受一下：

对比完之后你会发现，我需要的是一个介于Webhook和MQ之间的东西：它要有标准化的消息格式，要有路由能力，要能接不同渠道，又要轻到单个进程就能跑。这就是Agent-Reach的直接动机。

1.3 Agent-Reach的定义：一个专注于触达的轻量中间层

Agent-Reach的核心抽象可以概括为三条：

1. Agent到Agent：任意Agent可以向任意Agent发送结构化消息，无需关心对方的真实地址。
2. Agent到渠道：Agent可以通过连接器把消息推到飞书、企业微信、邮件、Webhook等外部系统。
3. 渠道到Agent：外部系统可以通过统一入口回调某个Agent，实现双向触达。

这样它就同时兼了消息路由和渠道适配两件事。你可能会问：这跟一个带Webhook的聊天机器人框架有什么区别？ChatBot框架通常预设了"人"作为交互对象，而Agent-Reach把交互对象泛化成Agent，人只是其中一种渠道。这个视角差异在后面做设计时会体现得很明显。

2. Agent-Reach核心设计：一个"消息信封"决定的架构

2.1 信封（Envelope）：所有消息的统一外壳

做Agent-Reach的时候，我第一个决定就是：绝对不能让Agent之间直接丢裸JSON过来。没有统一外壳的消息就像没贴面单的快递，路由系统根本没法处理。所以我定义了一个信封结构，用Python的dataclass表达出来就是这样的：

from __future__ import annotations import time import uuid from dataclasses import dataclass, field from typing import Any, Dict, Optional @dataclass class Envelope: msg_id: str = field(default_factory=lambda: uuid.uuid4().hex) trace_id: str = field(default_factory=lambda: uuid.uuid4().hex) sender: str = "" # 来源Agent ID recipient: str = "" # 目标Agent ID或路由Key channel: str = "agent" # 目标渠道类型，默认agent payload: Dict[str, Any] = field(default_factory=dict) event_type: str = "message" # 业务事件类型 ttl: int = 60 # 消息存活时间，秒 timestamp: float = field(default_factory=time.time) retry_count: int = 0 ack_required: bool = True

说说每个字段为什么存在。msg_id是幂等去重的依据；trace_id是一条消息从发出到最终投递的全局追踪ID，排查问题全靠它；sender和recipient是路由的起点和终点；channel决定这条消息是继续发给Agent还是从某个渠道发出去；ttl防止消息积压在队列里成为僵尸消息；retry_count配合ack_required实现可靠投递。

这里有一个容易被忽略的设计点：event_type。Agent-Reach不把消息定义成传统RPC里的"方法调用"，而是定义为"事件"。调用是你请求我、我返回结果；事件是你通知我、处理结果由我自主决定。Agent之间应该更偏事件驱动，因为对方Agent什么时候处理完、怎么处理，调用方其实不需要同步等待。把语义定位成事件，整个系统的解耦程度会高很多。

2.2 为什么先用Envelope路由而不是直接连MQTT

可能有人会想：MQTT不也是发布订阅吗？Agent订阅一组Topic，不也能实现触达？理论上可以，但MQTT的消息体对应用层是无感知的，它只保证字节级传输。Agent-Reach信封的核心价值在于把路由决策、鉴权、去重、追踪这些能力放到了消息协议层面，而不只是"发出去就完事"。

用生活类比来说，MQTT像一条公路，保证车能开过去，但车上装的是什么、该在哪个路口下来、下了车之后有没有人接，它不管。Agent-Reach更像一个有分拣中心、有配送员、有签收确认的快递系统，它的职责不只是"送达"，还包括"送到对的地方、确认对方收到、丢了会补发"。多Agent协作场景里，确认送达和路由准确的重要性，往往比传输速度更关键。

2.3 路由表与连接器：一个轻量交通枢纽

信封定义好之后，Agent-Reach的进程本体就好办多了。它内置一个路由表，维护每条路由规则：

routes: - from: "lead_agent" to: "content_agent" event_type: "lead.ready" - from: "content_agent" to: "reach_agent" event_type: "content.generated" - from: "reach_agent" to: "channel/slack" event_type: "notification.send"

路由规则分为两类：Agent到Agent的定向路由，和Agent到渠道的外发路由。路由匹配顺序是精确匹配优先，然后事件类型匹配，最后兜底丢进死信队列。

同时我还做了一个很轻的注册机制。Agent启动的时候，如果配置了agent.register，就到Agent-Reach这里登记自己的ID和健康状况。注册表直接放在内存里，加一个30秒一次的TTL过期，不做持久化。因为Agent-Reach本身不是注册中心，它只需要知道当前有哪些Agent处于可触达状态就够了。这个做法牺牲了一点强一致性，但换来了极高的吞吐和极低的运维复杂度。

3. Agent-Reach的代码落地：几个关键模块我这样写

3.1 消息路由核心：一个函数搞定分发

Agent-Reach的消息循环非常朴素：从入口队列里拿一条消息，解析信封，查路由表，决定下一跳。我把核心路由逻辑收敛到了一个处理函数里：

async def process_envelope(env: Envelope, router: Router) -> None: # 1. 幂等检查 if await dedup.is_duplicate(env.msg_id): logger.info(f"duplicated message ignored: {env.msg_id}") return # 2. TTL检查 if time.time() - env.timestamp > env.ttl: await dead_letter.push(env, reason="ttl_expired") return # 3. 路由匹配 targets = router.resolve(env) if not targets: await dead_letter.push(env, reason="no_route") return # 4. 按目标分发 for target in targets: try: connector = connector_registry.get(target.connector_type) await connector.dispatch(env, target) env.ack_required and await ack_queue.put(env) except Exception as exc: logger.exception("dispatch failed %s", env.msg_id) await retry_queue.put(env)

这个函数看起来简单，但每一步都有讲究。幂等检查必须在路由之前，否则渠道高并发的重复消息会把后面的环节冲垮；TTL检查放在路由前，能避免把过期的消息发给消费者；connector.dispatch失败之后统一丢给重试队列，由重试模块负责退避与封顶。

实际运行的时候，process_envelope会被包装成一个Worker，消费入口队列的并发数可以通过配置调整。我在单机上用8个Worker压过，每秒大概能处理2万条轻量消息，完全够Agent场景使用。

3.2 连接器（Connector）：每个渠道一个插件

连接器是我认为Agent-Reach里最体现"插拔"价值的部分。每个外部系统都对应一个Connector实现，核心接口就四个方法：

class Connector(ABC): @abstractmethod async def connect(self) -> None: ... @abstractmethod async def dispatch(self, env: Envelope, target: Target) -> bool: ... @abstractmethod async def healthcheck(self) -> bool: ... @abstractmethod async def close(self) -> None: ...

以飞书机器人这个渠道为例，connect阶段构建一个带签名的HTTP客户端；dispatch阶段把信封里的payload转成飞书消息卡片：

async def dispatch(self, env, target): resp = await self.http_client.post( f"https://open.feishu.cn/open-apis/im/v1/messages", params={"receive_id_type": "open_id"}, json={ "receive_id": target.address, "msg_type": "text", "content": json.dumps({"text": env.payload["text"]}) }, headers=self.headers ) return resp.status_code == 200

关键点在于dispatch方法的返回值。我让Connector在"成功投递到对方API"时返回True，而在网络错误、API超时这类不确定情况下抛异常或者返回False。Agent-Reach只根据这个返回值决定要不要进重试队列，至于业务上消息有没有真的被用户看到，那是由业务层自己负责的。把技术投递和业务送达分开，能避免很多"过度承诺"的麻烦。

3.3 重试队列与退避策略：别用蛮力，要有节奏

消息投递失败的时候，直接丢弃肯定不行，无限重试也肯定不行。Agent-Reach里我采用了一套指数退避加抖动的重试策略：

async def retry_worker(env: Envelope) -> None: for attempt in range(MAX_RETRY): try: await connector_registry.dispatch(env, env.target) return except Exception: wait_time = BASE_BACKOFF * (2 ** attempt) + random.uniform(0, 0.2) await asyncio.sleep(wait_time) await dead_letter.push(env, reason="max_retry_exceeded")

参数我设的是：BASE_BACKOFF=1s、MAX_RETRY=5。这样第一次重试等1秒，第二次2秒，第三次4秒，第四次8秒，第五次16秒，加上最大0.2秒的随机抖动，20秒左右就会停止。抖动不是玄学，是为了避免多个Agent同时失败后在同一时刻发起重试，形成请求风暴。这个话题在分布式系统里叫"惊群"，你如果自己实现过定时任务应该会有体感。

4. 接入真实业务：从两个Agent到一个完整闭环

4.1 我的第一个Demo：自动线索到内容的流转

光看代码不好建立体感，我分享一下第一个真正跑通的Demo链路。我当时技术选型用Python是因为团队已有模块都在这套技术栈里，Agent-Reach被做成一个独立服务，各Agent通过HTTP长连接与它通信。

Demo里两端是这样的：

• 线索Agent：把抓取完成的线索列表封装成lead.ready事件，调用Agent-Reach的/v1/messages接口推入。
• 内容Agent：通过WebSocket长连接订阅到lead.ready事件，拿到线索摘要后自动生成文案，再以content.generated事件回传。

Agent-Reach在这中间做的事情只有一个：把消息从A路由到B。但如果少了这层，内容Agent就要自己维护线索Agent的地址，一旦线索Agent换了端口、加了鉴权、改了消息格式，两边都要跟着改。有了Agent-Reach，这种变更被收敛到配置层，Agent代码完全不用动。

4.2 项目配置文件：一个YAML看清全部拓扑

我习惯把所有Agent和路由的声明收敛到一份配置文件里。看这个示例：

agents: - id: "lead_agent" entry: { type: "http", bind: ":9100" } - id: "content_agent" entry: { type: "websocket", bind: ":9200" } connectors: - name: "feishu_notifier" type: "feishu" config: app_id: "cli_xxx" app_secret_env: "FEISHU_SECRET" default_ack_timeout: 10 dead_letter: { type: "kafka", topic: "agent-reach.dlq" }

配置最多的是路由表和连接器配置，Agent本身的逻辑一点不用写进Agent-Reach。这个做法让我后来在加第三个Agent的时候只花了十分钟：加一条注册信息，加一条路由规则，完事。这也是我坚持把Agent-Reach做成一个独立服务而不是一个SDK的原因——SDK侵入每个Agent进程，独立服务则让触达层变成一个可以独立升级、独立监控的基础设施。

4.3 全链路追踪：用trace_id串联一切

消息一旦开始在多Agent之间流转，调试难度指数级上升。Agent-Reach最初版本没有任何追踪能力，我Debug的时候特别痛苦。后来加了全链路日志，每次消息进入系统或离开系统都打印一行结构日志，字段统一：

[agent-reach] dispatch msg_id=4f1a3c9d trace_id=9b8e7a6f sender=lead_agent recipient=content_agent event_type=lead.ready status=ok cost=23ms

有了这份日志，我可以在排查问题的时候做一件很简单的事：grep trace_id拿全部链路。哪一段延迟高、哪一段漏发、哪一段重试了，一目了然。给Agent系统做观测性，我认为第一优先级就是搭好这条trace链路，否则后面Agent数量一多，出问题基本只能靠猜。

5. 踩坑记录：我在Agent-Reach上栽过的几个跟头

5.1 消息顺序问题：不保证全局有序，但保证因果有序

第一个坑是关于消息顺序的。刚开始我天真地认为，只要消息是一条一条发的，接收方就会一条一条收到。实际跑了一阵子发现，HTTP异步请求导致的乱序是常态。比如内容Agent发送content.ready事件，中间走了一次重试，后面的事件反而先到了。

Agent-Reach最后采用的方案是：不承诺全局有序，只承诺同一个sender + recipient + event_type里，如果消息带seq字段，则按序投递。实现方式也不复杂：接收方每个窗口缓存一个expected_seq，乱序的消息先入缓冲区等待，超时了再触发重新拉取。这个妥协是必须的。如果你真的需要所有消息严格有序，那是消息队列的领域，不是触达中间件该干的事。

5.2 幂等性：渠道侧必须靠业务消息做去重

第二个坑在渠道侧暴露得很明显。某个连接器在投递飞书消息时，偶尔会碰到网络超时——消息其实已经发送成功，但响应没回来。Agent-Reach判断投递失败，会自动重试，结果就是用户在飞书上收到了两条一模一样的通知。这类重复投递问题在分布式系统里叫at-least-once语义，几乎可以肯定会发生。

修复方案是在所有渠道侧的消息卡片里带上msg_id。飞书这种IM平台天然支持一条消息的msg_id去重；Webhook渠道则是在消息体的扩展字段里放trace_id，让接收方自己决定要不要处理。记住一句话：去重逻辑一定要放在业务侧，不要指望Agent-Reach的队列机制帮你完全规避重复。

5.3 权限模型：触达能力不能随便开放

再多说一个权限坑。Agent-Reach初版只有简单的路由功能，任何Agent都可以给任何Agent发消息，这在内部环境没毛病。但当我通过Agent-Reach向企业微信、短信、邮件这些真实触达渠道发消息时，这个设计就危险了。一个被攻破的Agent理论上可以拿着Agent-Reach的令牌给你全量客户发垃圾短信。

最后我加了两层控制。第一层是每个Agent独占一个API Token，Agent-Reach根据Token解析sender身份；第二层是路由授权白名单，规定某个Agent只能投递到哪些Agent和哪些渠道。这两层加完，至少拦截了90%以上的误触达风险。如果你要把Agent-Reach推到生产环境，权限这一块千万不要省。

6. Agent-Reach的边界：它不是全能中间件

讲了这么多能做什么，聊聊它不做什么。Agent-Reach不适合当工作流编排引擎用，一个复杂业务流程需要条件分支、并行聚合、人工审批，那是工作流系统的事，你非要用消息路由硬凑会非常别扭。它也不适合做高吞吐日志管道，几GB级别的日志、需要批量刷盘的数据流，还是用更专业的数据管道更合适。它也没有完整的事务语义，Agent之间的消息投递是"尽力可靠"，不是分布式事务。

我的判断标准是：如果你的核心诉求是"把一条结构化的消息从一个Agent送到另一个Agent/渠道，并且你能看到它到了没有"，Agent-Reach就适合；如果你的核心诉求是"让一个流程严格地执行三百个步骤并在中途随时回滚"，请用它旁边的专用流程引擎。

最后分享一点实际体会。Agent-Reach真正让我的Agent协作从"演示Demo"变成"可依赖的工程"的关键，不是它有多聪明，而是它把触达变得标准化、可见、可控。先把信封结构定义下来，再一个连接器一个连接器地补，不要一开始就贪大求全。多Agent系统里，每个Agent都在变聪明，但让它们之间能稳定地喊到彼此，这件事本身就值得做扎实。如果你也在搭类似的东西，建议从最小闭环开始：两个Agent通过中间层互通消息，加一条路由规则，加一条链路追踪，跑通了再扩展。这个路径，我可以负责任地说，非常顺。