过去大半年我一直在维护一个叫 Agent-Reach 的项目。最开始它只是团队内部用于管理消息推送的小工具后来慢慢演变成一套覆盖“智能体任务编排—调度—多渠道触达—失败重试”的完整框架。现在之所以把它单独拿出来写一篇是因为我发现在多 Agent 协同越来越常见的今天很多人卡住的点并不是模型怎么调、Prompt 怎么写而是“怎么把智能体产生的结果稳定、不重复、不乱序地送到该去的地方”。Agent-Reach 就是从这个痛点里长出来的。如果你手里也有三五个或者几十个 AI 智能体并且需要把它们的输出可靠地交给用户、群机器人或下游系统那这篇文章值得你读完。我会从设计思路、核心代码、落地配置到踩坑过程把整个链路完整拆开。1. Agent-Reach 到底要解决什么问题1.1 多智能体时代的“最后一公里”困境我第一次意识到传统调用方式不行是在一个典型的周一早晨。团队在两周内把业务里的智能体从 3 个加到了 20 多个有做内容摘要的、有做用户分层的、有做知识库问答的还有几个负责把结果同步到客户群。问题集中暴发同一篇文章摘要被推送了三次凌晨两点某个智能体把测试数据发到了生产客户群还有一个耗时 15 秒的智能体因为上游超时被直接丢弃结果业务方第二天才看到告警。这些问题的本质不是智能体本身写得不好而是我们只关心了“如何得到结果”却完全忽略了“结果如何送达”。写代码的时候大家都默认用 HTTP 同步调用谁调用谁负责可一旦链路里出现 10 次调用失败率就变成了指数叠加重试和幂等各自为政根本无从维护。Agent-Reach 定位就是解决“最后一公里”把同一个智能体产生的结果按规则、按渠道、按时间可靠地触达给多个目标同时对这个过程做全生命周期管理。1.2 触达Reach不等于调用Call很多人会把“触达”简单理解成“调用接口”。实际上两者差别很大。调用Call是同步请求结果直接返回调用方等着拿数据。触达Reach则是一个异步过程要考虑目标是否可达、消息是否需要排队、失败后怎么办。比如一个智能体每 10 分钟生成一份销售简报需要推送到销售群、发邮件给管理层、写进内部知识库。如果用同步调用业务代码里全是散落的 HTTP 请求和 try-catch用 Agent-Reach你可以把“每 10 分钟生成简报”定义成一个触达计划Agent-Reach 自动判断三个渠道的优先级、失败策略和重试次数业务代码只需要提交一次任务。所以 Agent-Reach 的核心抽象就是把“触达”变成一个一等公民。你不再关注怎么发请求而是关注提交什么任务、希望什么结果、有哪些兜底规则。2. Agent-Reach 的核心设计把触达当一条流水线2.1 三层模型触发层、路由层、交付层在设计 Agent-Reach 的时候我最初想的是把消息发送逻辑塞进一个函数里后面发现根本行不通。因为来源多、目标多、失败场景复杂你必须把链路拆开。现在整个框架跑的是三层模型触发层Trigger负责接收来自智能体的输出、定时器、Webhook 或人工触发生成一个标准化的触达任务。路由层Router根据任务类型、目标渠道、优先级、限流策略决定任务的下一跳去向。交付层Delivery真正执行发送动作包括 HTTP 调用、邮件发送、消息队列写入、Webhook 转发等并对返回结果做确认。三层各司其职好处是任何一个环节出问题都可以单独排查。比如消息没到客户群你可以先查触发层有没有收到任务再查路由层是否把任务分配到了正确渠道最后查交付层是否真的调用成功。整个链路有日志、有状态、有超时而不是一句“消息丢了”。2.2 核心状态机从 pending 到 completedAgent-Reach 里的每个触达任务都有明确的状态流转。状态机是整个项目里最容易被人忽视但又最关键的骨架。我现在用的是六个状态状态含义下一步pending任务已创建等待调度进入路由层routed已确定目标渠道等待交付进入交付层delivering已发起触达请求等待确认结果succeeded触达成功任务完成终态failed触达失败等待重试或转死信重试或转死信dead超过最大重试次数需人工处理终态状态流转全部通过事件驱动完成。每当状态变化Agent-Reach 会发出一个事件执行对应回调。这个设计的价值在出问题时体现得最明显。比如某次触达失败你能看到任务停留在 failed 状态而且能直观看到失败原因而不是在日志里大海捞针。2.3 为什么选用消息队列而不是直接调用第一版 Agent-Reach 用的是直接调用智能体结果一出来马上同步发送到所有目标。但后来频繁出现拥堵一个邮件任务超时 30 秒后面所有任务全部卡住。我最终把同步模型改成了基于消息队列的异步模型。原因是异步模型天然适合触达场景。触达任务往往是“发完就行”的发送方和接收方不需要同时在线。用消息队列做缓冲可以把瞬间高并发变成平稳的队列消费也能方便地做延迟重试。我选 Redis Streams 作为默认队列因为它轻量团队已经很熟悉而且支持消费者组适合多实例消费。如果你已经有 Kafka 或 RabbitMQ也可以直接对接Agent-Reach 的队列层做了接口抽象。3. 从零搭建 Agent-Reach一套可直接复制的架构3.1 技术栈与基础设施目前 Agent-Reach 的技术栈并不复杂核心是 Python基础设施也只有三件套PostgreSQL 存储任务和明细数据Redis Streams 做消息队列Redis 另外一个库做分布式锁和限流。服务本身是 FastAPI 应用暴露一组 REST API 给智能体提交触达任务同时也支持 Python SDK 方式接入。选型的时候有三个考量。第一团队主力语言是 Python智能体框架也大多是 Python方便统一。第二PostgreSQL 能支撑复杂的任务查询和状态统计我们每天大约几十万条触达任务Postgres 毫无压力。第三Redis Streams 虽然没有 Kafka 那样的高吞吐但对触达任务这种量级已经足够而且运维成本极低。3.2 核心代码骨架一个触达任务怎么跑Agent-Reach 内部把触达任务抽象成ReachTask代码核心其实就两步提交任务消费任务。先看提交侧的关键逻辑# task_service.py class ReachTaskService: async def submit( self, task_type: str, payload: dict, targets: list[ReachTarget], policy: RetryPolicy ) - str: task_id uuid.uuid4().hex task ReachTask( idtask_id, task_typetask_type, payloadpayload, targetstargets, policypolicy, statepending ) await self._save_task(task) await self._queue.enqueue(reach:tasks, task_id) return task_id提交侧负责把任务落库然后丢进 Redis Streams。这里要特别注意落库和入队必须做到至少一次语义。我采用的方式是先写数据库再入队如果入队失败有一个补偿任务定期扫描 pending 状态超过 30 秒的任务重新入队。消费侧在主进程里跑一个 worker代码大致如下# worker.py class ReachWorker: async def process(self, task_id: str): task await self._get_task(task_id) await self._transition(task, routed) for target in task.targets: if target.pre_hook: hook_result await self._run_hook(target.pre_hook, task.payload) if not hook_result.ok: # 统一走失败分支 await self._handle_failure(task, target, hook_result.error) continue await self._deliver(task, target)消费侧会把任务从 pending 拉到 routed再对每个目标做实际投递。这里pre_hook的设计是后期加上的用来支持触达前的数据转换。比如某个智能体输出的 Markdown投递到钉钉前要转成纯文本。这比每个智能体各自做格式转换要省事得多。3.3 配置示例与任务定义Agent-Reach 里渠道和目标都通过 YAML 配置声明。我建议把所有触达渠道固化到配置文件里而不是散落在代码里这样运营或者后端同事可以直接调整而不需要改动程序。下面是个简化示例channels: enterprise_wechat: type: webhook endpoint: https://qyapi.weixin.qq.com/cgi-bin/webhook/send token_env: WECOM_WEBHOOK_KEY timeout: 5 email: type: smtp host_env: SMTP_HOST port: 465 from_env: SMTP_FROM reach_plans: daily_report: trigger: cron: 0 9 * * * payload: template: templates/daily_report.md targets: - channel: enterprise_wechat group: sales_group - channel: email recipients: [leadercompany.com] policy: max_retries: 3 retry_delay: 60 dead_on_final_failure: true这里的每个reach_plan都对应一种业务场景。写清楚之后新增一个触达场景只需要在 YAML 里加一段完全不需要写新的 Python 代码。3.4 部署与启动检查清单Agent-Reach 服务部署结构很简单一个 API 服务、一个 worker 服务、一组基础设施。我在部署完成后会习惯性检查几件事缺一个后面准出问题检查 Redis Streams 的消费者组是否创建成功否则多 instance 会抢消息。检查环境变量里的渠道密钥是否都配齐尤其是 webhook token 和邮件密码。配置一个测试reach_plan提交一条真实任务确认日志里状态能从 pending 走到 succeeded。确认 PostgreSQL 的表结构迁移已经执行特别是 task 表和 dead letter 表。检查 worker 的并发数设置我一般用 4-8 个并发 worker过大会把下游接口打崩。这套部署检查清单救过我很多次很多问题其实都出在“Redis Streams 消费者组没建”这种细节上。4. 上线后我踩过的四个大坑4.1 坑一触达任务重复执行第一次出现重复推送是在上线第二周。用户反馈同一份报告收到了两条内容完全一样的推送。排查之后发现根源在网络抖动导致 worker 处理任务超时生产端以为消息丢了于是重新从 Streams 里拉取再次执行了一次触达。Agent-Reach 在delivering状态停留的时候如果 worker 崩溃任务会被重新拉到另一个 worker。为了处理这个问题我引入了消费幂等每个目标投递之前在 Redis 里设置一个delivery:dedup:{task_id}:{target_id}的分布式锁带上触达动作的唯一 hash。如果同一任务同一目标在 5 分钟内重复触发直接跳过。关键代码如下async def _deliver(self, task, target): lock_key fdelivery:dedup:{task.id}:{target.id} acquired await self._redis.set( lock_key, 1, nxTrue, ex300 ) if not acquired: self._logger.warning(skipped duplicated delivery %s, lock_key) return # 真正执行触达 result await self._channel.send(target, task.payload) ...这个方案并不能消除所有重复但至少把绝大部分重复拦截住了。真正要做到精确一次需要下游配合做消息唯一 ID但大多数场景下分布式锁已经够了。4.2 坑二重试风暴打垮下游RetryPolicy 初版本里我把max_retries设成了 5retry_delay设成了 10 秒。结果有一次邮件服务短暂不可用几百条任务同时进入重试每 10 秒打一次直接把邮件服务彻底压垮形成了恶性循环。后来我把重试策略改成了指数退避加抖动。retry_delay从 30 秒起步每次乘 2再随机加减 10%这样即使一万条任务同时失败也不会在同一时间点集中爆发出流量。同时增加了全局限流器对同一个渠道的每秒请求数做限制默认企业微信 webhook 限制在 5 次/秒邮件限制在 2 次/秒。这里我的经验是重试不是越多次越好核心是让下游在恢复期内不被再次打死。指数退避 抖动是运维共识的解法但要落到代码里而不是只写在运维手册里。4.3 坑三死信消息没人管最初失败达到最大重试次数后任务直接被置为 dead但系统只打了日志没有主动告警。直到某个渠道配置错误连续一周没发送成功业务方才问“消息是不是丢了”。这是最尴尬的一种坑。我在 Agent-Reach 里专门增加了一张dead_letter表dead 状态的任务会连同失败原因、最后一次异常栈、原始 payload 都存进去同时通过一个简单的通知插件把死信情况推到团队告警群。这样每次发生失败至少会有人知道。死信表是触达系统的安全网。它不解决失败问题但能让失败变得可感知、可追溯。我建议所有做类似系统的人都要有这张表否则你的系统看起来在运行实际上在悄悄丢消息。4.4 坑四智能体返回格式不稳定导致解析失败智能体输出的格式真的很让人头疼。即便是同一条 Prompt今天返回 JSON明天在 JSON 前面多了一段说明文字后天干脆给你一个 Markdown 表格。Agent-Reach 对接的智能体越多这种解析失败就越频繁。我的处理方案是在路由层增加了一个PayloadNormalizer针对不同的智能体来源做格式规整。比如 OpenAI 类接口返回的内容里带着content字段和tool_callsClaude 可能有text和thinking不能一刀切去取某个字段。class PayloadNormalizer: staticmethod def normalize(source: str, raw_payload: dict) - dict: if source openai: return {text: raw_payload[choices][0][message][content]} if source claude: block raw_payload[content][0] return {text: block[text]} # 兜底逻辑 return {text: str(raw_payload)}这个代码看起来简单但它避免了 80% 的解析类故障。值得一提的还有尽量在智能体输出的源头就约定好结构化输出不要靠归一化兜底。Agent-Reach 里的归一化只是最后一道防线。5. 进阶玩法把 Agent-Reach 变成一个触达控制台5.1 跨渠道触达从企业微信到钉钉再到 API 回调Agent-Reach 目前的渠道插件已经支持企业微信机器人、钉钉机器人邮件、Webhook、Slack、飞书以及任意自定义 API 回调。每种渠道都实现了同一个接口send(target, payload) - DeliveryResult。新增一个渠道只需要写几十行代码注册到插件表里即可。渠道插件的接口设计要从一开始就抽象干净否则后面每加一个渠道就要改核心代码。Agent-Reach 目前的接口长这样class Channel(Protocol): async def send(self, target: ReachTarget, payload: dict) - DeliveryResult: ...各渠道自己负责协议细节。比如企业微信机器人需要把消息用 markdown 类型打包钉钉机器人则需要msgtype字段。这些细节全部封装在插件里路由层只负责播发不感知具体协议。如果你也想做一个多渠道触达系统最值得抄的就是这个抽象。5.2 可观测性一张图表看清触达情况Agent-Reach 自带了简单的监控面板核心指标有四个任务提交量、成功率、重试率、死信数。每个指标按渠道和智能体来源两个维度拆开。它不需要接入 Prometheus 也能跑因为数据都在 Postgres 里定时聚合到metrics_daily表即可。我个人重点关注的是 “触达成功率” 和 “死信数” 这两个指标。成功率低于 98% 就值得排查死信数大于 0 且无人处理就是事故隐患。有一个很有效的方法是在死信表里记录 channel 标签看哪个渠道最容易失败然后对那个渠道单独调参。5.3 多租户与配额不同业务线互不干扰我们内部有多条业务线A 业务线的智能体和 B 业务线的触达目标完全不同。如果没有隔离可能出现 A 业务线误触达 B 业务的客户群。Agent-Reach 引入了tenant概念每个任务创建时绑定一个tenant_id路由规则里也会限定租户可用渠道。这样既能在逻辑上隔离数据也能让不同租户使用独立的限流配额。多租户设计出来之后团队里新业务接入的成本大大降低。新增一条业务线只需要配置租户、渠道权限和默认配额即可完全不需要改核心代码。我自己使用 Agent-Reach 这大半年最大的体会是跨越多智能体系统真正稀缺的不是调用能力而是稳定且可控的触达能力。Agent-Reach 这个名字里“Agent”代表源源不断产生结果的智能体“Reach”代表让结果真正到达需要的人或系统。两者的连接点就是你构建这套系统时投入的设计和排错经验。如果你也在做类似的事我建议你从今天开始记录你的触达任务状态复用我上面提到的最小核心状态机、独立路由层、幂等键、死信表。这四件事做到了你的智能体结果基本不会再丢。