我是在第七次因为同一个线上告警被不同渠道轮番轰炸之后才决定认真对待“消息分发”这个问题的。那天下午同一份故障通知以三条完全不同的格式分别从工单系统、IM群机器人、邮件列表涌进来而我却花了十几分钟才把三份信息对起来期间还差点漏掉最关键的一行错误码。团队里没人有义务做这件事——把四处逃窜的消息理顺——但我意识到只要还靠人来对账这种混乱就会一直重演。于是就有了hermes-agent这个项目不是又一个 ChatBot也不是又一个任务调度器而是一个站在所有消息来源和所有执行目标之间的“信使”。Hermes 本来就是希腊神话里替众神传信的角色用它给这个代理命名至少在我心里是顺理成章的。如果你也在维护一堆业务系统每天被 webhook、告警、回调、机器人消息反复打断又不甘心把所有内部流程都交给某个第三方平台那这篇文章应该能给你一些可落地的思路。我会从“为什么需要这样一个代理”讲起拆解它的核心架构、路由决策、可靠性设计最后给出一套能在周末跑起来的最小部署方案。不吹不黑只说我在实际开发中踩过的坑和最终沉淀下来的设计。1. 为什么值得做一个自己的消息代理我的通知爆炸现场1.1 分散渠道带来的隐性成本很多团队的第一反应是消息多就多呗又不是什么大事。但真正做过运维或者负责过线上系统的人都知道问题从来不在“消息数量”而在“消息的离散程度”。我当时面临的局面是这样的监控告警走 A 平台的 webhookCI/CD 结果走 B 平台的 webhook客服反馈进了 C 系统的工单支付回调由第三方直接打到我们某个内部接口还有几个外部合作方各自用不同的格式推送数据。每一个渠道单独看都很简单但当它们同时涌进来的时候人就成了那个“手忙脚乱的中间件”。我统计过一周的数据平均每天有 200 多条外部事件需要处理每条事件至少要看一眼才能知道“要不要管、该谁管”。按每条 30 秒计算光“看一眼”这个动作就吃掉了将近两个小时而且这两个小时是碎片化的、随时被打断的根本无法用来做深度工作。更可怕的是人工处理意味着不同的人会对同一条消息做出不同的判断——有人觉得是 P0 告警有人觉得可以缓缓根本没有统一的口径。1.2 现成集成平台为什么不够用既然消息分发这么麻烦为什么不上 n8n、IFTTT 这类现成的自动化平台这个问题我在立项前认真调研过也实际试跑了一个月最终得出结论它们解决的是“流程编排”问题而我真正需要的是一个“持续运转的消息管道”。差别在哪里流程编排关心的是“触发 → 动作”比如收到邮件就创建工单这是离散的、一次性的逻辑。但消息管道关心的是“任意来源 → 统一处理 → 可靠投递”它需要处理乱序到达、重复推送、格式不规范、目标端暂时不可用这类真实世界里天天发生的问题。现成平台在这方面的抽象往往是浅层的一旦遇到需要自定义幂等策略、自定义重试语义、或者对消息体做深度清洗的场景就会感觉处处受制。另外还有一个很现实的因素数据驻留和可用性。我们有一部分业务回调包含敏感信息公司合规要求这些数据不能落到第三方平台的日志里。这就直接排除了所有纯 SaaS 方案。我需要的是一个能完全跑在自己服务器上的东西数据从进入到流出全程可控。1.3 hermes-agent 的设计目标经过那段时间的反复思考我把hermes-agent的设计目标明确成了四条后文所有架构决策都围绕这几条展开单一抽象所有输入源不管来自 webhook 还是 IM 机器人还是邮件进入系统后都变成同一种“信封”结构后续处理逻辑完全不关心消息最初是从哪来的。先可靠后智能第一优先级是保证消息不丢、不重、能追踪。AI 决策是加分项但绝不能因为 AI 的不确定性影响消息的确定性投递。确定性规则与 AI 并行能用规则解决的问题绝不用 AI规则覆盖不了的长尾场景才交给模型并且 AI 的决策结果必须可审计、可回滚。本地优先所有核心组件都支持单机 Docker 部署不强制依赖任何云服务。这四个目标听起来不复杂但真正落地的时候几乎每一个都在挑战“最小实现”的诱惑。尤其是“先可靠后智能”这条我前后推翻过两版设计就是为了不在基础管道还没稳的时候急着把大模型塞进去。2. 核心架构一条消息从进入到落地的完整旅程2.1 技术选型为什么用 Go Redis 而不是 Python项目最开始的原型是用 Python 写的因为快三两天就能把想法跑通。但原型跑通之后我立刻遇到了两个问题一是 Python asyncio 在高并发 webhook 接入时稍不注意就会出现回调阻塞二是部署形态太重为了跑一个代理还得维护 Python 运行时和一堆依赖和“本地优先”的目标有点背道而驰。后来我把核心重写成了 Go。选择 Go 不是因为它时髦而是因为它在这个场景下确实合适goroutine 模型天然适合处理大量并发的 webhook 长连接编译出来只有一个静态二进制扔进容器里连基础镜像都能省。我做过一个对比同样跑 500 个并发 webhook 推送Go 版本的 P99 延迟比 Python 版本低了将近一半内存占用反而只有原来的四分之一。存储层我选了 Redis理由也很实在消息管道需要的是一个高吞吐、低延迟、支持过期策略的临时存储而不是一个强一致的关系型数据库。Redis 的 List 可以做队列Stream 可以做消费者组Hash 可以做去重表一个组件就把管道最核心的几件事全包了。维度Python 原型Go 重写版并发模型asyncio 事件循环goroutine channelP99 延迟500并发210ms96ms内存占用稳态780MB190MB交付产物源码 依赖包单个静态二进制部署依赖Python 3.10无2.2 统一消息模型信封模式在hermes-agent里所有消息进入系统后的第一件事就是被装进一个统一的“信封”。信封结构大概是这样的type Envelope struct { ID string json:id TraceID string json:trace_id Source string json:source EventType string json:event_type Payload json.RawMessage json:payload Headers map[string]string json:headers ReceivedAt time.Time json:received_at Priority int json:priority IdempotencyKey string json:idempotency_key }ID是这条消息在系统内部的唯一标识TraceID用来串联一次业务事件的完整链路Source记录消息来自哪个渠道EventType是类型标签Payload是原始业务数据Headers带上调用方传入的自定义属性Priority表示优先级IdempotencyKey则专门用于去重。为什么非要这样统一不可因为后端的规则引擎、AI 分类器、路由器和投递器统统只认这一个大结构。新增一个消息源的时候只需要写一个 adapter把外部格式转换成 Envelope其余所有逻辑都能直接复用。我后来给项目加了不下五个新渠道每一个都只是写了一个几十行的转换函数完全没有动过核心管道。2.3 处理管道摄入、规范化、决策、投递整个消息处理流程被我拆成了四个阶段每个阶段之间用 Redis Stream 解耦摄入IngestHTTP 接口或者消息队列消费者接收到原始消息立即写入 Redis Stream返回 200 响应。这一步只做一件事快速确认收下了不让调用方等着。规范化Normalize从 Stream 里消费消息调用对应的 adapter 把原始数据转换成统一 Envelope然后重新写回另一个 Stream。决策Decide从 Envelope 里提取 EventType、Source、Payload 等特征依次经过规则引擎和 AI 分类器最终输出一个路由指令比如“转发到 team-a-alert 群并标记为 P1 优先级”。投递Deliver根据路由指令把消息发送到目标渠道。投递失败就进入重试流程重试超过上限就丢进死信队列。每个阶段都是独立的 worker 进程通过 Stream 的消费者组机制实现水平扩展。某一阶段处理能力不够的时候只需要多起几个 worker 实例不需要改任何业务代码。这套设计的收益在我后来接入高吞吐渠道时体现得非常明显某个渠道高峰期每秒能推 2000 条 webhook摄入 worker 会自动积压到 Stream 里后续阶段慢慢消费整个系统没有任何一个环节被打崩。3. 路由决策规则引擎和 AI 的协作边界3.1 为什么我坚持规则优先现在聊 AI 的博文很多一上来就让你把所有业务逻辑交给大模型。我的态度恰恰相反规则优先AI 兜底。原因很简单——AI 是有概率的。哪怕准确率做到 99%在每天上万条消息的规模下意味着每天有一百条消息可能被错误路由。如果这些消息是普通告警也就罢了万一是支付失败的异常回调被 AI 误判成了“需要人工介入”的低优先级消息后果可能就严重了。而规则引擎是确定性的同样的输入永远产生同样的输出这让消息处理的行为是可预期的。我更喜欢用“闸门 分拣员”的类比来解释这个设计规则引擎是一道道闸门把那些特征明确、处理方式固定的消息直接放行到对应的通道AI 是站在闸门后面的分拣员只处理那些特征模糊、没有明确规则定义的长尾消息。这样一来AI 的影响面被控制在一个可控范围内即使它偶尔判断失误也不会影响核心链路的确定性。3.2 规则引擎的具体设计规则引擎的输入是前面定义好的 Envelope输出是一个RouteDecision结构type RouteDecision struct { Action string json:action Target string json:target Priority int json:priority Params map[string]string json:params MatchedRule string json:matched_rule AIAssisted bool json:ai_assisted }规则本身用 YAML 配置支持基于 Source、EventType、Payload 字段、Priority 的匹配条件rules: - id: payment-failed-high-priority desc: 支付失败且金额大于1万的订单需要立即通知 when: event_type: payment.failed payload.amount: { gt: 10000 } then: action: notify target: ops-critical priority: 1 - id: ci-build-success-silent desc: CI构建成功只记录不通知 when: event_type: ci.build.success then: action: log-only规则之间是有优先级顺序的从上到下依次匹配命中即停。这段逻辑本身不复杂但真正写起来有几个容易踩的坑一是 YAML 里的when条件如果涉及嵌套字段取值路径的写法一定要统一我一开始同时支持点号和括号两种写法结果写规则的人经常分不清后来干脆统一成点号路径二是条件操作符不要一开始就追求大而全eq、neq、gt、lt、contains、exists这几个基本够用多了反而增加心智负担。3.3 AI 兜底怎么用 LLM 而不被它带偏规则覆盖不了的消息会进入 AI 分类流程。我在这个环节的设计原则是让 AI 做选择题而不是做翻译题。换句话说不指望 AI 生成一段目标渠道名而是让它在预定义的有限选项里做选择然后通过程序强制校验选项的合法性。具体实现上我把可路由目标、优先级、动作类型都枚举出来拼进 prompt 里要求模型只返回 JSON{ target: ops-critical, priority: P1, reason: 订单退款失败次数超过阈值疑似支付通道异常 }模型输出之后还会有一个校验层检查target是否存在于合法枚举列表里priority是否属于 P0/P1/P2 三档。只要校验不通过这条消息就会被降级为“待人工处理”而不是按照模型的原样输出硬着头皮投递。另外还有一个很关键的小细节AI 判断的reason会跟着消息一起存进审计日志。后续一旦发现某条消息被错误路由能根据模型给出的理由回溯是 prompt 设计的问题还是模型本身理解错了这比面对一个黑盒输出要省心得多。3.4 灰度与审计新规则上线会不会误伤AI 分类准确率到底够不够这些都是带病上线的问题。我的做法是给每条经过决策的消息都打上全量审计日志包括命中的规则 ID、AI 给出的理由、最终路由决策和执行结果。在这个基础上我专门做了一个审计面板可以按规则 ID 查看最近一周的命中次数、目标分布和异常率。每次新加规则都会先以log-only模式发布几天只记录“如果这条规则生效会怎么路由”不实际投递确认无误后再切换成notify模式。这套流程看起来保守但它帮我避免了好几次“看似合理的规则实际会造成消息风暴”的事故。4. 可靠性在不可靠的环境里做可靠的交付4.1 重试策略指数退避加抖动别让重试变成二次灾难投递环节是最容易出问题的。目标系统可能临时宕机、可能返回 5xx、可能网络闪断这些都不是处理器本身能控制的。我一开始用的是固定间隔重试每 30 秒重试一次结果某次目标系统挂了二十分钟重试消息在 Redis 里堆成了一座小山。后来换成了指数退避加抖动第一次重试等待 1 秒第二次 2 秒第三次 4 秒以此类推上限 300 秒同时每次等待时间加上一个 0 到 500 毫秒的随机抖动。这样做的原因是如果同时有一大批消息投递失败固定间隔重试会导致它们在同一个时间点集中冲击目标系统相当于人为制造了一次重试风暴加入抖动之后重试请求被天然打散目标系统恢复时会好受很多。重试次数也不是无限大的。单条消息最多重试 5 次超过之后就进入死信队列等待人工处理。这个上限是根据“目标系统常规故障恢复时间”反推出来的5 次重试最多需要十几分钟正常情况下系统早就缓过来了。4.2 幂等让同一条消息重复到达也安全“消息至少到达一次”几乎是所有分布式系统的默认现实但“至少一次”意味着调用方可能重试、渠道可能重复推送、中间环节可能重新消费。这时候如果没有幂等保护同一条告警就可能被发到群里三遍同一条支付回调就可能被业务系统处理三次。我在 Envelope 里专门设计了IdempotencyKey字段调用方可以在传入时指定如果没指定服务端会自动用Source EventType Payload的MD5生成一个。每次消息进入投递阶段前先在 Redis 的 Set 里检查这个 key 是否已存在存在说明这条消息已经处理过了直接丢弃不进入投递。不存在写入 Set 并设置过期时间我设的是 24 小时继续投递流程。这个实现说来只有几行代码但它解决的是消息管道里最让人头疼的重复投递问题。加了幂等保护之后我再也不怕渠道方抽风重复推 webhook 了。4.3 死信队列给处理不了的异常留一条人类通道不管重试机制设计得多完善总会有一些消息是永远也投递不出去的。可能是目标渠道已经不存在的群聊可能是 payload 结构坏到连规范化阶段都无法解析也可能是 AI 校验层判定为非法路由。这些消息如果一直卡在队列里会把后面的正常消息都堵住。所以我在hermes-agent里单独建了一个死信队列所有超过重试上限或者校验失败的消息都会被打上原因标签推送到这里。死信队列的设计目标不是自动解决所有问题而是给人工处理提供一个清晰的工作台能看到原始消息、能看到完整的处理轨迹、能看到失败的具体原因。我见过很多团队的 MQ 死信队列形同虚设因为没人看、没人管死信堆积成千上万条才发现。我的经验是死信队列一定要配一个简单的告警只要死信数量超过阈值比如 5 条就立刻通知管理员。死信量应该长期趋近于零一旦明显增长说明系统里大概率有 bug需要马上去排查。4.4 全链路追踪一条消息的一生前面提到过 Envelope 里有TraceID这个字段从消息被摄入的那一刻起就贯穿始终。我基于它实现了一张“消息生命周期表”每经过一个处理阶段就会写入一条记录阶段状态耗时备注ingest成功2ms原始 webhook 写入 Streamnormalize成功5msadapter 转换为统一信封decide成功120ms命中规则 payment-failed-high-prioritydeliver成功86ms已推送到 ops-critical 群排查线上问题的时候这张表几乎是第一手工具。某条消息没收到直接查 TraceID看它卡在了哪个阶段。是没进来还是决策错了还是投递失败重试完了每一步都清清楚楚不用靠猜。5. 实操在一个周末把 hermes-agent 跑起来5.1 最小部署这套系统的核心跑起来只需要两个容器一个是hermes-agent本体一个是 Redis。我用 Docker Compose 做了一个最小编排配置大概长这样services: redis: image: redis:7-alpine command: redis-server --appendonly yes volumes: - redis-data:/data hermes-agent: image: hermes-agent:latest depends_on: - redis environment: HERMES_REDIS_ADDR: redis:6379 HERMES_CONFIG_PATH: /etc/hermes/config.yaml HERMES_WORKER_INGEST: 2 HERMES_WORKER_DECIDE: 2 HERMES_WORKER_DELIVER: 3 ports: - 8080:8080 volumes: - ./config.yaml:/etc/hermes/config.yaml:roHERMES_WORKER_*这几个环境变量分别控制摄入、决策、投递三个阶段的 worker 数量。单机起步阶段不需要调得太大默认值各 2 个就已经能扛住日均几十万条消息的量级。5.2 第一个接入webhook 转发到群机器人跑通系统之后我建议第一个接入场景选一个最简单的把一份 webhook 告警转发到一个 IM 群机器人。这个过程虽然简单但能完整走一遍消息管道的所有环节非常适合用来验证系统是不是真的按预期工作。第一步在配置文件里加一个来源 adaptersources: - id: ops-webhook type: http path: /webhook/ops adapter: generic-json targets: - id: team-im-robot type: webhook url: https://your-im-robot-url headers: Content-Type: application/json第二步加一条转发规则rules: - id: ops-alert-notify when: source: ops-webhook payload.level: { gte: 2 } then: action: notify target: team-im-robot priority: 1第三步往POST http://localhost:8080/webhook/ops推送一条测试消息。如果一切正常几秒钟后目标群机器人就会收到封装后的通知。就这个最小闭环已经能覆盖“摄入 → 规范化 → 决策 → 投递”的完整链路后续再接入其他渠道本质上都只是加来源和加规则的事情。5.3 我踩过的三个坑第一个坑是时区问题。不同来源方推送的时间格式五花八门有的是 UTC、有的是带时区的 ISO8601、有的是纯数字时间戳。一开始我在规范化阶段只按一种格式解析结果某些消息的ReceivedAt直接解析失败。后来的做法是引入一个统一的时间解析函数支持常见的几种格式并且在解析失败时不抛异常而是降级为当前时间并记录日志。第二个坑是 webhook 请求体过大。有些外部系统动辄把几十 MB 的日志打包推过来如果直接存进 Redis Stream会把内存打爆。我的方案是给摄入接口加一个请求体大小上限默认 1MB超过上限的消息直接拒绝并返回 413同时记录一条审计日志。真正需要传大文件的场景应该走对象存储而不是塞进消息管道。第三个坑是重试风暴我前面提到过。当时是某个目标渠道连续故障加上固定间隔重试一小时产生了上万条重试请求Redis 的 CPU 飙到 90%。换成指数退避加抖动并且加了重试上限之后这个隐患才算彻底解决。现在每次接入新的目标渠道我都会先确认对方的错误响应语义再看要不要给这个目标单独设置更保守的重试参数。6. 后续演进方向与个人体会hermes-agent的第一版只做到了“能跑、可靠、可追踪”距离“好用”还有很多路要走。我目前正在折腾的有三件事一是把规则编辑器做成可视化界面让不熟悉 YAML 的同学也能安全地创建和修改路由规则二是做多租户隔离让不同团队可以共享同一个代理但各自的数据和配置完全隔离三是研究一下基于消息特征的自动健康检测在目标渠道出现异常的第一时间就自动停止投递并切换备用通道而不是傻等着重试耗尽。最后说点个人的真心话。做这个项目的过程里我最大的收获不是学会了 Go也不是深入理解了 Redis Stream而是真正想明白了一个道理工具的价值在于把人的注意力从“重复确认”中解放出来而不是制造新的需要确认的东西。消息代理听起来不像一个大模型项目那么性感但当它真正稳定运行、把你从每天几百条消息的轰炸中捞出来的时候那种踏实感是任何炫技都换不来的。如果你也在被类似的问题困扰我的建议是从最小的闭环开始不要一上来就规划几十个渠道、上百条规则。先把一个来源、一个目标、一条规则跑通感受一下消息在自己的控制下有序流动是什么体验然后再逐步扩展。你会发现原来让消息“长了腿自己跑”这件事并不复杂只是需要有人先动手。
我是在第七次因为同一个线上告警被不同渠道轮番轰炸之后才决定认真对待“消息分发”这个问题的。那天下午同一份故障通知以三条完全不同的格式分别从工单系统、IM群机器人、邮件列表涌进来而我却花了十几分钟才把三份信息对起来期间还差点漏掉最关键的一行错误码。团队里没人有义务做这件事——把四处逃窜的消息理顺——但我意识到只要还靠人来对账这种混乱就会一直重演。于是就有了hermes-agent这个项目不是又一个 ChatBot也不是又一个任务调度器而是一个站在所有消息来源和所有执行目标之间的“信使”。Hermes 本来就是希腊神话里替众神传信的角色用它给这个代理命名至少在我心里是顺理成章的。如果你也在维护一堆业务系统每天被 webhook、告警、回调、机器人消息反复打断又不甘心把所有内部流程都交给某个第三方平台那这篇文章应该能给你一些可落地的思路。我会从“为什么需要这样一个代理”讲起拆解它的核心架构、路由决策、可靠性设计最后给出一套能在周末跑起来的最小部署方案。不吹不黑只说我在实际开发中踩过的坑和最终沉淀下来的设计。1. 为什么值得做一个自己的消息代理我的通知爆炸现场1.1 分散渠道带来的隐性成本很多团队的第一反应是消息多就多呗又不是什么大事。但真正做过运维或者负责过线上系统的人都知道问题从来不在“消息数量”而在“消息的离散程度”。我当时面临的局面是这样的监控告警走 A 平台的 webhookCI/CD 结果走 B 平台的 webhook客服反馈进了 C 系统的工单支付回调由第三方直接打到我们某个内部接口还有几个外部合作方各自用不同的格式推送数据。每一个渠道单独看都很简单但当它们同时涌进来的时候人就成了那个“手忙脚乱的中间件”。我统计过一周的数据平均每天有 200 多条外部事件需要处理每条事件至少要看一眼才能知道“要不要管、该谁管”。按每条 30 秒计算光“看一眼”这个动作就吃掉了将近两个小时而且这两个小时是碎片化的、随时被打断的根本无法用来做深度工作。更可怕的是人工处理意味着不同的人会对同一条消息做出不同的判断——有人觉得是 P0 告警有人觉得可以缓缓根本没有统一的口径。1.2 现成集成平台为什么不够用既然消息分发这么麻烦为什么不上 n8n、IFTTT 这类现成的自动化平台这个问题我在立项前认真调研过也实际试跑了一个月最终得出结论它们解决的是“流程编排”问题而我真正需要的是一个“持续运转的消息管道”。差别在哪里流程编排关心的是“触发 → 动作”比如收到邮件就创建工单这是离散的、一次性的逻辑。但消息管道关心的是“任意来源 → 统一处理 → 可靠投递”它需要处理乱序到达、重复推送、格式不规范、目标端暂时不可用这类真实世界里天天发生的问题。现成平台在这方面的抽象往往是浅层的一旦遇到需要自定义幂等策略、自定义重试语义、或者对消息体做深度清洗的场景就会感觉处处受制。另外还有一个很现实的因素数据驻留和可用性。我们有一部分业务回调包含敏感信息公司合规要求这些数据不能落到第三方平台的日志里。这就直接排除了所有纯 SaaS 方案。我需要的是一个能完全跑在自己服务器上的东西数据从进入到流出全程可控。1.3 hermes-agent 的设计目标经过那段时间的反复思考我把hermes-agent的设计目标明确成了四条后文所有架构决策都围绕这几条展开单一抽象所有输入源不管来自 webhook 还是 IM 机器人还是邮件进入系统后都变成同一种“信封”结构后续处理逻辑完全不关心消息最初是从哪来的。先可靠后智能第一优先级是保证消息不丢、不重、能追踪。AI 决策是加分项但绝不能因为 AI 的不确定性影响消息的确定性投递。确定性规则与 AI 并行能用规则解决的问题绝不用 AI规则覆盖不了的长尾场景才交给模型并且 AI 的决策结果必须可审计、可回滚。本地优先所有核心组件都支持单机 Docker 部署不强制依赖任何云服务。这四个目标听起来不复杂但真正落地的时候几乎每一个都在挑战“最小实现”的诱惑。尤其是“先可靠后智能”这条我前后推翻过两版设计就是为了不在基础管道还没稳的时候急着把大模型塞进去。2. 核心架构一条消息从进入到落地的完整旅程2.1 技术选型为什么用 Go Redis 而不是 Python项目最开始的原型是用 Python 写的因为快三两天就能把想法跑通。但原型跑通之后我立刻遇到了两个问题一是 Python asyncio 在高并发 webhook 接入时稍不注意就会出现回调阻塞二是部署形态太重为了跑一个代理还得维护 Python 运行时和一堆依赖和“本地优先”的目标有点背道而驰。后来我把核心重写成了 Go。选择 Go 不是因为它时髦而是因为它在这个场景下确实合适goroutine 模型天然适合处理大量并发的 webhook 长连接编译出来只有一个静态二进制扔进容器里连基础镜像都能省。我做过一个对比同样跑 500 个并发 webhook 推送Go 版本的 P99 延迟比 Python 版本低了将近一半内存占用反而只有原来的四分之一。存储层我选了 Redis理由也很实在消息管道需要的是一个高吞吐、低延迟、支持过期策略的临时存储而不是一个强一致的关系型数据库。Redis 的 List 可以做队列Stream 可以做消费者组Hash 可以做去重表一个组件就把管道最核心的几件事全包了。维度Python 原型Go 重写版并发模型asyncio 事件循环goroutine channelP99 延迟500并发210ms96ms内存占用稳态780MB190MB交付产物源码 依赖包单个静态二进制部署依赖Python 3.10无2.2 统一消息模型信封模式在hermes-agent里所有消息进入系统后的第一件事就是被装进一个统一的“信封”。信封结构大概是这样的type Envelope struct { ID string json:id TraceID string json:trace_id Source string json:source EventType string json:event_type Payload json.RawMessage json:payload Headers map[string]string json:headers ReceivedAt time.Time json:received_at Priority int json:priority IdempotencyKey string json:idempotency_key }ID是这条消息在系统内部的唯一标识TraceID用来串联一次业务事件的完整链路Source记录消息来自哪个渠道EventType是类型标签Payload是原始业务数据Headers带上调用方传入的自定义属性Priority表示优先级IdempotencyKey则专门用于去重。为什么非要这样统一不可因为后端的规则引擎、AI 分类器、路由器和投递器统统只认这一个大结构。新增一个消息源的时候只需要写一个 adapter把外部格式转换成 Envelope其余所有逻辑都能直接复用。我后来给项目加了不下五个新渠道每一个都只是写了一个几十行的转换函数完全没有动过核心管道。2.3 处理管道摄入、规范化、决策、投递整个消息处理流程被我拆成了四个阶段每个阶段之间用 Redis Stream 解耦摄入IngestHTTP 接口或者消息队列消费者接收到原始消息立即写入 Redis Stream返回 200 响应。这一步只做一件事快速确认收下了不让调用方等着。规范化Normalize从 Stream 里消费消息调用对应的 adapter 把原始数据转换成统一 Envelope然后重新写回另一个 Stream。决策Decide从 Envelope 里提取 EventType、Source、Payload 等特征依次经过规则引擎和 AI 分类器最终输出一个路由指令比如“转发到 team-a-alert 群并标记为 P1 优先级”。投递Deliver根据路由指令把消息发送到目标渠道。投递失败就进入重试流程重试超过上限就丢进死信队列。每个阶段都是独立的 worker 进程通过 Stream 的消费者组机制实现水平扩展。某一阶段处理能力不够的时候只需要多起几个 worker 实例不需要改任何业务代码。这套设计的收益在我后来接入高吞吐渠道时体现得非常明显某个渠道高峰期每秒能推 2000 条 webhook摄入 worker 会自动积压到 Stream 里后续阶段慢慢消费整个系统没有任何一个环节被打崩。3. 路由决策规则引擎和 AI 的协作边界3.1 为什么我坚持规则优先现在聊 AI 的博文很多一上来就让你把所有业务逻辑交给大模型。我的态度恰恰相反规则优先AI 兜底。原因很简单——AI 是有概率的。哪怕准确率做到 99%在每天上万条消息的规模下意味着每天有一百条消息可能被错误路由。如果这些消息是普通告警也就罢了万一是支付失败的异常回调被 AI 误判成了“需要人工介入”的低优先级消息后果可能就严重了。而规则引擎是确定性的同样的输入永远产生同样的输出这让消息处理的行为是可预期的。我更喜欢用“闸门 分拣员”的类比来解释这个设计规则引擎是一道道闸门把那些特征明确、处理方式固定的消息直接放行到对应的通道AI 是站在闸门后面的分拣员只处理那些特征模糊、没有明确规则定义的长尾消息。这样一来AI 的影响面被控制在一个可控范围内即使它偶尔判断失误也不会影响核心链路的确定性。3.2 规则引擎的具体设计规则引擎的输入是前面定义好的 Envelope输出是一个RouteDecision结构type RouteDecision struct { Action string json:action Target string json:target Priority int json:priority Params map[string]string json:params MatchedRule string json:matched_rule AIAssisted bool json:ai_assisted }规则本身用 YAML 配置支持基于 Source、EventType、Payload 字段、Priority 的匹配条件rules: - id: payment-failed-high-priority desc: 支付失败且金额大于1万的订单需要立即通知 when: event_type: payment.failed payload.amount: { gt: 10000 } then: action: notify target: ops-critical priority: 1 - id: ci-build-success-silent desc: CI构建成功只记录不通知 when: event_type: ci.build.success then: action: log-only规则之间是有优先级顺序的从上到下依次匹配命中即停。这段逻辑本身不复杂但真正写起来有几个容易踩的坑一是 YAML 里的when条件如果涉及嵌套字段取值路径的写法一定要统一我一开始同时支持点号和括号两种写法结果写规则的人经常分不清后来干脆统一成点号路径二是条件操作符不要一开始就追求大而全eq、neq、gt、lt、contains、exists这几个基本够用多了反而增加心智负担。3.3 AI 兜底怎么用 LLM 而不被它带偏规则覆盖不了的消息会进入 AI 分类流程。我在这个环节的设计原则是让 AI 做选择题而不是做翻译题。换句话说不指望 AI 生成一段目标渠道名而是让它在预定义的有限选项里做选择然后通过程序强制校验选项的合法性。具体实现上我把可路由目标、优先级、动作类型都枚举出来拼进 prompt 里要求模型只返回 JSON{ target: ops-critical, priority: P1, reason: 订单退款失败次数超过阈值疑似支付通道异常 }模型输出之后还会有一个校验层检查target是否存在于合法枚举列表里priority是否属于 P0/P1/P2 三档。只要校验不通过这条消息就会被降级为“待人工处理”而不是按照模型的原样输出硬着头皮投递。另外还有一个很关键的小细节AI 判断的reason会跟着消息一起存进审计日志。后续一旦发现某条消息被错误路由能根据模型给出的理由回溯是 prompt 设计的问题还是模型本身理解错了这比面对一个黑盒输出要省心得多。3.4 灰度与审计新规则上线会不会误伤AI 分类准确率到底够不够这些都是带病上线的问题。我的做法是给每条经过决策的消息都打上全量审计日志包括命中的规则 ID、AI 给出的理由、最终路由决策和执行结果。在这个基础上我专门做了一个审计面板可以按规则 ID 查看最近一周的命中次数、目标分布和异常率。每次新加规则都会先以log-only模式发布几天只记录“如果这条规则生效会怎么路由”不实际投递确认无误后再切换成notify模式。这套流程看起来保守但它帮我避免了好几次“看似合理的规则实际会造成消息风暴”的事故。4. 可靠性在不可靠的环境里做可靠的交付4.1 重试策略指数退避加抖动别让重试变成二次灾难投递环节是最容易出问题的。目标系统可能临时宕机、可能返回 5xx、可能网络闪断这些都不是处理器本身能控制的。我一开始用的是固定间隔重试每 30 秒重试一次结果某次目标系统挂了二十分钟重试消息在 Redis 里堆成了一座小山。后来换成了指数退避加抖动第一次重试等待 1 秒第二次 2 秒第三次 4 秒以此类推上限 300 秒同时每次等待时间加上一个 0 到 500 毫秒的随机抖动。这样做的原因是如果同时有一大批消息投递失败固定间隔重试会导致它们在同一个时间点集中冲击目标系统相当于人为制造了一次重试风暴加入抖动之后重试请求被天然打散目标系统恢复时会好受很多。重试次数也不是无限大的。单条消息最多重试 5 次超过之后就进入死信队列等待人工处理。这个上限是根据“目标系统常规故障恢复时间”反推出来的5 次重试最多需要十几分钟正常情况下系统早就缓过来了。4.2 幂等让同一条消息重复到达也安全“消息至少到达一次”几乎是所有分布式系统的默认现实但“至少一次”意味着调用方可能重试、渠道可能重复推送、中间环节可能重新消费。这时候如果没有幂等保护同一条告警就可能被发到群里三遍同一条支付回调就可能被业务系统处理三次。我在 Envelope 里专门设计了IdempotencyKey字段调用方可以在传入时指定如果没指定服务端会自动用Source EventType Payload的MD5生成一个。每次消息进入投递阶段前先在 Redis 的 Set 里检查这个 key 是否已存在存在说明这条消息已经处理过了直接丢弃不进入投递。不存在写入 Set 并设置过期时间我设的是 24 小时继续投递流程。这个实现说来只有几行代码但它解决的是消息管道里最让人头疼的重复投递问题。加了幂等保护之后我再也不怕渠道方抽风重复推 webhook 了。4.3 死信队列给处理不了的异常留一条人类通道不管重试机制设计得多完善总会有一些消息是永远也投递不出去的。可能是目标渠道已经不存在的群聊可能是 payload 结构坏到连规范化阶段都无法解析也可能是 AI 校验层判定为非法路由。这些消息如果一直卡在队列里会把后面的正常消息都堵住。所以我在hermes-agent里单独建了一个死信队列所有超过重试上限或者校验失败的消息都会被打上原因标签推送到这里。死信队列的设计目标不是自动解决所有问题而是给人工处理提供一个清晰的工作台能看到原始消息、能看到完整的处理轨迹、能看到失败的具体原因。我见过很多团队的 MQ 死信队列形同虚设因为没人看、没人管死信堆积成千上万条才发现。我的经验是死信队列一定要配一个简单的告警只要死信数量超过阈值比如 5 条就立刻通知管理员。死信量应该长期趋近于零一旦明显增长说明系统里大概率有 bug需要马上去排查。4.4 全链路追踪一条消息的一生前面提到过 Envelope 里有TraceID这个字段从消息被摄入的那一刻起就贯穿始终。我基于它实现了一张“消息生命周期表”每经过一个处理阶段就会写入一条记录阶段状态耗时备注ingest成功2ms原始 webhook 写入 Streamnormalize成功5msadapter 转换为统一信封decide成功120ms命中规则 payment-failed-high-prioritydeliver成功86ms已推送到 ops-critical 群排查线上问题的时候这张表几乎是第一手工具。某条消息没收到直接查 TraceID看它卡在了哪个阶段。是没进来还是决策错了还是投递失败重试完了每一步都清清楚楚不用靠猜。5. 实操在一个周末把 hermes-agent 跑起来5.1 最小部署这套系统的核心跑起来只需要两个容器一个是hermes-agent本体一个是 Redis。我用 Docker Compose 做了一个最小编排配置大概长这样services: redis: image: redis:7-alpine command: redis-server --appendonly yes volumes: - redis-data:/data hermes-agent: image: hermes-agent:latest depends_on: - redis environment: HERMES_REDIS_ADDR: redis:6379 HERMES_CONFIG_PATH: /etc/hermes/config.yaml HERMES_WORKER_INGEST: 2 HERMES_WORKER_DECIDE: 2 HERMES_WORKER_DELIVER: 3 ports: - 8080:8080 volumes: - ./config.yaml:/etc/hermes/config.yaml:roHERMES_WORKER_*这几个环境变量分别控制摄入、决策、投递三个阶段的 worker 数量。单机起步阶段不需要调得太大默认值各 2 个就已经能扛住日均几十万条消息的量级。5.2 第一个接入webhook 转发到群机器人跑通系统之后我建议第一个接入场景选一个最简单的把一份 webhook 告警转发到一个 IM 群机器人。这个过程虽然简单但能完整走一遍消息管道的所有环节非常适合用来验证系统是不是真的按预期工作。第一步在配置文件里加一个来源 adaptersources: - id: ops-webhook type: http path: /webhook/ops adapter: generic-json targets: - id: team-im-robot type: webhook url: https://your-im-robot-url headers: Content-Type: application/json第二步加一条转发规则rules: - id: ops-alert-notify when: source: ops-webhook payload.level: { gte: 2 } then: action: notify target: team-im-robot priority: 1第三步往POST http://localhost:8080/webhook/ops推送一条测试消息。如果一切正常几秒钟后目标群机器人就会收到封装后的通知。就这个最小闭环已经能覆盖“摄入 → 规范化 → 决策 → 投递”的完整链路后续再接入其他渠道本质上都只是加来源和加规则的事情。5.3 我踩过的三个坑第一个坑是时区问题。不同来源方推送的时间格式五花八门有的是 UTC、有的是带时区的 ISO8601、有的是纯数字时间戳。一开始我在规范化阶段只按一种格式解析结果某些消息的ReceivedAt直接解析失败。后来的做法是引入一个统一的时间解析函数支持常见的几种格式并且在解析失败时不抛异常而是降级为当前时间并记录日志。第二个坑是 webhook 请求体过大。有些外部系统动辄把几十 MB 的日志打包推过来如果直接存进 Redis Stream会把内存打爆。我的方案是给摄入接口加一个请求体大小上限默认 1MB超过上限的消息直接拒绝并返回 413同时记录一条审计日志。真正需要传大文件的场景应该走对象存储而不是塞进消息管道。第三个坑是重试风暴我前面提到过。当时是某个目标渠道连续故障加上固定间隔重试一小时产生了上万条重试请求Redis 的 CPU 飙到 90%。换成指数退避加抖动并且加了重试上限之后这个隐患才算彻底解决。现在每次接入新的目标渠道我都会先确认对方的错误响应语义再看要不要给这个目标单独设置更保守的重试参数。6. 后续演进方向与个人体会hermes-agent的第一版只做到了“能跑、可靠、可追踪”距离“好用”还有很多路要走。我目前正在折腾的有三件事一是把规则编辑器做成可视化界面让不熟悉 YAML 的同学也能安全地创建和修改路由规则二是做多租户隔离让不同团队可以共享同一个代理但各自的数据和配置完全隔离三是研究一下基于消息特征的自动健康检测在目标渠道出现异常的第一时间就自动停止投递并切换备用通道而不是傻等着重试耗尽。最后说点个人的真心话。做这个项目的过程里我最大的收获不是学会了 Go也不是深入理解了 Redis Stream而是真正想明白了一个道理工具的价值在于把人的注意力从“重复确认”中解放出来而不是制造新的需要确认的东西。消息代理听起来不像一个大模型项目那么性感但当它真正稳定运行、把你从每天几百条消息的轰炸中捞出来的时候那种踏实感是任何炫技都换不来的。如果你也在被类似的问题困扰我的建议是从最小的闭环开始不要一上来就规划几十个渠道、上百条规则。先把一个来源、一个目标、一条规则跑通感受一下消息在自己的控制下有序流动是什么体验然后再逐步扩展。你会发现原来让消息“长了腿自己跑”这件事并不复杂只是需要有人先动手。
