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 实现逐词实时输出以及如何通过WatermarkingConfigWatermarkDetector完成“加水印—检测机器生成文本”的完整闭环并理解二者在源码中的真实调用链路。一、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_tensorspt) streamer TextStreamer(tokenizer) _ model.generate(**inputs, streamerstreamer, max_new_tokens20)注意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_promptFalse是否跳过 prompt 部分的输出聊天机器人场景很有用decode_kwargs无透传给tokenizer.decode的额外参数如skip_special_tokensTrueTextStreamer.put内部维护一个token_cache每次把新 token 加入缓存后整体重新 decode并采用启发式策略决定打印到哪里若解码结果以换行\n结尾打印全部并清空缓存若最后一个字符是 CJK 字符中日韩直接打印汉字之间无空格否则只打印到最后一个空格为止text.rfind( )避免把可能被下一个 token 改变的不完整单词打到屏幕上。end()会把缓存中剩余内容冲刷输出并打印换行。此外TextStreamer只支持 batch size 为 1 的输入batch1 会直接抛ValueError。1.4 进阶 streamer迭代器与异步版本除基础的TextStreamer外streamers.py 还提供了面向应用的两个变体均继承自TextStreamerTextIteratorStreamer把可打印文本放入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_tensorspt) streamer TextIteratorStreamer(tok) # 在独立线程运行 generate以便非阻塞地取回文本 generation_kwargs dict(inputs, streamerstreamer, max_new_tokens20) thread Thread(targetmodel.generate, kwargsgeneration_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], paddingTrue, return_tensorspt) input_len inputs[input_ids].shape[-1] watermarking_config WatermarkingConfig(bias2.5, seeding_schemeselfhash) out model.generate(**inputs, watermarking_configwatermarking_config, do_sampleFalse, max_length20)detector WatermarkDetector(model_configmodel.config, devicecpu, watermarking_configwatermarking_config) detection_out detector(out, return_dictTrue) 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_ratio0.25绿色 token 数与词表大小之比必须在 0.0 到 1.0 之间bias2.0加在绿色 token logits 上的偏置若生成质量下降建议调低推荐 0.52.0hashing_key15485863哈希键第 100 万个质数。实际部署时建议换成私有键seeding_schemelefthash绿色 token 的选定算法lefthash依赖上一个 token论文 Algorithm 2selfhash依赖当前候选 token 自身Algorithm 3需考察所有可能的下一 token因此比 lefthash 更慢context_width1播种时使用的上文 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 等模型信息 devicecpu, # 必须与生成时一致 watermarking_config..., # 必须与生成时一致 ignore_repeated_ngramsFalse, # 每个 unique n-gram 只计一次 max_cache_size128, # 播种/采样算法的 LRU 缓存上限 )调用detector(input_ids, z_threshold3.0, return_dictFalse)时内部流程是跳过起始的 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_dictTrue时返回WatermarkDetectorOutput包含num_tokens_scored、num_green_tokens、green_fraction、z_score、p_value、prediction、confidence七个字段——上面示例中detection_out.prediction得到的array([True, True])即来自此结构。此外仓库还提供了一类基于学习的检测器SynthIDTextWatermarkingConfigSynthIDTextWatermarkLogitsProcessorBayesianDetectorModel走的是 tournament 式多层 n-gram 水印参数ngram_len、keys等与贝叶斯后验打分路线官方 z-score 检测器WatermarkDetector则与WatermarkingConfig配套是最常用的组合。三、小结Streaming任何带put/end两方法的对象都能作为streamer传给generate内置的TextStreamer家族文本打印 / 同步迭代器 / 异步迭代器覆盖了从终端展示到 Gradio 异步 Demo 的主流场景其“最后一个空格”与 CJK 特判策略保证了逐词输出的可读性WatermarkingWatermarkingConfigbias、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 剥离建议均以文档与上述源码为准。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考