串流與 Hooks

源码版本v2026.7.20

職責

① 串流 (stream) 消費——agent 的 worker 執行緒同步地 stream_delta_callback(text),而平台發送是非同步的;GatewayStreamConsumer 把同步增量橋到非同步,漸進式編輯同一條平台訊息,讓使用者看到「打字機」效果。② Hooks——在訊息生命週期節點(收到、生成前、生成後、投遞)插入自訂邏輯,如安全過濾、計費、kanban 同步。③ Relay——跨實例/跨 profile 的訊息中繼。

設計動機

agent 的 run_conversation 在 worker 執行緒裡跑,token 是同步 yield 出來的——這是底層 SDK 決定的(LLM 串流介面通常是 blocking iterator)。而平台發送是 asyncio 協程,跨執行緒直接呼叫會踩事件迴圈。如果沒有中間層,要麼把整段回答攢完再發(使用者看著「正在輸入…」幾十秒看不到內容),要麼在 worker 執行緒裡 asyncio.run_coroutine_threadsafe 每次都建立 future(開銷大、錯誤處理難)。GatewayStreamConsumer 用 queue.Queue 做執行緒邊界——worker 執行緒只 queue.put,消費協程從 asyncio 側 get_nowait,兩邊的時鐘域解耦,節流、合併、回退 (fallback) 都在消費側加。Hooks 單列出來是因為安全/計費這種橫切關注點不該污染主訊息管道。

關鍵檔案

• class GatewayStreamConsumer:83-120 — 非同步消費器本體,on_delta 作為回調餵給 agent
• class StreamConsumerConfig:55-83 — 單次消費的執行期設定(native draft 串流策略、最終編輯延遲等)
• 模組文件:1-40 — 說明同步→非同步橋接原理與 sentinel 完成信號
• stream_dispatch — 串流事件分發 (dispatch)
• stream_events — 串流事件類型定義
• hooks — 生命週期 hook 註冊框架
• builtin_hooks/ 目錄 — 內建 hook 實作
• relay/ 目錄 — 跨實例中繼
• slash_commands — / 命令路由
• authz_mixin — 授權(被 GatewayRunner 混入)
• profile_routing — 多 profile 路由

資料流

1. GatewayRunner 啟動 _start_stream_consumer:21218。
2. 建構 GatewayStreamConsumer,把 consumer.on_delta 作為 stream_delta_callback 傳給 AIAgent(run_agent.py:400)。
3. agent 在 worker 執行緒同步呼叫 on_delta(text) → 消費器把增量塞進非同步佇列。
4. 消費器協程從佇列取增量,按 StreamConsumerConfig:55 策略:
   • auto/draft:優先原生 draft 串流(send_draft:2650),回退到普通編輯
   • 最終編輯可能延遲(慢推理模型場景),避免高頻抖動
5. 串流完成後 sentinel 信號觸發最終編輯,交給 delivery_ledger:155 記帳。
6. 全程各節點觸發 hooks(收到/生成前/生成後/投遞),內建 hook(gateway/builtin_hooks/)執行安全、計費、kanban 等。

on_delta 是消費器對外的入口,只做一件事——把 worker 執行緒的 token 塞進 queue.Queue:

def on_delta(self, text: str) -> None:
    """Thread-safe callback — called from the agent's worker thread.

    When *text* is ``None``, signals a tool boundary: the current message
    is finalized and subsequent text will be sent as a new message so it
    appears below any tool-progress messages the gateway sent in between.
    """
    if text:
        self._queue.put(text)
    elif text is None:
        self.on_segment_break()

def finish(self) -> None:
    """Signal that the stream is complete."""
    self._queue.put(_DONE)

None 是 tool boundary 信號——後續 token 走新訊息,插在 tool 進度氣泡下方。_DONE 是 sentinel,消費協程見到就觸發最終編輯。「值 = token、None = 邊界、sentinel = 完成」讓一個佇列承載全部控制流。

消費協程 run() 批量取增量,按策略下發:

async def run(self) -> None:
    """Async task that drains the queue and edits the platform message."""
    _len_fn = (
        self.adapter.message_len_fn
        if isinstance(self.adapter, _BasePlatformAdapter)
        else len
    )
    _raw_limit = self._raw_message_limit()
    _safe_limit = max(500, _raw_limit - _len_fn(self.cfg.cursor) - 100)
    self._use_draft_streaming = self._resolve_draft_streaming()
    try:
        while True:
            if not self._run_still_current():
                return
            got_done = False
            while True:
                try:
                    item = self._queue.get_nowait()
                    if item is _DONE:
                        got_done = True
                        break

幾個細節:message_len_fn 來自轉接器 (adapter)(Telegram 用 utf16 長度),保證溢出判斷和平台真實限制一致;_safe_limit 給 cursor 和富文字格式化留 100 字元餘量;_run_still_current() 每次循環檢查,session 被 /new 或 /stop 就提前退出,避免陳舊 token 繼續推到螢幕上。

GatewayEventDispatcher(gateway/stream_dispatch.py)是更結構化的封裝——agent 產出 typed events(MessageChunk、ToolCallChunk、Commentary…),dispatcher 根據 adapter 能力決定如何渲染,平台不能渲染 tool chrome 時回傳 None 直接吃掉事件。這是給「平台需要原生 UI(釘釘 AI Card、Slack Block Kit)而非純文字編輯」留的擴展點:

class GatewayEventDispatcher:
    """Route typed stream events through an adapter onto a delivery sink."""
    # Message/commentary/segment events flow into the consumer (native draft
    # on Telegram DMs, edit-in-place elsewhere).  Tool events are formatted
    # by the adapter — which may return None to *eat* the event on platforms
    # that can't render tool chrome — and the rendered line is enqueued onto
    # the same tool progress queue the gateway already drains, so the two no
    # longer race through independent code paths.

邊界與失敗

• 串式中斷:session 被 /new 或 /stop 後,消費協程下次循環檢查 _run_still_current() 直接 return。已 queue.put 的增量會丟,所以最終編輯前 _flush_think_buffer() 把殘留的半截 think 標籤沖掉。
• Flood control 擊穿:_MAX_FLOOD_STRIKES = 3,連續 3 次 429 永久降級到非漸進式,本流剩餘 token 攢批一次性發。
• 平台不支援 edit:Signal SUPPORTED_MESSAGE_EDITING = False,消費器走非 edit 回退——只發一次完整訊息,串式過程中看不到打字機。這是能力位兜底,不是 bug。
• sentinel 漏發:worker 執行緒崩在 on_delta 後沒調 finish(),消費協程會卡在 queue.get()。run_still_current 是兜底——session 被清理也觸發退出。

小結

串流消費器是「同步 agent 回調 ↔ 非同步平台投遞」的緩衝與節流層,解決兩個時鐘域不匹配。Hooks 是橫切關注點(安全/計費/同步)的插入點。兩者讓網關 (gateway) 在不污染 agent 主幹的前提下,支援即時打字、安全過濾與跨實例 relay。