Pipeline 快速入门

Pipeline 是 Transformers 库提供的高层 API，让你只需几行代码就能使用预训练模型完成各种 AI 任务。本章将介绍 Pipeline 的基本用法和常见任务类型。

什么是 Pipeline？

Pipeline 将复杂的模型推理流程封装成简单的接口，自动处理：

• 模型和分词器的加载
• 输入数据的预处理
• 模型推理
• 输出结果的后处理

┌─────────────────────────────────────────────────────────────┐
│                      Pipeline 流程                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   原始输入        预处理         模型推理        后处理       │
│      │              │              │             │         │
│      ▼              ▼              ▼             ▼         │
│   "Hello!"  →  Token IDs  →  Hidden States  →  结果       │
│                                                             │
│   Pipeline 自动完成所有步骤，你只需关注输入和输出            │
│                                                             │
└─────────────────────────────────────────────────────────────┘

基本用法

创建 Pipeline

from transformers import pipeline

# 创建情感分析 Pipeline
classifier = pipeline("sentiment-analysis")

# 使用 Pipeline
result = classifier("I love using Transformers!")
print(result)
# [{'label': 'POSITIVE', 'score': 0.9998}]

指定模型

# 使用特定的预训练模型
classifier = pipeline(
    "sentiment-analysis",
    model="distilbert-base-uncased-finetuned-sst-2-english"
)

result = classifier("This movie was absolutely fantastic!")
print(result)

批量处理

# 同时处理多条文本
texts = [
    "I love this product!",
    "This is terrible.",
    "It's okay, nothing special."
]

results = classifier(texts)
for text, result in zip(texts, results):
    print(f"{text} -> {result['label']} ({result['score']:.4f})")

支持的任务类型

自然语言处理任务

1. 情感分析 (sentiment-analysis)

from transformers import pipeline

classifier = pipeline("sentiment-analysis")

# 单条文本
result = classifier("I really enjoyed this movie!")
print(result)
# [{'label': 'POSITIVE', 'score': 0.9998}]

# 多条文本
texts = [
    "The food was amazing!",
    "I waited for an hour and the service was terrible."
]
results = classifier(texts)

2. 文本生成 (text-generation)

generator = pipeline("text-generation", model="gpt2")

# 生成文本
result = generator(
    "The future of artificial intelligence is",
    max_length=50,
    num_return_sequences=2,
    temperature=0.7
)

for i, res in enumerate(result):
    print(f"生成结果 {i+1}: {res['generated_text']}\n")

3. 问答 (question-answering)

qa_pipeline = pipeline("question-answering")

context = """
Transformers is a library maintained by Hugging Face. 
It provides general-purpose architectures for natural language processing, 
computer vision, and audio tasks.
"""

question = "Who maintains the Transformers library?"

result = qa_pipeline(question=question, context=context)
print(f"答案: {result['answer']}")
print(f"置信度: {result['score']:.4f}")
print(f"起始位置: {result['start']}, 结束位置: {result['end']}")

4. 命名实体识别 (ner)

ner_pipeline = pipeline("ner", grouped_entities=True)

text = "Hugging Face is a company based in New York."
results = ner_pipeline(text)

for entity in results:
    print(f"实体: {entity['word']}")
    print(f"类型: {entity['entity_group']}")
    print(f"置信度: {entity['score']:.4f}\n")

5. 文本摘要 (summarization)

summarizer = pipeline("summarization")

text = """
The Hugging Face Transformers library is one of the most popular 
machine learning libraries for natural language processing. 
It provides thousands of pretrained models to perform tasks on 
texts such as classification, information extraction, question 
answering, summarization, translation, text generation, and more 
in over 100 languages. Its aim is to make cutting-edge NLP 
accessible to everyone.
"""

summary = summarizer(
    text,
    max_length=50,
    min_length=20,
    do_sample=False
)

print(summary[0]['summary_text'])

6. 翻译 (translation)

# 英译中
translator = pipeline("translation_en_to_zh")

result = translator("Hello, how are you today?")
print(result[0]['translation_text'])

# 中译英
translator = pipeline("translation_zh_to_en")
result = translator("今天天气真好！")
print(result[0]['translation_text'])

7. 填空 (fill-mask)

unmasker = pipeline("fill-mask", model="bert-base-uncased")

# BERT 使用 [MASK] 标记进行填空
result = unmasker("The capital of France is [MASK].")[MASK].")

for item in result[:3]:
    print(f"{item['token_str']}: {item['score']:.4f}")

8. 零样本分类 (zero-shot-classification)

classifier = pipeline("zero-shot-classification")

text = "This is a contract about the sale of goods."
candidate_labels = ["legal", "finance", "marketing", "technology"]

result = classifier(text, candidate_labels)

for label, score in zip(result['labels'], result['scores']):
    print(f"{label}: {score:.4f}")

计算机视觉任务

图像分类

from transformers import pipeline

# 图像分类
classifier = pipeline("image-classification")

result = classifier("path/to/image.jpg")

for item in result[:3]:
    print(f"类别: {item['label']}, 置信度: {item['score']:.4f}")

目标检测

detector = pipeline("object-detection")

result = detector("path/to/image.jpg")

for item in result:
    print(f"检测到: {item['label']}")
    print(f"置信度: {item['score']:.4f}")
    print(f"位置: {item['box']}\n")

音频任务

自动语音识别 (ASR)

# 语音识别
asr = pipeline("automatic-speech-recognition")

result = asr("path/to/audio.wav")
print(result['text'])

Pipeline 高级配置

设备指定

# 使用 GPU
classifier = pipeline("sentiment-analysis", device=0)

# 使用 CPU
classifier = pipeline("sentiment-analysis", device=-1)

# 自动选择设备
classifier = pipeline("sentiment-analysis", device_map="auto")

批处理大小

# 设置批处理大小以优化性能
classifier = pipeline(
    "sentiment-analysis",
    batch_size=16  # 根据 GPU 显存调整
)

# 大量数据处理
texts = ["Sample text"] * 1000
results = classifier(texts)

自定义配置

from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification

# 加载自定义模型和分词器
model_name = "distilbert-base-uncased-finetuned-sst-2-english"

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(model_name)

# 创建自定义 Pipeline
classifier = pipeline(
    "sentiment-analysis",
    model=model,
    tokenizer=tokenizer,
    truncation=True,
    max_length=512,
    device=0
)

Pipeline 参数详解

文本生成参数

文本生成是 Pipeline 中最复杂的任务之一，涉及多种解码策略的选择。理解这些参数对于控制生成质量至关重要。

generator = pipeline("text-generation", model="gpt2")

result = generator(
    "Once upon a time",
    max_length=100,           # 最大生成长度
    min_length=20,            # 最小生成长度
    do_sample=True,           # 使用采样而非贪婪解码
    temperature=0.7,          # 温度参数（越高越随机）
    top_k=50,                 # Top-K 采样
    top_p=0.95,               # Top-P (nucleus) 采样
    num_return_sequences=3,   # 返回的序列数量
    repetition_penalty=1.2,   # 重复惩罚
    pad_token_id=50256,       # Padding token ID
)

常用生成策略

贪婪解码：每步选择概率最高的词，适合需要确定性输出的场景。

result = generator("Hello", do_sample=False)

采样生成：从概率分布中随机选择，产生更多样化的输出。

result = generator("Hello", do_sample=True, temperature=0.7)

Top-P 采样：在累积概率达到 P 的词集中采样，是当前最常用的策略。

result = generator("Hello", do_sample=True, top_p=0.9)

关于文本生成策略的详细讲解，包括束搜索、温度调节、Top-K/Top-P 采样等高级技巧，请参阅 文本生成策略 章节。

摘要参数

summarizer = pipeline("summarization")

result = summarizer(
    long_text,
    max_length=150,           # 摘要最大长度
    min_length=30,            # 摘要最小长度
    do_sample=False,          # 贪婪解码
    num_beams=4,              # Beam Search
    early_stopping=True,      # 提前停止
)

自定义 Pipeline

创建自定义任务 Pipeline

from transformers import Pipeline
from transformers.pipelines import PIPELINE_REGISTRY

class MyCustomPipeline(Pipeline):
    def _sanitize_parameters(self, **kwargs):
        # 参数预处理
        return {}, {}, {}
    
    def preprocess(self, inputs, **kwargs):
        # 输入预处理
        return self.tokenizer(inputs, return_tensors="pt")
    
    def _forward(self, model_inputs):
        # 模型推理
        return self.model(**model_inputs)
    
    def postprocess(self, model_outputs, **kwargs):
        # 后处理
        return model_outputs

# 注册自定义 Pipeline
PIPELINE_REGISTRY.register_pipeline(
    "my-custom-task",
    pipeline_class=MyCustomPipeline,
    pt_model=AutoModelForSequenceClassification,
)

性能优化

使用 Fast Tokenizer

# 使用 Rust 实现的高速分词器
classifier = pipeline(
    "sentiment-analysis",
    use_fast=True  # 默认启用
)

启用推理优化

import torch

classifier = pipeline(
    "sentiment-analysis",
    torch_dtype=torch.float16,  # 半精度推理
    model_kwargs={
        "attn_implementation": "flash_attention_2"  # Flash Attention
    }
)

管道并行

from transformers import pipeline
import torch

# 大模型使用设备映射
model_id = "meta-llama/Llama-2-7b-hf"
pipe = pipeline(
    "text-generation",
    model=model_id,
    torch_dtype=torch.float16,
    device_map="auto",  # 自动分配到多个 GPU
)

完整示例

情感分析 Web 服务

from transformers import pipeline
from flask import Flask, request, jsonify

app = Flask(__name__)

# 加载模型
classifier = pipeline("sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english")

@app.route('/analyze', methods=['POST'])
def analyze():
    data = request.json
    text = data.get('text', '')
    
    if not text:
        return jsonify({'error': 'No text provided'}), 400
    
    result = classifier(text)
    return jsonify(result[0])

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

批量处理脚本

from transformers import pipeline
import json

def batch_process(input_file, output_file, batch_size=32):
    # 加载 Pipeline
    classifier = pipeline(
        "sentiment-analysis",
        batch_size=batch_size,
        device=0
    )
    
    # 读取数据
    with open(input_file, 'r', encoding='utf-8') as f:
        texts = [line.strip() for line in f if line.strip()]
    
    # 批量处理
    results = classifier(texts)
    
    # 保存结果
    with open(output_file, 'w', encoding='utf-8') as f:
        for text, result in zip(texts, results):
            f.write(json.dumps({
                'text': text,
                'label': result['label'],
                'score': result['score']
            }, ensure_ascii=False) + '\n')
    
    print(f"处理完成: {len(texts)} 条文本")

# 使用
batch_process('input.txt', 'output.jsonl')

下一步

掌握 Pipeline 后，你可以继续学习：

• 模型加载与使用 - 深入理解模型架构
• 分词器详解 - 掌握文本预处理