반응형
1편에서 기본 호출까지 했습니다. 2편은 프로덕션에서 쓰는 패턴입니다. 프로바이더가 죽어도 자동 전환하고, 여러 배포를 밸런싱하고, 토큰 비용을 실시간으로 추적하고, 예산을 초과하면 자동으로 막습니다.
[2편 핵심 요약]
→ 단순 폴백: completion()의 fallbacks 파라미터 — 가장 빠른 방법
→ Router: 여러 배포를 하나의 모델 그룹으로 묶어 관리하는 핵심 클래스
→ 라우팅 전략: simple-shuffle(기본) / least-busy / latency-based / usage-based
→ 3종 폴백: 일반 실패 / 컨텍스트 초과 / 콘텐츠 정책 위반 각각 따로 설정
→ order 파라미터: 배포 우선순위 — order=1 실패 시 order=2로 자동 에스컬레이션
→ completion_cost(): 요청당 실제 비용 계산 (달러)
→ max_budget: 전역 예산 한도 — 초과 시 BudgetExceededError
→ 캐싱: 동일 프롬프트 반복 요청 시 API 호출 없이 캐시 반환
실전 1 — 단순 폴백 (가장 빠른 방법)
Router 없이 completion() 파라미터 하나로 폴백 체인을 만듭니다.
import litellm
import os
# ── 기본 폴백 ──────────────────────────────────────────
response = litellm.completion(
model="anthropic/claude-opus-4-7", # 1순위
messages=[{"role": "user", "content": "안녕"}],
fallbacks=[
"anthropic/claude-sonnet-4-6", # 2순위 (Opus 실패 시)
"openai/gpt-5.4", # 3순위
"google/gemini-3-flash", # 4순위 (마지막 보루)
],
num_retries=2, # 각 모델 최대 2번 재시도
timeout=30,
)
print(response.choices[0].message.content)
print(f"실제 사용 모델: {response.model}")
# ── 에러 유형별 폴백 분리 ─────────────────────────────
response = litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "매우 긴 문서 분석..."}],
# 일반 실패 폴백 (429, 500, timeout 등)
fallbacks=["openai/gpt-5.4", "google/gemini-3-flash"],
# 컨텍스트 초과 폴백 → 더 큰 컨텍스트 모델로
context_window_fallbacks=[
{"anthropic/claude-sonnet-4-6": ["google/gemini-3.1-pro-preview"]},
# Sonnet(200K) 초과 시 → Gemini(1M)으로
],
# 콘텐츠 정책 위반 폴백 (Azure 필터 등)
content_policy_fallbacks=[
{"anthropic/claude-sonnet-4-6": ["openai/gpt-5.4"]},
],
)
# ── 재시도 전략 세부 설정 ─────────────────────────────
response = litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "안녕"}],
num_retries=3,
# 재시도 간격: exponential backoff 자동 적용
# 1차: 1s → 2차: 2s → 3차: 4s
)
실전 2 — Router: 프로덕션 핵심 클래스
Router는 여러 배포를 하나의 모델 그룹으로 묶어 자동 분산·폴백합니다. 단순 completion()보다 훨씬 세밀한 제어가 가능합니다.
기본 Router 세팅
from litellm import Router
# 모델 그룹 정의
model_list = [
# ── Claude 그룹 (claude라는 이름으로 묶음) ────────
{
"model_name": "claude", # 호출 시 쓸 이름
"litellm_params": {
"model": "anthropic/claude-sonnet-4-6",
"api_key": os.environ["ANTHROPIC_API_KEY"],
},
"rpm": 100, # 분당 요청 한도
"tpm": 100000, # 분당 토큰 한도
},
{
"model_name": "claude", # 같은 그룹 — 자동 로드밸런싱
"litellm_params": {
"model": "bedrock/anthropic.claude-sonnet-4-6",
"aws_access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
"aws_secret_access_key": os.environ["AWS_SECRET_ACCESS_KEY"],
"aws_region_name": "us-east-1",
},
"rpm": 50,
},
# ── GPT 그룹 ──────────────────────────────────────
{
"model_name": "gpt",
"litellm_params": {
"model": "openai/gpt-5.4",
"api_key": os.environ["OPENAI_API_KEY"],
},
"rpm": 200,
},
# ── 저렴한 폴백 모델 ──────────────────────────────
{
"model_name": "fallback-cheap",
"litellm_params": {
"model": "google/gemini-3-flash",
"api_key": os.environ["GEMINI_API_KEY"],
},
},
]
router = Router(
model_list=model_list,
routing_strategy="simple-shuffle", # 기본값
num_retries=2,
timeout=30,
# 폴백 설정
fallbacks=[
{"claude": ["gpt", "fallback-cheap"]}, # claude 실패 시
{"gpt": ["claude", "fallback-cheap"]}, # gpt 실패 시
],
context_window_fallbacks=[
{"claude": ["google/gemini-3.1-pro-preview"]}, # 컨텍스트 초과 시
],
)
# 사용 — completion()과 동일한 인터페이스
response = router.completion(
model="claude", # 그룹 이름으로 호출
messages=[{"role": "user", "content": "안녕"}]
)
print(response.choices[0].message.content)
# 비동기
response = await router.acompletion(
model="claude",
messages=[{"role": "user", "content": "안녕"}]
)
실전 3 — 라우팅 전략 4가지
from litellm import Router
# ── 1. simple-shuffle (기본) ──────────────────────────
# 랜덤 + 쿨다운 배포 제외
router = Router(
model_list=model_list,
routing_strategy="simple-shuffle",
)
# ── 2. least-busy ─────────────────────────────────────
# 현재 진행 중인 요청이 가장 적은 배포 선택
router = Router(
model_list=model_list,
routing_strategy="least-busy",
)
# ── 3. latency-based-routing ──────────────────────────
# 최근 응답 시간 기반으로 가장 빠른 배포 선택
router = Router(
model_list=model_list,
routing_strategy="latency-based-routing",
)
# ── 4. usage-based-routing ───────────────────────────
# TPM/RPM 한도 기반으로 여유 있는 배포 선택 (Redis 필요)
router = Router(
model_list=model_list,
routing_strategy="usage-based-routing",
redis_host=os.environ.get("REDIS_HOST", "localhost"),
redis_port=6379,
redis_password=os.environ.get("REDIS_PASSWORD", ""),
)
order로 배포 우선순위 지정
# order 파라미터 — 낮을수록 우선순위 높음
model_list = [
{
"model_name": "gpt-4",
"litellm_params": {
"model": "azure/gpt-4-primary",
"api_key": os.environ["AZURE_API_KEY_PRIMARY"],
"api_base": os.environ["AZURE_ENDPOINT_PRIMARY"],
},
"order": 1, # 1순위 — 항상 먼저 시도
},
{
"model_name": "gpt-4",
"litellm_params": {
"model": "azure/gpt-4-secondary",
"api_key": os.environ["AZURE_API_KEY_SECONDARY"],
"api_base": os.environ["AZURE_ENDPOINT_SECONDARY"],
},
"order": 2, # 2순위 — order=1 실패 시
},
{
"model_name": "gpt-4",
"litellm_params": {
"model": "openai/gpt-4",
"api_key": os.environ["OPENAI_API_KEY"],
},
"order": 3, # 3순위 — 마지막 보루
},
]
router = Router(model_list=model_list)
# order=1 실패 → order=2 자동 에스컬레이션 → order=3
실전 4 — 비용 추적
import litellm
# ── 요청 후 비용 계산 ─────────────────────────────────
response = litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "긴 분석 태스크..."}]
)
# 실제 비용 (달러)
cost = litellm.completion_cost(completion_response=response)
print(f"이번 요청 비용: ${cost:.6f}")
# → $0.001234 같이 작은 값
# 토큰별 세부 비용
input_cost = litellm.completion_cost(
model="anthropic/claude-sonnet-4-6",
prompt="입력 텍스트",
completion="",
)
print(f"입력 비용: ${input_cost:.6f}")
# ── 콜백으로 모든 요청 자동 추적 ─────────────────────
from collections import defaultdict
from datetime import datetime
# 비용 누적 저장소
cost_tracker = defaultdict(lambda: {
"total_cost": 0.0,
"requests": 0,
"input_tokens": 0,
"output_tokens": 0,
})
def track_cost(kwargs, completion_response, start_time, end_time):
"""모든 성공 요청에서 비용 자동 집계"""
model = kwargs.get("model", "unknown")
cost = litellm.completion_cost(completion_response=completion_response)
usage = completion_response.usage
cost_tracker[model]["total_cost"] += cost
cost_tracker[model]["requests"] += 1
cost_tracker[model]["input_tokens"] += usage.prompt_tokens
cost_tracker[model]["output_tokens"] += usage.completion_tokens
print(f"[{model}] ${cost:.6f} | "
f"in:{usage.prompt_tokens} / out:{usage.completion_tokens} tokens")
# 전역 콜백 등록
litellm.success_callback = [track_cost]
# 이후 모든 completion() 호출 시 자동 추적
litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "안녕"}]
)
litellm.completion(
model="openai/gpt-5.4",
messages=[{"role": "user", "content": "안녕"}]
)
# 일별 리포트
def print_cost_report():
print(f"\n{'='*50}")
print(f"비용 리포트 ({datetime.now().strftime('%Y-%m-%d')})")
print(f"{'='*50}")
total = 0
for model, data in cost_tracker.items():
print(f"\n📌 {model}")
print(f" 요청: {data['requests']} | 비용: ${data['total_cost']:.4f}")
print(f" 토큰: {data['input_tokens']:,} in / {data['output_tokens']:,} out")
total += data["total_cost"]
print(f"\n총 비용: ${total:.4f}")
print_cost_report()
# ── 모델별 비용 미리 확인 ─────────────────────────────
# 실제 API 호출 없이 예상 비용 계산
models_to_check = [
"anthropic/claude-opus-4-7",
"anthropic/claude-sonnet-4-6",
"openai/gpt-5.4",
"google/gemini-3-flash",
"deepseek/deepseek-chat",
]
# 동일 프롬프트로 모델별 예상 비용 비교
test_prompt = "파이썬으로 REST API 서버 만드는 법 설명해줘"
test_completion = "FastAPI를 사용하면..." * 100 # 약 600 토큰 출력 가정
print("모델별 예상 비용 (입력+출력 합산):")
for model in models_to_check:
try:
cost = litellm.completion_cost(
model=model,
prompt=test_prompt,
completion=test_completion,
)
print(f" {model}: ${cost:.6f}")
except Exception:
print(f" {model}: 가격 정보 없음")
실전 5 — 예산 제한
import litellm
from litellm.exceptions import BudgetExceededError
# ── 전역 예산 설정 ────────────────────────────────────
litellm.max_budget = 1.0 # $1 이상 사용 시 에러
litellm.budget_duration = "1d" # 1일 기준 (리셋 주기)
# 옵션: "1h", "1d", "7d", "30d", "1mo"
try:
response = litellm.completion(
model="anthropic/claude-opus-4-7",
messages=[{"role": "user", "content": "매우 긴 태스크..."}]
)
except BudgetExceededError as e:
print(f"예산 초과! {e}")
# 저렴한 모델로 폴백
response = litellm.completion(
model="google/gemini-3-flash", # 무료/저렴한 모델
messages=[{"role": "user", "content": "매우 긴 태스크..."}]
)
# ── Router 레벨 예산 ──────────────────────────────────
router = Router(
model_list=model_list,
budget_manager=litellm.BudgetManager(
project_name="my-project",
client_type="local", # 로컬 저장 (프로덕션은 "hosted")
)
)
# 사용자별 예산 설정
router.budget_manager.create_budget(
total_budget=0.5, # $0.5 한도
user="user-alice",
duration="daily", # 일별 리셋
)
router.budget_manager.create_budget(
total_budget=5.0, # $5 한도
user="user-bob",
duration="monthly", # 월별 리셋
)
# 사용자별 비용 추적 요청
response = await router.acompletion(
model="claude",
messages=[{"role": "user", "content": "안녕"}],
user="user-alice", # 이 사용자 예산에서 차감
)
# 잔여 예산 확인
remaining = router.budget_manager.get_current_cost("user-alice")
print(f"Alice 사용량: ${remaining:.4f}")
실전 6 — 캐싱
import litellm
from litellm.caching import Cache
# ── 인메모리 캐시 (개발·테스트용) ─────────────────────
litellm.cache = Cache()
# 첫 번째 호출 — API 실제 호출
response1 = litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "파이썬이란?"}],
caching=True, # 캐싱 활성화
)
print(f"캐시 히트: {response1._hidden_params.get('cache_hit', False)}")
# → False (첫 호출)
# 두 번째 호출 — 캐시에서 반환 (API 미호출)
response2 = litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "파이썬이란?"}],
caching=True,
)
print(f"캐시 히트: {response2._hidden_params.get('cache_hit', False)}")
# → True (캐시 반환)
python
# ── Redis 캐시 (프로덕션용) ────────────────────────────
from litellm.caching import Cache
litellm.cache = Cache(
type="redis",
host=os.environ.get("REDIS_HOST", "localhost"),
port=6379,
password=os.environ.get("REDIS_PASSWORD", ""),
ttl=3600, # 캐시 유효 시간 (초) — 1시간
)
# 이후 모든 caching=True 요청이 Redis 캐시 사용
response = litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "자주 묻는 질문"}],
caching=True,
)
# ── 시맨틱 캐시 (의미 유사 프롬프트도 캐시 히트) ────────
litellm.cache = Cache(
type="redis-semantic",
host=os.environ.get("REDIS_HOST", "localhost"),
port=6379,
similarity_threshold=0.8, # 유사도 80% 이상이면 캐시 히트
embedding_model="text-embedding-3-small",
)
# "파이썬이란?" 캐시 후
# "Python이 뭐야?" 요청 → 시맨틱 유사도 높으면 캐시에서 반환
response = litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "Python이 뭐야?"}],
caching=True,
)
python
# ── 캐시 키 커스터마이징 ──────────────────────────────
# 특정 파라미터를 캐시 키에서 제외하거나 추가
response = litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "정적인 FAQ 답변"}],
caching=True,
cache={
"no-cache": False, # 캐시 사용 (기본)
"no-store": False, # 캐시 저장 (기본)
"ttl": 86400, # 이 요청만 24시간 캐시
}
)
# 특정 요청 캐시 무효화
response = litellm.completion(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": "실시간 정보 필요"}],
caching=True,
cache={"no-cache": True}, # 캐시 무시하고 항상 새 요청
)
실전 7 — 모델 별칭(Alias)
from litellm import Router
router = Router(
model_list=[
{
"model_name": "claude-sonnet-4-6",
"litellm_params": {
"model": "anthropic/claude-sonnet-4-6",
"api_key": os.environ["ANTHROPIC_API_KEY"],
},
},
{
"model_name": "gpt-5.4",
"litellm_params": {
"model": "openai/gpt-5.4",
"api_key": os.environ["OPENAI_API_KEY"],
},
},
],
# 별칭 설정 — 팀에서 쓸 짧은 이름
model_group_alias={
"claude": "claude-sonnet-4-6", # "claude"로 호출 → Sonnet
"gpt": "gpt-5.4", # "gpt"로 호출 → GPT-5.4
"default": "claude-sonnet-4-6", # "default" → Claude
"fast": "gpt-5.4", # "fast" → GPT (더 빠름)
"cheap": "claude-sonnet-4-6", # "cheap" → Claude
}
)
# 별칭으로 호출
response = router.completion(
model="claude", # "anthropic/claude-sonnet-4-6" 호출
messages=[{"role": "user", "content": "안녕"}]
)
response = router.completion(
model="default", # 기본 모델
messages=[{"role": "user", "content": "안녕"}]
)
# ── 환경별 별칭 전략 ──────────────────────────────────
import os
ENV = os.environ.get("ENVIRONMENT", "development")
if ENV == "development":
# 개발: 저렴한 모델
alias = {
"claude": "anthropic/claude-haiku-4-5",
"gpt": "openai/gpt-4o-mini",
"default": "anthropic/claude-haiku-4-5",
}
elif ENV == "staging":
# 스테이징: 중간 모델
alias = {
"claude": "anthropic/claude-sonnet-4-6",
"gpt": "openai/gpt-5.4",
"default": "anthropic/claude-sonnet-4-6",
}
else:
# 프로덕션: 최고 모델
alias = {
"claude": "anthropic/claude-opus-4-7",
"gpt": "openai/gpt-5.4",
"default": "anthropic/claude-sonnet-4-6",
}
router = Router(
model_list=model_list,
model_group_alias=alias
)
# 코드는 항상 "default"로 호출
# 환경에 따라 다른 모델 실행
response = router.completion(
model="default",
messages=[{"role": "user", "content": "안녕"}]
)
마무리
✅ 2편에서 한 것
→ 단순 폴백: fallbacks + context_window_fallbacks + content_policy_fallbacks
→ Router: 모델 그룹, 로드밸런싱, order 우선순위
→ 라우팅 전략 4가지: simple-shuffle / least-busy / latency / usage-based
→ 비용 추적: completion_cost() + success_callback 자동 집계
→ 예산 제한: max_budget + BudgetManager 사용자별 예산
→ 캐싱: 인메모리 / Redis / 시맨틱 캐시
→ 모델 별칭: 환경별 모델 전략 + 짧은 이름 매핑
❌ 3편에서 다룰 것
→ LiteLLM Proxy 서버 모드 구축
→ Docker Compose 배포
→ 가상 키(Virtual Keys) 관리
→ 팀별·프로젝트별 예산 분리
→ Admin 대시보드
→ 로깅 및 Langfuse 연동
관련 글
- LiteLLM 완전 가이드 1편 — 100개+ LLM을 코드 한 줄로 갈아타는 오픈소스 AI 게이트웨이
- OpenRouter 완전 가이드 2편 — 폴백 라우팅, 프로바이더 제어, 비용 최적화
반응형
'AI 개발' 카테고리의 다른 글
| LiteLLM 완전 가이드 4편 — LangChain·LangGraph 통합, 가드레일, Prometheus 모니터링, 프로덕션 운영 (0) | 2026.05.19 |
|---|---|
| LiteLLM 완전 가이드 3편 — Proxy 서버 모드: 팀 공용 LLM 게이트웨이 구축 실전 (0) | 2026.05.19 |
| LiteLLM 완전 가이드 1편 — 100개+ LLM을 코드 한 줄로 갈아타는 오픈소스 AI 게이트웨이 (0) | 2026.05.19 |
| OpenRouter 완전 가이드 4편 — 모니터링, 레이트 리밋 관리, OAuth PKCE, 팀 운영, ZDR (0) | 2026.05.19 |
| OpenRouter 완전 가이드 3편 — 스트리밍, 툴 콜링, 멀티모달, 구조화 출력, LangChain·LangGraph 통합 (0) | 2026.05.19 |