平台适配器 - Hermes Agent

架构概述

平台适配器层位于外部消息平台与 Hermes Agent 核心引擎之间。它将所有传入消息规范化为统一格式，并将传出消息路由回相应平台。

外部平台
(Telegram/Discord/飞书/QQ)

→

平台适配器
(gateway/platforms/)

→

统一消息
(标准化格式)

→

核心引擎
(消息处理)

位于 gateway/platforms/，每个适配器实现标准接口以处理特定平台的协议。

可用适配器

✈️ Telegram

完整支持 Telegram Bot API，包括文本消息、媒体、贴纸、内联查询和回调查询。支持 Webhook 和轮询模式。

telegram.py Webhook 轮询

🎮 Discord

通过 Discord.py 网关集成 Discord 机器人。支持斜杠命令、消息组件、嵌入消息和服务器特定配置。

discord.py 斜杠命令

📱 飞书

企业消息平台飞书/Lark 支持。处理文本、富文本、卡片和交互式消息，支持租户隔离。

feishu.py 企业版

💬 QQ

腾讯 QQ 平台适配器，支持群消息、私聊消息、图片和 CQ 码。兼容 QQNT 和传统协议。

qq.py CQ 码

🔗 Webhook

通用 Webhook 适配器，用于自定义集成。接收 JSON 数据包，支持身份验证，可触发操作或接收通知。

webhook.py 自定义

统一消息接口

所有平台适配器在处理前将消息规范化为标准的 UnifiedMessage 格式：

class UnifiedMessage:
    # 身份标识
    platform: str          # "telegram", "discord", "feishu", "qq", "webhook"
    message_id: str         # 平台特定唯一ID
    
    # 发送者
    user_id: str            # 平台用户标识符
    username: str           # 显示名称或昵称
    chat_id: str            # 频道/群组标识符
    
    # 内容
    content: str            # 文本内容（已标准化）
    raw_content: Any        # 平台特定原始数据
    
    # 元数据
    message_type: str      # "text", "image", "audio", "video", "file"
    timestamp: float         # Unix 时间戳
    metadata: Dict[str, Any]  # 平台特定扩展信息
            

接收消息处理

当收到来自任何平台的消息时，适配器执行以下步骤：

1. 接收 — 平台特定的 Webhook 或轮询捕获原始事件
2. 解析 — 从数据 payload 中提取平台特定字段
3. 标准化 — 转换为 UnifiedMessage 格式
4. 验证 — 检查消息完整性和必填字段
5. 分发 — 转发到核心消息路由器

# 示例：Telegram 接收消息流程
async def handle_update(update: Update):
    message = update.message
    
    # 解析平台特定字段
    unified = UnifiedMessage(
        platform="telegram",
        message_id=str(message.message_id),
        user_id=str(message.from_user.id),
        username=message.from_user.username,
        chat_id=str(message.chat.id),
        content=message.text or "",
        raw_content=message,
        message_type="text",
        timestamp=message.date.timestamp()
    )
    
    # 分发到核心引擎
    await message_router.route(unified)
            

发送消息处理

响应通过适配器层路由回正确的平台：

1. 格式化 — 核心引擎返回平台无关的响应数据
2. 适配 — 适配器转换为平台特定的消息格式
3. 发送 — 通过平台 API 投递（REST、WebSocket 等）
4. 确认 — 更新消息状态并处理错误

# 示例：平台适配器回复接口
class PlatformAdapter(ABC):
    @abstractmethod
    async def send_message(
        self,
        chat_id: str,
        content: str,
        **kwargs
    ) -> MessageResult:
        """发送消息到指定会话"""
        pass
    
    @abstractmethod
    async def send_image(
        self,
        chat_id: str,
        image_url: str,
        **kwargs
    ) -> MessageResult:
        """发送图片到指定会话"""
        pass
            

统一消息路由

MessageRouter 是核心调度器，负责在平台和核心代理之间路由消息：

class MessageRouter:
    def __init__(self, adapters: Dict[str, PlatformAdapter]):
        self.adapters = adapters
        self.middlewares = []
    
    async def route(self, message: UnifiedMessage) -> Response:
        # 应用中间件链
        for mw in self.middlewares:
            message = await mw.process(message)
        
        # 使用核心代理处理
        response = await self.agent.process(message)
        
        # 将响应路由回平台
        adapter = self.adapters.get(message.platform)
        await adapter.send_message(message.chat_id, response.content)
        
        return response
            

添加新平台

要添加对新消息平台的支持，请实现 PlatformAdapter 接口：

from abc import ABC, abstractmethod

class PlatformAdapter(ABC):
    """所有平台适配器的基类"""
    
    @property
    abstract def platform_name(self) -> str:
        """该平台的唯一标识符"""
        pass
    
    @abstractmethod
    async def initialize(self, config: Dict) -> None:
        """使用配置初始化适配器"""
        pass
    
    @abstractmethod
    async def start(self) -> None:
        """开始监听消息（webhook/轮询）"""
        pass
    
    @abstractmethod
    async def stop(self) -> None:
        """优雅地停止适配器"""
        pass
            

将新适配器放置在 gateway/platforms/your_platform.py 并在适配器注册表中注册。