Nozzle：一个按需缓冲、尽早输出的文本流处理库

一个文本流式处理库，能输出时就输出，必要时才缓冲。

Nozzle 是一个小型 TypeScript 库，用于处理从 LLM 流式输出的文本。它的各个函数只在必要时缓冲，其余情况下会尽快输出数据块。例如，正则表达式只会缓冲一个可能的匹配，直到该匹配被确认或排除，而匹配前后的内容会立即输出。

npm i nozzle-js # 或 pnpm / bun / yarn

import { nz } from "nozzle-js" const parts = nz(textStream).parse(/img-(\w+)/g, (m) => ({ src: m[1] })) for await (const part of parts) console.log(part) /*

• 输入: "Here is im" "g-abc" "123 and im" "g-xyz" "789."
• 输出: "Here is " { src: "abc123" } " and " { src: "xyz789" } "." */

nz() 接受任何同步或异步可迭代对象。

// Vercel AI SDK const { textStream } = streamText({ model, prompt }) nz(textStream) // OpenAI const stream = await openai.chat.completions.create({ ...args, stream: true }) nz(stream).map((chunk) => chunk.choices[0]?.delta?.content ?? "") // Anthropic const stream = anthropic.messages.stream({ ...args }) nz(stream).map((e) => (e.type === "content_block_delta" && e.delta.type === "text_delta" ? e.delta.text : "")) // fetch const res = await fetch(url) nz(res.body!.pipeThrough(new TextDecoderStream()))

用法示例

提取生成的某个片段

在模型还在写的时候，就把代码块流式发送到预览，然后再保存。

const code = await nz(stream) .after("ts\n") .before("") .tap(sendToPreview) .consume() await saveSnippet(code.string()) /*

• 输入: "Sure!\n" "`ts\nconst x = " "await f()\n" "return x\n" "`\nDone!"
• 输出: "const x = " "await f()\n" "return x\n" */

平滑处理供应商输出中的大块内容

供应商经常以突发方式交付文本。可以重新按单词切分，并以稳定的节奏释放。

const smooth = nz(stream) .splitAfter(" ") .compact() .minInterval(40) /*

• 输入: The quick ...... brown fox jumps over ................ the lazy dog.
• 输出: The .. quick .. brown .. fox .. jumps .. over .. the .. lazy .. dog. */

保证

管道是惰性的：除非你使用 for await 或 .consume()，否则不会执行。生成器源只能使用一次。

提前跳出循环会关闭上游迭代器。取消进行中的 I/O 由源/供应商负责。

来自源的错误（包括在计时器中抛出的错误）会在你 await 的位置浮现，因此 try/catch 可以正常工作。

生成器的返回值会穿过各个转换（参见 mapReturn、consume().return()）。

API 一览

分组 函数

正则 parse · scan · match · replace

切分 split · splitAfter · splitBefore · after · before · chunk

计时 minInterval · throttle

元素 map · asyncMap · filter · find

转换 flatMap · flatten

过滤 compact · takeWhile · takeUntil

索引 first · last · at · head · tail · initial · slice

累积 accumulate · diff · reduce

缓冲 batch · buffer · withContext · aperture

副作用 tap · tee

错误 recover · wrap · unwrap

转换 consume · toReadableStream · fromList · mapReturn

参考

正则

这些函数使用 JavaScript 正则表达式来搜索、匹配和转换流式文本。

Nozzle 通过在内部缓冲文本来处理跨数据块边界的匹配模式。确定不属于匹配的文本会被立即输出，而可能的匹配会被保留，直到确认。

以下特性会抛出错误，因为它们在流式处理中无法可靠工作：

某些模式会迫使 nozzle 缓冲比你预期更长的文本：

位于模式末尾的开放式量词：/hello.+/g 会缓冲 "hello" 之后的所有内容，直到流结束——.+ 总能匹配更多内容。请改用分隔符：/hello[^!]+/g 会一直匹配到 !，从而更早输出。

具有共享前缀的备选项：/cat|caterpillar/g 会缓冲 "cat"，直到有足够文本到达、排除 "caterpillar"。请把较长的备选项放在前面：/caterpillar|cat/g。

可选后缀：/items?/g 会缓冲 "item"，以检查后面是否跟有 "s"。这通常没问题，但 /data.*?end/g 会从 "data" 一直缓冲到 "end" 出现。

全局（/pattern/g）：在整个流中查找所有匹配

非全局（/pattern/）：只查找第一个匹配，然后将剩余文本原样传递

match

nz(["a", "b", "b", "a"]).match(/a([ab]*)a/g) // => ["abba", "bb"]（带有捕获组的匹配数组）

从输入流中提取正则模式的匹配项。

使用 earliestPossibleMatchIndex 高效跳过确定不匹配的 token，同时保留可能的匹配，直到能够判断是否匹配。

构建于：scan(input, regex).filter(x => 'match' in x).map(x => x.match)

function match<R = any>(input: AsyncIterable<string, R>, regex: RegExp): AsyncGenerator<RegExpExecArray, R, undefined>;

参数

参数 类型 说明

input AsyncIterable<string, R> 字符串的异步可迭代对象。

regex RegExp 要匹配的正则表达式模式。

parse

// 从文本中提取 UUID nz(["Now I'm taking uuid-asdf-flkj and adding it to uuid-fslkj-alkjlsf."]) .parse(/uuid-(?<id>\w+)-\w+/g, m => ({ id: m.groups!.id })) // 产出: "Now I'm taking ", { id: "asdf" }, " and adding it to ", { id: "fslkj" }, "." // 从文本中解析数字 nz(["The answer is 42 and also 123"]) .parse(/\d+/g, m => parseInt(m[0], 10)) // 产出: "The answer is ", 42, " and also ", 123

解析输入中的正则匹配，原样输出文本并转换匹配项。

此转换适合从文本流中提取结构化数据。不匹配的文本会作为字符串透传，而匹配项会使用提供的函数进行转换。

function parse<T, R = any>(input: AsyncIterable<string, R>, regex: RegExp, transform: (match: RegExpExecArray) => T): AsyncGenerator<string | T, R, undefined>;

参数

参数 类型 说明

input AsyncIterable<string, R> 字符串的异步可迭代对象。

regex RegExp 要匹配的正则表达式模式。

transform (match: RegExpExecArray) => T 把每个匹配项转换为所需类型的函数。

replace

nz(["a", "b", "b", "a"]).replace(/a[ab]*a/g, "X") // => "X"

（原文较长，此处译出前 6000 字，完整内容请点「打开原文」。）