비용 인식 LLM 파이프라인은(는) 무엇을 하나요?
LLM API 비용을 제어하면서 품질을 유지하기 위한 패턴입니다. 모델 라우팅, 예산 추적, 재시도 로직, 프롬프트 캐싱을 결합하여 구성 가능한 파이프라인을 제공합니다.
활성화 시점
- LLM API(Claude, GPT 등)를 호출하는 애플리케이션 구축
- 다양한 복잡도의 항목을 일괄 처리
- API 지출 예산 범위 내 유지 필요
- 복잡한 작업에서 품질 저하 없이 비용 최적화
핵심 개념
1. 작업 복잡도별 모델 라우팅
간단한 작업에는 자동으로 더 저렴한 모델을 선택하고, 복잡한 작업에는 고비용 모델을 예약합니다.
MODEL_SONNET = "claude-sonnet-5"
MODEL_HAIKU = "claude-haiku-4-5-20251001"
_SONNET_TEXT_THRESHOLD = 10_000 # chars
_SONNET_ITEM_THRESHOLD = 30 # items
def select_model(
text_length: int,
item_count: int,
force_model: str | None = None,
) -> str:
"""작업 복잡도에 따라 모델을 선택합니다."""
if force_model is not None:
return force_model
if text_length >= _SONNET_TEXT_THRESHOLD or item_count >= _SONNET_ITEM_THRESHOLD:
return MODEL_SONNET # 복잡한 작업
return MODEL_HAIKU # 간단한 작업 (3-4배 저렴함)
2. 변경 불가능한 비용 추적
불변 데이터클래스로 누적 지출을 추적합니다. 각 API 호출은 새로운 추적기를 반환하며 상태를 절대 변경하지 않습니다.
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class CostRecord:
model: str
input_tokens: int
output_tokens: int
cost_usd: float
@dataclass(frozen=True, slots=True)
class CostTracker:
budget_limit: float = 1.00
records: tuple[CostRecord, ...] = ()
def add(self, record: CostRecord) -> "CostTracker":
"""레코드가 추가된 새 추적기를 반환합니다(자체를 변경하지 않음)."""
return CostTracker(
budget_limit=self.budget_limit,
records=(*self.records, record),
)
@property
def total_cost(self) -> float:
return sum(r.cost_usd for r in self.records)
@property
def over_budget(self) -> bool:
return self.total_cost > self.budget_limit
3. 세분화된 재시도 로직
일시적 오류에 대해서만 재시도합니다. 인증 오류나 잘못된 요청 오류가 발생하면 빠르게 실패합니다.
from anthropic import (
APIConnectionError,
InternalServerError,
RateLimitError,
)
_RETRYABLE_ERRORS = (APIConnectionError, RateLimitError, InternalServerError)
_MAX_RETRIES = 3
def call_with_retry(func, *, max_retries: int = _MAX_RETRIES):
"""일시적 오류에 대해서만 재시도하고, 다른 오류는 즉시 실패합니다."""
for attempt in range(max_retries):
try:
return func()
except _RETRYABLE_ERRORS:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # 지수 백오프
# AuthenticationError, BadRequestError 등 → 즉시 발생
4. 프롬프트 캐싱
긴 시스템 프롬프트를 캐싱하여 모든 요청에서 다시 보내지 않도록 합니다.
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": system_prompt,
"cache_control": {"type": "ephemeral"}, # 캐시
},
{
"type": "text",
"text": user_input, # 가변 부분
},
],
}
]
구성
네 가지 기술을 모두 하나의 파이프라인 함수로 결합합니다:
def process(text: str, config: Config, tracker: CostTracker) -> tuple[Result, CostTracker]:
# 1. 모델 라우팅
model = select_model(len(text), estimated_items, config.force_model)
# 2. 예산 확인
if tracker.over_budget:
raise BudgetExceededError(tracker.total_cost, tracker.budget_limit)
# 3. 재시도 + 캐싱으로 호출
response = call_with_retry(lambda: client.messages.create(
model=model,
messages=build_cached_messages(system_prompt, text),
))
# 4. 비용 추적 (변경 불가)
record = CostRecord(model=model, input_tokens=..., output_tokens=..., cost_usd=...)
tracker = tracker.add(record)
return parse_result(response), tracker
가격 참조 (2026)
| 모델 | 입력 ($/100만 토큰) | 출력 ($/100만 토큰) | 상대 비용 |
|---|---|---|---|
| Haiku 3.5 (레거시) | $0.80 | $4.00 | 0.8x |
| Haiku 4.5 | $1.00 | $5.00 | 1x |
| Sonnet 5 | $2.00 | $10.00 | 2x |
| Sonnet 4.6 | $3.00 | $15.00 | 3x |
| Opus 4.8 | $5.00 | $25.00 | 5x |
| Fable 5 / Mythos 5 | $10.00 | $50.00 | 10x |
| Opus 4.0 / 4.1 (레거시) | $15.00 | $75.00 | 15x |
모범 사례
- 가장 저렴한 모델로 시작하고 복잡도 임계값이 충족된 경우에만 고비용 모델로 라우팅하세요.
- 배치를 처리하기 전에 명시적인 예산 한도를 설정하세요. — 초과 지출보다는 조기에 실패하는 것이 낫습니다.
- 모델 선택 결정을 로깅하여 실제 데이터를 기반으로 임계값을 조정할 수 있도록 하세요.
- 1024 토큰 이상의 시스템 프롬프트에는 프롬프트 캐싱을 사용하세요. 비용과 지연 시간을 모두 절약할 수 있습니다.
- 인증 또는 유효성 검사 오류에서는 절대 재시도하지 마세요. — 일시적 오류(네트워크, 속도 제한, 서버 오류)에만 재시도하세요.
피해야 할 안티 패턴
- 복잡도와 관계없이 모든 요청에 가장 비싼 모델 사용
- 모든 오류에서 재시도 (영구적 오류에 예산 낭비)
- 비용 추적 상태 변경 (디버깅 및 감사 어려움)
- 코드베이스 전체에 모델 이름 하드코딩 (상수 또는 구성 사용)
- 반복적인 시스템 프롬프트에 대한 프롬프트 캐싱 무시
사용 시기
- Claude, OpenAI 또는 유사 LLM API를 호출하는 모든 애플리케이션
- 비용이 빠르게 증가하는 일괄 처리 파이프라인
- 지능형 라우팅이 필요한 멀티 모델 아키텍처
- 예산 가드레일이 필요한 프로덕션 시스템