流式与 Hooks

源码版本v2026.7.20

职责

① 流式消费——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,两边的时钟域解耦,节流、合并、回退都在消费侧加。Hooks 单列出来是因为安全/计费这种横切关注点不该污染主消息管道。

关键文件

• 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 等。

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 来自适配器(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 是横切关注点(安全/计费/同步)的插入点。两者让网关在不污染 agent 主干的前提下,支持实时打字、安全过滤与跨实例 relay。