대규모 언어모델(LLM)을 실제 서비스 환경에 배포할 때 가장 먼저 직면하는 문제는 추론 속도, GPU 메모리 관리, 동시 요청 처리입니다. 이러한 문제를 해결을 좀더 쉽게 처리하기 위해 다양한 추론 최적화 도구들이 소개 되고 있습니다. 다만, 다른 분야 패키지에 비해 지배적으로 사용되고 있는 추론 최적화 도구가 없을 뿐만 아니라 이에 대한 파이썬 코드 작성 스타일 또한 관련 파이썬 코드 작성 방식도 개발자마다 상이해 일관된 스타일에 기반한 협업이 쉽지 않은 상황입니다. 이러한 환경을 고려하여, 추론 최적화 도구인 vLLM을 추천하고 추천하는 이유를 본 포스팅에서 설명합니다. 아울러, 통일된 스타일을 제안하기 위한 코드 예시와 vLLM의 한계에 따른 주의점까지 본포스팅에서 다루고자 합니다.
vLLM 활용 이유
- PagedAttention 기반의 고효율 메모리 관리
- TensorRT-LLM 등 대체 솔루션에 비해, Python 기반 API가 단순하고 일관적임
- SSE 기반 스트리밍에 최적화된 구조 제공
- Hugging Face 생태계와 완전한 호환성
vLLM 코드 스타일 제안
아래 코드는 vllm을 활용하여 sse 통신이 가능하도록 작성해둔 코드 이며, 객체 지향으로 작성하여 효율성을 높이고자 노력하였습니다. 아래 코드 스타일을 참고하여 주세요. 단 출력 형태는 SSE(Server-Sent Events) 출력시 출력 표준제안 를 참고하여 코드를 완성하였습니다.
import asyncio
import json
import traceback
import uuid
from dataclasses import dataclass, field
from typing import Any, AsyncGenerator, Dict, Optional
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from transformers import AutoTokenizer
from vllm.engine.arg_utils import AsyncEngineArgs
from vllm.engine.async_llm_engine import AsyncLLMEngine
from vllm.sampling_params import SamplingParams
# --- 1. 설정 관리 클래스 (Configuration) ---
@dataclass
class VLLMConfig:
"""vLLM 엔진 및 모델 설정을 관리하는 데이터 클래스"""
model_path: str = "{model_path}"
dtype: str = "bfloat16"
gpu_memory_utilization: float = 0.75
tensor_parallel_size: int = 2
max_num_seqs: int = 16
max_model_len: int = 4096
limit_mm_per_prompt: Dict[str, int] = field(default_factory=lambda: {"image": 0})
trust_remote_code: bool = True
# --- 2. 엔진 매니저 클래스 (Engine Manager) ---
class VLLMEngineManager:
"""vLLM 엔진의 생명주기(초기화, 접근)를 담당하는 클래스"""
def __init__(self, config: VLLMConfig):
self.config = config
self._engine: Optional[AsyncLLMEngine] = None
self._tokenizer = None
self._lock = asyncio.Lock()
async def get_engine(self) -> AsyncLLMEngine:
"""싱글톤 패턴을 사용하여 엔진 인스턴스를 반환 (Double-checked locking)"""
if self._engine is not None:
return self._engine
async with self._lock:
if self._engine is not None:
return self._engine
print(">>> Initializing vLLM Engine...")
try:
engine_args = AsyncEngineArgs(
model=self.config.model_path,
dtype=self.config.dtype,
gpu_memory_utilization=self.config.gpu_memory_utilization,
tensor_parallel_size=self.config.tensor_parallel_size,
max_num_seqs=self.config.max_num_seqs,
max_model_len=self.config.max_model_len,
limit_mm_per_prompt=self.config.limit_mm_per_prompt,
trust_remote_code=self.config.trust_remote_code,
disable_log_stats=False
)
self._engine = AsyncLLMEngine.from_engine_args(engine_args)
print(">>> vLLM Engine is ready.")
except Exception as e:
print(f"!!! Engine initialization failed: {e}")
raise e
return self._engine
def get_tokenizer(self):
"""토크나이저 로드 (Lazy Loading)"""
if self._tokenizer is None:
try:
self._tokenizer = AutoTokenizer.from_pretrained(self.config.model_path)
except Exception as e:
print(f"!!! Tokenizer loading failed: {e}")
raise e
return self._tokenizer
# --- 3. 채팅 서비스 클래스 (Service Logic) ---
class ChatCompletionService:
"""사용자 요청을 처리하고 스트리밍 응답을 생성하는 서비스 클래스"""
def __init__(self, engine_manager: VLLMEngineManager):
self.engine_manager = engine_manager
def _build_prompt(self, question: str) -> str:
"""프롬프트 템플릿 구성"""
tokenizer = self.engine_manager.get_tokenizer()
# System Prompt를 User 메시지에 병합 (Gemma 호환성)
system_instruction = (
"You are Gemma-3, a helpful, honest, and concise AI assistant. "
"Answer clearly, avoid hallucinations, and think step-by-step internally."
"Ask for clarification when needed and follow the user’s instructions precisely."
)
conversation = [
{"role": "user", "content": f"{system_instruction}\n\nQuestion: {question}"}
]
try:
#return tokenizer.apply_chat_template(
# conversation,
# tokenize=False,
# add_generation_prompt=True
#)
return (
f"<bos><start_of_turn>system\n{system_instruction}<end_of_turn>\n"
f"<start_of_turn>user\n{question}<end_of_turn>\n"
f"<start_of_turn>model\n"
)
except Exception as e:
print(f"Template apply failed, using raw string: {e}")
return f"<start_of_turn>user\n{question}<end_of_turn>\n<start_of_turn>model\n"
async def generate_response_sse(self, question: str, request_id: str = None) -> AsyncGenerator[str, None]:
"""SSE 포맷으로 답변 스트리밍"""
if request_id is None:
request_id = str(uuid.uuid4())
try:
engine = await self.engine_manager.get_engine()
prompt = self._build_prompt(question)
print("QUESTION")
print(prompt)
sampling_params = SamplingParams(
max_tokens=2048,
temperature=0.5,
top_p=0.9,
repetition_penalty=1,
skip_special_tokens=True
)
results_generator = engine.generate(
prompt,
sampling_params,
request_id=request_id
)
generated_text = ""
async for request_output in results_generator:
for output in request_output.outputs:
new_text = output.text
if len(new_text) > len(generated_text):
delta = new_text[len(generated_text):]
generated_text = new_text
# SSE Event: message
yield "event: message\n"
yield f"data: {json.dumps({'text': delta}, ensure_ascii=False)}\n\n"
# SSE Event: DONE
yield "event: done\n"
yield "data: [DONE]\n\n"
except Exception as e:
traceback.print_exc()
yield "event: error\n"
yield f"data: {json.dumps({'error': str(e)}, ensure_ascii=False)}\n\n"
# 1. 설정 로드
config = VLLMConfig() # 필요 시 인자로 설정 변경 가능
# 2. 매니저 및 서비스 인스턴스화 (Dependency Injection)
engine_manager = VLLMEngineManager(config)
chat_service = ChatCompletionService(engine_manager)
print(">>> Starting Stream Process...")
app = FastAPI(title="Gemma-3 Chat API", version="1.0.0")
class ChatRequest(BaseModel):
question: str
@app.post("/chat")
async def chat(request: ChatRequest):
# 3. 서비스 실행
return StreamingResponse(
chat_service.generate_response_sse(request.question),
media_type="text/event-stream",
)
vLLM 활용시 주의점
vLLM은 서버에 장착된 GPU 자원을 최대한 활용해 추론 엔진을 초기화합니다. 이 과정에서 지정된 GPU를 독점적으로 점유하기 때문에, 동일한 GPU를 사용하는 다른 프로세스나 서비스와 자원 충돌이 발생할 수 있습니다. 따라서 운영 환경에서 vLLM을 사용할 때는 GPU에 상주하는 모델 수를 최소화하는 것이 중요합니다. 기본적으로 GPU 개수만큼만 모델 엔진이 생성되므로, 불필요한 모델 로딩을 피해야 합니다.
프롬프트 엔지니어링 기반의 파이프라인을 사용하는 경우라면 이미 로딩된 vanilla 모델을 재사용해 불필요한 모델 초기화를 방지할 수 있습니다. 반면, 파인튜닝이 필요한 경우에는 LoRA 방식을 적용하는 것이 효율적입니다. 예를 들어 저는 Gemma 모델에 LoRA 가중치 두 개를 동시에 로드해두고, 조건문을 통해 필요한 가중치만 선택해 사용하는 방식으로 운영합니다.
Release Note
2025-12-10 - ver 0.1.0
- 초안 업로드