企业官网建设流程全解析

上个月排查一条线上异常任务的时候，我在跟踪日志里翻了四十多分钟，最后发现根本不是模型回答出了问题，而是两个 Agent 之间的任务消息在传递过程中被静默弄丢了。就是那种代码里完全查不到异常、但整条链路已经断掉的状态。也是从那次之后，我把 Agent-Reach 这个项目从“有时间再做的事”提前到“立刻必须做”。

这篇文章想把 Agent-Reach 的设计思路和我在真实环境里跑通的过程完整展开讲一遍。它不依赖任何特定的大模型厂商，也不是要你学一套新语言，核心就解决一件事：智能体之间发任务消息，怎么发得出去、发得有回执、丢了能查出来。如果你正在做多 Agent 编排、自动化任务流，或者被“某个 Agent 调了另一个 Agent 却没反应”这种事折磨过，这篇内容应该能帮你少踩几个坑。

1. 多智能体协作里最容易丢消息的位置：Agent-Reach 想解决的痛点

1.1 一次审计日志里的完整丢包过程

先说当时的场景。我维护的自动化客服链路里有三个 Agent：意图识别 Agent、知识检索 Agent、生成回答 Agent。用户提问进来之后，意图识别 Agent 判断出这是一个售后问题，于是把检索任务发给知识检索 Agent；检索完成后再把结果交给生成回答 Agent。

整个链路看起来挺顺的，问题出在某天晚上的一次线路抖动。知识检索 Agent 处理到一半，向生成回答 Agent 发起了一次调用，结果超时。由于我当时写的是最简单的 HTTP 同步调用，超时之后既没有重试，也没有把任务存下来，代码只是打印了一行 warning 就继续往下走了。

结果就是：用户那边只看到“系统繁忙”，而日志里看不到任何报错记录。

那行 warning 还是我在当天白天为了调调试加的，如果不加，这个失败的调用连痕迹都不会有。也就是说，大多数多 Agent 系统里最先“丢”的，不是模型能力，而是 Agent 之间的协作消息。这个教训让我意识到：多个 Agent 串成一条链路的时候，真正值得优先做好的工程问题，是它们之间怎么把任务可靠地递到对方手里。

1.2 Agent-Reach 管什么、不管什么

我想了很久，给这类问题划了一条边界。Agent-Reach 不是去替代 Agent 的推理能力，也不管你底层的 LLM 是哪一家，它只负责“任务消息可靠触达”这件事。

具体来说，它管四件事：

• 投递：把任务消息从发起方送到接收方，不直接依赖对方进程是否在线。
• 回执：接收方处理完或处理失败之后，把状态返回给发起方或审计系统。
• 重试：消息处理失败时，按既定策略重新投递，而不是直接放弃。
• 可观测：每条消息在所有环节的状态都有记录，能查、能重放。

它不管三件事：

• 不管 Agent 内部的 Prompt 怎么写、模型怎么推理。
• 不管业务结果本身对不对，它只保证“结果有明确终态”。
• 不解决所有极端一致性问题，比如分布式事务最终一致性只做到“不静默丢消息”这一层。

这层边界很重要。如果什么都想做，这套系统就会变成一个沉重的流程引擎，反而不利于在项目里落地。

1.3 为什么我不想在 Kafka / RabbitMQ 上裸写任务语义

很多人会说：消息可靠性问题，用现成的 MQ 不就行了吗？我在实践里发现，只靠 MQ 远远不够。

Kafka 非常擅长高吞吐的流式数据，但它默认的消费模型是“消费者组内竞争消费”，而 Agent 协作更像“请求—应答”。你需要在 Kafka 之上自己实现一套回执、超时、幂等等概念，这套东西代码量相当可观。

RabbitMQ 有路由、有确认机制，功能很全，但“全”也意味着决策成本高。你要自己想清楚：消息确认之后如果处理失败怎么办？重试是通过死信队列还是消息 TTL？重投时如何避免对方重复执行？两个 Agent 之间如果互相等回执，超时由谁负责？

每个问题都有好几种标准做法，但组合在一起就形成一个非常自由的“弹性空间”。每个人凭直觉选，选完就是一堆隐式约定，只能靠口口相传。后来团队扩大的时候，新同事接手的成本非常高。

所以 Agent-Reach 做的事情是把这层语义固化下来：触达令牌、意图路由、三层回执。MQ 组件可以换，但 Agent 之间的协作约定是统一的。这样不管是 Kafka、RabbitMQ 还是 Redis Stream，底层用什么只是实现细节，上层暴露的接口一直是同一套。

2. Agent-Reach 的触达握手模型：令牌、意图路由与三层回执

2.1 触达令牌：给每条任务发一张实名快递单

我在 Agent-Reach 里定义了一个数据结构，叫 ReachToken。你可以把它理解成一张快递单：一张单子对应一次任务投递，单子上写了寄件人、收件人、物品描述、有效期和唯一单号。

ReachToken 的核心字段如下：

有一个细节值得单独说：trace_id 和 msg_id 要分开，不要混用。Trace_id 是一条完整业务链路的ID，从用户提问到最终回答都是同一个；而 msg_id 只是当前这一段“A 发给 B”消息的唯一标识。B 处理完之后，它再发给 C 时会产生新的 msg_id，但 trace_id 保持不变。

这样做的好处是，排障时可以按 trace_id 查出整条链路所有跳数的回执时间，也能按 msg_id 看某一跳是否发生过重试、是不是同一批消息重复消费了。

2.2 意图路由：按“想做什么”分发，而不是按“找哪个进程”

第一版 Agent-Reach 的路由我做成过 target_agent 直连，类似“我就是要发给 agent_b”。但在灰度验证的时候立刻发现问题：如果我临时把某个 Agent 换成新版本，或者原地扩容多了一个实例，路由规则就得跟着改。发消息方鲁棒性太差。

所以后来我改成了按intent路由。投递端只关心“我想发什么类型的任务”，不关心“谁注册了这个类型的消费”。下游消费方启动时注册自己感兴趣的 intent，上游完全解耦。

这个设计有点像我常举的例子：你网购的时候只需要写“家电类订单”，快递公司会根据当前运力安排配送员，而不是你必须在下单时指名道姓指定某个配送员。Agent 版本升级、实例扩缩容，对上游来说是无感知的。

intent命名我推荐用两级结构，比如域.动作：retrieval.request、policy.check、notification.push。这样后面做权限控制和审计都比随便起名字清晰得多。

2.3 三层回执：RECEIVED / ACCEPTED / COMPLETED 各代表什么

Agent-Reach 里最让我受益的设计就是三层回执。一开始我想得很简单：A 给 B 发消息，B 处理完回一个 ACK 不就行了？但实际跑起来发现不够。

原因在于“收到了”和“处理完了”是两件完全不同的事。消息进了 B 的消息队列，不代表 B 已经开始处理；B 开始处理了，也不代表处理结果是对的。如果只有一层 ACK，A 看到 ACK 就会以为任务已经完成，但等它拿结果时才发现 B 其实还在排队。

Agent-Reach 的三层回执完整定义如下：

A 发起任务后，只需要盯着回执状态流转。如果在预期时间内还没到 COMPLETED，就能判断当前卡在哪一层：

• 卡在 RECEIVED：消息还没被消费者取走，说明消费方可能挂了或者队列积压。
• 卡在 ACCEPTED：消费者在处理中，可能业务执行比较慢，需要结合耗时来分析。
• 一直没有 RECEIVED 之外的回执：消息可能压根没进队列，问题在投递端或网络。

这三层状态让“排查”这一步变得极其高效。

2.4 状态机里那条容易漏掉的 FAILED 路径

三层回执之外，有一条路径最开始我没设计：COMPLETED 出了异常分支怎么办？

后来我在真实环境里发现，Agent 的 handler 抛异常的概率远高于单机服务。因为 Agent 处理逻辑里经常要调用 LLM 接口，模型限流、响应超时、返回格式异常，都会导致处理中断。

所以我给最终状态补了一个分支：COMPLETED（处理成功）和FAILED（处理失败），FAILED 状态下回执里必须携带error_code和error_message。同时，FAILED 之后是否重试，由投递端根据策略决定，而不是由接收方自动无限重试。这样避免了一个处理不成功的任务在系统里来回弹跳，直到打满资源。

3. 30 分钟跑通一个小闭环：基于 Redis Stream 的实现记录

3.1 环境选型：为什么是 Redis Stream 而不是 MQ

选 Redis Stream 而不是 Kafka 或 RabbitMQ，主要原因是落地成本。如果项目里已经用了 Redis，加 Stream 只是几行配置的事，不需要额外维护一套消息集群。对于 Agent 协作这种“单跳消息量不大、但对可靠性和可观测性要求高”的场景，Redis Stream 的持久化能力和消费组模型完全够用。

另一个原因是 Redis Stream 的消费组模型非常直观。每个 Agent 对应一个消费组，组内多个消费者竞争同一批消息，一个消息只能被组内一个消费者处理，天然支持水平扩容。

当然也要清楚它的边界：Redis 如果只开默认持久化配置，极端故障下可能丢消息。建议至少开启 AOF 持久化，并且给 Redis 做主从。如果你对持久性要求极高，把底层的存储换成 Kafka 也不影响 Agent-Reach 的上层模型，这是它设计上刻意保持的松耦合。

3.2 投递端：xadd 与回执写入的事务一致性

先看投递端核心实现。

import json import time import uuid import redis class ReachToken: def __init__(self, intent, source_agent, payload, target_agent=None, ttl_ms=10000): self.msg_id = str(uuid.uuid4()) self.trace_id = None self.intent = intent self.source_agent = source_agent self.target_agent = target_agent self.payload = payload self.ttl_ms = ttl_ms self.created_at_ms = int(time.time() * 1000) def dispatch(redis_conn, token: ReachToken) -> None: """把任务消息写入意图对应的 Stream，并写入 RECEIVED 回执。""" if not token.trace_id: token.trace_id = token.msg_id queue_key = f"queue:{token.intent}" receipt_key = f"receipt:{token.msg_id}" pipe = redis_conn.pipeline(transaction=True) pipe.xadd(queue_key, {"token": json.dumps( token.__dict__, ensure_ascii=False)}, maxlen=100000) pipe.xadd(receipt_key, { "status": "RECEIVED", "intent": token.intent, "at_ms": int(time.time() * 1000), "msg_id": token.msg_id, "trace_id": token.trace_id, }) pipe.execute()

关键点在于pipeline(transaction=True)。这一步保证：任务消息进队列和 RECEIVED 回执写入同时成功或同时失败。如果分开执行，可能出现消息进队列了、回执没写上的情况，到时候发起方以为没发出去，重试一次就变成两条重复任务。

maxlen=100000是给队列设的一个上限。Agent 协作消息不需要无限保留，保留太多处理不过来的历史反而浪费内存。按 10 万条来算，正常量级的 Agent 协作场景至少可以覆盖数天的消息量。

3.3 消费端：xreadgroup、幂等抢占与 TTL 校验

消费端比投递端复杂一些，因为要处理幂等、TTL 和回执三个问题。

import time import json import redis REDIS_CONN = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True) def mark_receipt(token, status, error_code=None, error_message=None): REDIS_CONN.xadd( f"receipt:{token.msg_id}", { "status": status, "at_ms": int(time.time() * 1000), "msg_id": token.msg_id, "trace_id": token.trace_id, "error_code": error_code or "", "error_message": error_message or "", }, maxlen=1000, ) def check_ttl(token) -> bool: now_ms = int(time.time() * 1000) return (now_ms - token.created_at_ms) <= token.ttl_ms def try_acquire(token) -> bool: """幂等抢占：同一 msg_id 只允许一个消费者开始处理。""" acquired = REDIS_CONN.set( f"lock:{token.msg_id}", token.source_agent, nx=True, ex=60 ) return acquired is not None and acquired is True def consume_loop(agent_name: str, intents, handler): streams = {f"queue:{intent}": ">" for intent in intents} for intent in intents: try: REDIS_CONN.xgroup_create(f"queue:{intent}", agent_name, id="0", mkstream=True) except redis.ResponseError: pass # 消费组已存在 print(f"[{agent_name}] 开始监听: {intents}") while True: results = REDIS_CONN.xreadgroup( agent_name, agent_name, streams, count=5, block=2000 ) for queue_key, messages in results: for msg_id, fields in messages: token = json.loads(fields["token"]) token_obj = ReachToken(**token) if not check_ttl(token_obj): mark_receipt(token_obj, "FAILED", error_code="TTL_EXPIRED", error_message="token exceeded ttl") REDIS_CONN.xack(queue_key, agent_name, msg_id) continue if not try_acquire(token_obj): # 说明同一条 msg_id 已经在被另一个实例处理 REDIS_CONN.xack(queue_key, agent_name, msg_id) continue mark_receipt(token_obj, "ACCEPTED") try: result = handler(token_obj) mark_receipt(token_obj, "COMPLETED", error_message=json.dumps(result, ensure_ascii=False)) except Exception as e: mark_receipt(token_obj, "FAILED", error_code=type(e).__name__, error_message=str(e)) finally: REDIS_CONN.xack(queue_key, agent_name, msg_id)

几个设计点值得说明。

第一，try_acquire用SET NX做幂等抢占。如果任务因为网络原因被重复投递，只有一个消费者能获取锁，其余的直接跳过并xack。这一步避免了业务 handler 被重复执行。

第二，TTL 判断用的是now_ms - created_at_ms，没有拿两个不同机器上的时间戳做截点比较。这里埋了一个坑，后面第四章会专门展开。

第三，xack放在finally里。不管 handler 成功还是失败，这条 Redis Stream 消息都被确认。那失败之后怎么重试？

答案是：Agent-Reach 的重试不是 Redis Stream 层面的重投，而是投递端的重试。消费端把 FAILED 回执写给投递端，投递端拿到 FAILED 后根据策略决定是否生成一个新的 ReachToken 再次投递。这样实现了“消费层确认”和“业务层重试”的解耦，不会因为一个 FAILED 消息反复阻塞后续消息的消费。

3.4 观察一次完整握手：从 RECEIVED 到 COMPLETED

我把这套最小闭环跑在了两个 Agent 上：agent_a 作为发起方，agent_b 作为消费方。

操作步骤：

# 终端1：启动 agent_b 消费端 python demo/agent_b.py # 终端2：用 agent_a 发起一次检索请求 python demo/agent_a.py

agent_a 的任务逻辑是：构造一个retrieval.request类型的 ReachToken，把问题字符串放进 payload，调用dispatch()投递出去，然后轮询回执 key。

回执流大概长这个样子：

消息 9f2e1a 状态变化： RECEIVED -> 时间 1721... ACCEPTED -> 时间 1721... COMPLETED -> 时间 1721... 结果: {"answer": "..."}

三条回执的时间戳间隔只有几十毫秒，说明 agent_b 的 handler 执行非常快。如果某天你发现 ACCEPTED 和 COMPLETED 的间隔从几十毫秒变成几秒甚至几十秒，那多半是 handler 里有慢调用，比如 LLM 接口响应时间变长，这时候回执时间戳就能帮你快速定位瓶颈。

跑通这条链路后，Agent-Reach 的最小闭环就算成立了。接下来会遇到的问题，才是我最想分享的部分。

4. 上线后碰到的一堆实际问题与调参记录

4.1 重试风暴：当我错误地吃掉 ack 之后

第一次上线时，我的重试策略写得很粗暴：收到 FAILED 回执，就立刻生成新 token 投递同一任务，最多重试 3 次。结果第三方模型接口连续报限流，一批任务全部 FAILED，然后全部立刻重试。重试任务又把限流接口打得更满，形成恶性循环。

这个问题本质上是我把重试触发的条件和失败的后果绑定在了一起。显然，重试的风暴会顺着触达层扩散到整个下游。接口在限流的时候，你的重试只会延长它的恢复时间。

后来的做法是引入退避和熔断：每次 FAILED 重试之前，先递增退避时间。第一次重试等 1 秒，第二次 5 秒，第三次 30 秒。同时加了一个最小间隔限制，同一 intent 的失败率超过 50% 时，暂停 2 分钟不投递新任务，只处理存量任务。

调整之后，限流高峰期的任务成功率反而提高了一点。因为暂停新任务投递的这段时间，第三方服务的队列压力得到了释放，存量任务能更顺利地执行完。

4.2 TTL 时钟偏移：时间判断不能只信本机

第一次写 TTL 校验的时候，我犯了个错：用机器本地时间做比较。

看起来没毛病吧？但线上系统如果不是全部部署在同一台物理机上，多实例 Agent 各自所在机器的时钟可能相差几十秒甚至更多。如果发起方把任务发到另一个时钟走得慢的实例上，这个任务可能一进来就被判成过期，直接 FAILED。

排查过程很有意思。回执流显示某几个 Agent 收到的任务几乎全部TTL_EXPIRED，但消息明明才刚投递不到 2 秒。后来把各个节点的本机时间打出来一一对照，才发现是时钟偏移。有些云实例如果之前做过快照迁移，系统时间可能会复位到快照创建时刻，偏移量能大到好几分钟。

我最终的处理方案是：Agent-Reach 里所有跨节点的过期判断，都基于 Redis 的 TIME 命令或消息里携带的 created_at_ms 来做，不用消费端本机时间当权威。每一次握手回执里也强制带上标准时间戳，方便全局对齐。

另外，TTL 不要设置得太死。如果任务处理时间本身就要 5 秒，而 TTL 给 6 秒，那几乎每次处理都会在过期边缘徘徊。我建议 TTL 至少是预期处理耗时的 2 到 3 倍，要给排队等待、微服务抖动这些情况留出缓冲。

4.3 跨 Agent 回调死锁：一版排障排了三个小时

这个坑最让我头疼，值得单独拿出来讲。

场景是这样的：Agent A 发起一个任务给 Agent B，Agent B 处理到一半，发现自己需要 Agent A 掌握的某些数据才能继续，于是 B 又调用 A 的查询接口。结果 A 正在等 B 的完成回执，B 在等 A 的查询响应，两边互等，全部卡死。

我在回执流里看到的异常状态是：A 发起的任务停在 ACCEPTED 没有后续，B 那侧同样有一批任务停在 ACCEPTED。一开始我还以为只是慢，等了十分钟还是没动静，才意识到是死锁。

最后用 trace_id 把所有回执拼接起来，画出一条依赖环，真相才浮现。其实就一段依赖链：

agent_a 等待 agent_b 回执 agent_b 等待 agent_a 查询接口

解决思路是把这种同步依赖打散。跨 Agent 的任务约定里加了一条：任何 Agent 在业务处理过程中，不允许同步调用发起方的接口来获取继续处理所需的参数。如果确实需要，拆成三阶段：

1. Agent A 先把 B 可能需要的上下文打包进 payload，作为初始数据传过去。
2. B 收到后发现数据不全，把“缺失数据请求”作为一条异步任务发给 A，而不是同步等待。
3. A 处理完请求之后，把补充数据作为新任务投递给 B，B 继续工作。

这条约定牺牲了一点“实时性”，但换来了整个栈的可靠性。在设计 Agent 协作模式时，这一点是非常容易被忽视的，尤其是来自单体服务开发背景的人。单体里面方法调用是同步的，天然不会互相等死；但是消息化 / Agent 化之后，所有同步等待都变成了跨节点等待，不加约束就容易死锁。

4.4 线上初始参数：一组我实测下来的配置参考

我在正式环境用了以下几个参数作为初始值，你如果做类似的触达层可以根据自己的场景调整。

# Agent-Reach 关键参数初始值 ttl_ms: 30000 # 任务令牌有效期，按单跳预期耗时的2-3倍设置 maxlen: 100000 # 单意图 Stream 最大消息数 retry_max: 3 # 最大重试次数 retry_backoff_sec: [1, 5, 30] # 重试退避间隔，递增 lock_ttl_sec: 60 # 幂等锁有效期 failed_ratio_threshold: 0.5 # 意图失败率熔断阈值 circuit_break_sec: 120 # 熔断暂停时间

这套参数在我当时的场景下表现还不错。如果你们的任务是短平快类型，TTL 可以收到 10 秒；如果涉及 LLM 多轮处理或第三方接口，TTL 放宽到 60 秒也可以。重点在于：先定住一组可解释的参数，再根据线上回执时间分布去调整，不要一上来就追求完美。

5. 演进方向与给初次尝试者的建议

5.1 回执日志与事件溯源：让无声故障变得可审计

跑了两个月后，我对 Agent-Reach 最大的感受是：它最大的价值可能不是让消息更可靠，而是让故障从“无声”变成“有声”。

每一层回执，本质上就是一条事件日志。把这堆回执持续落盘到普通文件或对象存储，就形成了一个 message 级别的事件溯源系统。之后任何一个 Agent 出了问题，我只要拿 trace_id 去查回执流水，就能看到它在哪里停住、停了多少时间、最后一条回执是什么状态。

这个能力在我做系统巡检和线上问题复盘时用得非常频繁。它不只是给开发人员看，也可以让运维同学在出问题时快速给出“拓扑级”的判断，减少反复拉人进群的次数。

另外，如果要加更高级的调度策略，比如根据意图消息堆积量做动态扩容，回执日志也能当作计量数据源，比单独做埋点更简单。

5.2 对初次使用 Agent-Reach 的人的三条建议

第一条建议：先定义清楚终态，再写第一行代码。消息处理是成功、失败，还是允许“过期放弃”，这必须在一开始就定下来。如果设计成无限重试，结果就是一个永远处理不完的任务永远占用你的队列资源。

第二条建议：幂等键是底线，不能省。网络不稳、消费重启、消费组 rebalance，每个环节都可能造成重复投递。你可以在大部分情况下靠概率赌它不会重复，但只要线上发生一次重复扣款或重复外呼，你就会后悔省掉了这一步。

第三条建议：从第一版就把 trace_id 传下去。链路追踪这种事，早期不加，后面数据量大了再想补，几乎是推翻重做。无论你最后用了什么队列、什么语言，这条建议都不会变。

现在这个项目已经在我这边稳定运行了一段时间。每次看到回执流里整整齐齐的 RECEIVED、ACCEPTED、COMPLETED，我都会想起那次日志里翻不到异常的深夜。如果你也在做多 Agent 系统，并且已经开始被消息触达的问题困扰，我建议早点把触达层当成一等公民去设计。消息不丢、有回执、能排查，这九个字值得你认真对待一次。