【Hermes Agent源码系列】 多平台网关:一个进程如何同时服务16+个消息平台

前面拆了对话循环和工具系统，今天拆 Hermes Agent 最"重"的一个模块——多平台消息网关。

gateway/run.py 有 856KB，是整个项目最大的单文件。它用一个进程同时对接 Telegram、Discord、Slack、飞书、企业微信、WhatsApp、Signal 等 16+ 个消息平台。用户在任何平台发消息，Agent 都能收到并回复——体验完全一致。

概述

Hermes Agent 的 Gateway 系统是一个多平台消息路由层，支持 20+ 个消息平台（Telegram、Discord、Slack、WhatsApp、Signal、Matrix、Feishu、WeCom 等），将不同平台的消息协议统一转换为标准化的内部事件流，并将 Agent 响应路由回原平台。

核心价值：

• • 统一接口：所有平台适配器实现相同的 BasePlatformAdapter 接口
• • 会话管理：跨平台的会话持久化与上下文注入
• • 媒体处理：图片/音频/文档的本地缓存与 SSRF 防护
• • 安全隔离：平台级锁、PII 脱敏、URL 安全检查

1. 架构设计

1.1 整体架构

┌─────────────────────────────────────────────────────────────┐│                      Gateway Runner                          ││  (gateway/run.py - 启动所有平台适配器)                       │└────────────┬────────────────────────────────────────────────┘             │             ├─► Platform Adapter 1 (Telegram)             ├─► Platform Adapter 2 (Discord)             ├─► Platform Adapter 3 (Slack)             ├─► Platform Adapter 4 (WhatsApp)             └─► ... (20+ 平台)                      │                      ▼         ┌────────────────────────┐         │  BasePlatformAdapter   │  ◄─── 统一接口         │  - connect()           │         │  - send_message()      │         │  - handle_message()    │         └────────────────────────┘                      │                      ▼         ┌────────────────────────┐         │   MessageEvent         │  ◄─── 标准化消息         │   - text               │         │   - media_urls         │         │   - source (SessionSource) │         └────────────────────────┘                      │                      ▼         ┌────────────────────────┐         │  Session Manager       │  ◄─── 会话持久化         │  - build_session_key() │         │  - SessionContext      │         └────────────────────────┘                      │                      ▼         ┌────────────────────────┐         │   run_agent.py         │  ◄─── 对话循环         │   (AIAgent)            │         └────────────────────────┘

1.2 核心组件

2. BasePlatformAdapter 接口

2.1 核心抽象方法

class BasePlatformAdapter(ABC):    @abstractmethod    async def connect(self) -> None:        """连接到平台（登录、建立 WebSocket 等）"""    @abstractmethod    async def send_message(        self,         chat_id: str,         text: str,         **kwargs) -> SendResult:        """发送消息到指定聊天"""    @abstractmethod    async def send_typing(self, chat_id: str) -> None:        """发送"正在输入"指示器"""    @abstractmethod    def truncate_message(self, text: str) -> str:        """根据平台限制截断消息"""

2.2 公共工具方法

媒体缓存（Image/Audio/Document Cache）

# 图片缓存async def cache_image_from_url(url: str, ext: str = ".jpg") -> str:    """下载图片到本地缓存，返回文件路径"""    # 1. SSRF 防护：检查 URL 是否安全    # 2. 重试机制：429/5xx 错误自动重试    # 3. 魔术字节检查：验证是否为真实图片    # 4. 唯一文件名：img_{uuid12}.jpg# 音频缓存async def cache_audio_from_url(url: str, ext: str = ".ogg") -> str:    """下载音频到本地缓存（用于 STT）"""# 文档缓存def cache_document_from_bytes(data: bytes, filename: str) -> str:    """保存文档到本地缓存，防止路径遍历"""

设计亮点：

• • 本地化媒体：平台 URL 通常有时效性（Telegram 1 小时过期），缓存到本地确保 Vision Tool 可访问
• • SSRF 防护：is_safe_url() 检查 + 重定向守卫（_ssrf_redirect_guard）
• • 重试策略：429/5xx 错误指数退避重试，避免 CDN 抖动丢失媒体

代理支持

def resolve_proxy_url(platform_env_var: str | None = None) -> str | None:    """解析代理 URL（支持 SOCKS/HTTP）"""    # 优先级：    # 1. 平台专用环境变量（DISCORD_PROXY）    # 2. 通用环境变量（HTTPS_PROXY）    # 3. macOS 系统代理（scutil --proxy）def proxy_kwargs_for_bot(proxy_url: str | None) -> dict:    """为 discord.py Bot 构建代理参数"""    # SOCKS → {"connector": ProxyConnector(..., rdns=True)}    # HTTP  → {"proxy": url}

设计亮点：

• • rdns=True：强制通过代理进行 DNS 解析，绕过 GFW DNS 污染
• • macOS 自动检测：读取系统代理配置，无需手动设置

UTF-16 长度计算

def utf16_len(s: str) -> int:    """计算 UTF-16 编码单元数（Telegram 限制 4096）"""    return len(s.encode("utf-16-le")) // 2def _prefix_within_utf16_limit(s: str, limit: int) -> str:    """二分查找最长前缀，不破坏代理对"""

设计亮点：

• • Telegram 的 4096 字符限制是 UTF-16 编码单元，而非 Unicode 码点
• • Emoji（😀）占 2 个 UTF-16 单元，Python len() 只计 1 个
• • 二分查找确保不会在代理对中间截断

3. MessageEvent 标准化

3.1 数据结构

@dataclassclass MessageEvent:    # 消息内容    text: str    message_type: MessageType  # TEXT, PHOTO, AUDIO, DOCUMENT, COMMAND    # 来源信息    source: SessionSource  # 平台、聊天 ID、用户 ID    # 媒体附件（本地文件路径）    media_urls: List[str]    media_types: List[str]    # 回复上下文    reply_to_message_id: Optional[str]    reply_to_text: Optional[str]  # 被回复消息的文本    # 自动加载技能（频道绑定）    auto_skill: Optional[str | list[str]]    # 内部标志（绕过授权检查）    internal: bool = False

3.2 消息类型

class MessageType(Enum):    TEXT = "text"    LOCATION = "location"    PHOTO = "photo"    VIDEO = "video"    AUDIO = "audio"    VOICE = "voice"    DOCUMENT = "document"    STICKER = "sticker"    COMMAND = "command"  # /new, /reset

3.3 照片爆发合并

def merge_pending_message_event(    pending_messages: Dict[str, MessageEvent],    session_key: str,    event: MessageEvent,) -> None:    """合并照片爆发（相册）到同一事件"""    existing = pending_messages.get(session_key)    if (        existing        and existing.message_type == MessageType.PHOTO        and event.message_type == MessageType.PHOTO    ):        # 合并媒体 URL        existing.media_urls.extend(event.media_urls)        existing.media_types.extend(event.media_types)        # 合并标题        if event.text:            existing.text = _merge_caption(existing.text, event.text)        return    # 非照片消息替换待处理事件    pending_messages[session_key] = event

设计亮点：

• • 用户发送相册时，平台通常发送多个 PHOTO 事件
• • 合并到单个事件，Agent 一次性看到所有图片
• • 非照片消息不合并，保持时序

4. 会话管理

4.1 SessionSource（会话来源）

@dataclassclass SessionSource:    platform: Platform  # TELEGRAM, DISCORD, SLACK, ...    chat_id: str    chat_name: Optional[str]    chat_type: str  # "dm", "group", "channel", "thread"    user_id: Optional[str]    user_name: Optional[str]    thread_id: Optional[str]  # 论坛主题、Discord 线程    chat_topic: Optional[str]  # 频道描述    @property    def description(self) -> str:        """人类可读描述：DM with Alice, group: DevTeam"""

4.2 SessionContext（会话上下文）

@dataclassclass SessionContext:    source: SessionSource    connected_platforms: List[Platform]  # 所有已连接平台    home_channels: Dict[Platform, HomeChannel]  # 定时任务投递目标    session_key: str  # "telegram:12345"    session_id: str   # UUID    created_at: datetime    updated_at: datetime

4.3 会话密钥构建

def build_session_key(source: SessionSource) -> str:    """构建唯一会话密钥"""    # 基础格式：platform:chat_id    key = f"{source.platform.value}:{source.chat_id}"    # 论坛主题/线程：追加 thread_id    if source.thread_id:        key += f":{source.thread_id}"    return key

设计亮点：

• • 同一聊天的不同线程是独立会话
• • Telegram 论坛主题、Discord 线程各自独立上下文

4.4 PII 脱敏

def _hash_sender_id(value: str) -> str:    """哈希用户 ID → user_<12hex>"""    return f"user_{hashlib.sha256(value.encode()).hexdigest()[:12]}"def _hash_chat_id(value: str) -> str:    """哈希聊天 ID，保留平台前缀"""    # telegram:12345 → telegram:<hash>

适用平台：

• • WhatsApp、Signal、Telegram、BlueBubbles（电话号码敏感）
• • Discord 不脱敏：提及系统需要真实 ID（<@123456>）

5. 平台适配器实现

5.1 已支持平台（26 个）

5.2 Telegram 适配器关键实现

连接与认证

async def connect(self) -> None:    # 1. 获取代理配置    proxy_url = resolve_proxy_url("TELEGRAM_PROXY")    # 2. 创建 Bot 实例    self.bot = Bot(        token=self.config.token,        **proxy_kwargs_for_bot(proxy_url)    )    # 3. 设置命令菜单    await self.bot.set_my_commands([        BotCommand("new", "Start new conversation"),        BotCommand("reset", "Reset conversation"),        # ...    ])    # 4. 启动轮询    await self.application.run_polling()

消息处理

async def _handle_telegram_message(    self,     update: Update,     context: ContextTypes.DEFAULT_TYPE) -> None:    message = update.message or update.edited_message    # 1. 提取来源信息    source = SessionSource(        platform=Platform.TELEGRAM,        chat_id=str(message.chat_id),        chat_name=message.chat.title,        chat_type=self._map_chat_type(message.chat.type),        user_id=str(message.from_user.id),        user_name=message.from_user.username,        thread_id=str(message.message_thread_id) if message.is_topic_message else None,    )    # 2. 处理媒体附件    media_urls = []    if message.photo:        # 下载最高分辨率照片        file = await message.photo[-1].get_file()        url = file.file_path        local_path = await cache_image_from_url(url, ext=".jpg")        media_urls.append(local_path)    # 3. 构建 MessageEvent    event = MessageEvent(        text=message.text or message.caption or "",        message_type=self._detect_message_type(message),        source=source,        media_urls=media_urls,        raw_message=message,        message_id=str(message.message_id),    )    # 4. 调用通用处理器    await self.handle_message(event)

发送消息

async def send_message(    self,     chat_id: str,     text: str,     **kwargs) -> SendResult:    try:        # 1. 截断消息（Telegram 限制 4096 UTF-16 单元）        text = self.truncate_message(text)        # 2. 提取线程 ID        thread_id = kwargs.get("thread_id")        # 3. 发送消息        sent = await self.bot.send_message(            chat_id=int(chat_id),            text=text,            message_thread_id=int(thread_id) if thread_id else None,            parse_mode="Markdown",        )        return SendResult(            success=True,            message_id=str(sent.message_id),            raw_response=sent,        )    except Exception as e:        return SendResult(            success=False,            error=str(e),            retryable=self._is_retryable_error(e),        )

5.3 Discord 适配器关键实现

线程支持

async def _handle_discord_message(self, message: discord.Message) -> None:    # 检测是否在线程中    thread_id = None    if isinstance(message.channel, discord.Thread):        thread_id = str(message.channel.id)    source = SessionSource(        platform=Platform.DISCORD,        chat_id=str(message.channel.id),        chat_name=message.channel.name,        chat_type="thread" if thread_id else "channel",        user_id=str(message.author.id),        user_name=message.author.name,        thread_id=thread_id,    )

提及处理

def _process_mentions(self, text: str, message: discord.Message) -> str:    """将 <@123456> 转换为 @username"""    for user in message.mentions:        text = text.replace(f"<@{user.id}>", f"@{user.name}")    for role in message.role_mentions:        text = text.replace(f"<@&{role.id}>", f"@{role.name}")    return text

6. 流式输出消费

6.1 StreamConsumer

class StreamConsumer:    """消费 Agent 输出流，实时推送到平台"""    async def consume_stream(        self,        stream: AsyncIterator[str],        adapter: BasePlatformAdapter,        chat_id: str,        **kwargs) -> None:        buffer = ""        last_send_time = time.time()        async for chunk in stream:            buffer += chunk            # 每 2 秒或遇到句子结束符时推送            if (                time.time() - last_send_time > 2.0                or chunk.endswith((".", "!", "?", "\n"))            ):                await adapter.send_message(chat_id, buffer, **kwargs)                buffer = ""                last_send_time = time.time()        # 推送剩余内容        if buffer:            await adapter.send_message(chat_id, buffer, **kwargs)

设计亮点：

• • 实时推送：用户看到 Agent 逐步生成响应
• • 批量优化：避免每个 token 都发送一次消息
• • 句子边界：在自然断点处推送，提升可读性

7. 安全机制

7.1 SSRF 防护

async def _ssrf_redirect_guard(response):    """重定向守卫：防止 302 跳转到内网"""    if response.is_redirect and response.next_request:        redirect_url = str(response.next_request.url)        if not is_safe_url(redirect_url):            raise ValueError(f"Blocked redirect to private address: {redirect_url}")

攻击场景：

• • 攻击者发送 https://evil.com/image.jpg
• • 服务器 302 重定向到 http://169.254.169.254/latest/meta-data/
• • 未防护的客户端会访问 AWS 元数据服务

7.2 平台锁

def _acquire_platform_lock(self, scope: str, identity: str) -> bool:    """获取平台级锁，防止多实例冲突"""    acquired, existing = acquire_scoped_lock(        scope, identity, metadata={'platform': self.platform.value}    )    if not acquired:        owner_pid = existing.get('pid')        self._set_fatal_error(            f'{scope}_lock',            f'Resource already in use (PID {owner_pid})',            retryable=False        )        return False    return True

用途：

• • 防止同一 Telegram Bot Token 被多个进程使用
• • 防止 Discord Bot 重复连接

7.3 路径遍历防护

def cache_document_from_bytes(data: bytes, filename: str) -> str:    # 1. 清理文件名    safe_name = Path(filename).name  # 去除目录部分    safe_name = safe_name.replace("\x00", "").strip()    # 2. 生成唯一文件名    cached_name = f"doc_{uuid.uuid4().hex[:12]}_{safe_name}"    filepath = cache_dir / cached_name    # 3. 验证路径在缓存目录内    if not filepath.resolve().is_relative_to(cache_dir.resolve()):        raise ValueError(f"Path traversal rejected: {filename!r}")    filepath.write_bytes(data)    return str(filepath)

8. 与 LifeOS 的对比

可借鉴点：

1. 1. 媒体缓存策略：LifeOS 可为 /read-pdf 添加 URL 下载 + SSRF 防护
2. 2. 会话上下文注入：LifeOS 可在 system prompt 中注入当前 Vault 路径、活跃项目
3. 3. 流式输出：LifeOS 可为 /research 等长时任务添加进度推送
4. 4. 平台锁机制：LifeOS 可防止同一 Vault 被多个 MCP 实例同时访问

9. 关键设计模式

9.1 适配器模式（Adapter Pattern）

# 统一接口class BasePlatformAdapter(ABC):    @abstractmethod    async def send_message(...) -> SendResult: ...# 各平台实现class TelegramAdapter(BasePlatformAdapter): ...class DiscordAdapter(BasePlatformAdapter): ...

优势：

• • 新增平台只需实现 BasePlatformAdapter
• • Gateway Runner 无需关心平台细节

9.2 事件驱动（Event-Driven）

# 平台适配器 → MessageEvent → 通用处理器event = MessageEvent(text="Hello", source=source)await self.handle_message(event)

优势：

• • 解耦平台协议与业务逻辑
• • 便于添加中间件（日志、授权、限流）

9.3 策略模式（Strategy Pattern）

class SessionResetPolicy(Enum):    NEVER = "never"    DAILY = "daily"    WEEKLY = "weekly"    ON_COMMAND = "on_command"

优势：

• • 不同聊天可配置不同重置策略
• • 易于扩展新策略

10. 性能优化

10.1 媒体缓存清理

def cleanup_image_cache(max_age_hours: int = 24) -> int:    """删除超过 24 小时的缓存图片"""    cache_dir = get_image_cache_dir()    cutoff = time.time() - (max_age_hours * 3600)    removed = 0    for f in cache_dir.iterdir():        if f.is_file() and f.stat().st_mtime < cutoff:            f.unlink()            removed += 1    return removed

10.2 重试策略

_RETRYABLE_ERROR_PATTERNS = (    "connecterror",    "connectionreset",    "connecttimeout",    "network",    "broken pipe",)# 指数退避for attempt in range(retries + 1):    try:        return await client.get(url)    except Exception as e:        if attempt < retries:            wait = 1.5 * (attempt + 1)            await asyncio.sleep(wait)

10.3 并发处理

# 每个会话独立处理，不阻塞其他会话self._background_tasks.add(    asyncio.create_task(self._process_message(event)))

11. 错误处理

11.1 致命错误标记

def _set_fatal_error(self, code: str, message: str, *, retryable: bool) -> None:    self._running = False    self._fatal_error_code = code    self._fatal_error_message = message    self._fatal_error_retryable = retryable    # 写入状态文件    write_runtime_status(        platform=self.platform.value,        platform_state="fatal",        error_code=code,        error_message=message,    )

11.2 可重试错误判断

def _is_retryable_error(self, exc: Exception) -> bool:    error_str = str(exc).lower()    return any(        pattern in error_str        for pattern in _RETRYABLE_ERROR_PATTERNS    )

12. 总结

12.1 核心优势

1. 1. 统一抽象：BasePlatformAdapter 隐藏平台差异
2. 2. 安全优先：SSRF 防护、平台锁、PII 脱敏
3. 3. 媒体本地化：解决平台 URL 时效性问题
4. 4. 会话持久化：跨平台统一会话管理
5. 5. 流式输出：实时推送提升用户体验

12.2 可改进点

1. 1. 消息队列：当前是内存队列，重启丢失待处理消息
2. 2. 负载均衡：单进程处理所有平台，可拆分为多进程
3. 3. 监控告警：缺少 Prometheus 指标导出
4. 4. 测试覆盖：平台适配器测试依赖真实 Token

12.3 对 LifeOS 的启示

1. 1. 添加 HTTP 工具：支持 URL 下载 + SSRF 防护
2. 2. 会话上下文注入：在 system prompt 中注入 Vault 元信息
3. 3. 流式进度推送：长时任务（/research）可推送中间结果
4. 4. 多 Vault 隔离：防止同一 Vault 被多个 MCP 实例访问

下一步：Phase 4 - 高级特性分析（Memory System、Plugin Architecture、MCP Integration）

📌 欢迎关注「老王的AI局」每周分享 AI Agent 架构解析、工具实战和行业洞察。

同时承接 Agent项目开发，小程序，web站，后台管理saas系统等 欢迎私聊