Hermes Agent 源码探秘 (6)：多平台网关 — 一个 Agent 服务所有平台
系列：Hermes Agent 源码探秘 作者：元思未来 字数：约3200字
前五篇我们一直在讨论 Hermes 在终端（CLI） 下的工作方式。但你可能不知道：Hermes 真正的"杀手级功能"是——它可以同时接入 15+ 聊天平台，同一个 Agent 服务微信、Telegram、Discord、Slack 等所有渠道。
这篇就来拆 Gateway（网关） 的架构。
一、问题：一个 Agent 怎么服务多个平台？
先想一下，如果让你设计"一个 AI 跑在多个聊天平台上"，你会怎么做？
最简单的方案：
每个平台部署一个独立的 Agent 实例
WeChat → Agent 实例1
Telegram → Agent 实例2
Discord → Agent 实例3
问题很明显：
- 每个实例都有自己的上下文和状态——用户在微信上聊了半天的内容，切到 Telegram 就"失忆"了
- 维护成本高——每个实例单独管理
Hermes 的做法是：
WeChat ──┐
Telegram ─┼── Gateway ──→ 同一个 AIAgent 实例
Discord ──┘
所有平台的消息都汇聚到 Gateway，Gateway 统一路由给 同一个 Agent。Agent 处理完后，Gateway 把回复发回对应的平台。
这就是网关模式。
二、Gateway 的整体架构
Gateway 的代码在 gateway/ 目录下：
gateway/
├── run.py # 入口，启动 Gateway 服务
├── session.py # 会话管理
├── base.py # 平台适配器基类
├── platforms/ # 各平台适配器
│ ├── telegram.py # Telegram
│ ├── discord.py # Discord
│ ├── slack.py # Slack
│ ├── wecom.py # 企业微信
│ ├── weixin.py # 个人微信
│ ├── feishu.py # 飞书
│ ├── dingtalk.py # 钉钉
│ ├── whatsapp.py # WhatsApp
│ ├── signal.py # Signal
│ ├── email.py # 邮件
│ ├── sms.py # 短信
│ ├── matrix.py # Matrix
│ ├── homeassistant.py # 智能家居
│ └── ... # 还有更多
架构图
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Telegram │ │ WeChat │ │ Discord │ ... 用户平台
└─────┬────┘ └─────┬────┘ └─────┬────┘
│ │ │
└─────────────┼─────────────┘
│
▼
┌────────────────┐
│ Gateway │
│ (gateway/run) │
└───────┬────────┘
│
▼
┌──────────────────┐
│ Session Manager │
│ (会话路由) │
└───────┬──────────┘
│
▼
┌──────────────────┐
│ AIAgent │
│ (核心循环) │
└──────────────────┘
三、核心机制拆解
3.1 Platform Adapter（平台适配器）
每个平台对应一个适配器文件。适配器继承自基类 BasePlatform：
# gateway/base.py
class BasePlatform:
"""所有平台适配器的基类"""
@property
def platform_name(self) -> str:
return "base"
async def send_message(self, message: str, chat_id: str, **kwargs):
"""发送消息到平台"""
raise NotImplementedError
async def send_file(self, file_path: str, chat_id: str, **kwargs):
"""发送文件到平台"""
raise NotImplementedError
async def start_polling(self, message_handler):
"""开始轮询/监听消息"""
raise NotImplementedError
具体平台的适配器实现这个接口。以 Telegram 为例：
# gateway/platforms/telegram.py
class TelegramPlatform(BasePlatform):
platform_name = "telegram"
def __init__(self, config):
self.token = config["telegram"]["bot_token"]
self.api_base = f"https://api.telegram.org/bot{self.token}"
async def send_message(self, message, chat_id, **kwargs):
url = f"{self.api_base}/sendMessage"
payload = {
"chat_id": chat_id,
"text": message,
"parse_mode": "Markdown"
}
async with httpx.AsyncClient() as client:
resp = await client.post(url, json=payload)
return resp.json()
async def start_polling(self, message_handler):
"""轮询 Telegram API 获取新消息"""
offset = 0
while True:
url = f"{self.api_base}/getUpdates"
resp = await httpx.post(url, json={
"offset": offset,
"timeout": 30
})
updates = resp.json().get("result", [])
for update in updates:
if "message" in update:
msg = update["message"]
# 统一消息格式后传给 handler
await message_handler({
"platform": "telegram",
"chat_id": str(msg["chat"]["id"]),
"user_id": str(msg["from"]["id"]),
"text": msg.get("text", ""),
"raw": update
})
offset = update["update_id"] + 1
await asyncio.sleep(0.5)
每个适配器的核心职责只有两件事：
- 接收消息 → 转成统一格式 → 交给 Gateway
- 发送消息 → 从 Gateway 拿到回复 → 发回平台
3.2 Gateway 主循环
Gateway 的入口在 gateway/run.py，核心逻辑很简单——一条消息处理流水线：
async def handle_message(platform_msg):
"""处理来自任意平台的消息"""
# 1. 统一消息格式
unified = {
"platform": platform_msg["platform"],
"chat_id": platform_msg["chat_id"],
"user_id": platform_msg["user_id"],
"text": platform_msg["text"],
}
# 2. 会话路由 —— 找到或创建对应的 Agent 会话
session = session_manager.get_or_create(
platform=unified["platform"],
chat_id=unified["chat_id"]
)
# 3. 交给 Agent 处理
response = await session.agent.run_conversation(unified["text"])
# 4. 回复发回原平台
platform_adapter = get_adapter(unified["platform"])
await platform_adapter.send_message(
response["final_response"],
unified["chat_id"]
)
这个流水线清晰地分层：
平台消息 → 统一格式化 → 会话路由 → Agent处理 → 回复发送
3.3 会话管理（Session Manager）
一个关键设计问题是：不同平台的用户消息，怎么路由到正确的 Agent 会话？
class SessionManager:
def __init__(self):
self._sessions: Dict[str, AgentSession] = {}
def _make_key(self, platform, chat_id):
return f"{platform}:{chat_id}"
def get_or_create(self, platform, chat_id):
key = self._make_key(platform, chat_id)
if key in self._sessions:
return self._sessions[key]
# 创建新 Agent 会话
agent = AIAgent(
platform=platform,
session_id=str(uuid.uuid4()),
...
)
session = AgentSession(agent=agent, key=key)
self._sessions[key] = session
return session
每个 "platform:chat_id" 对应一个独立的 Agent 实例。 这样：
- 微信用户 A 有自己的 Agent 实例，保持对话上下文
- Telegram 用户 B 有自己的 Agent 实例，互不干扰
- 同一用户在不同平台上的会话是独立的
3.4 启动 Gateway
hermes gateway run
这个命令会：
# gateway/run.py 中简化后的启动逻辑
async def run_gateway():
# 1. 读取配置, 加载已启用的平台
config = load_config()
enabled_platforms = config.get("gateway", {}).get("platforms", [])
# 2. 为每个平台创建适配器实例
adapters = {}
for name in enabled_platforms:
adapter = create_platform_adapter(name, config)
adapters[name] = adapter
# 3. 启动所有平台的监听
tasks = []
for name, adapter in adapters.items():
task = asyncio.create_task(
adapter.start_polling(handle_message)
)
tasks.append(task)
# 4. 持续运行
await asyncio.gather(*tasks)
所有平台的监听是并发的——使用 asyncio 让多个平台同时运行。
四、从 WeCom（企业微信）适配器看实现细节
我们看看 gateway/platforms/wecom.py 是怎么工作的。
企业微信的接入方式比较特殊。它使用 Webhook 回调机制，而不是轮询：
用户发消息 → 企业微信服务器 → HTTP POST → Hermes Gateway
↓
Hermes Gateway → HTTP POST → 企业微信服务器 → 用户收到回复
核心交互：
# gateway/platforms/wecom.py (简化)
class WeComPlatform(BasePlatform):
platform_name = "wecom"
async def handle_callback(self, request):
"""处理企业微信的回调请求"""
# 1. 解密消息（企业微信的消息是加密的）
encrypted = request.xml["Encrypt"]
decrypted = self.decrypt_message(encrypted)
# 2. 解析消息内容
msg_type = decrypted["MsgType"]
content = decrypted["Content"]
from_user = decrypted["FromUserName"]
# 3. 统一格式后交给 Gateway 处理
await handle_message({
"platform": "wecom",
"chat_id": from_user,
"user_id": from_user,
"text": content
})
async def send_message(self, message, chat_id, **kwargs):
"""发送消息到企业微信"""
# 企业微信使用主动发送消息的 API
access_token = await self._get_access_token()
url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}"
payload = {
"touser": chat_id,
"msgtype": "markdown",
"markdown": {"content": message}
}
await httpx.post(url, json=payload)
这里有个有趣的细节：企业微信的消息是加密的，需要解密后才能处理。每个平台的适配器都要处理自己平台的"方言"（加密方式、消息格式、API差异），然后转成统一的内部格式。
五、适配器模式的价值
看完 Gateway 的源码，最值得学习的设计模式就是适配器模式。
传统写法（反例）
def handle_platform_message(platform, message):
if platform == "telegram":
# Telegram 特殊的处理逻辑
...
elif platform == "wecom":
# 微信特殊的处理逻辑
...
elif platform == "discord":
# Discord 特殊的处理逻辑
...
# 每加一个平台，就要改这个函数
问题：违反了"开闭原则"（对扩展开放，对修改封闭）。每加一个平台，就要改核心代码。
适配器模式（正解）
# 核心代码只依赖抽象基类
class BasePlatform:
async def send_message(self, ...): ...
async def start_polling(self, ...): ...
# 新增平台 = 新建一个类实现接口
class NewPlatform(BasePlatform):
async def send_message(self, ...): ...
async def start_polling(self, ...): ...
# 核心代码完全不需要改
新增一个平台就是新建一个文件，不改已有代码。 这就是适配器模式的魅力。
六、总结：Gateway 的核心设计哲学
| 设计点 | 实现方式 | 好处 |
|---|---|---|
| 平台适配器 | 每个平台一个类，继承 BasePlatform | 新增平台不改核心代码 |
| 统一消息格式 | 各平台消息转成统一 dict | 消息处理逻辑与平台无关 |
| 会话隔离 | platform:chat_id → Agent 实例 | 各用户上下文互不干扰 |
| 异步并发 | asyncio 同时监听多平台 | 一个进程服务所有用户 |
| 即插即用 | 配置启用/禁用平台 | 按需加载，资源效率高 |
七、下一篇预告
Agent 会思考（核心循环），会干活（工具系统），有自我认知（System Prompt），还能在多平台服务（Gateway）。但还有一个关键能力：它能学习和记忆。
第七篇我们拆 记忆系统和技能系统：
- Agent 怎么记住你是谁、你喜欢什么？
- "技能"到底是什么？怎么通过 Markdown 文件教 Agent 新技能？
- 长期记忆和短期记忆怎么协同工作？
代码位置：
~/.hermes/hermes-agent/gateway/
平台数量： 15+（还在增长中）
核心模式： 适配器模式
元思未来 · 行稳致远，进而有为