串流與 Hooks

源码版本v2026.7.20

職責

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

關鍵檔案

• class GatewayStreamConsumer:83-120 — 非同步消費器本體,on_delta 作為回調餵給 agent
• class StreamConsumerConfig:55-83 — 單次消費的執行期設定(native draft 串流策略、最終編輯延遲等)
• 模組文件:1-40 — 說明同步→非同步橋接原理與 sentinel 完成信號
• stream_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 等。

小結

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