パイプライン

Pipelineは、Hugging Face Hubのあらゆるモデルで、多様な機械学習タスクにすぐに使える、シンプルかつ強力な推論APIです。

タスク固有のパラメータ、例えば会議メモの文字起こし用の自動音声認識（ASR）パイプラインにタイムスタンプを追加することなどにより、Pipelineをタスクに合わせて調整できます。Pipelineは、推論の高速化とメモリ節約のために、GPU、Apple Silicon、および半精度の重みをサポートしています。

Transformersには、汎用的なPipelineと、TextGenerationPipelineのようなタスク固有のパイプラインが多数用意されています。これらの個別のパイプラインを読み込むには、Pipelineのtaskパラメータにタスク識別子を指定します。各パイプラインのタスク識別子は、それぞれのAPIドキュメントで確認できます。

各タスクは、デフォルトの事前学習済みモデルとプリプロセッサを使用するように設定されていますが、別のモデルを使用する場合は、model パラメータで上書きできます。

例えば、Gemma 2でTextGenerationPipelineを使用するには、task="text-generation"とmodel="google/gemma-2-2b"を設定します。

from transformers import pipeline

pipeline = pipeline(task="text-generation", model="google/gemma-2-2b")
pipeline("the secret to baking a really good cake is ")
[{'generated_text': 'the secret to baking a really good cake is 1. the right ingredients 2. the'}]

入力データが複数ある場合は、リストとして渡します。

from transformers import pipeline
from accelerate import Accelerator

device = Accelerator().device

pipeline = pipeline(task="text-generation", model="google/gemma-2-2b", device=device)
pipeline(["the secret to baking a really good cake is ", "a baguette is "])
[[{'generated_text': 'the secret to baking a really good cake is 1. the right ingredients 2. the'}],
 [{'generated_text': 'a baguette is 100% bread.\n\na baguette is 100%'}]]

このガイドでは、Pipelineの概要を紹介し、その機能や各種パラメータの設定方法について説明します。

タスク

Pipelineは、さまざまなモダリティにわたる多くの機械学習タスクに対応しています。パイプラインに適切な入力を渡すだけで、残りはすべて処理されます。

ここでは、Pipelineをさまざまなタスクやモダリティに適用する例をいくつか示します。

自動音声認識

画像分類

視覚的質問応答

パラメータ

最低限、Pipelineにはタスク識別子、モデル、および適切な入力が必要です。ただし、タスク固有のパラメータからパフォーマンスの最適化まで、パイプラインを設定するための多くのパラメータが利用可能です。

このセクションでは、より重要なパラメータの一部をご紹介します。

デバイス

Pipelineは、GPU、CPU、Apple Silicon など、多くのハードウェアタイプに対応しています。ハードウェアタイプはdeviceパラメータで設定します。デフォルトでは、deviceが設定されていない場合、Pipelineはモデルを最初に見つかったアクセラレーター（CUDA GPU、Apple Silicon MPS、XPU など）に自動で配置し、アクセラレーターが利用できない場合のみ CPU にフォールバックします。CPU での実行を強制するには、device="cpu"を指定してください。

GPU

Apple silicon

バッチ推論

Pipelineは、batch_sizeパラメータを使用して、入力の一括処理（バッチ処理）も実行できます。バッチ推論は、特にGPU上で速度の向上につながる場合がありますが、必ずしも保証されるものではありません。ハードウェア、データ、モデル自体などの他の変数も、バッチ推論が速度向上に寄与するかどうかに影響を与える可能性があります。そのため、バッチ推論はデフォルトで無効になっています。

以下の例では、入力数が4つでbatch_sizeが2に設定されている場合、Pipelineは一度に2つの入力からなるバッチをモデルに渡します。

from transformers import pipeline
from accelerate import Accelerator

device = Accelerator().device

pipeline = pipeline(task="text-generation", model="google/gemma-2-2b", device=device, batch_size=2)
pipeline(["the secret to baking a really good cake is", "a baguette is", "paris is the", "hotdogs are"])
[[{'generated_text': 'the secret to baking a really good cake is to use a good cake mix.\n\ni’'}],
 [{'generated_text': 'a baguette is'}],
 [{'generated_text': 'paris is the most beautiful city in the world.\n\ni’ve been to paris 3'}],
 [{'generated_text': 'hotdogs are a staple of the american diet. they are a great source of protein and can'}]]

Pipelineでストリーミングデータを処理する場合も、バッチ推論の優れたユースケースの一つです。

from transformers import pipeline
from accelerate import Accelerator
from transformers.pipelines.pt_utils import KeyDataset
import datasets

device = Accelerator().device

# KeyDataset is a utility that returns the item in the dict returned by the dataset
dataset = load_dataset("stanfordnlp/imdb", name="plain_text", split="unsupervised")
pipeline = pipeline(task="text-classification", model="distilbert/distilbert-base-uncased-finetuned-sst-2-english", device=device)
for out in pipeline(KeyDataset(dataset, "text"), batch_size=8, truncation="only_first"):
    print(out)

バッチ推論がパフォーマンスの向上に寄与するかどうかを判断する際は、以下の一般的な指針を念頭に置いてください。

確実に知る唯一の方法は、ご自身のモデル、データ、ハードウェア上でパフォーマンスを測定することです。
レイテンシが制約となる場合（ライブ推論プロダクトなど）は、推論のバッチ処理を行わないでください。
CPU を使用している場合は、推論のバッチ処理を行わないでください。
データのsequence_lengthが不明な場合は、推論のバッチ処理を行わないでください。パフォーマンスを測定し、sequence_lengthに段階的に追加していき、失敗からの復旧のためにメモリ不足（OOM）チェックを含めてください。
sequence_lengthが一定の場合はバッチ推論を行い、OOMエラーが発生するまでバッチサイズを大きくし続けてください。GPUが大きいほど、バッチ推論の恩恵は大きくなります。
バッチ推論を行うことを決めた場合は、OOMエラーを処理できることを必ず確認してください。

タスク固有のパラメータ

Pipeline は、各タスクのパイプラインがサポートするパラメータをすべて受け取ります。利用可能なパラメータの種類を確認するには、各タスクのパイプラインを必ずチェックしてください。ユースケースに有用なパラメータが見つからない場合は、GitHub の issue を開いてリクエストしてください。

以下の例では、利用可能なタスク固有のパラメータの一部を示しています。

自動音声認識

テキスト生成

チャンクバッチング

データをチャンク単位で処理する必要がある場合があります。

一部のデータ型では、単一の入力（非常に長い音声ファイルなど）を処理する前に、複数のチャンクに分割する必要がある場合があります。
ゼロショット分類や質問応答などのタスクでは、1つの入力に対して複数のフォワードパスが必要になることがあり、batch_sizeパラメータに問題を引き起こす可能性があります。

ChunkPipeline クラスは、これらのユースケースに対応するために設計されています。両方のパイプラインクラスは同じ方法で使用されますが、ChunkPipeline はバッチ処理を自動的に処理できるため、入力によってトリガーされるフォワードパスの数を気にする必要はありません。代わりに、入力とは独立して batch_size を最適化できます。

以下の例は、Pipelineとの違いを示しています。

# ChunkPipeline
all_model_outputs = []
for preprocessed in pipeline.preprocess(inputs):
    model_outputs = pipeline.model_forward(preprocessed)
    all_model_outputs.append(model_outputs)
outputs =pipeline.postprocess(all_model_outputs)

# Pipeline
preprocessed = pipeline.preprocess(inputs)
model_outputs = pipeline.forward(preprocessed)
outputs = pipeline.postprocess(model_outputs)

大規模なデータセット

大規模なデータセットでの推論では、データセット自体を直接イテレートできます。これにより、データセット全体に対してメモリを即座に割り当てることを避け、バッチの作成を自分で気にする必要もなくなります。パフォーマンスが向上するかどうかを確認するために、batch_size パラメータを指定してバッチ推論を試してください。

from transformers.pipelines.pt_utils import KeyDataset
from transformers import pipeline
from accelerate import Accelerator
from datasets import load_dataset

device = Accelerator().device

dataset = load_dataset("stanfordnlp/imdb", name="plain_text", split="unsupervised")
pipeline = pipeline(task="text-classification", model="distilbert/distilbert-base-uncased-finetuned-sst-2-english", device=device)
for out in pipeline(KeyDataset(dataset, "text"), batch_size=8, truncation="only_first"):
    print(out)

Pipeline を用いて大規模データセットで推論を実行する他の方法としては、イテレータやジェネレータを使用する方法があります。

def data():
    for i in range(1000):
        yield f"My example {i}"

pipeline = pipeline(model="openai-community/gpt2", device=0)
generated_characters = 0
for out in pipeline(data()):
    generated_characters += len(out[0]["generated_text"])

大規模モデル

Accelerateは、Pipelineで大型モデルを実行するためのいくつかの最適化を可能にします。まず、Accelerateがインストールされていることを確認してください。

!pip install -U accelerate

device_map="auto" の設定は、利用可能な場合、より高速なデバイス（GPU）に優先的にモデルを分散配置し、その後、より低速なデバイス（CPU、ハードドライブ）に割り当てるのに役立ちます。

Pipeline は半精度の重み（torch.float16）をサポートしており、大幅な高速化とメモリ削減が期待できます。ほとんどのモデル、特に大規模なモデルでは、性能の低下はほぼ無視できる程度です。ハードウェアが対応している場合は、より広い範囲を扱える torch.bfloat16 を代わりに有効にできます。

入力値は内部でtorch.float16に変換され、PyTorch バックエンドを持つモデルでのみ動作します。

最後に、Pipelineは量子化モデルも受け入れるため、メモリ使用量をさらに削減できます。まず、bitsandbytesライブラリがインストールされていることを確認し、その後、パイプラインのmodel_kwargsにquantization_configを追加してください。

import torch
from transformers import pipeline, BitsAndBytesConfig

pipeline = pipeline(model="google/gemma-7b", dtype=torch.bfloat16, device_map="auto", model_kwargs={"quantization_config": BitsAndBytesConfig(load_in_8bit=True)})
pipeline("the secret to baking a good cake is ")
[{'generated_text': 'the secret to baking a good cake is 1. the right ingredients 2. the right'}]

Update on GitHub