hermes-agent实践：用消息代理与AI Agent构建统一消息中枢

我发现团队聊天工具越来越多，监控告警、工单通知、发布流水线消息从四面八方涌来。那段时间我最常做的事，就是从一个窗口切到另一个窗口，把消息复制、转达、再确认。后来我干脆写了一个名为hermes-agent的个人自动化代理，专门负责统一收消息、判断优先级、调用工具去执行，最后把结果回传给对应的人。这篇文章想把这些天折腾下来的设计思路、代码结构和踩坑记录整理出来，给正在做类似“消息中枢 / AI Agent”方向的读者一个参考。

hermes-agent的名字取自希腊神话里的信使神赫尔墨斯，它本身不生产业务数据，只负责可靠地把消息从一个地方送到另一个地方，并在途中完成必要的转换和动作。如果你也在考虑搭建一个团队内部使用的消息代理，或者想把自己的 AI 能力从“单轮问答”升级成“自动执行任务”，这篇博文应该能给你一些可落地的思路。

1. 创建 hermes-agent 前，我到底被什么问题逼疯了

1.1 消息分散是表象，上下文断裂才是真痛

事情最开始其实没有这么宏大。我只是发现自己每天三分之二的时间都花在“倒腾消息”上：IM 群里有人问线上服务是不是挂了，我要先去查监控；飞书文件夹里出现一条新的工单，我要去翻知识库找解决方案；发布系统推送了“构建失败”，我又要跳到日志平台拉错误堆栈。

单独看每一条消息都不难处理，难的是每条消息都需要一段背景上下文。之前的解决方式全凭我的个人记忆，状态一多就出错。有一次凌晨三点，告警系统连续发了十几条磁盘告警，我醒来后一条条看完，才发现其中三条其实来自同一台机器的同一波抖动，白折腾了半天。最让人崩溃的，是拿着刚通过的消息去执行时又发现权限不够、参数已经过期，整个链路重新来一遍。

所以我最核心的需求不是“把消息聚合到一个页面”，而是让消息带着足够的上下文流动起来，并且到了该执行的一步就直接执行，不要再让我来回翻译。

1.2 Hermes 这个名字要表达的三层意思

起名hermes-agent的时候，我对这个名字有三层期待：

• 信使，不决策：它负责把消息准确送达，不在中途夹带私货，也不擅自省略信息。
• 翻译官：不同系统的消息格式千差万别，它能把外部消息翻译成内部统一语言。
• 执行力：不只是传话，必要的时候它能调用工具、触发流程，做到“消息到，事情毕”。

这三点后来直接决定了项目的整体架构。一个消息代理最忌讳的就是把自己做成一个巨大的业务逻辑堆砌区，所以我在设计上刻意让“路由判断”和“工具执行”解耦，让模型只负责理解意图，真正的动作仍由可审计的代码完成。

2. 系统架构设计：以“消息包裹”为核心的流转模型

2.1 核心思想：一切皆消息，一切消息都有状态

要想大幅度降低复杂度，必须先定义一个简单到不可能搞错的核心抽象。在hermes-agent里，我把一切进入系统的数据都封装成统一的HermesMessage对象，它会经历完整的生命周期：

Received -> Parsed -> Routed -> Executing -> Completed | -> Failed -> Retrying

你可以把它类比成快递包裹：从菜鸟驿站（入口）收件，经历分拣中心（路由），最后由快递员（执行器）送到收件人手上。中间的每个节点都有留痕、都有重试机制，这就是可靠性的基础。

对比一下我之前的习惯：直接在代码里写一堆 if/else 处理不同来源的数据，刚开始很爽，后来每加一个渠道就要改动核心代码，线上行为不可预测，每次发布都心惊胆战。用统一消息模型之后，新渠道只是做一个“适配器”，核心链路完全不用变。

2.2 功能模块拆分：接入层、路由层、执行层、存储层

整体划分为四个模块，每个模块只负责一件事：

这四个模块之间通过一个内部事件总线通信。我最初的是想用消息队列把每一层彻底解耦，后来意识到个人项目不用背负“微服务”的包袱，所以实际落地时采用了进程内事件总线加 Redis Stream 的混合方案。单机运行的时候是同步调用，横向扩容时再切换到 Stream，这样兼顾了开发和部署的灵活性。

2.3 为什么不做成纯 LLM 应用

开始时有朋友建议：直接用大模型做端到端，把消息一股脑丢给它，让它自己决定怎么回。我也试过这条路，但它有几个很难解决的问题：

• 不可控的输出结构：模型可能临场发明不存在的工具名，或生成一段含糊行动描述。
• 上下文窗口限制：把大量日志和工单全塞进提示词既不经济也不稳定。
• 权限边界模糊：让模型直接拥有数据库操作权限，风险太高。

所以hermes-agent的做法是：让 LLM 做人话到结构化指令的翻译，让代码去执行具体的动作。模型不再拥有直接执行能力，它只输出一个 JSON 格式的“意图 + 参数”，后续是否执行、如何执行仍由路由层和执行层的代码严格把关。这就好像你让助手帮你“订外卖”，助手只需要理解你的意图并转录成标准订单，真正下单仍走固定流程。

3. 核心链路拆解：一条消息从接入到闭环的全过程

3.1 接入层：把外部世界的乱象挡在外面

不同系统最典型的差异体现在消息格式上。比如飞书消息是 JSON，GitLab Webhook 也是 JSON，但字段命名差异极大；企业微信的消息结构又完全不一样。我在接入层做了一个标准接口，所有适配器都实现同一个抽象类：

from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional @dataclass class HermesMessage: msg_id: str source: str # 来源，如 feishu, gitlab, prometheus msg_type: str # 消息类型，如 alert, command, question content: str # 统一后的文本内容 raw_data: dict # 原始数据，保留用于审计 created_at: datetime = field(default_factory=datetime.utcnow) trace_id: str = "" # 链路追踪 ID sender_id: str = "" # 发送人 target_channel: str = "" # 回复目标渠道 metadata: dict = field(default_factory=dict) class BaseIngressAdapter(ABC): @abstractmethod async def receive(self, request: Any) -> HermesMessage: """将外部请求统一转换成 HermesMessage"""

以飞书机器人为例，适配器要做的事情包括：解析 event 里的message.content，提取open_id，判断是否属于需要代理响应的会话，然后把文本内容填入HermesMessage.content。这样路由层从来不需要关心“来自飞书还是来自 Webhook”，它看到的永远是结构体。

3.2 路由层：意图识别与规则兜底的双轨方案

路由层是代理的“大脑”，负责回答一个问题：这条消息要触发什么动作。我采用的是双轨制：

1. 快速规则路径：先检测关键词和正则，如果命中高置信度规则，直接走对应动作。
2. 模型推理路径：规则没命中时，调用 LLM 把自然语言转换成意图 JSON。

为什么要保留规则路径？因为很多日常操作是高度确定的，例如“查一下磁盘使用率”“把服务重启一下”。这种场景让模型参与完全是浪费时间和金钱，而且模型还可能产生幻觉。规则路径的响应时间可以压到毫秒级，模型路径则通常需要一到三秒。两者的输出都收敛到同一个Intent结构：

@dataclass class Intent: action: str # 例如 query_disk_usage params: dict # 例如 {"host": "web-01"} confidence: float # 置信度 fallback_reason: str = "" # 路由命中的依据

实际测试中，大约 60% 的常用操作通过规则路径就能解决，剩下 40% 才真正需要模型。这个比例极大地节约了成本，也让整体响应速度更稳定。

路由层还有一个关键点：做权限前置校验。任何人通过 IM 发消息都能让代理去执行命令，这显然不合理。我增加了一层简单的访问控制列表，校验发送者的身份和动作的权限等级。最高危的操作（比如删库、上线发布）除了要求发送者具备权限，还必须二次确认。

3.3 执行层：工具注册中心与统一调用协议

当路由层输出了Intent，执行层就该上场了。在hermes-agent里，每一个可被代理调用的能力都被封装成“工具”，统一注册到中心。工具定义遵循一个简单协议：

@dataclass class ToolDefinition: name: str description: str parameters_schema: dict executor: callable timeout_seconds: float = 10.0 permission_level: int = 1

在向模型提供工具清单时，hermes-agent会动态筛选当前发送者有权限看到的工具，防止低权限用户触达敏感操作。执行器本身由普通 Python 函数实现，例如一个查询发布窗口的工具：

async def query_release_window(env: str, service: str) -> str: # 这里读取内部的发布日历接口 result = await fetch_json( f"https://release-calendar.internal/api/window", params={"env": env, "service": service} ) return result.to_plain_text()

调用完成后，结果会被包装成标准回复文本，再交给反馈模块送回用户渠道。整个过程支持同步和异步两种模式：轻量查询同步返回，耗时任务则先回复“任务已受理”，再通过回调把结果推送到用户。

执行层还有一个容易被人忽略的设计：每次执行都要写入审计日志。什么时候、谁、通过什么渠道、调用了哪个工具、输入参数是什么、结果如何，全部落库。这对“代理”类项目至关重要，因为代理本质上是代表用户行使操作权，没有审计就没法追溯责任。

3.4 可观测性：代理不能是个黑盒

我在一开始就决定给hermes-agent打满日志和指标。每个消息的唯一trace_id会贯穿接入、路由、执行全过程，日志里随时能 grep 出来。指标方面主要关注四个东西：

• 消息接入速率（每秒处理多少条入站消息）
• 路由耗时分布（规则路径与模型路径分开统计）
• 工具调用成功率与延时
• 每个来源渠道的错误量

上线第一天朋友问我：代理看起来回话挺快，你怎么知道它没在乱来？我说因为每一次动作都有日志、有 trace。出现问题不是靠猜，直接查msg_id就能还原整个过程。这是从无数线上事故里学到的教训：任何自动化系统，只要看不到内部状态，最终都会变成一个“薛定谔的代理”。

4. 技术选型复盘：为什么留下这些依赖而不是那些

4.1 Python + FastAPI：个人项目里最务实的组合

我没有选择 Go，也没有选择 Node.js，核心原因是这个项目需要快速接入大量 AI 生态和数据处理库，Python 在这方面的优势无可替代。FastAPI 提供了出色的异步性能、自动生成 OpenAPI 文档，非常适合做接入层。即使是个人项目，我也希望能随时用浏览器看一眼接口文档，这个需求 FastAPI 天然满足。

开发过程中最大的体会是：选型不是选最热门的，而是选你最能在深夜里维护的那个。Python 是我最熟悉的语言，遇到诡异 bug 时我能在几分钟内定位问题。对于一个需要长期维护的工具型项目，这一点比微小的性能优势重要得多。

4.2 Redis Stream 而不是 Celery：可控的队列才是好队列

消息代理的核心是队列选型。我用过 Celery，也用过 RabbitMQ，但最终选择了 Redis Stream。原因有三：

• 部署简单：Redis 几乎是每台服务器必装的组件，不需要额外运维一套中间件。
• 消费者组语义：Stream 的XREADGROUP天然支持多个 worker 分摊消息，也支持未确认消息的重新投递。
• 可观察性好：Redis 命令行直接能看到 backlog、pending 消息，定位问题非常方便。

对比一下市面方案：

随着项目增长，如果你需要更复杂的工作流编排，可以考虑引入 Temporal。但就现阶段而言，Redis Stream 完全够用，而且它让整体架构保持在“一眼能看懂”的复杂度层级。我个人非常推崇这种设计原则：先把简单方案用到极限，再考虑引入重型工具。

4.3 自研状态机为什么比 Workflow 框架更顺手

很多人会问：状态流转这么麻烦，直接上 Airflow 或者 Temporal 不是更好吗？我的答案是：如果你的核心任务是“处理单条消息并回传结果”，引入工作流框架是在制造隐性的分布式复杂度。hermes-agent的状态机只有五个状态、六条转移边，用纯 Python 枚举加一个状态转移表就能表达清楚：

class MessageState(Enum): RECEIVED = "received" PARSED = "parsed" ROUTED = "routed" EXECUTING = "executing" COMPLETED = "completed" FAILED = "failed" TRANSITIONS = { MessageState.RECEIVED: {MessageState.PARSED, MessageState.FAILED}, MessageState.PARSED: {MessageState.ROUTED, MessageState.FAILED}, MessageState.ROUTED: {MessageState.EXECUTING, MessageState.FAILED}, MessageState.EXECUTING: {MessageState.COMPLETED, MessageState.FAILED}, MessageState.FAILED: {MessageState.EXECUTING}, # 重试 }

每个状态转移都会触发一个持久化事件，写入 PostgreSQL。这样即使进程崩溃，重启后也能从最后状态继续恢复。自研并不是因为框架不好，而是因为这个状态机简单到用框架反而像在杀鸡用牛刀。

5. 踩坑实录：代理运行路上最值得说的三个问题

5.1 重复消费问题：同一个告警为什么执行了三次

第一次压测时我发现了严重问题：一个消息被消费者组处理了三次，数据库里写入了三条重复记录。排查后发现根因是XREADGROUP的自动确认机制配合超时重试导致消息被重新投递。

当时的代码逻辑是：从 Stream 里读到消息后，立刻执行XACK确认，再开始处理业务逻辑。这样的问题在于：如果业务逻辑抛异常，重新投递机制不会认为这条消息处理完成，还是会把它重新发给消费者。而另一个 worker 接单后执行成功，但异常后重试的 worker 又执行了一遍，造成重复。

解决方案是在执行器外围包一层幂等防护：每条消息的msg_id作为唯一键，先检查 PostgreSQL 里是否已有同 ID 的执行记录，有就直接跳过。执行结果写入前做一次冲突检查。这个改动之后，重复消费不再产生副作用。

从这件事得到的经验是：任何队列系统都不能保证恰好一次投递，系统设计必须预设“可能会重复，但重复无副作用”。现在hermes-agent的所有工具函数都被要求入参携带幂等键，把它当成一个基本约定。

5.2 上下文爆炸问题：提示词里塞了太多日志，结果模型开始胡说八道

最初做“告警根因分析”功能时，我的做法是把原始日志全量塞进提示词，让模型“自由发挥”。结果模型经常被大量冗余日志干扰，产生错误的根因推测，甚至编造出日志里根本没有的进程名。

后来我反思：模型需要的是高质量特征，而不是海量数据。现在hermes-agent处理告警时，会先用规则把日志压缩成几个关键信息：错误码出现次数、时间线、关联服务、异常堆栈的第一行和最后一行。只有压缩后的摘要才进入模型。

同时针对长对话场景，我实现了一个轻量记忆管理器：最近的对话以完整形式保留，更早的对话自动摘要成状态卡片，避免上下文窗口被塞满。这个机制上线后，模型回复的准确率明显提升，响应耗时也下降了近 40%。

5.3 工具调用黑洞：第三方接口没响应，整个任务卡死了

代理调用外部工具时，最怕遇到“黑洞”：接口既不返回成功也不返回失败，TCP 连接就这么挂着，直到超时。初期我没有设置工具级超时，导致一个第三方服务抖动时，hermes-agent的 worker 线程全部被占满，其他正常消息也处理不了。

后来我在工具注册中心强制每个工具必须配置超时时间，并引入两级熔断：

• 第一级：单次调用的超时控制，超过时间直接返回错误。
• 第二级：连续 10 次失败后打开熔断器，后续请求直接快速失败，不再进入调用逻辑；经过冷却时间后放一半流量试探恢复情况。

这两级保护上线后，即使外部系统彻底不可用，代理也能在几毫秒内给出“当前服务异常，请稍后再试”的反馈。这正是代理和普通脚本的本质区别：脚本可以崩，代理必须优雅降级。

6. 实测结果与部署形态：跑了一个多月，数据说明问题

6.1 性能数字：千条消息压测下的表现

用一个模拟工具压测了hermes-agent的处理能力，数据如下：

需要说明的是，模型路径的 QPS 主要受限于模型 API 的响应速度而非代理本身。如果换用流式输出，客户端首字延迟可以降到 500ms 内，但完整意图解析仍需等完整 JSON 返回，所以这个数字是符合预期的。

运行一个多月以来，hermes-agent每天大约处理 1500 到 3000 条消息，处理成功率 99.3%。失败主要集中在第三方接口不稳定这一环节，代理本身的崩溃次数为零——这是状态机设计带来的好处，任何异常都能被捕获并进入重试状态。

6.2 部署形态：一台 2C4G 的云服务器足够

部署用的是 Docker Compose，里面跑五个容器：接口服务、worker、Redis、PostgreSQL、Prometheus。整机资源占用约 1.5 个 CPU 核心和 2.1GB 内存。日志按天轮转，保留 30 天。我用 GRAFANA 做了一个简易面板，展示消息量、耗时和错误率。

对个人或小团队来说，这个部署规模非常友好，一台 2C4G 的入门云服务器就能稳定运行，每月成本很低。如果消息量涨到每天几万条，可以先把 Redis 和 PostgreSQL 拆到独立节点，再横向扩容 worker，架构上不会遇到瓶颈。

7. 后续规划：代理不应该停在“传话”这一步

目前的hermes-agent已经帮我撑起了日常消息处理和轻度自动化，但距离“团队级代理”还有几段路要走。

首先是多租户隔离。现在所有消息和工具共用一个环境，权限控制比较粗糙。下一步打算引入工作空间概念，不同小组的数据严格隔离，避免出现 A 组成员向代理发命令操作 B 组资源的情况。

其次是主动感知能力。当前代理依然是被动响应，消息来一条处理一条。我希望它能定时巡检业务状态，发现异常主动发送告警。这会让代理从“传话员”进化成“哨兵”，价值会大很多。

最后是更完善的工具生态。社区里已经有很多现成的 API 连接器，如果能以插件形式扩展，hermes-agent的复用价值会进一步提升。我会继续维护这个项目，也会把更多实践中验证有效的模式沉淀成文档。

最后说一点个人体会：写hermes-agent最大的收获，不是代码量或功能列表，而是彻底改变了我处理“消息风暴”的方式。当我把所有入口统一成一套逻辑后，再多的新渠道接入也只是增加一个适配器而已。如果你所在的团队也面临消息混乱、重复操作、上下文撕裂的问题，不妨从一个小而美的消息代理开始，先让一条消息自动流转起来，再逐步扩大边界。这个方向能带来的效率提升，远比你想象中更大。