AI Transformers 批量文本分类 怎样处理

批量文本分类的关键不是把字符串简单地塞进一个列表，而是同时处理好三个问题：输入规模、批内长度差异和结果回写。Transformers 的 text-classification Pipeline 可以直接接收字符串列表；当数据量继续增大时，更适合用数据集或生成器逐条提供文本，再用 batch_size 控制每次推理的数量。

一个稳妥的起点是：选择已经针对序列分类任务微调的模型，给批量输入明确设置 truncation=True，让批内按最长文本补齐，并从较小的 batch_size 开始观察内存和吞吐。这样既不会把“批量”误解成一次性读入全部数据，也能保留每条结果与原文的对应关系。

少量文本传列表即可；大批量文本用迭代器分批喂给 Pipeline。padding 解决同一批长度不齐，truncation 解决单条文本过长，batch_size 只控制一次送入模型的数量。

先把批量入口分成两种场景

如果只有几十条文本，列表输入最直观，Pipeline 会为每条输入返回一个分类结果。列表的优点是代码短、结果顺序清晰；缺点是所有文本已经在内存中。如果数据来自文件、数据库或消息流，不必先拼成一个超大列表，可以使用生成器，让生产速度和推理速度通过迭代边界隔开。

这里的“批量”仍然是推理接口的输入组织方式，不代表模型会自动替你完成数据清洗、去重或标签映射。原始记录最好带一个稳定的业务编号，输出时将编号一起保存。

padding 和截断要一起设定

同一批文本通常长短不同，模型需要规则的张量形状，因此短文本要补齐。使用 padding=True 或 padding="longest" 时，补齐长度通常跟随当前批次最长文本；这比所有批次都固定补到模型最大长度更省无效计算。

另一方面，truncation=True 会把超过允许长度的输入截断。max_length 应根据模型、文本语言和业务字段确定，不要把任意数字当成通用答案。截断会丢掉文本尾部信息，客服长描述、日志和合同片段尤其需要先决定保留头部、尾部，还是在业务层切片后汇总。

若分类模型有多个标签，Pipeline 默认会把多标签 logits 转成 softmax 分数；单标签模型则走 sigmoid。文章中的“分数”应当被当作模型输出的置信度信号，而不是未经校准的业务概率。

用小批量开始，再决定 batch_size

下面的示例适合先验证输入和结果映射。示例故意保存 id，因为后续换成生成器或分块读取时，不能依赖“第几次调用”来定位原记录。

from transformers import pipeline

# 使用已经针对文本分类任务微调的模型；实际项目应替换成自己的模型名
classifier = pipeline(
    task="text-classification",
    model="distilbert/distilbert-base-uncased-finetuned-sst-2-english",
    device=-1,  # -1 表示先用 CPU；有 GPU 时再按设备配置调整
)

records = [
    {"id": "a-001", "text": "The delivery was fast and the package was intact."},
    {"id": "a-002", "text": "The response arrived late and did not solve the issue."},
]

# 列表输入适合小批量；batch_size 先取小值，避免一开始就占满显存或内存
outputs = classifier(
    [item["text"] for item in records],
    batch_size=2,
    padding=True,       # 让同一批输入对齐到当前批次的最长文本
    truncation=True,    # 防止超长文本超过模型可接受的输入范围
    max_length=256,     # 业务上确认过的截断上限，不是所有模型的固定值
)

result = [
    {"id": item["id"], "label": output["label"], "score": output["score"]}
    for item, output in zip(records, outputs)
]
print(result)  # 用稳定 id 回写结果，避免分批后与原始记录错位

如果输入规模更大，可以把 records 换成按行读取的生成器，并在输出时立即写入结果文件。这样降低峰值内存，但也要注意：生成器场景通常不能随意重放，发生中途失败时应保留已写出的最后位置或业务游标。

from transformers import pipeline

classifier = pipeline(
    "text-classification",
    model="distilbert/distilbert-base-uncased-finetuned-sst-2-english",
    device=-1,
)

def text_stream(lines):
    # 逐条产生文本，让 Pipeline 能按批消费，而不是一次加载整个文件
    for line_no, line in enumerate(lines, start=1):
        text = line.strip()
        if text:
            yield {"id": line_no, "text": text}

source = [
    "The service recovered quickly.",
    "The instructions were confusing.",
    "The result was easy to verify.",
]

stream = text_stream(source)
for batch in classifier(
    (item["text"] for item in stream),
    batch_size=2,
    padding=True,
    truncation=True,
    max_length=256,
):
    # 迭代器输出的结果应及时落盘；这里仅打印，生产代码应写入持久化存储
    print(batch)

第二段只演示迭代边界，若要严格回写原始 id，应在自己的分块读取器中同时维护文本和编号，并按同一批次保存。对 Hugging Face Datasets，可以使用其字段迭代方式，把目标文本列交给 Pipeline；不要为了追求“批量”而把几百万条记录先转换成 Python 列表。

怎样判断批量参数合适

先用一小段固定样本比较单条与批量输出是否仍然一一对应，再逐步增加 batch_size。在 GPU 上批量可能提高吞吐，在 CPU 或文本长度差异很大时也可能变慢，因为补齐带来了额外计算。出现显存不足时，优先降低 batch_size 或 max_length；不要直接关闭截断而把问题推给模型输入限制。

最后要区分“能批量调用”和“业务分类可靠”两个问题。Pipeline 负责预处理、模型推理和结果后处理，但标签语义、阈值、长文本切分和低置信度人工复核，都需要结合自己的训练数据与业务规则确定。

常见问题

列表输入和 batch_size 是一回事吗？ 不是。列表描述输入集合，batch_size 描述推理时每次送入模型的数量；大列表仍可能被分成多个小批次。

batch_size 越大是不是越快？ 不一定。硬件、模型、文本长度分布和数据读取速度都会影响结果，应以固定样本测吞吐和峰值内存。

为什么设置了 padding 还要 truncation？ padding 解决同一批文本长短不齐，truncation 解决单条文本超过上限，它们处理的是两个不同边界。

批量结果能直接当业务概率吗？ 不能直接等同。它是模型后处理后的分数，阈值和概率校准仍需用业务验证集决定。

小结

少量文本用列表快速接入，大量文本用数据集或生成器控制内存；用 padding 处理批内对齐，用 truncation 和 max_length 管住输入边界；从较小的 batch_size 开始测吞吐，最后用稳定 id 把标签和分数安全写回原数据。