LLM 서비스 운영 방법 — 프로덕션 AI 서비스 안정화 가이드
LLM 기반 서비스를 프로덕션에서 안정적으로 운영하는 방법. 비용 관리, 레이턴시 최적화, 장애 대응, 모니터링까지 실전 경험 정리.
TestForge Team ·
LLM 서비스의 특수한 도전
일반 API와 다른 점:
- 레이턴시: 수 초~수십 초 (스트리밍 없으면 사용자 이탈)
- 비용: 트래픽 × 토큰 수 = 예측 불가 폭탄
- 비결정성: 같은 입력도 다른 출력
- Rate Limit: API 제공사의 분당 요청/토큰 제한
1. 스트리밍으로 UX 개선
사용자가 결과를 기다리는 체감을 줄이는 가장 효과적인 방법입니다.
# FastAPI SSE 스트리밍
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from anthropic import Anthropic
client = Anthropic()
app = FastAPI()
@app.post("/chat/stream")
async def chat_stream(message: str):
async def generate():
with client.messages.stream(
model="claude-haiku-4-5-20251001",
max_tokens=1024,
messages=[{"role": "user", "content": message}]
) as stream:
for text in stream.text_stream:
yield f"data: {text}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
// 프론트엔드
const response = await fetch('/chat/stream', { method: 'POST', body: ... });
const reader = response.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const text = new TextDecoder().decode(value);
appendToUI(text);
}
2. 프롬프트 캐싱으로 비용 절감
같은 시스템 프롬프트를 반복 전송하는 비용을 줄입니다.
# Anthropic Prompt Caching
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=[
{
"type": "text",
"text": long_system_prompt, # 수천 토큰의 고정 프롬프트
"cache_control": {"type": "ephemeral"} # 5분 캐시
}
],
messages=[{"role": "user", "content": user_message}]
)
# 캐시 히트 시 입력 비용 90% 절감
3. 모델 라우팅 전략
모든 요청에 비싼 모델을 쓸 필요는 없습니다.
def select_model(task_type: str, complexity: int) -> str:
"""
complexity: 1(단순) ~ 5(복잡)
"""
if task_type == "classification" or complexity <= 2:
return "claude-haiku-4-5-20251001" # 빠르고 저렴
elif complexity <= 4:
return "claude-sonnet-4-6" # 균형
else:
return "claude-opus-4-6" # 복잡한 추론
# 비용 비교 (Input/Output per 1M tokens)
# Haiku: $0.80 / $4
# Sonnet: $3 / $15
# Opus: $15 / $75
4. Rate Limit 대응
import asyncio
from tenacity import retry, wait_exponential, retry_if_exception_type
from anthropic import RateLimitError, APIConnectionError
@retry(
retry=retry_if_exception_type((RateLimitError, APIConnectionError)),
wait=wait_exponential(multiplier=1, min=2, max=60),
stop=stop_after_attempt(5),
)
async def call_llm_with_retry(messages: list) -> str:
response = await client.messages.create(...)
return response.content[0].text
# 동시 요청 제한 (Semaphore)
semaphore = asyncio.Semaphore(10) # 최대 10개 동시 요청
async def safe_llm_call(messages):
async with semaphore:
return await call_llm_with_retry(messages)
5. 비용 모니터링
# 요청마다 토큰 사용량 기록
import time
async def tracked_llm_call(user_id: str, messages: list):
start = time.time()
response = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=messages,
)
# 비용 계산
input_cost = response.usage.input_tokens * 3 / 1_000_000
output_cost = response.usage.output_tokens * 15 / 1_000_000
# 메트릭 기록
metrics.record({
"user_id": user_id,
"model": "claude-sonnet-4-6",
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens,
"cost_usd": input_cost + output_cost,
"latency_ms": (time.time() - start) * 1000,
})
return response
6. Fallback 전략
MODELS_BY_PRIORITY = [
"claude-sonnet-4-6", # 1순위
"claude-haiku-4-5-20251001", # Fallback
]
async def call_with_fallback(messages: list) -> str:
for model in MODELS_BY_PRIORITY:
try:
response = await client.messages.create(
model=model, messages=messages, max_tokens=1024
)
return response.content[0].text
except RateLimitError:
continue # 다음 모델 시도
raise RuntimeError("All models rate limited")
7. 모니터링 대시보드 필수 지표
| 지표 | 정상 | 알림 기준 |
|---|---|---|
| 평균 레이턴시 | < 3초 | > 10초 |
| P99 레이턴시 | < 15초 | > 30초 |
| 에러율 | < 1% | > 5% |
| Rate Limit 비율 | < 0.1% | > 1% |
| 일일 비용 | 예산 내 | 예산 80% 초과 |
| 평균 토큰/요청 | 기준치 내 | 2배 초과 |
운영 체크리스트
- 스트리밍 응답 구현 (UX)
- 프롬프트 캐싱 적용 (비용)
- 재시도 + 지수 백오프 (안정성)
- 동시 요청 수 제한 (Rate Limit 방어)
- 모델별 비용 트래킹
- 일일 비용 알림 설정
- Fallback 모델 지정
- 타임아웃 설정 (30초)
- 프롬프트 인젝션 방어