본문 바로가기

AI 개발

LiteLLM 완전 가이드 2편 — 폴백·재시도, Router 로드밸런싱, 비용 추적, 예산·캐싱 실전

반응형

 

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 연동

관련 글

 

반응형