Pipeline

[Pipeline]은 간단하지만 강력한 추론 API예요. Hugging Face Hub의 어떤 모델이든 다양한 머신러닝 작업에 즉시 쓸 수 있어요.

작업에 맞는 파라미터로 [Pipeline]을 조정할 수 있어요. 예를 들어 회의록을 받아쓰는 자동 음성 인식(ASR) 파이프라인에 타임스탬프를 추가하는 식이죠. [Pipeline]은 GPU, Apple Silicon, 반정밀도 가중치를 지원해서 추론을 가속하고 메모리를 아껴요.

Transformers에는 제네릭 [Pipeline]과 [TextGenerationPipeline] 같은 많은 작업별 파이프라인, 두 종류의 파이프라인 클래스가 있어요. 개별 파이프라인은 [Pipeline]의 task 파라미터에 작업 식별자를 설정하면 로드돼요. 각 파이프라인의 작업 식별자는 API 문서에서 찾을 수 있어요.

각 작업은 기본 사전 학습 모델과 전처리기를 쓰도록 설정돼 있지만, 다른 모델을 쓰고 싶다면 model 파라미터로 덮어쓸 수 있어요.

예를 들어 [TextGenerationPipeline]을 Gemma 2와 함께 쓰려면 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]을 소개하고, 기능을 보여주며, 다양한 파라미터를 어떻게 설정하는지 알려줄게요.

작업 (Tasks)

[Pipeline]은 다양한 모달리티에 걸친 많은 머신러닝 작업과 호환돼요. 파이프라인에 적절한 입력을 넘기면 나머지는 알아서 처리해요.

다른 작업·모달리티에 [Pipeline]을 쓰는 예시 몇 가지를 볼게요.

자동 음성 인식:

from transformers import pipeline

pipeline = pipeline(task="automatic-speech-recognition", model="openai/whisper-large-v3")
pipeline("https://huggingface.co/datasets/Narsil/asr_dummy/resolve/main/mlk.flac")
{'text': ' I have a dream that one day this nation will rise up and live out the true meaning of its creed.'}

이미지 분류:

from transformers import pipeline

pipeline = pipeline(task="image-classification", model="google/vit-base-patch16-224")
pipeline(images="https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/pipeline-cat-chonk.jpeg")
[{'label': 'lynx, catamount', 'score': 0.43350091576576233},
 {'label': 'cougar, puma, catamount, mountain lion, painter, panther, Felis concolor',
  'score': 0.034796204417943954},
 {'label': 'snow leopard, ounce, Panthera uncia',
  'score': 0.03240183740854263},
 {'label': 'Egyptian cat', 'score': 0.02394474856555462},
 {'label': 'tiger cat', 'score': 0.02288915030658245}]

시각 질의응답:

from transformers import pipeline

pipeline = pipeline(task="visual-question-answering", model="Salesforce/blip-vqa-base")
pipeline(
    image="https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/idefics-few-shot.jpg",
    question="What is in the image?",
)
[{'answer': 'statue of liberty'}]

파라미터

[Pipeline]은 최소한 작업 식별자, 모델, 적절한 입력만 필요해요. 하지만 작업별 파라미터부터 성능 최적화까지 설정할 수 있는 파라미터가 많이 있어요.

이 섹션에서는 좀 더 중요한 파라미터 몇 가지를 소개해요.

장치 (Device)

[Pipeline]은 GPU, CPU, Apple Silicon 등 다양한 하드웨어와 호환돼요. device 파라미터로 하드웨어 종류를 설정해요. device를 설정하지 않으면 기본적으로 [Pipeline]은 모델을 첫 번째 사용 가능한 가속기(CUDA GPU, Apple Silicon MPS, XPU, ...)에 올리고, 가속기가 없을 때만 CPU로 폴백해요. device="cpu"를 넘기면 CPU에서 강제 실행돼요.

GPU에서 [Pipeline]을 실행하려면 device에 해당 CUDA 장치 아이디를 설정해요. 예를 들어 device=0은 첫 GPU에서 실행돼요.

from transformers import pipeline

pipeline = pipeline(task="text-generation", model="google/gemma-2-2b", device=0)
pipeline("the secret to baking a really good cake is ")

분산 훈련용 라이브러리인 Accelerate를 쓰면 모델 가중치를 적절한 장치에 어떻게 로드·저장할지 자동으로 선택하게 할 수도 있어요. 장치가 여러 개일 때 특히 유용해요. Accelerate는 가장 빠른 장치에 가중치를 먼저 로드·저장하고, 필요하면 CPU·하드 드라이브 같은 다른 장치로 옮겨요. device_map="auto"를 설정하면 Accelerate가 장치를 선택해요.

[!TIP] Accelerate가 설치돼 있는지 확인해요.
!pip install -U accelerate

from transformers import pipeline

pipeline = pipeline(task="text-generation", model="google/gemma-2-2b", device_map="auto")
pipeline("the secret to baking a really good cake is ")

Apple Silicon에서 실행하려면 device="mps"를 설정해요.

from transformers import pipeline

pipeline = pipeline(task="text-generation", model="google/gemma-2-2b", device="mps")
pipeline("the secret to baking a really good cake is ")

배치 추론 (Batch inference)

[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은 데이터셋이 반환한 dict에서 항목을 반환하는 유틸리티예요
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 오류를 처리할 수 있는지 꼭 확인하세요.

작업별 파라미터 (Task-specific parameters)

[Pipeline]은 각 개별 작업 파이프라인이 지원하는 모든 파라미터를 받아들여요. 각 작업 파이프라인을 확인해 어떤 파라미터가 가능한지 확인해 보세요. 사용 사례에 유용한 파라미터를 찾지 못했다면 GitHub 이슈를 열어 요청해도 돼요!

아래 예시는 일부 작업별 파라미터를 보여줘요.

자동 음성 인식에서는 return_timestamps="word" 파라미터를 넘기면 각 단어가 언제 말해졌는지 반환해요.

from transformers import pipeline

pipeline = pipeline(task="automatic-speech-recognition", model="openai/whisper-large-v3")
pipeline(audio="https://huggingface.co/datasets/Narsil/asr_dummy/resolve/main/mlk.flac", return_timestamps="word")
{'text': ' I have a dream that one day this nation will rise up and live out the true meaning of its creed.',
 'chunks': [{'text': ' I', 'timestamp': (0.0, 1.1)},
  {'text': ' have', 'timestamp': (1.1, 1.44)},
  {'text': ' a', 'timestamp': (1.44, 1.62)},
  {'text': ' dream', 'timestamp': (1.62, 1.92)},
  {'text': ' that', 'timestamp': (1.92, 3.7)},
  {'text': ' one', 'timestamp': (3.7, 3.88)},
  {'text': ' day', 'timestamp': (3.88, 4.24)},
  {'text': ' this', 'timestamp': (4.24, 5.82)},
  {'text': ' nation', 'timestamp': (5.82, 6.78)},
  {'text': ' will', 'timestamp': (6.78, 7.36)},
  {'text': ' rise', 'timestamp': (7.36, 7.88)},
  {'text': ' up', 'timestamp': (7.88, 8.46)},
  {'text': ' and', 'timestamp': (8.46, 9.2)},
  {'text': ' live', 'timestamp': (9.2, 10.34)},
  {'text': ' out', 'timestamp': (10.34, 10.58)},
  {'text': ' the', 'timestamp': (10.58, 10.8)},
  {'text': ' true', 'timestamp': (10.8, 11.04)},
  {'text': ' meaning', 'timestamp': (11.04, 11.4)},
  {'text': ' of', 'timestamp': (11.4, 11.64)},
  {'text': ' its', 'timestamp': (11.64, 11.8)},
  {'text': ' creed.', 'timestamp': (11.8, 12.3)}]}

텍스트 생성에서는 return_full_text=False를 넘기면 전체 텍스트(프롬프트 + 생성 텍스트) 대신 생성된 텍스트만 반환해요.

[~TextGenerationPipeline.__call__]은 [~GenerationMixin.generate] 메서드의 추가 키워드 인자도 지원해요. 생성 시퀀스를 둘 이상 반환하려면 num_return_sequences를 1보다 큰 값으로 설정해요.

from transformers import pipeline

pipeline = pipeline(task="text-generation", model="openai-community/gpt2")
pipeline("the secret to baking a good cake is", num_return_sequences=4, return_full_text=False)
[{'generated_text': ' how easy it is for me to do it with my hands. You must not go nuts, or the cake is going to fall out.'},
 {'generated_text': ' to prepare the cake before baking. The key is to find the right type of icing to use and that icing makes an amazing frosting cake.\n\nFor a good icing cake, we give you the basics'},
 {'generated_text': " to remember to soak it in enough water and don't worry about it sticking to the wall. In the meantime, you could remove the top of the cake and let it dry out with a paper towel.\n"},
 {'generated_text': ' the best time to turn off the oven and let it stand 30 minutes. After 30 minutes, stir and bake a cake in a pan until fully moist.\n\nRemove the cake from the heat for about 12'}]

청크 배칭 (Chunk batching)

데이터를 청크로 처리해야 하는 경우가 있어요.

어떤 데이터 타입은 단일 입력(예: 아주 긴 오디오 파일)을 처리하기 전에 여러 부분으로 나눠야 할 수 있어요.
zero-shot 분류나 질의응답 같은 일부 작업에서는 단일 입력에 여러 번의 forward pass가 필요해서 batch_size 파라미터에 문제가 생길 수 있어요.

ChunkPipeline 클래스는 이런 사용 사례를 처리하도록 설계됐어요. 두 파이프라인 클래스 모두 같은 방식으로 쓰지만, ChunkPipeline이 배칭을 자동 처리하므로 입력이 몇 번의 forward pass를 유발하는지 신경 쓸 필요가 없어요. 대신 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)

대형 데이터셋 (Large datasets)

대형 데이터셋으로 추론할 때는 데이터셋 자체를 직접 순회할 수 있어요. 이러면 전체 데이터셋에 대한 메모리를 즉시 할당하지 않아도 되고, 배치를 직접 만들 걱정도 사라져요. 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"])

대형 모델 (Large models)

Accelerate는 [Pipeline]으로 대형 모델을 실행할 때 유용한 몇 가지 최적화를 가능하게 해요. 먼저 Accelerate를 설치해 두세요.

!pip install -U accelerate

device_map="auto" 설정은 모델을 가장 빠른 장치(GPU)에 먼저 자동 분배하고, 가능하면 그다음 느린 장치(CPU, 하드 드라이브)로 보내는 데 유용해요.

[Pipeline]은 반정밀도 가중치(torch.float16)를 지원하는데, 훨씬 빠르고 메모리를 절약할 수 있어요. 대부분의 모델, 특히 큰 모델에서는 성능 손실이 무시할 만해요. 하드웨어가 지원한다면 더 넓은 범위를 위해 torch.bfloat16을 활성화할 수도 있어요.

[!TIP] 입력은 내부적으로 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'}]

출처: Hugging Face Transformers — Pipeline

더 알아보기 (Learn more)

[Pipeline] API 레퍼런스 — 지원 작업과 파라미터 전체 목록
Accelerate — 분산 훈련·대형 모델 최적화
Quantization — 양자화 백엔드와 설정