모델 서빙: FastAPI, TorchServe, vLLM

.pth 파일에서 HTTP 엔드포인트까지

운영 복잡도가 낮은 순서로 세 가지 서빙 방식을 살펴보자:

FastAPI + uvicorn: 서버를 직접 작성해. 인프라는 단순하면서 유연성은 가장 커서 시제품, 내부 도구, 작은 운영 환경에 좋아.
TorchServe: AWS와 Meta가 만든 관리형 추론 서버야. 배치 처리, 모델 버전 관리, 평가지표, 여러 모델 서빙 기능을 내장해. 운영 환경에서 여러 모델을 관리해야 할 때 알맞아.
vLLM / TGI / SGLang: LLM 전용 서빙 도구야. 연속 배치 처리, PagedAttention, 최적화된 커널을 제공해 언어 모델을 서빙할 때 잘 맞아.

모든 서빙 구성에 필요한 기본 요소

서비스 시작 시 모델을 한 번만 불러오고 요청마다 다시 불러오지 마.
model.eval()을 설정하고 매 요청을 torch.inference_mode() 안에서 처리해.
지연 시간 예산이 허용하면 요청을 배치로 묶어 처리량을 크게 높여.
요청 처리 중 CPU와 GPU의 불필요한 동기화를 피해. 버퍼를 미리 할당하고 non_blocking 전송을 사용해.
상태 확인 엔드포인트가 오케스트레이터에 서비스가 살아 있는지 알려 주게 해.
평가지표 엔드포인트에서 지연 시간, 처리량, 오류율, GPU 활용도를 제공해.

Code

FastAPI: 가장 단순 HTTP 서버·python

# pip install fastapi uvicorn[standard]
from fastapi import FastAPI
from pydantic import BaseModel
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification

app = FastAPI()

# Load once at startup
model_name = "distilbert-base-uncased-finetuned-sst-2-english"
tok = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(model_name)
model.eval()
device = "cuda" if torch.cuda.is_available() else "cpu"
model = model.to(device)
labels = ['NEGATIVE', 'POSITIVE']

class Req(BaseModel):
    text: str

class Resp(BaseModel):
    label: str
    score: float

@app.post("/predict", response_model=Resp)
@torch.inference_mode()
async def predict(req: Req) -> Resp:
    inputs = tok(req.text, return_tensors='pt', truncation=True, max_length=512).to(device)
    logits = model(**inputs).logits
    probs = logits.softmax(-1)[0]
    idx = int(probs.argmax())
    return Resp(label=labels[idx], score=float(probs[idx]))

@app.get('/health')
async def health(): return {'status': 'ok'}

# Run: uvicorn app:app --host 0.0.0.0 --port 8000

TorchServe: 운영 환경급 옵션·python

# Bash, not Python — TorchServe is a separate process

# 1. Package model
# torch-model-archiver \
#     --model-name resnet50 \
#     --version 1.0 \
#     --model-file resnet_model.py \
#     --serialized-file resnet50.pth \
#     --handler image_classifier \
#     --export-path model_store \
#     --force

# 2. Start server
# torchserve --start --model-store model_store --models resnet50=resnet50.mar

# 3. Call API
# curl -X POST http://localhost:8080/predictions/resnet50 -T image.jpg

# Built-ins: batching, scaling, metrics, model versioning, multi-model
# Drawback: more setup vs FastAPI; opinionated about the handler interface

vLLM: LLM에 맞춘 옵션·python

# pip install vllm
# vLLM is the de facto LLM serving runtime — continuous batching, PagedAttention,
# OpenAI-compatible API.

# Start a server (CLI)
# python -m vllm.entrypoints.openai.api_server \
#     --model meta-llama/Llama-3.2-3B-Instruct \
#     --host 0.0.0.0 --port 8000

# Or programmatically
from vllm import LLM, SamplingParams

llm = LLM(model="gpt2")
sampling = SamplingParams(temperature=0.7, max_tokens=64)

prompts = ["Once upon a time", "The capital of France is"]
outputs = llm.generate(prompts, sampling)

for output in outputs:
    print(output.prompt, '→', output.outputs[0].text)

단순한 요청 묶음 처리: 지연 시간 예산 허락 시·python

import asyncio
import torch
from collections import defaultdict
from fastapi import FastAPI

app = FastAPI()
queue = asyncio.Queue()

async def batched_worker():
    while True:
        # Pull up to 32 requests OR wait 20ms, whichever first
        first = await queue.get()
        batch = [first]
        try:
            for _ in range(31):
                more = await asyncio.wait_for(queue.get(), timeout=0.02)
                batch.append(more)
        except asyncio.TimeoutError:
            pass

        # Run model.forward on the whole batch
        inputs = [b['input'] for b in batch]
        with torch.inference_mode():
            outputs = model(inputs)              # batched call

        # Resolve each request's future
        for b, out in zip(batch, outputs):
            b['future'].set_result(out)

@app.on_event("startup")
async def start_worker():
    asyncio.create_task(batched_worker())

@app.post("/predict")
async def predict(text: str):
    fut = asyncio.get_event_loop().create_future()
    await queue.put({'input': text, 'future': fut})
    return await fut

Exercise

작은 분류기 하나를 FastAPI 엔드포인트로 감싸 봐. HF distilbert나 직접 만든 TinyMLP 등 무엇이든 좋아. uvicorn으로 실행하고 여러 입력을 /predict 엔드포인트에 curl로 보내 응답을 검증해. /health 엔드포인트도 추가해. 첫 요청과 이후 요청의 시간을 비교하면 초기 준비 비용을 확인할 수 있어.

.pth 파일에서 HTTP 엔드포인트까지

모든 서빙 구성에 필요한 기본 요소

Code

External links

Exercise

Progress

댓글 0