1. 为什么要做 Agent-Reach：多智能体协作里最容易被忽略的最后一公里
1.1 智能体多起来了，问题也跟着多起来了
当你把一个 Agent 做成单机玩具的时候，你关心的是 Prompt、工具调用、上下文窗口。当你有三五个 Agent 一起跑业务流水线的时候，你就会发现一个特别朴素但又特别致命的问题：消息到底怎么送到该送的地方？
我去年年中接手了一个内部项目，客户侧的诉求很简单：让不同的 AI 角色在一条业务链路里接力干活。比如客服 Agent 接住用户问题，判断需要工单，就通知工单 Agent 生成单据；工单 Agent 处理完又得把结果推回给客服 Agent，再由后者整理成用户能看懂的答复。Demo 阶段大家用 Python 函数直接互相调用，跑得很欢。一上真实环境就崩了——网络抖动、渠道限流、Agent 进程重启、消息没送达没人知道。
这其实就是我所说的最后一公里问题：Agent 之间可以靠 API 或函数调用通信，但当 Agent 要触达真实世界的用户和外部系统时，你需要一个可靠的、有状态的消息触达层。我给这个项目起名叫 Agent-Reach，功能上就干三件事：把消息可靠送到指定渠道，把外部回执可靠收回来，把投递过程完整记录下来。
1.2 Agent-Reach 不是什么新式编排框架
先说清楚边界。市面上已经有成熟的 Agent 编排框架（LangGraph、AutoGen 这些），它们解决的是“控制流怎么走”“哪个 Agent 先执行”。Agent-Reach 不在这一层，它解决的是“消息怎么传”“传丢了怎么补”“渠道变了怎么路由”。
我用一个生活类比来理解这件事：编排框架是公司的业务流程——谁先签批、谁后复核；Agent-Reach 是公司的内部快递站——每一份文件放进统一的信封，写清楚收件人和地址，快递站负责送、负责签收、负责出问题追查。至于业务决策，那是流程层面的事，跟快递站无关。
这个定位很重要，如果你试图把路由和触达逻辑硬编码到每个 Agent 里，出现的情况就是：客服 Agent 里写了一堆钉钉 Webhook 推送代码，工单 Agent 里又写了一遍飞书 Bot 推送，以后要接入企业微信，你还得在所有 Agent 里再做一轮改动。我经历过这种维护噩梦，教训就是：触达能力必须下沉成基础设施。
2. 核心设计与整体架构：一个中转站，而不是一整套平台
2.1 统一消息协议长什么样
Agent-Reach 的第一版设计里，我定义了一套很薄的统一消息协议。它不是那种追求大而全的规范，只包含五个字段。
| 字段 | 说明 |
|---|---|
| message_id | 全局唯一 ID，生成规则为reach_+ 雪花 ID |
| source_agent | 消息来源 Agent 名称 |
| target | 目标描述，形如channel:wecom:agent_ops |
| payload_type | 内容类型标识，如 plain_text、card、json_event |
| payload | 实际内容，统一为 JSON 字符串 |
| trace_id | 链路追踪 ID，便于串联完整流程 |
为什么 payload 统一用 JSON 字符串而不是直接用对象？因为我希望所有渠道适配器面对的是同一个内容契约，而不是各写各的参数结构。后续要做消息内容版本兼容、缓存、重试，都是基于字符串做，至少少踩一半的坑。每条消息还允许附加最多 8 个 custom_tags，用于业务方打标签，按照用户ID、工单号或渠道类型做后续查询统计。
这套协议我建议做成消息中心内部的通用规范即可，不用去外面引入一套大型消息标准。按需求够用、可演进为原则，能用简单的字段组合表达需求，就不要引入过重的框架。
2.2 渠道适配器与可达性路由
适配器是 Agent-Reach 里最重要的概念。每个渠道对应一个 Adapter，至少实现三个接口：发送（Send）、回执（Ack）、状态回调（StatusCallback）。
我最初想过做成插件热加载架构，后来实际落地时发现没必要。Agent-Reach 里维护了一个 AdaapterRegistry，所有适配器在启动时注册，路由时从注册表里取对应实例。配置则集中在一个 channel.json 中，每新增一个渠道只需增加一个配置块，不修改主流程代码。
路由这块有个细节值得展开。目标地址不是死的，而是可以带有条件表达式。Agent 在构造 target 的时候可以指定channel:wecom:agent_ops|priority:high|fallback:email，意思是优先用企业微信发送到 agent_ops 队列，如果不可达则降级为邮件。路由层会首先解析这些约束，决定走哪条路径。这个点在实际运维中救了我很多次——被限流时自动降级，不会影响业务链路。
可达性路由的判定依据有三条：
- 配置层面：该渠道是否启用，是否处于维护模式。
- 健康状态：该渠道近 5 分钟的失败率是否超过阈值（我设的是 20%）。
- 业务约束：优先级、限时送达、是否允许降级。
如果三条全部通过，就尝试发送。任一条不满足，就走 fallback 路径。一次投递失败后不会立刻换渠道，而是先进入重试队列，达到重试上限后才触发降级，避免因为瞬时抖动导致渠道频繁切换。
2.3 幂等、重试与去重的取舍
消息发送和普通写请求不一样，天然存在“发送成功但回执丢失”的歧义。所以幂等是必须的，而不是可选项。Agent-Reach 在存储层索引了 message_id，相同 message_id 的发送请求只执行首次，后续直接返回原结果。
重试策略我采用的是带指数的退避，附带随机抖动。具体参数：
- 最大重试次数：5
- 初始退避：1 秒
- 退避倍数：2
- 抖动范围：±20%
- 最大退避上限：30 秒
为什么要有抖动，比较容易被忽视。如果没有随机性，批量投递失败后的首次重试会在同一个时间点扎堆触发，渠道侧看起来就是一波突刺流量。加上抖动后，重试流量在时间轴上摊开，大幅降低渠道限流风险。实际观察中，这一项就把限流触发概率降低了一半以上。
去重和幂等略有区别。幂等管的是相同请求只执行一次；去重管的是不同 Agent 实际上发了一样的内容时，是否要合并。我最终没做内容级去重，只做 ID 级幂等。原因是内容级去重容易误伤：两个 Agent 可能因为上下文不同，生成了看似相同但语义完全不同的消息。为了减少消息量而做内容去重，我建议慎重。
3. 实操实现：从零搭一个最小可用的 Agent-Reach
3.1 环境与目录结构
这里用 Go 实现，因为部署就是个单二进制，后续丢到任何容器里都方便。你也可以用 Python 或其他语言脚本来做，核心逻辑是同一套。
我的目录结构大体是这样的：
agent-reach/ ├── cmd/ │ └── server/main.go ├── internal/ │ ├── adapter/ │ │ ├── registry.go │ │ ├── wecom.go │ │ └── email.go │ ├── protocol/ │ │ └── message.go │ ├── store/ │ │ ├── message_store.go │ │ └── retry_queue.go │ ├── router/ │ │ └── router.go │ └── config/ │ └── channel.json ├── go.mod └── README.md存储我选择了短期内存加长期落盘的方式。消息进入后先写 MySQL（或 PostgreSQL），然后进入内存队列等待发送。发送成功后异步更新状态为 delivered；最终状态写入消息详情表。Redis 可以用来做分布式锁，如果单机部署且流量不大，锁这项可以先省略。
3.2 统一消息协议和存储设计的落地细节
消息结构体我直接用 Go struct 表达。注意一点：所有字段都保留 JSON 标签，方便后续接入方通过 HTTP JSON 接口投递。
package protocol type Message struct { MessageID string `json:"message_id"` SourceAgent string `json:"source_agent"` Target string `json:"target"` PayloadType string `json:"payload_type"` Payload json.RawMessage `json:"payload"` TraceID string `json:"trace_id"` CustomTags map[string]string `json:"custom_tags,omitempty"` CreatedAt time.Time `json:"created_at"` } func (m *Message) Validate() error { if m.MessageID == "" { return errors.New("message_id is required") } if m.Target == "" { return errors.New("target is required") } return nil }存储设计里有张表，我建议重点关注两个索引。第一是ix_message_id，用于幂等判断；第二是ix_target_status，用于查询某个渠道或某个接收方的投递情况。其余的跟普通消息表没太大区别。
物理表结构（简化版）：
CREATE TABLE reach_message ( id BIGINT PRIMARY KEY AUTO_INCREMENT, message_id VARCHAR(64) NOT NULL, source_agent VARCHAR(64) NOT NULL, target VARCHAR(256) NOT NULL, payload_type VARCHAR(32) NOT NULL, payload JSON NOT NULL, trace_id VARCHAR(64) NOT NULL, status TINYINT NOT NULL DEFAULT 0, retry_count TINYINT NOT NULL DEFAULT 0, next_retry_at DATETIME DEFAULT NULL, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, UNIQUE KEY uk_message_id (message_id), KEY ix_target_status (target(64), status) );状态字段我用枚举：0 为 pending，1 为 sending，2 为 delivered，3 为 failed，4 为 dropped。其中 dropped 表示达到最大重试次数后的最终失败，跟 failed 区分开，因为 failed 还可以被人工手动触发重投，dropped 则在业务语义里已经终结了。
3.3 示例：对接一个 Webhook 渠道并做路由
Webhook 渠道是最容易上手的渠道类型，推送就是发一个 HTTP POST。我这里拿一个企业微信 Bot Webhook 做例子。渠道适配器本质上要做的事，就是定义好如何把统一消息转换成 Webhook 需要的 JSON 格式，再发送。
package adapter type WebhookAdapter struct { Name string WebhookURL string client *http.Client } func (a *WebhookAdapter) Send(msg *protocol.Message) error { // 根据 payload_type 决定消息内容结构 switch msg.PayloadType { case "plain_text": body, _ := json.Marshal(map[string]interface{}{ "msgtype": "text", "text": map[string]string{ "content": string(msg.Payload), }, }) req, _ := http.NewRequest("POST", a.WebhookURL, bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") resp, err := a.client.Do(req) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode >= 400 { return fmt.Errorf("webhook returned status %d", resp.StatusCode) } return nil default: return fmt.Errorf("unsupported payload_type %s", msg.PayloadType) } } func (a *WebhookAdapter) Ack(msg *protocol.Message) error { // Webhook 是即发即收，通常不需要额外 Ack return nil } func (a *WebhookAdapter) StatusCallback(msg *protocol.Message) (string, error) { return "delivered", nil }在 router 里，解析 target 得到渠道类型:渠道标识，然后从 registry 里取 adapter，调用 Send，再根据结果更新状态。
路由的核心逻辑，我建议留一个很薄的中间层：
func (r *Router) Route(ctx context.Context, msg *protocol.Message) error { if err := msg.Validate(); err != nil { return err } // 幂等检查 if r.store.Exists(msg.MessageID) { return ErrDuplicateMessage } target, fallback, err := r.ParseTarget(msg.Target) if err != nil { return err } adapter := r.adapters.Get(target.Channel) if adapter == nil { return ErrUnsupportedChannel } if err := adapter.Send(msg); err != nil { r.enqueueRetry(msg) } return nil }上传后，只要 Agent 内部调用 Agent-Reach 的 HTTP 接口传一个 JSON 消息，就能完成一次触达。对外提供的 HTTP 接口接受消息结构，并返回 message_id 作为异步确认。
3.4 可观测性：如何确认消息真的送达到了
我踩过的最大教训是：消息发送层把 HTTP 200 当成终态，结果用户根本没收到。原因是 Webhook 中间经过了一层网关，网关返回 200 但实际消费失败了。所以 Agent-Reach 里我引入了回执概念——每个渠道适配器除了 Send，还必须实现 StatusCallback 来返回最终状态。
比如企业微信用的是 Bot Webhook，没有真正的回执通道，那么 StatusCallback 可以基于发送时拿到的消息ID去调用查询接口拿最终状态；如果查询接口也不可用，至少要在日志里打印明确的级别与建议排查方向，而不是默默吞掉错误。
Agent-Reach 自带一个轻量级指标端点，暴露三个核心指标：
reach_send_total发送请求总数reach_send_failure_total发送失败总数（按渠道分类）reach_delivery_latency_seconds投递耗时直方图
这三个指标配合 Grafana 能直观看出哪个渠道最近在抖动、是不是有积压。比捞日志排查快一个量级。我强烈建议你从第一天就把指标端点加进去，哪怕不当仪表盘用，排查问题的时候也会非常有用。
4. 真实项目里踩过的坑与排查技巧实录
4.1 高频重试把 Webhook 推送打到限流
第一次上真实流量时，我设置的重试策略比较简单，失败后等 1 秒就重试。结果某个渠道因为瞬时故障连续失败，一批消息重试后集中再发，渠道侧直接把我们的 Webhook 拉进限流名单。影响到了其他正常业务消息，场面相当被动。
后来我把重试策略改成上面的指数退避加抖动，并且加了一条硬规则：任何渠道的单条消息 1 分钟内最多重试 5 次，超出后进入人工补偿队列。另外，在渠道配置里增加了并发限制，比如企业微信渠道最多同时 10 个在飞请求，超出就排队，防止瞬时打满。
经验就是：不要把重试参数做成全局统一的，要按渠道特性分别配置。邮件渠道的失败多半是DNS或MX服务器的问题，重试间隔可以拉长到分钟级；Webhook 渠道的失败多是限流或网络抖动，秒级重试加上退避就够了。
4.2 回调风暴：Agent 触发 Agent 的连锁反应
Agent-Reach 上线后跑了一周，出现了一个特别诡异的现象：某个 Agent 发出一条消息，另一条 Agent 收到后自动处理，又回发出消息，循环往复，消息量指数级上涨，险些把整个消息层打爆。排查下来发现原因特别简单——两个 Agent 的业务逻辑都包含了“如果收到内容含确认字样，则回复确认”，一个说“确认”，另一个也回“确认”，车轱辘话转不停。
我的解法是加了两个机制。
第一个是触发深度限制。每条消息带一个hop_count字段，从 0 开始，每经过一次 Agent 处理就加 1。超过MAX_HOPS=8时，消息进入 dead_letter 队列并打告警，由人工介入判断到底是链路设计问题还是真的需要那么多跳。第二个是链路追踪。因为每条消息带 trace_id，我把同一条 trace_id 下所有 Agent 的处理记录串起来，在排查时能一眼看到环路出现在哪一步。
这个坑不会发生在消息层本身，但消息系统必须有兜底能力，否则它就会成为风暴放大器。Agent-Reach 后来在消息入队时强制检查 hop_count，并配合规则引擎做动作收敛。建议所有做多 Agent 协同的同学，尽早把跳数限制和链路追踪加上，省得线上出事故时才手忙脚乱。
4.3 路由判定的边界问题
路由层有一个很隐蔽的坑：当两个渠道都显示健康时，你按什么规则选？如果按固定优先级，高优先级渠道可能会一直被流量占满；如果按随机轮询，又会出现同一会话的消息在不同渠道间跳动，用户感知割裂。
我最终采用的方式是混合策略：同一个目标在短时间内保持粘滞，默认 10 分钟窗口内走同一渠道；窗口到期后才重新做路由决策。这样既保证了渠道负载均衡，又不至于频繁切换。
实现上其实就是在内存里维护了一个小 Map，key 是目标地址，value 是渠道与窗口截止时间。Agent-Reach 单机节点大约维护 5 万个粘滞键，内存占用可控，如果做集群部署，可以换成 Redis 集中存储。粘滞性这块不妥善处理，用户侧体验会非常差。
4.4 会话状态与消息粘滞性
消息系统本身是裸投递，但在 Agent 的真实使用场景里，会话状态很关键。你做在线客服 Agent，用户上次的问题和这次的补充说明是要连起来的。一开始我以为 Agent-Reach 可以不去管这个问题，后来发现如果不做，用户一个简单追问就会触发一轮全新的 Agent 调用，上下文断裂，回答质量直线下降。
所以 Agent-Reach 增加了一个轻量化会话模块，在消息协议里引入 session_id 字段，并在存储层做会话上下文索引。Agent 侧调用 Agent-Reach 时带上 session_id，消息层把最近 N 条历史摘要作为上下文附带返回给 Agent。这个 N 我目前设的是 10，按 token 数和业务需求可调节。这样的好处是：Agent 处理每条消息时天然获得了连续性，不需要自己维护复杂的记忆存储。
要注意的是，会话状态与消息投递是两套生命周期。消息送达后是否清理会话数据，应该由业务控制，而不是由 Agent-Reach 自动决定。我见过有团队把消息层和会话记忆强行绑定，结果消息正常送达，Agent 却拿不到历史，排查半天才发现是消息清理策略把会话也给清了。
5. 落地这两个月，我的实际体会
Agent-Reach 从设计到落地差不多花了两周时间，真正稳定运行并扛住业务流量，用了将近一个月的持续打磨。核心稳定下来的节点就是重试策略、路由粘滞和可观测性这三块。如果你准备在自己项目里做类似的东西，不用一上来就追求大平台化。先把统一协议、幂等存储、渠道适配器这三件事做好，已经能覆盖 80% 的触达问题。
按照我目前的规划，Agent-Reach 后续想扩展两块内容。一块是支持更多渠道类型，比如短信运营商、企业自建工单系统、邮件网关，这个其实只是新增适配器的工作量；另一块是做一个简单的编排触发规则，让 Agent 之间不必硬编码相互调用的关系，而是基于消息内容事件触发下一个 Agent。这个会牵扯到规则引擎的设计，我还在研究中。
最后分享一个已经被我列为惯例的小技巧：在开发环境里，用一个简单的模拟渠道，随机模拟 5% 的发送失败、10% 的回执延迟。因为真实系统的故障不会优雅地提示你，只会悄悄丢消息，提前用混沌手段把脆弱点暴露出来，比上线后救火轻松太多。
