Transformers streaming逐步输出生成文本

Transformers要逐步输出生成文本，可以把TextIteratorStreamer传给model.generate，在生成线程之外迭代读取文本片段。生成、解码队列和下游展示需要分开运行；否则先等generate全部结束，再读取队列，界面仍会表现为一次性输出。

这里用单条输入的本地推理调用说明完整边界：何时开始输出、怎样结束、后台异常怎样回传，以及客户端不再需要答案后如何请求停止。它是应用接入示例，未宣称已经在本机执行，也不是一套完整HTTP服务。

官方接口地址：https://huggingface.co/docs/transformers/main/internal/generation_utils

界定流式输出的触发与输入

用户提交问题后，应用先完成授权、输入长度限制和模型容量分配，再开始一次生成。模型与分词器在服务启动阶段加载，不能每输出一个片段就重新加载。对于对话模型，先使用该模型要求的聊天模板准备输入，不要把普通文本拼接方式套到所有模型。

本文函数接收已经加载并设置eval的model、配套tokenizer及准备好的inputs。inputs的张量必须与模型运行设备相匹配；自动设备分配模型应按自身部署方式准备输入。这里只接受单请求、单序列和非束搜索，避免把多个用户的输出混到一条文本流里。

流式传输改善的是读者开始看到内容的时机，并不自动提高模型生成吞吐。模型预填充、首步计算和文本解码都可能造成开头一段等待，不应以“尚无文本”立即判定推理失败。

组织生成线程与文本队列

生成线程负责调用generate，TextIteratorStreamer负责接收生成内容并提供可消费文本，消费端通过emit把片段交给展示层。另外保留错误队列和停止事件：错误队列记录后台失败，停止事件让生成逻辑在可检查的位置协作退出。文本队列的结束与业务成功不是同一个信号。

普通TextStreamer适合直接输出，TextIteratorStreamer更便于应用消费。它迭代出来的是已解码文本片段，不承诺每个片段对应一个token、一个汉字或一个词；空片段也可能出现。应用应按返回文本顺序拼接，不要把每次迭代次数当作生成token数。

实现带超时和错误回传的输出函数

下面给出原创函数骨架。emit是调用方提供的同步文本回调，例如写入本地显示组件；若它用于网络发送，需要自行限制阻塞时间并处理连接异常。函数只使用内部固定生成参数，不允许外部任意参数覆盖streamer、停止规则或序列数量。

from queue import Queue, Empty
from threading import Thread, Event
from time import monotonic
import torch
from transformers import (
    TextIteratorStreamer, StoppingCriteria, StoppingCriteriaList,
)

class StopOnEvent(StoppingCriteria):
    def __init__(self, event):
        self.event = event

    def __call__(self, input_ids, scores, **kwargs):
        # 每一行返回一个停止标志；这里只允许单条输入。
        return torch.full(
            (input_ids.shape[0],), self.event.is_set(),
            dtype=torch.bool, device=input_ids.device,
        )

def stream_once(model, tokenizer, inputs, emit, budget=30.0):
    # model须已eval，inputs须在匹配设备上；同一模型不要无控制并发调用。
    if inputs["input_ids"].shape[0] != 1 or budget = deadline:
                raise TimeoutError("本次生成超过应用时间预算")
            try:
                piece = next(streamer)
            except Empty:
                # 队列暂时无文本不代表完成；后台失败也不能无限等待。
                if done.is_set():
                    break
                continue
            except StopIteration:
                break
            if piece:
                emit(piece)
                chunks.append(piece)
    finally:
        stop.set()
        # 协作式回收：不能强杀卡在底层计算中的线程。
        worker.join()
    if not errors.empty():
        raise RuntimeError("模型生成失败，已输出内容仅为部分结果") from errors.get()
    return "".join(chunks)

timeout=0.5限制的是本次队列等待，出现Empty时继续检查状态；budget是应用自己增加的整体时间判断。它们都不是可以强制中断GPU内核的硬超时。finally里的join会等待协作退出，所以函数实际返回时间仍可能超过budget。若需要强隔离与硬截止，应由推理工作进程和服务调度层设计终止、重启与设备恢复，不能依赖Python线程强杀。

max_new_tokens限制新增token数量，避免一次请求无上限生成。skip_prompt用于不把输入提示词当作答案输出；skip_special_tokens属于解码选项。取消依据只查看停止事件，不依赖scores，所以这里不需要为了取消而额外请求全部生成分数。

示例不包含下载模型和依赖安装，因为模型选择、权重位置与硬件配置由调用项目确定。应锁定transformers与torch的实际版本，并查对应版本文档；main是开发文档入口，不代表当前环境安装的稳定版本接口完全相同。

定义完成事件与失败处理

建议展示层区分“正在输出”“完成”“失败”“取消”四种状态。只有函数正常返回后，才发出业务完成事件并把拼接文本作为完整答案保存。后台异常、整体超时或emit抛出异常时，前面已经显示的文字只能保留为部分内容。不能因为队列收到结束标志就把错误答案计为成功。

如果HTTP适配层采用SSE，应把每段文本封装成事件，而非直接把任意文本当作协议行写入；换行和编码由协议层正确处理。代理、服务框架和浏览器端都可能缓冲，即使本地消费及时，页面也可能最后才一起刷新。先分别记录生成侧与发送侧时间，再判断问题处在哪一段。

接入服务时的观察指标与常见问题

每个请求使用独立的streamer、停止事件和状态记录。共享模型需要容量控制，不能用“每请求一个线程”代替调度策略。默认队列也不能被理解为自动具备有界背压；如果消费端明显慢于生成端，需要限制请求长度、监测排队与内存，或实现匹配业务的有界通道。

为什么输出不是一个字一个字出来？分词粒度与显示粒度不同，解码还可能积攒上下文后才给出可显示片段。应用可以调整展示动画，但不要切坏Unicode字符，也不要将动画速度当成模型速度。

设置timeout后为什么仍未停止模型？streamer的等待超时只影响读取队列，消费端退出也不天然终止生成。本例额外使用停止事件与StoppingCriteria，但它仍是协作取消；底层计算尚未返回时需要等待。

为什么最后有文本，却没有完成标记？文本和状态应走明确的应用协议。客户端断连、发送失败、后台异常都可能发生在部分文本已经输出之后。完成事件只能在正常收尾后发送，不要在finally里无条件发送成功。

应该记录哪些指标？区分模型调用开始、首个非空文本片段、首次网络发送、生成结束和资源回收。首个文本片段时间不等同于模型首token时间。按成功、失败、取消分别统计总耗时及输出长度，日志保留错误类型和请求标识即可，避免无必要保存完整提示词与用户内容。