平台适配器

源码版本v2026.7.20

职责

每个平台(Telegram、Discord、Slack、Signal、WhatsApp、微信、QQ…)各有一个适配器,把平台原生事件翻译成统一的 MessageEvent,并把统一发送动作翻译回平台 API。适配器继承 BasePlatformAdapter,实现少量抽象方法即可。适配器可来自内置 gateway/platforms/ 或插件 plugins/platforms/,都经 PlatformRegistry 注册。

设计动机

平台原生 API 差异极大:Telegram Bot API 用 sendMessage + MarkdownV2 转义,Discord 走 webhook + 雪花 ID,Signal 要跟本地 signal-cli daemon 说话,微信连富文本都不支持。把这些差异塞进 agent 主循环,agent 就永远写不完。适配器层把这些吸收掉——对上吐统一的 MessageEvent,对下接收统一的 send(chat_id, content) 抽象。配合 PlatformRegistry 的 deferred 加载,启动 hermes chat 这种根本不碰平台的命令,也不会被二十多个平台 SDK 拖慢。

关键文件

• class BasePlatformAdapter(ABC):2317 — 适配器抽象基类
• 抽象方法 send / send_draft:2983-3015 — 子类必须实现的发送接口
• MessageType / ProcessingOutcome / MessageEvent:1737-1759 — 统一事件模型
• CachedMedia:1638-1737 — 媒体缓存(避免重复上传)
• SendResult / EphemeralReply:1898-2068 — 发送结果与临时回复
• 通用发送辅助:3173-3326 — send_slash_confirm / send_clarify / send_typing / send_multiple_images
• gateway/platforms/ 目录 — signal / whatsapp_cloud / weixin / bluebubbles / yuanbao / qqbot…
• plugins/platforms/ 目录 — Telegram / Discord / Slack(以插件形式)
• ADDING_A_PLATFORM.md — 新增平台指南
• PlatformRegistry:162 — 注册表(register / register_deferred / unregister)

数据流

1. 启动期 GatewayRunner(gateway/run.py:3029)经 _start_one_profile_adapter:9389 拉起适配器。
2. 适配器从 PlatformRegistry(gateway/platform_registry.py:162)按名查询,支持 deferred 加载(便宜先占位,用到才真导入)。
3. 适配器监听平台事件,翻译成 MessageEvent:1759 交给 _handle_message:9947。
4. agent 产出流式输出,适配器经统一接口(send:3008 / send_draft)把内容投递回平台;媒体走 CachedMedia:1638 缓存。
5. 平台差异(消息长度、富文本、图片上限)全由适配器内部吸收,网关核心不感知。

PlatformRegistry 维护两套登记——_entries 已实装、_deferred 便宜占位。被查询时才把后者转成前者,核心字段:

class PlatformRegistry:
    """Central registry of platform adapters.

    Thread-safe for reads (dict lookups are atomic under GIL).
    Writes happen at startup during sequential discovery.
    """
    def __init__(self) -> None:
        self._entries: dict[str, PlatformEntry] = {}
        # Deferred platform loaders: name -> zero-arg callable that imports the
        # owning plugin module (which calls register() and populates _entries).
        self._deferred: dict[str, Callable[[], None]] = {}

    def get(self, name: str) -> Optional[PlatformEntry]:
        """Look up a platform entry by name."""
        if name not in self._entries:
            self._resolve(name)
        return self._entries.get(name)

deferred 的动机写在注释里——平台适配器模块顶层就 import 重型 SDK(lark_oapi、discord.py、slack_bolt…),全量加载会让 hermes chat 也慢上几秒。占位 loader 只在第一次 get(name) 时真正导入。

BasePlatformAdapter 用 ABC 强制子类实现三件套——connect、disconnect、send,这是适配器的最小契约:

@abstractmethod
async def connect(self, *, is_reconnect: bool = False) -> bool:
    """Connect to the platform and start receiving messages.

    Args:
        is_reconnect: False on a cold first boot; True when the reconnect
            watcher is re-establishing a platform that was previously running
            and dropped after an outage. Adapters that buffer a server-side
            update queue (e.g. Telegram's Bot API) should preserve that queue
            when ``is_reconnect`` is True so messages sent during the outage
            are delivered rather than silently discarded.
    """
    pass

@abstractmethod
async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:

is_reconnect 让重连时保留服务端 update 队列(Telegram 那种),否则掉线期间发的消息会被静默丢掉。

具体适配器(Signal)就长成这样。send 把统一 content 翻译成 signal-cli 的 RPC 参数:

class SignalAdapter(BasePlatformAdapter):
    """Signal messenger adapter using signal-cli HTTP daemon."""

    platform = Platform.SIGNAL
    SUPPORTS_MESSAGE_EDITING = False  # Signal 已发消息没有 edit API

    async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:
        """Send a text message with native Signal formatting."""
        await self._stop_typing_indicator(chat_id)
        plain_text, text_styles = self._markdown_to_signal(content)
        params: Dict[str, Any] = {"account": self.account, "message": plain_text}
        if text_styles:
            params["textStyles"] = text_styles
        if chat_id.startswith("group:"):
            params["groupId"] = chat_id[6:]
        else:
            params["recipient"] = [await self._resolve_recipient(chat_id)]
        result = await self._rpc("send", params)

SUPPORTED_MESSAGE_EDITING = False 这种能力位告诉流式消费器别 edit 已发消息,避免在 Signal 客户端留下「编辑失败」占位框。平台特性差异全靠适配器字段表达,而非 if-else 分支。

边界与失败

• 平台离线/掉线:connect 用 is_reconnect 区分冷启动与恢复,掉线期间用户消息可能积在服务端。若适配器没保留 update 队列(Telegram)或没 SSE 重连(Signal),消息会丢。
• 重型 SDK 误加载:deferred 注册错了名或 _resolve_all 在不该调用的路径被触发,会让 CLI chat 拖入 discord.py。register 优先于 deferred 是兜底。
• 能力位不一致:SUPPORTED_MESSAGE_EDITING=False 但流式消费器仍走 edit 路径,客户端会留下「编辑失败」占位框。新增能力位所有适配器必须显式声明默认值。
• 多 profile 重复 listener:同一 Signal 号在多 profile 都注册会启两套 SSE。Signal 用 _acquire_platform_lock('signal-phone', self.account, ...) 加进程级锁防重复。

小结

适配器 = 「平台方言 ↔ 统一事件」的翻译器。新增平台只需继承 BasePlatformAdapter 实现几个抽象方法并注册,参考 gateway/platforms/ADDING_A_PLATFORM.md。Telegram/Discord/Slack 走插件路径,与内置适配器同等对待。