Liking cljdoc? Tell your friends :D

Token 流变换链设计(:token-xform)

状态:✅ 已实施(2026-07-14,全套 253 tests / 1039 assertions / 0)。 2026-09-03 补 §2.1 线程契约:异步入口落地后 token 不再保证在调用线程上 派发(单次流内仍串行)。见 async-chat-model-design.md §6。 本文是 filter 体系第五钩子 :token-xform 的权威参考。四链体系 (:tool/:chat/:turn)见 filter-chain-design.md;本文只讲 token 粒度。 动机来自与 Spring AI StreamAdvisor 的对照(§5)。

实现:core/filter.clj(契约 + 内置 filter)、core/chat_client.clj (invoke-chat-stream terminal 组装点)。


0. 动机:流式路径缺"逐 token 变换"

现状(见 filter-chain-design.md §2.2):invoke-chat-stream 与同步路径 共用同一条 :chat 洋葱;:on-token 作为 request 普通字段透传进 terminal。 这给了三档能力:

  1. 请求侧改写——完全支持(与同步零差异);
  2. 对最终响应 around/后置——支持(stream-fn 流末返回完整归一化响应);
  3. 逐 token 介入——只能靠 chat filter 包 :on-token 做副作用观测。

第 3 档的缺口:on-token 是推入的回调(push/sink),不是可返回可变换的值。 逐 token 的变换(改写/吞掉/缓冲后批量放行)只能在包装回调里手写状态机, 框架不提供组合形式。对照 Spring AI:它的流是一等 Flux<ChatClientResponse>, advisor 用 Reactor 算子对流做 map/filter/buffer,逐 chunk 变换是原生能力。

补齐这档能力需要满足三条(缺一不可,曾考虑的"on-token 返回值契约"因缺 后两条被否):

  • 1→N:一个 token 进、0/1/N 个 token 出(缓冲-批量放行的形状);
  • 跨 chunk 状态:敏感词被切在两片之间、按句重组,需要状态机;
  • 流末 flush:流结束时缓冲区残留必须有机会冲出——需要 end-of-stream 信号。

1. 结论:第五钩子 :token-xform,用 transducer 承接

Clojure 的 transducer 恰好原生满足全部三条:

需求transducer 对应物
1→Nstep 二元 arity 可调下游 0/1/N 次(mapcat/partition-all 现成)
跨 chunk 状态有状态 transducer(volatile! 闭包)
流末 flushcompletion 一元 arity 就是 end-of-stream 信号

这就是"不引 Reactor 拿到 Reactor 算子"的答案——吸收 Spring 的算子思想, 不搬它的响应式栈与 Call/Stream 双接口(一条 :chat 链服务同步与流式 是我们的既有优势,保持不动)。

2. 契约

filter map 增加可选键 :token-xform,与 :tool/:chat/:iteration/:turn 并存:

{:name :redact
 :token-xform (map #(update % :token redact-secrets))}   ;; 无状态改写

{:name :sentence-guard
 :token-xform (my-stateful-xform)}                           ;; 有状态:缓冲/重组
  • 值是一个 transducer,作用于 token-data map 流。token-data 形状即 on-token 现行契约:{:token "..."} / {:reasoning-token "..."}(可带 :index 等),filter 应对不认识的键透传((if (:token %) (update ...) %));
  • 顺序 = :filters 注册顺序:(apply comp xforms) 下靠前者最先见原始 token——与三链"靠前者最先见 req"语义一致,不引入 order;
  • 状态作用域 = 单次 LLM 流:(xform rf) 在 terminal 每次调用现场实例化, 工具循环每轮 LLM 调用各自新状态。存在 filter map 里的 transducer 值本身 无状态,可安全复用(这是 transducer 的标准性质);
  • 同步路径(invoke-chat)完全忽略 :token-xform;:tool/:iteration/:turn 链与 token 流无关。

2.1 ⚠️ 线程契约:sink 的线程不保证(2026-09-03 起)

invoke-chat-stream(同步入口)下 token 在调用线程串行 emit——这条从来 没写进契约,但实现上一直如此,写 sink 的人自然会依赖它。

invoke-chat-stream-async / agent/chat-async 落地后不再成立:

入口token 派发在哪根线程
invoke-chat-stream(同步)调用线程(不变)
异步 + provider 无原生异步async/vthread 起的那根虚拟线程
异步 + provider 原生异步(IAsyncLLMProvider)传输层 HttpClient 的 executor(虚拟线程池)

不变的部分(这些仍然是契约):

  • 单次流内串行:同一条流的 token 不会并发 emit,:token-xform 的有状态 transducer 因此不需要自己加锁;作用域仍是「单次 LLM 流」;
  • flush 时机不变:正常完流调 completion arity,异常(error channel)不 flush;
  • 顺序不变:token 按到达顺序过链。

变的部分(写 sink 的人要知道):token 回调可能跑在任意工作线程上。 往 UI / 请求作用域 / ThreadLocal 里写的 sink 必须自己切回去;跨流共享的 可变状态要自己保证线程安全。这是可观察语义变化,故显式记在这里, 而不是当作实现细节。

3. 组装点与数据流

组装在 chat-client/invoke-chat-stream 的 terminal 内(chat 洋葱最内层):

provider 原始 token → :token-xform 链(注册顺序) → 最终 on-token(sink)

要点:

  • chat filter 之后:chat filter 若包/换了 :on-token,xform 链包裹的是 链上存活下来的那个——语义自洽:xform 链定义出站流,on-token 包装者观测 交付物(看到的是变换后的 token)。要观测原始 token,用带副作用的 :token-xform((map #(do (log %) %)));
  • flush 只在正常完流:stream-fn 正常返回后调一次 completion arity, 缓冲残留冲给 sink;stream-fn 抛异常则不调(缓冲丢弃,半截答案不外泄);
  • reduced? 早停:下游 step 返回 reduced(如 (take n))后不再喂 token,但 completion 照常调一次(对齐 transduce 语义);
  • 无 :token-xform 时零开销退化:on-token 原样直通,路径与现状 bit 级一致。

3.1 硬边界:token 链不改"答案",只改"交付"

:token-xform 变换的是送给 on-token 的出站流;stream-fn 返回的最终 归一化响应不经过它——memory 落库、turn 结果、后续工具循环用的都是 原始完整答案。分工:

要改什么用哪条链
用户实时看到什么(脱敏/吞半截/缓冲放行):token-xform
这个 turn 的最终答案是什么(校验重试/改写):turn(validation-turn-filter)

两者可组合:hold-release 管"没审完不外泄",validation-turn-filter 管 "不合格重问"——同一 guardrail 的交付面与内容面。

4. 内置 filter

filter说明
(token-redact-filter re replacement)无状态逐 token 正则脱敏。已知限制:秘密被切在两个 chunk 之间时漏检(跨 chunk 检测需有状态缓冲,用户按需自写或用 hold-release)
(hold-release-filter check-fn)先审后放:缓冲整流不外泄;完流时 check-fn 收全文(:token 拼接),返回 nil=通过(缓冲原样放行)| 字符串=不通过(只 emit 一个 {:token 替换文本})。代价即语义:用户在流结束前看不到任何 token——"完整答案没成形就无法审"这一根本矛盾任何机制都消不掉(Spring 的 Flux buffer 同样毁流式 UX),本 filter 只是把缓冲逻辑标准化

5. 与 Spring AI StreamAdvisor 的对照(吸收记录)

Spring AI我们结论
StreamAdvisor.adviseStream 返回 Flux<ChatClientResponse>,流是一等值:token-xform transducer 变换出站 token 流吸收算子思想(1→N/有状态/flush),不搬 Reactor
CallAdvisor / StreamAdvisor 双接口(BaseAdvisor 统一 before/after)同一条 :chat 链服务同步与流式;token 粒度独立成 :token-xform不跟双接口——粒度用钩子声明,不用接口拆分
MessageAggregator 聚合 Flux 供 memory(只读)stream-fn 流末返回完整归一化响应,memory 天然拿到全文同构解法,我们免费
guardrail-on-stream:collectList 缓冲再放hold-release-filter语义对称——缓冲毁流式两边一样,Spring 只是表达顺手;我们标准化成内置 filter 后表达同样顺手
逐 chunk 有状态变换(跨 chunk 敏感词等)有状态 transducer两边都要手写状态机,同量级

6. 测试锚点

  1. 1→N 与 flush:(partition-all 3) 作 xform,7 个 token 进 → 正常完流 sink 收到 3 批(2+1 的尾巴由 completion 冲出);
  2. 异常不 flush:stream-fn 抛异常 → sink 只收到异常前已放行的,缓冲丢弃;
  3. 组合顺序:两个改写 xform 按注册顺序作用(先注册先见原始 token);
  4. hold-release 两分支:通过 → 全部缓冲在完流时按原序放行;不通过 → 只收到一个替换 token;
  5. 退化路径:无 :token-xform 的 chat-client,sink 收到原始 token(行为与现状 一致);同步 invoke-chat 带 :token-xform filter 不受影响;
  6. reasoning-token 透传:{:reasoning-token ...} 不被 :token 类 filter 误伤;
  7. 最终响应不被变换(§3.1):hold-release 吞掉全部 token,返回的 :response 仍是完整原文。

相关文档

  • filter-chain-design.md——三链体系权威参考(本文是其 token 粒度扩展)
  • agent-loop-concurrency-design.md §14——turn 链与 Spring AI 调研

Can you improve this documentation?Edit on GitHub

cljdoc builds & hosts documentation for Clojure/Script libraries

Keyboard shortcuts
Ctrl+kJump to recent docs
←Move to previous article
→Move to next article
Ctrl+/Jump to the search field
× close