流式与 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。