Agent-Reach：大模型代理的主动触达通知层设计与实战

刚把一个 Agent 项目从“不会主动找人”改造成“该找谁就找谁、该什么时候找就什么时候找”这就是 Agent-Reach 的核心价值。一句话说清楚它是一个轻量级的触达增强层专门解决大模型 Agent 只能被动等消息、没法主动联系用户或外部系统的问题。适合正在做 Agent 应用、智能客服、定时任务通知、多租户系统的开发者拿来即用也可以作为内部基础组件学习借鉴。这个项目最初的痛点很实在。我们的 Agent 已经能回答问题、执行工具调用但所有结果都堆在会话里用户不打开页面就永远不知道任务完成了。我们试过让 Agent 在最后一步自己调 Webhook结果模型偶尔忘了传签名、偶尔重复发通知偶尔把内部细节拼给用户看根本没法上生产。Agent-Reach 就是在这一堆混乱里长出来的一个规范层由系统接管触达策略把“什么时候通知、通知谁、通过什么渠道、说什么内容”四件事固化下来Agent 只负责生成意图和上下文。1. 整体设计与思路拆解1.1 为什么给 Agent 加一层触达控制先聊一个容易被忽视的问题Agent 天然是“被动”的。无论是对话型还是编排型它都在响应请求。但真实业务里有大量场景需要 Agent 主动发起联系比如异步任务执行完要回执、长时间未操作要提醒、异常要告警、跨系统的流程要有人审批。没有触达层的 Agent就像一个只会接电话但从不会主动打出去的员工。但要给 Agent 打通“主动触达”的能力三条路都不好走直接让模型调 HTTP 接口发通知不可控。模型可能漏参数、错渠道、丢上下文而且每次触发都要重写 prompt成本高。把所有通知逻辑写在业务代码里太死板。渠道一换、策略一变就要改代码Agent 本身的能力没有成长。让 Agent 调用统一消息平台又太重。引一堆消息中间件、微服务小团队根本维护不过来。Agent-Reach 的思路是站在中间Agent 只负责“表达意图”比如任务 T-1024 已完成结果摘要如下需要通知项目负责人触达层负责“执行意图”决定用邮件还是飞书、要不要重试、内容模板怎么渲染、敏感字段如何处理。这个分离带来的直接好处是同一套 Agent 逻辑可以无缝对接不同的触达渠道而 Agent 本身不需要知道渠道细节。1.2 触达层与业务层的边界怎么切设计时我反复纠结一件事触达层到底管到什么程度。管太宽又变成一个消息中心管太窄Agent 还是要自己拼内容。最后定下来五条边界这个划分在多次实操里证明是稳的意图识别归 AgentAgent 输出结构化触达意图包括目标对象、事件类型、优先级、简要内容。策略决策归 Reach根据事件的类型、用户的偏好、时段规则决定触达方式。内容渲染归模板意图和占位符进模板渲染出渠道友好的正文Agent 不关心最终文案。发送与重试归通道适配器每个渠道一个适配器统一处理超时、限流、失败重试。审计与追踪归记录器每次触达都留痕便于排查模型干了什么、触达层干了什么。这五条边界在后续接入时省了大力气。比如某次线上事故Agent 误判了一次紧急告警重复触达了 14 次。由于边界清晰我们能快速判断是意图层出错直接在上游加了置信度阈值触达层完全不用改。1.3 核心运行流程Agent 意图到触达落地的全链路把流程完整跑一遍大家就明白这个组件的脾气。假设你有一个“自动化测试 Agent”跑完一轮回归后希望向研发群发结果并在失败时额外给负责人发邮件# Agent 输出的触达意图示例 intent: action: notify event_type: test_run_completed targets: - role: dev_group - role: owner condition: result failed content: summary: 回归测试完成通过率 96.3% failed_count: 3 duration_minutes: 42 priority: normalAgent-Reach 拿到这段结构化意图后做几步处理载荷校验检查 action 是否支持、targets 是否合法、event_type 是否在注册表里。这三样不过都直接返回错误不进入后续流程。策略解析查询触达策略表确定 dev_group 走群机器人owner 走邮件夜间23:00-07:00自动延后普通级通知紧急级立即发送。内容渲染按渠道分别渲染模板。群机器人模板关心“通过率失败列表”邮件模板关心“详细摘要修复负责人”。发送与可靠性把任务放进内存队列由 sender 协程逐条发送失败按指数退避重试重试上限内不丢弃。记录结果发送成功、失败、重试次数、最终状态全部写入日志表还提供 webhook 回调。这个流程所有环节都可以单测比让 Agent 自己发通知可观测性强得多。接下来展开每个环节的实操细节。2. 核心细节解析与实操要点2.1 触达事件注册表一切触达类型都要先声明Agent-Reach 要求所有事件先注册再使用这是我一开始就坚持的规矩。没有注册表的话Agent 随口说一个 event_typeReach 根本不知道该用哪个模板、走什么策略最后必然出乱子。注册表非常简单一个字典即满足绝大多数场景REACH_REGISTRY { test_run_completed: { templates: { feishu: templates/test_run_completed.feishu.md, email: templates/test_run_completed.email.md, }, policy: default_policy, allowed_targets: [dev_group, owner, pm], max_retry: 3, require_audit: True, }, anomaly_alert: { templates: { call: templates/anomaly_alert.call.md, email: templates/anomaly_alert.email.md, }, policy: urgent_policy, allowed_targets: [oncall, owner], max_retry: 5, require_audit: True, }, }实际经验有两条第一模板一定用文件路径而不是内嵌字符串方便不同环境覆盖第二allowed_targets 这个字段极其关键。没有它Agent 可能会向任意角色发高优通知有它即使 Agent 意图错误也只是被拒绝不会真发出去。注册要放在服务启动时校验一次模板文件缺失、策略不存在直接 fail-fast。别等第一个事件到了才发现模板路径写错那样生产环境就尴尬了。2.2 优先级与时段策略怎么避免通知轰炸触达系统做不好就是骚扰工具。我见过一个团队直接把 Agent 所有输出都转发到全员群三天后群里只剩投诉。问题根源不是 Agent 太吵而是策略层缺少过滤。Agent-Reach 的策略对象长这样class ReachPolicy: def decide(self, event_type, priority, target, current_hour): if priority urgent: return Decision(sendTrue, delay0) if target dev_group and current_hour in range(9, 21): return Decision(sendTrue, delay0) if priority normal and current_hour in range(22, 24): return Decision(sendFalse, reasonquiet_hours) return Decision(sendTrue, delay30) # 非工作时间延迟提醒这里面最值得说的就是 quiet_hours 机制。白天正常发晚上收到普通通知先存起来第二天早上统一补发。对 Agent 任务来说90%的通知晚几个小时完全没问题但用户满意度却大幅提升。曾经有用户反馈“以前晚上十一点还能收到机器人消息现在没了”这就是策略层的价值。另一个细节是冷却时间同一个 event_type 在五分钟内向同一角色触达超过 3 次后续请求直接丢弃并打日志。这个配置是为了防住 Agent 死循环调发送接口的意外情况实测非常有用。2.3 渠道适配器封装规范不同渠道的 API 差异很大但适配器的接口必须收敛到四个方法否则维护成本会迅速膨胀class ChannelAdapter(ABC): def send(self, message: str, target: str, **kwargs) - bool: ... def validate(self, target: str) - bool: ... def get_rate_limit(self) - int: ... def health_check(self) - bool: ...飞书适配器里send 封装的是自定义机器人的签名校验和消息推送邮件适配器封装的是 SMTP 发送Webhook 适配器封装的是 POST 请求。每个适配器只做一件事把统一的 message 和 target 转成渠道要求的格式然后发送。有一个非常容易踩的坑飞书或钉钉的自定义机器人有验签Webhook 也有超时。适配器内部必须统一处理超时异常不向上层抛不可控错误。我在第一版代码里没做这层结果某次渠道 API 超时直接让整个 sender 协程崩掉消息队列全部卡死。后来加了 per-channel timeout 装饰器这种情况全部转成重试逻辑。适配器还需要在启动时做 health_check至少检查 token 是否有效、邮箱服务器是否可达。别指望等到发的时候再发现渠道侧挂掉你才发现是最坏的时间点。3. 实操过程与核心环节实现3.1 项目骨架与依赖选型Agent-Reach 我用 Python 3.11 asyncio 实现依赖保持在很小范围httpx渠道请求、Jinja2模板渲染、pydantic意图校验、apscheduler定时任务。为什么不引入 Celery 或 Redis 队列因为对于多数 Agent 应用场景触达量远没到需要持久化队列的地步引入消息队列反而增加部署复杂度而 asyncio.Queue 单机处理每秒几百条触达完全够用。如果你要部署在容器环境建议把 Agent-Reach 做成一个独立服务通过 HTTP 接口接收触达意图内部处理和发送完全解耦。这样 Agent 那边只需要一个简单的 post 请求不需要安装任何依赖。agent-reach/ ├── agent_reach/ │ ├── __init__.py │ ├── api.py # FastAPI 入口 │ ├── registry.py # 事件注册表 │ ├── policy.py # 策略模块 │ ├── renderer.py # 模板渲染 │ ├── supervisor.py # 队列与调度 │ ├── adapters/ │ │ ├── __init__.py │ │ ├── feishu.py │ │ ├── email.py │ │ └── webhook.py │ └── models.py # pydantic 模型 ├── templates/ │ ├── test_run_completed.feishu.md │ └── test_run_completed.email.md ├── config.yaml └── main.py选择 FastAPI 而不是 Flask 的原因很简单它天生支持 async和 asyncio.Queue 配合非常自然pydantic 校验也是自带的省了一个依赖。3.2 核心代码实现意图接收与校验API 层只做一件事接收意图并校验然后扔进队列。代码非常短但边界处理要严谨。from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app FastAPI(titleAgent-Reach) class TargetSpec(BaseModel): role: str condition: str | None None class ReachIntent(BaseModel): action: str Field(..., pattern^notify$) event_type: str targets: list[TargetSpec] content: dict priority: str normal app.post(/intent) async def accept_intent(intent: ReachIntent): if intent.event_type not in REACH_REGISTRY: raise HTTPException(status_code400, detailunknown event_type) for t in intent.targets: if t.role not in REACH_REGISTRY[intent.event_type][allowed_targets]: raise HTTPException(status_code400, detailftarget {t.role} not allowed) await supervisor.submit(intent) return {status: queued, intent_id: intent_id}校验穿透到 allowed_targets 这层很重要。实际项目里出过一次问题Agent 意图里带了一个内部的部门缩写理论上这种角色根本不该接收通知。如果没有这层校验消息会发到一个错误的渠道事后排查非常麻烦。其实还能更严格解析 content 里有没有包含不应该触达的敏感字段。比如邮件通知不应该包含内部 token设置 content 字段的 allowlist 可以进一步降低风险。我给 Agent-Reach 加了 content_schema 校验效果很好。3.3 调度模块队列、重试与幂等防止supervisor 是整个组件的核心它管理一个 asyncio.Queue对每条消息负责调度。实现里有一个点很多人容易忽略发送失败后的重试不能简单地 sleep 再重发队列的消费顺序会被卡住。我的做法是每个任务独立一个 asyncio.Task彼此之间不阻塞async def worker(self): while True: job await self.queue.get() try: await self.process_with_retry(job) except Exception as e: logger.error(job failed permanently: %s, e, exc_infoTrue) finally: self.queue.task_done() async def process_with_retry(self, job): max_retry REACH_REGISTRY[job.event_type][max_retry] for attempt in range(max_retry 1): try: adapter get_adapter(job.channel) await adapter.send(job.message, job.target) await self.record(job, statussuccess, attemptattempt) return except ChannelError as e: await self.record(job, statusretrying, attemptattempt, errorstr(e)) await asyncio.sleep(2 ** attempt) # 指数退避 await self.record(job, statusfailed, attemptmax_retry)指数退避按 attempt 递增第一次重试等 1 秒第二次 2 秒第三次 4 秒最高封顶 8 秒。这个小策略很有效既不太长也不太短。实际线上一轮失败恢复后重试基本一次成功。幂等防止用过两次方案第一种是对每一条触达生成唯一的 intent_id在发送前检查已发记录第二种是为每个渠道的消息内容生成 content_hash记录五分钟内相同 hash 的发送次数超过阈值直接丢弃。建议两个都用前者防重放后者防死循环轰炸。class IdempotencyGuard: def __init__(self, redisNone, ttl300): self._seen {} self._ttl ttl self._redis redis def is_duplicate(self, intent_id: str) - bool: if self._redis: return self._redis.exists(freach:{intent_id}) 0 return intent_id in self._seen and time.time() - self._seen[intent_id] self._ttl def mark_seen(self, intent_id: str): if self._redis: self._redis.setex(freach:{intent_id}, self._ttl, 1) else: self._seen[intent_id] time.time()单进程多 worker 部署时建议换成 Redis否则幂等只能在进程内生效。我在测试环境的单进程下够用生产环境还是接上了 Redis量不大开销很小。3.4 模板渲染与敏感信息过滤渲染层看起来简单实际上有一个容易忽略的坑模板不能只依赖 Agent 给的 content 字典还要有默认值和降级机制。飞书渠道模板长这样{% if summary %} **回归测试结果** - 通过率: {{ pass_rate }} - 失败数量: {{ failed_count }} - 耗时: {{ duration_minutes }} 分钟 {% if failed_tests %} 失败用例: {% for t in failed_tests %} - {{ t }} {% endfor %} {% endif %} {% else %} 回归测试已完成详情请查看任务列表。 {% endif %}这套模板除了渲染好看还有一个关键安全点就是模板里不能出现不可信的自由文本。所有 Agent 生成的描述性内容在进入渲染前要做转义和清洗尤其是邮件模板防止内容注入。加一个 sanitize 函数处理掉 HTML 标签和链接对我的场景够用。def sanitize_content(text: str) - str: import html text html.escape(text) text text.replace(http://, ).replace(https://, ) return text[:500] # 截断超长内容避免消息过宽渲染完成后还有个后置步骤用渠道限制去检查消息长度。飞书自定义机器人限制 10 万字符邮件建议单封不超过 5000 字Webhook 就不需要了。超出时不是直接截断而是优先丢弃冗余的日志附件保留摘要部分这个逻辑写在 renderer 的 post_process 里。3.5 部署配置与基础用法以 docker-compose 部署为例Agent-Reach 服务本身只需要几个环境变量version: 3.8 services: agent-reach: build: . ports: - 8040:8000 environment: - REACH_CONFIG/app/config.yaml - REDIS_URLredis://redis:6379 depends_on: - redis redis: image: redis:7-alpine然后在 Agent 侧只要把触达意图 POST 给这个服务curl -X POST http://agent-reach:8040/intent \ -H Content-Type: application/json \ -d { action: notify, event_type: test_run_completed, targets: [{role: dev_group}], content: {summary: 回归测试完成, pass_rate: 96.3%, failed_count: 3}, priority: normal }服务返回 intent_id后续就靠这 id 查状态。Agent 完全不用感知触达细节只要管好结构化意图即可。这套对接方式让新接入的 Agent 项目几乎没有心智负担大部分开发者半小时内能跑通。4. 常见问题与排查技巧实录4.1 队列堆积触达任务突发暴涨有一次我们对接了一个批量任务 Agent它一次性给策略层塞了三千多条触达意图。supervisor 的队列瞬间堆积上万个未处理任务发送延迟从几百毫秒拉到两分钟。排查后发现有两条优化路径一是 sender worker 的数量可配置。单 worker 串行发送肯定是瓶颈可以在 supervisor 里开多个 worker彼此消费同一个队列。这个改动要小心并发数量提高后渠道的限流可能被打爆。我的经验是 sender 默认开 3 个 worker单渠道限流最大值 10 条/秒超过就排队。二是给队列设置 backpressure。当队列堆积超过阈值比如 5000入口 API 直接返回 429要求 Agent 侧等一下再提交而不是无脑吞掉。加了这层之后系统会主动反馈“我忙不过来”不再是隐含的超时问题。4.2 重试风暴与死信处理触达系统最怕的是渠道整体不可用如果这时候还在重试就会造成重试风暴。我在生产环境踩过一次SMTP 服务升级Agent-Reach 不断重试发邮件每分钟产生几百条失败日志Redis 里塞满了待重试的 key。对策是在 adapter 层增加 circuit breaker如果连续失败超过 5 次熔断器打开后新请求直接不再重试每一个任务而是进入死信队列等渠道恢复后统一再发。这个逻辑用 Python 的 pybreaker 库实现很快breaker pybreaker.CircuitBreaker(fail_max5, reset_timeout60) try: breaker async def guarded_send(adapter, message, target): return await adapter.send(message, target) except pybreaker.CircuitError: await supervisor.dead_letter(job, reasonchannel_down)熔断比单纯重试更省资源而且保护了下游渠道不会在对方恢复时被瞬间刷爆。死信队列我建议用数据库表存起来每天跑一个定时任务重新投递一次。Agent-Reach 的 dead_letter 表带上原 intent_id、失败原因、时间戳方便事后审计和补发。4.3 模板渲染 MissingError 与内容缺失有一个比较隐蔽的问题Agent 输出的 content 字段遗漏了模板里必需的变量Jinja2 默认会抛 UndefinedError 还是渲染为空取决于配置。我把环境设置成 StrictUndefined这样变量缺失直接失败不会渲染出一封“通过率: ”这种残缺消息。from jinja2 import Environment, FileSystemLoader, StrictUndefined env Environment( loaderFileSystemLoader(templates), undefinedStrictUndefined )同时给 API 层的 content 字段加了最小必填校验比如 test_run_completed 事件必须有 summary 和 pass_rate。有了双层校验残缺消息事件基本归零。4.4 常见问题速查表现象可能原因处理建议通知从未发出event_type 未注册检查注册表和模板文件是否存在通知发出但内容残缺content 字段缺必填项检查内容 schema 校验同一告知重复发送幂等 key 失效或未开启检查 intent_id 幂等逻辑夜间收到普通通知策略未配置 quiet_hours在 policy 中增加时段判断某渠道所有消息失败渠道限流/证书过期检查 health_check 和最近日志队列堆积严重worker 并发不足调高 worker 数量或加熔断对了关于“消息发出去了但用户看到的渠道里没有”——这个大概率是渠道侧的业务限制比如飞书自定义机器人必须在群里、邮件必须通过配置的发送域名白名单。此时去适配器日志里看响应体通常会有明确提示。5. 经验沉淀与两点补充建议5.1 触达系统的可观测性设计做触达层最容易忽略的是观测。消息可能发在三个地方业务系统、渠道服务商、用户聊天界面。任何一环出问题都查起来费劲。Agent-Reach 里我加了结构化日志每次触达都记录 intent_id、channel、target、elapsed_ms、retry_count、result。配合简单的统计接口按小时统计成功率。实际排查事故时这些字段帮了大忙。有一次飞书渠道被限流我通过日志发现发送成功率从 99.7% 掉到 84%再进去看适配器响应码马上定位是 rate_limited5 分钟就恢复了。另外给一个建议把 Agent 触达的意图原文也记录下来。看 Agent 到底打算干什么比只看最终发出的消息更能发现模型侧的问题。5.2 权限与审计的通用做法如果 Agent-Reach 要服务多个团队那么每个团队的事件注册表、模板目录、渠道配置应该隔离。我采用一个简单的 team_id 字段做区分注册表和策略都按 team_id 过滤避免团队之间互相触达。审计日志是这类基础组件的生命线。所有发生在触达系统里的“决策-修改-发送-失败”我建议全部落表留痕。这不仅是排查故障的需要也是团队协作时区分责任的依据。表设计不需要复杂意图原文 JSON、目标角色、事件类型、渠道、结果、时间戳就够了。5.3 最后分享一点个人的实操体会做 Agent-Reach 最深的体会是不要把“智能”放错位置。Agent 已经很强了不需要再让它操心“该用飞书还是邮件、晚上要不要提醒用户”这种事。真正专业的做法是让 Agent 只表达意图把策略和执行的脏活留给确定性的组件。这套思路不仅在触达场景适用评估回流、审批提醒、数据变更通知都能复用。实际接入的几个 Agent 项目里能把触达意图输出稳定在结构化格式的团队都跑得比较顺反过来总想让 Agent 直接生成消息文案的团队后期基本都在补安全漏洞和格式错误。确定性逻辑负责兜底生成式模型负责创意这个分工是目前实践下来最稳的模式。还有一个小的实战技巧上线初期在 Agent 侧全量开启意图日志连续观测两周你会发现有些触达请求根本不是业务要的而是模型在测试阶段“多此一举”。筛选掉这些噪声意图比事后白名单拦截要省力得多。如果你也在做一个会异步干活的 Agent强烈建议先把触达层单独抽出来。别等 Agent 变得复杂了再补一开始把它设计成一个明确的服务后面会轻松特别多。