Transformers 生成高级特性实战：流式输出（Streaming）与文本水印（Watermarking）-平芜编程栈

Transformers 生成高级特性实战：流式输出（Streaming）与文本水印（Watermarking）

【免费下载链接】transformers🤗 Transformers: the model-definition framework for state-of-the-art machine learning models in text, vision, audio, and multimodal models, for both inference and training.项目地址: https://gitcode.com/GitHub_Trending/tra/transformers

本文基于 Transformers 官方文档 generation_features.md 展开，系统讲解 [GenerationMixin.generate] 之上构建应用的两大高级特性：流式输出（Streaming）与文本水印（Watermarking）。读完本文，你将掌握如何在生成接口中挂接自定义 streamer 实现逐词实时输出，以及如何通过WatermarkingConfig+WatermarkDetector完成“加水印—检测机器生成文本”的完整闭环，并理解二者在源码中的真实调用链路。

一、Streaming：让生成结果“边生成边输出”

流式输出（Streaming）的价值在于：文本一旦生成就立刻返回给用户，而不必等整个响应全部完成。对于面向用户的应用（聊天机器人、交互式 Demo），它能显著降低感知延迟，让用户看到生成过程实时推进。

Transformers 的接入方式非常直接：用 tokenizer 创建一个 [TextStreamer] 实例，把它作为streamer参数传给 [~GenerationMixin.generate]，即可逐词流式输出：

from transformers import AutoModelForCausalLM, AutoTokenizer, TextStreamer tokenizer = AutoTokenizer.from_pretrained("openai-community/gpt2") model = AutoModelForCausalLM.from_pretrained("openai-community/gpt2") inputs = tokenizer(["The secret to baking a good cake is "], return_tensors="pt") streamer = TextStreamer(tokenizer) _ = model.generate(**inputs, streamer=streamer, max_new_tokens=20)

注意：generate本身照常返回完整结果张量；streamer 是“旁路”副作用通道——文档中的_ =只是强调返回值可被忽略。

1.1 streamer 的契约：只需`put`与`end`两个方法

streamer参数并不要求必须是TextStreamer：从源码看，任何实现了put和end方法的对象都兼容（见 BaseStreamer）。其契约是：

put(value)：由generate在每一步解码后调用，把新 token 推给流式端；
end()：由generate在生成结束时调用，标志流式输出终止。

这意味着你可以完全自定义 streamer（例如把 token 转发到 WebSocket、写入文件或做 TTS 合成），只要提供这两个方法即可。

1.2 源码解析：generate 内部如何调用 streamer

在生成主循环 generation/utils.py 中可以看到真实调用点：

# update generated ids, model inputs, and length for next step input_ids = torch.cat([input_ids, next_tokens[:, None]], dim=-1) if streamer is not None: streamer.put(next_tokens.cpu()) ... # 循环结束后 if streamer is not None: streamer.end()

即：每步选出next_tokens并拼入input_ids后，立刻调用streamer.put(next_tokens.cpu())；整个循环退出前统一调用streamer.end()。这也解释了为什么 streamer 拿到的是CPU 上的 token id 张量——解码（decode 成文字）的工作完全交给 streamer 侧完成。

1.3 TextStreamer 的参数与分词边界策略

从 TextStreamer 实现看，它接受三个关键参数：

参数	默认值	说明
`tokenizer`	必填	用于将累积 token 解码为文字的分词器
`skip_prompt`	`False`	是否跳过 prompt 部分的输出，聊天机器人场景很有用
`decode_kwargs`	无	透传给`tokenizer.decode`的额外参数，如`skip_special_tokens=True`

TextStreamer.put内部维护一个token_cache，每次把新 token 加入缓存后整体重新 decode，并采用启发式策略决定打印到哪里：

若解码结果以换行\n结尾：打印全部并清空缓存；
若最后一个字符是 CJK 字符（中日韩）：直接打印（汉字之间无空格）；
否则：只打印到最后一个空格为止（text.rfind(" ")），避免把可能被下一个 token 改变的不完整单词打到屏幕上。

end()会把缓存中剩余内容冲刷输出并打印换行。此外，TextStreamer只支持 batch size 为 1 的输入（batch>1 会直接抛ValueError）。

1.4 进阶 streamer：迭代器与异步版本

除基础的TextStreamer外，streamers.py 还提供了面向应用的两个变体，均继承自TextStreamer：

TextIteratorStreamer：把可打印文本放入queue.Queue，下游以迭代器方式非阻塞读取（典型用法：在独立线程中运行generate，主线程for new_text in streamer:逐段消费）。额外参数timeout控制队列阻塞超时，便于在generate抛异常时跳出等待；
AsyncTextIteratorStreamer：异步版本，基于asyncio.Queue，供 Gradio 等异步框架使用。注意它必须在协程内部初始化（因为要获取当前事件循环），并通过loop.call_soon_threadsafe跨线程安全地把文本投递回事件循环。

典型的迭代器用法（源码 docstring 示例）：

from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer from threading import Thread tok = AutoTokenizer.from_pretrained("openai-community/gpt2") model = AutoModelForCausalLM.from_pretrained("openai-community/gpt2") inputs = tok(["An increasing sequence: one,"], return_tensors="pt") streamer = TextIteratorStreamer(tok) # 在独立线程运行 generate，以便非阻塞地取回文本 generation_kwargs = dict(inputs, streamer=streamer, max_new_tokens=20) thread = Thread(target=model.generate, kwargs=generation_kwargs) thread.start() generated_text = "" for new_text in streamer: generated_text += new_text

二、Watermarking：给机器生成的文本“打指纹”

水印（Watermarking）用于回答一个问题：这段文本是否由机器生成？Transformers 实现的水印策略（参考论文 2306.04634）思路是：随机把词表中一部分 token 染成“绿色”（green）；生成时，绿色 token 的 logits 会获得一个小的正偏置（bias），使其被选中的概率更高。检测时，只需比较文本中绿色 token 的比例是否显著高于人类文本的典型比例，即可判断是否带水印。

两个关键工程优势：

适用于 Transformers 中的任何生成模型，不改动模型结构；
检测不需要额外的分类模型，水印信息内嵌于 token 序列本身。

2.1 完整示例：加水印生成 + 检测

from transformers import AutoTokenizer, AutoModelForCausalLM, WatermarkDetector, WatermarkingConfig model = AutoModelForCausalLM.from_pretrained("openai-community/gpt2") tokenizer = AutoTokenizer.from_pretrained("openai-community/gpt2") tokenizer.pad_token_id = tokenizer.eos_token_id tokenizer.padding_side = "left" inputs = tokenizer(["This is the beginning of a long story", "Alice and Bob are"], padding=True, return_tensors="pt") input_len = inputs["input_ids"].shape[-1] watermarking_config = WatermarkingConfig(bias=2.5, seeding_scheme="selfhash") out = model.generate(**inputs, watermarking_config=watermarking_config, do_sample=False, max_length=20)

detector = WatermarkDetector(model_config=model.config, device="cpu", watermarking_config=watermarking_config) detection_out = detector(out, return_dict=True) detection_out.prediction array([True, True])

两个细节值得注意：

批量生成时建议设置pad_token_id = eos_token_id且padding_side = "left"，这是文档示例中的标准做法；
官方提示（TIP）：[WatermarkDetector] 统计的是整段序列的绿色 token 比例，因此如果 prompt 远长于生成部分，建议先剥离 prompt 文本再检测，padding 也会干扰结果——变量input_len正是留给开发者截取纯生成段（out[:, input_len:]）用的。

2.2 WatermarkingConfig 参数全解

WatermarkingConfig定义了水印生成的全部超参，参数、默认值与含义如下（结合 validate 方法可确认取值约束）：

参数	默认值	取值范围 / 说明
`greenlist_ratio`	`0.25`	绿色 token 数与词表大小之比，必须在 0.0 到 1.0 之间
`bias`	`2.0`	加在绿色 token logits 上的偏置；若生成质量下降建议调低，推荐 0.5～2.0
`hashing_key`	`15485863`	哈希键（第 100 万个质数）。实际部署时建议换成私有键
`seeding_scheme`	`"lefthash"`	绿色 token 的选定算法：`"lefthash"`依赖上一个 token（论文 Algorithm 2）；`"selfhash"`依赖当前候选 token 自身（Algorithm 3），需考察所有可能的下一 token，因此比 lefthash 更慢
`context_width`	`1`	播种时使用的上文 token 数，必须 ≥ 1；上下文越长，水印越稳健

GenerationConfig会把watermarking_config直接纳入生成参数体系（configuration_utils.py 中支持 dict 形式传入并自动转换为WatermarkingConfig实例），并在validate()阶段对水印参数做合法性校验。

2.3 底层机制：WatermarkLogitsProcessor 如何“染色”

水印在生成循环中通过一个 logits processor 生效——WatermarkLogitsProcessor。其工作过程可以概括为三步：

播种（seeding）：以hashing_key初始化随机数生成器，并用一个固定随机排列表fixed_table（大小 1,000,003）为 token 序列生成种子——lefthash用最后一个 token 计算种子，selfhash则用固定表对序列做乘积哈希，使绿色列表只取决于候选 token 本身；
选绿（greenlist）：用种子驱动torch.randperm(vocab_size)，取排列的前int(vocab_size * greenlist_ratio)个位置作为绿色 token 集合；
加偏置（bias）：把bias值加到这些绿色 token 的 logits 上，绿色 token 因此更容易被采样选中。

selfhash方案由于要为每个候选 token 单独计算绿色列表，源码中采用拒绝采样（_score_rejection_sampling，最多迭代 40 步）来加速，这也印证了文档中“selfhash 可能更慢”的说明。

2.4 检测原理：z 分数与 p 值

WatermarkDetector是“无模型”检测器：它需要与生成时完全相同的WatermarkingConfig、设备与词表大小，才能复现绿色列表并计数。构造参数为：

WatermarkDetector( model_config=..., # 提供 vocab_size 等模型信息 device="cpu", # 必须与生成时一致 watermarking_config=..., # 必须与生成时一致 ignore_repeated_ngrams=False, # 每个 unique n-gram 只计一次 max_cache_size=128, # 播种/采样算法的 LRU 缓存上限 )

调用detector(input_ids, z_threshold=3.0, return_dict=False)时，内部流程是：

跳过起始的 bos token，要求除去播种所需的前context_width个 token 后至少还有 1 个 token 可打分，否则抛ValueError；
对每个 n-gram 前缀重算绿色列表并统计num_tokens_scored与num_green_tokens（支持重复 n-gram 去重）；
计算 z 分数：z = (green_count - p*N) / sqrt(N*p*(1-p))，其中p = greenlist_ratio；当z > z_threshold（默认 3.0）时预测为机器生成。z_threshold越高，检测越保守（敏感度越低）；
return_dict=True时返回WatermarkDetectorOutput，包含num_tokens_scored、num_green_tokens、green_fraction、z_score、p_value、prediction、confidence七个字段——上面示例中detection_out.prediction得到的array([True, True])即来自此结构。

此外，仓库还提供了一类基于学习的检测器：SynthIDTextWatermarkingConfig+SynthIDTextWatermarkLogitsProcessor+BayesianDetectorModel，走的是 tournament 式多层 n-gram 水印（参数ngram_len、keys等）与贝叶斯后验打分路线；官方 z-score 检测器（WatermarkDetector）则与WatermarkingConfig配套，是最常用的组合。

三、小结

Streaming：任何带put/end两方法的对象都能作为streamer传给generate；内置的TextStreamer家族（文本打印 / 同步迭代器 / 异步迭代器）覆盖了从终端展示到 Gradio 异步 Demo 的主流场景，其“最后一个空格”与 CJK 特判策略保证了逐词输出的可读性；
Watermarking：WatermarkingConfig（bias、greenlist_ratio、seeding_scheme、context_width、hashing_key 五参数）控制加水印过程，由WatermarkLogitsProcessor在 logits 上完成“染色”；检测侧用同一份配置构造WatermarkDetector，通过 z 分数统计检验判断文本是否机器生成，全程无需额外分类模型；
相关实现集中在 src/transformers/generation/ 目录：streamers.py、watermarking.py、configuration_utils.py、logits_process.py 与 utils.py，可进一步对照阅读；流式与水印特性的行为边界（如 batch 限制、z_threshold敏感度调节、prompt 剥离建议）均以文档与上述源码为准。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Transformers 生成高级特性实战：流式输出（Streaming）与文本水印（Watermarking）