平台轉接器

源码版本v2026.7.20

職責

每個平台(Telegram、Discord、Slack、Signal、WhatsApp、微信、QQ…)各有一個轉接器 (adapter),把平台原生事件翻譯成統一的 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. 平台差異(訊息長度、富文字、圖片上限)全由轉接器內部吸收,網關 (gateway) 核心不感知。

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 走外掛路徑,與內建轉接器同等對待。