【系列：手搓自主 AI Agent：Hermes 架构原理剖析 · 第 9 篇】

Gateway 与平台适配器：一套循环接入十五个平台

导读： 同一个 agent，CLI 里能用，Telegram、Discord、企业微信里也能用——核心循环一行没变，变的只是消息从哪来、回复往哪送。本文拆解 Hermes 架构的 Gateway 与平台适配器：会话隔离、统一消息格式、并发串行化、断线重连，一套机制打通 15+ 平台。

从最笨的微信机器人说起

假设你写好了自己的 agent，它能在终端里和你对话、调用工具、记住上下文。一切都很美好，直到你同事说：“能不能让它在企业微信里也用上？”

最简单粗暴的想法是什么？连上微信的 WebSocket，死循环收消息，调一次 run_conversation，把回复发回去。大概 40 行代码就能跑起来。

但这个"最笨的实现"有三个致命问题。

三个致命问题

第一，没有记忆。 每条消息进来都是一次全新的对话，agent 不记得你五分钟前说过什么。之前做的持久化全部白费。

第二，只能接微信。 明天你想接 Telegram，怎么办？改这 40 行代码？那后天接 Discord 呢？每接一个平台就 fork 一份代码，最后维护成本爆炸。

第三，并发消息。 agent 正在思考的时候，用户又发了一条消息。是排队处理？直接丢弃？还是打断当前思考？三种选择背后是完全不同的设计。

这三个问题，就是 s12 要解决的核心。

问题一：session key 会话隔离

先解决记忆问题。思路很简单：给每个会话一个唯一标识，用这个标识去数据库里拉历史。

def build_session_key(source: SessionSource, agent_name: str = "main") –> str:
parts = [f"agent:{agent_name}:{source.platform}:{source.chat_type}:{source.chat_id}"]
if source.chat_type == "group":
parts.append(source.user_id)
return ":".join(parts)

私聊的 key 长这样：

agent:main:wecom:dm:zhangsan

群聊的 key 长这样：

agent:main:wecom:group:grp_001:zhangsan

注意群聊的 key 里多了 user_id。这意味着张三和李四在同一个群里 @agent，各自拥有独立的对话历史——互不干扰。

这一步解决的是"每条消息全新对话"的问题。agent 拿到 session key，从数据库加载历史，继续之前的上下文。

问题二：MessageEvent 统一格式

接下来解决"只能接微信"的问题。

核心思路：不管消息来自哪个平台，进入核心循环之前，先翻译成同一种格式。

class MessageType(Enum):
TEXT = "text"
PHOTO = "photo"
VOICE = "voice"
DOCUMENT = "document"

@dataclass
class SessionSource:
platform: str # "console", "telegram", "wecom", …
chat_id: str
chat_type: str # "dm" or "group"
user_id: str
user_name: str = ""

@dataclass
class MessageEvent:
message_id: str
text: str
source: SessionSource
message_type: MessageType = MessageType.TEXT
media_urls: list[str] = field(default_factory=list)

微信的消息、Telegram 的消息、Discord 的消息，翻译完之后都是同一个 MessageEvent。下游代码根本不需要知道消息来自哪个平台。

这就是适配器的三职责：

每个平台适配器只做这两件翻译工作。

GatewayRunner：启动、路由、管理

有了统一格式，还需要一个东西来管理所有适配器。这就是 GatewayRunner。

它的工作有三件：

• 启动所有适配器：connect 全部跑起来
• 路由消息：适配器收到消息 → 调 handle_message → 找到对应 session → 交给 agent
• 管理会话：session 的创建、缓存、过期

两个关键设计值得注意。

第一，agent 实例按 session key 缓存复用。 不是每条消息都创建一个新的 agent，而是同一个 session 复用同一个 agent 实例。否则每条消息都要重新加载全部工具、重新初始化记忆系统。

第二，history 每次都从数据库重新拉取。 不依赖 agent 内部记忆。为什么？因为历史可能被 /undo、/compress、会话过期修改过。agent 内部记忆是运行时的，数据库才是权威。每次从数据库拉，保证拿到的是最新状态。

问题三：并发消息串行化

最后一个问题：用户连发三条消息，agent 正在思考第一条，怎么办？

Hermes 的做法是串行化 + 优雅中断：

self._active_sessions: dict[str, asyncio.Event] = {}
self._pending_messages: dict[str, MessageEvent] = {}

• 一个 session 同一时间只有一个 agent 在跑
• 新消息暂存，只保留最后一条（平台用户连发多条通常是补充同一个意思）
• 给正在运行的 agent 发中断信号：agent._interrupt_requested = True

关键在"优雅中断"：不是粗暴地杀掉任务，而是：

中断后内容不丢。部分回复、已执行工具、被跳过工具的占位，全部存数据库。下一轮从数据库加载完整脉络，接着聊。

从 Gateway 到适配器：共性问题浮出水面

到这里，Gateway 的架构已经清楚了。但新的问题来了：一个适配器容易写，第二个、第三个呢？

接微信的时候发现消息可能被截断成半截；接企业微信发现媒体 URL 是临时的，过一会儿就失效；接 Discord 发现同样的消息可能被重复推送。

这些不是某个平台的特殊情况，而是所有平台都会遇到的共性问题。s13 把这些共性机制抽出来，做成一套基础设施。

BasePlatformAdapter：抽象基类

先看抽象基类，这是所有适配器的骨架：

class BasePlatformAdapter(ABC):
def __init__(self, platform_name: str):
self.platform_name = platform_name
self._on_message: Callable | None = None # injected by GatewayRunner
self._running = False

@abstractmethod
async def connect(self) –> bool: ...
@abstractmethod
async def disconnect(self): ...
@abstractmethod
async def send(self, chat_id: str, content: str) –> bool: ...

async def handle_message(self, event: MessageEvent):
if self._on_message:
await self._on_message(event)

三个必须实现的方法：连接、断开、发送。_on_message 回调由 GatewayRunner 启动时注入——适配器不需要知道消息怎么处理，只需要把翻译好的 MessageEvent 交出去。

三大共性机制

TextBatcher：消息分片合并。

微信个人号有 1500 字符截断，企业微信是 4000。agent 回复长文时，平台会截断成半截。TextBatcher 的规则：

• 长度 ≥ 3900（接近企微截断 4000）→ 等 2.0 秒，大概率被截断，等续片
• 长度远小于 → 等 0.6 秒，正常人打不出这么快两条独立消息

实现是三个字典加任务管理：每收到一条消息就重新倒计时，倒计时跑完才交出去。

MessageDeduplicator：消息去重。

按 message_id 去重，FIFO 淘汰旧记录，max_size 上限 1000。防止平台重复推送导致 agent 重复响应。

媒体缓存：URL 是临时的。

入站：下载 → 解密 → 本地缓存。因为平台给的 URL 是临时的，不立刻下载就过期了。出站：加密 → 分块上传 → 发消息。

断线重连：指数退避

平台连接断掉怎么办？指数退避重连：

[2, 5, 10, 30, 60] 秒

连上之后重置计数。不慌不忙，越等越久，避免在平台不稳定时疯狂重连打爆对方服务器。

平台差异对照

不同平台的差异比想象中大得多：

这些差异全部封装在适配器内部。核心循环看到的只有 MessageEvent，完全不知道背后的平台是什么。

避坑：两章 8 错精选

两章各列了 4 个初学者常犯的错误，合并精选最要命的：

小结

Gateway 解决的是"消息从哪来"的问题，适配器解决的是"怎么接住"的问题。两者合在一起，让同一个 agent 核心循环接入 15+ 平台，核心循环一行没变。

下一篇预告：第 10 篇，执行环境抽象与定时任务。 agent 不能只会被动响应，还要能主动做事。

你在接多平台的时候，踩过最深的坑是什么？欢迎在评论区聊聊。

参考文献

• Hermes Agent 教学仓库：agents/s12_gateway_architecture.py（本文代码素材，53287 字节真实可运行）
• Hermes Agent 教学仓库：agents/s13_platform_adapters.py（本文代码素材，63026 字节真实可运行）
• Hermes Agent 教学仓库：docs/zh/s12-gateway-architecture.md 与 docs/zh/s13-platform-adapters.md（Gateway 架构、适配器模式详解）

📥 源码获取：如需本系列全部源码，请在以下链接克隆： https://gitcode.com/ganxin7932508/learn-hermes-agent.git