批处理推理用批处理吞吐换取响应延迟的实现方法

离线推理有一批固定文本、图片或音频要处理时，批处理的价值在于让 GPU 一次接收多个样本，减少重复的调度和数据搬运。它并不是“batch_size 越大越快”：样本长度不均、显存不足、单批等待时间都会改变结果。实用的做法是把数据做成流式输入，先用小批次测吞吐，再把显存和失败恢复写进调度器。

官方地址：https://huggingface.co/docs/transformers/main/en/main_classes/pipelines

先固定离线任务的批次边界

先把任务分成“可等待的离线队列”和“必须即时返回的在线请求”。前者可以积攒 8、16 或 32 条样本，让设备保持较高利用率；后者若为了凑批次而等待，用户感知到的就是更长的首响应时间。本文只讨论前一种情况，输出顺序仍按输入序号保存。

还要先处理样本长度。文本长度差异很大时，一个超长样本会把同批其他样本一起垫到更大的张量尺寸，吞吐可能反而下降。因此队列可以先按长度区间分桶，再在每个桶内组批。

用流式数据集输入 pipeline

不要为了批处理先把全部输入复制到内存。Transformers 的 pipeline 可以把 Dataset 或生成器作为输入，在迭代时交给 DataLoader 组批；这样既能控制内存，也方便把处理进度写入外部队列。

from datasets import Dataset
from transformers import pipeline

# 用小样本先建立可重复的离线输入，生产环境可替换为文件扫描器
records = Dataset.from_dict({"text": [
    "第一条待分类文本",
    "第二条待分类文本",
    "第三条待分类文本",
]})

# device=0 表示使用第一张 CUDA 卡；模型和设备要固定，方便比较批次结果
pipe = pipeline("text-classification", device=0)

# batch_size 只影响离线组批，不改变输出结构；输出要按输入顺序写回
for offset, result in enumerate(pipe(records["text"], batch_size=8, truncation=True)):
    # offset 可作为断点游标，异常重启时从未完成的位置继续
    print(offset, result)

示例中的 8 只是起始值，不是通用答案。实际任务里建议把输入迭代器和结果写入拆开：每完成一批就落盘一个游标，单批失败时只重试当前范围，不让已经完成的结果重复计算。

用吞吐、显存和等待时间共同决定批大小

批大小至少要看三项指标：每秒处理样本数、峰值显存和单批完成时间。固定模型与设备后，用 1、4、8、16 逐档测试；当吞吐增长已经变小，或显存逼近上限，就停止扩大。样本长度规律时，批处理更容易获得收益；长度波动大时，应优先分桶而不是继续加大批次。

为显存不足和失败任务保留降级路径

离线任务的调度器不应把 CUDA out of memory 当作整批任务终止信号。为每批保存起止偏移、当前 batch_size 和重试次数；第一次失败时将批次减半，重新处理同一偏移范围。若最小批次仍失败，再把该样本单独转入异常队列，并保留错误信息。

恢复逻辑要有上限：例如同一批最多降级两次，避免坏样本造成循环。成功写回结果后再推进游标，不能在推理开始前提前提交“已完成”状态。

常见问题

离线批处理为什么不能直接用于在线接口？

在线接口首先受首响应和尾延迟约束，等待更多请求凑批可能抵消 GPU 的吞吐收益。除非业务明确允许排队，否则应使用单条或有严格等待上限的动态批处理。

batch_size 越大是不是一定越省时间？

不是。长度不均会放大填充开销，过大的批次还会触发显存不足。只有在真实数据上测得吞吐继续增长、显存稳定且单批时间可接受时，增大批次才有意义。

怎样判断批处理参数已经可以上线？

用接近生产分布的数据重复跑多轮，记录吞吐、峰值显存、P95 单批时间、失败重试次数和队列积压。参数应以最差一档可恢复为准，而不是只取一次最快结果。