LLM API 비용 최적화: 배치 처리·캐싱·프롬프트 압축 실전 전략

AI 기술

LLM 비용 최적화Prompt Caching배치 API토큰 최적화AI 운영

이 글은 누구를 위한 것인가

  • LLM API 비용이 예상보다 빠르게 늘어 걱정인 스타트업 CTO·개발자
  • 대량 문서 처리에 GPT/Claude API를 쓰는 팀
  • 프롬프트 캐싱, 배치 API를 아직 적용 안 한 팀

들어가며

LLM을 프로덕션에 올리면 트래픽이 늘수록 비용이 폭발적으로 증가한다. "프롬프트를 줄여라"는 조언은 맞지만, 비용 최적화에는 더 구조적인 접근이 필요하다.

Anthropic Claude의 프롬프트 캐싱은 반복 호출 비용을 최대 90% 줄일 수 있다. 배치 API는 비실시간 작업 비용을 50% 절감한다. 이 두 가지만 적용해도 월 비용이 크게 달라진다.

이 글은 bluefoxdev.kr의 LLM 운영 비용 가이드 를 참고하고, 실전 비용 절감 전략 관점에서 확장하여 작성했습니다.


1. LLM 비용 구조 이해

[API 비용 계산 구조]

비용 = (입력 토큰 × 입력 단가) + (출력 토큰 × 출력 단가)

Claude Sonnet 4 (2026년 기준 예시):
  입력: $3 / 1M 토큰
  출력: $15 / 1M 토큰
  캐시 쓰기: $3.75 / 1M 토큰
  캐시 읽기: $0.30 / 1M 토큰  ← 90% 할인!

비용 최적화 우선순위:
  1. 프롬프트 캐싱 (반복 시스템 프롬프트)  → 입력 비용 90% 절감
  2. 배치 API (비실시간 작업)              → 전체 비용 50% 절감
  3. 모델 라우팅 (단순 요청은 소형 모델)    → 비용 70-90% 절감
  4. 출력 길이 제어 (max_tokens)           → 출력 비용 절감
  5. 프롬프트 압축 (RAG 청크 최적화)       → 입력 비용 절감

2. Anthropic 프롬프트 캐싱 구현

import anthropic

client = anthropic.Anthropic()

# 프롬프트 캐싱 활성화
# 시스템 프롬프트에 cache_control 추가
def call_with_caching(user_message: str, documents: list[str]) -> str:
    """
    시스템 프롬프트 + 문서를 캐싱하여 반복 호출 비용 절감
    캐시 TTL: 5분 (동일 시스템 프롬프트 반복 시 캐시 적중)
    """
    
    # 문서를 단일 블록으로 결합
    docs_text = "\n\n---\n\n".join(
        f"[문서 {i+1}]\n{doc}" for i, doc in enumerate(documents)
    )
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": "당신은 문서 분석 전문가입니다. 주어진 문서를 기반으로 질문에 정확하게 답하세요.",
                "cache_control": {"type": "ephemeral"}  # 캐싱 활성화
            },
            {
                "type": "text",
                "text": f"참고 문서:\n\n{docs_text}",
                "cache_control": {"type": "ephemeral"}  # 문서도 캐싱
            }
        ],
        messages=[
            {"role": "user", "content": user_message}
        ]
    )
    
    # 캐시 사용량 확인
    usage = response.usage
    print(f"캐시 생성 토큰: {usage.cache_creation_input_tokens}")
    print(f"캐시 읽기 토큰: {usage.cache_read_input_tokens}")
    print(f"일반 입력 토큰: {usage.input_tokens}")
    
    return response.content[0].text

3. 배치 API 구현

import anthropic
import time

client = anthropic.Anthropic()

def process_batch(items: list[dict]) -> list[dict]:
    """
    배치 API: 비실시간 대량 처리
    비용 50% 절감, 처리 시간 최대 24시간
    """
    
    # 배치 요청 생성
    requests = [
        {
            "custom_id": f"item-{i}",
            "params": {
                "model": "claude-haiku-4-5-20251001",  # 단순 작업은 Haiku
                "max_tokens": 200,
                "messages": [
                    {
                        "role": "user",
                        "content": item["prompt"]
                    }
                ]
            }
        }
        for i, item in enumerate(items)
    ]
    
    # 배치 생성
    batch = client.messages.batches.create(requests=requests)
    batch_id = batch.id
    print(f"배치 생성: {batch_id}")
    
    # 완료 대기 (폴링)
    while True:
        status = client.messages.batches.retrieve(batch_id)
        
        if status.processing_status == "ended":
            break
        
        print(f"처리 중... {status.request_counts}")
        time.sleep(30)  # 30초마다 확인
    
    # 결과 수집
    results = []
    for result in client.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            results.append({
                "id": result.custom_id,
                "output": result.result.message.content[0].text
            })
        else:
            results.append({
                "id": result.custom_id,
                "error": result.result.error.message
            })
    
    return results

# 사용 예시 (상품 설명 자동 생성)
products = [
    {"prompt": f"상품명: 무선 이어폰 AX-100\n특징: ANC, 30시간 배터리\n→ 50자 이내 상품 설명 작성"}
    for _ in range(1000)  # 1000개 상품
]

results = process_batch(products)

4. 모델 라우팅 (Model Routing)

from enum import Enum

class Complexity(Enum):
    SIMPLE = "simple"
    MEDIUM = "medium"
    COMPLEX = "complex"

def classify_complexity(prompt: str) -> Complexity:
    """요청 복잡도 분류"""
    
    # 단순 분류 규칙 (실제로는 더 정교하게)
    word_count = len(prompt.split())
    
    # 복잡한 작업 키워드
    complex_keywords = ["분석", "비교", "추론", "계획", "설계", "코드 작성"]
    # 단순 작업 키워드
    simple_keywords = ["번역", "요약", "분류", "추출", "형식 변환"]
    
    if any(kw in prompt for kw in complex_keywords) or word_count > 500:
        return Complexity.COMPLEX
    elif any(kw in prompt for kw in simple_keywords) or word_count < 100:
        return Complexity.SIMPLE
    else:
        return Complexity.MEDIUM

MODEL_MAP = {
    Complexity.SIMPLE: "claude-haiku-4-5-20251001",   # 가장 저렴
    Complexity.MEDIUM: "claude-sonnet-4-6",            # 중간
    Complexity.COMPLEX: "claude-opus-4-7",             # 가장 강력
}

COST_PER_1M = {
    "claude-haiku-4-5-20251001": {"input": 0.25, "output": 1.25},
    "claude-sonnet-4-6":         {"input": 3.0,  "output": 15.0},
    "claude-opus-4-7":           {"input": 15.0, "output": 75.0},
}

def route_and_call(prompt: str) -> tuple[str, dict]:
    complexity = classify_complexity(prompt)
    model = MODEL_MAP[complexity]
    
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    
    # 비용 계산
    costs = COST_PER_1M[model]
    estimated_cost = (
        response.usage.input_tokens / 1_000_000 * costs["input"] +
        response.usage.output_tokens / 1_000_000 * costs["output"]
    )
    
    return response.content[0].text, {
        "model": model,
        "complexity": complexity.value,
        "cost_usd": round(estimated_cost, 6),
    }

5. 비용 모니터링

from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict

@dataclass
class CostTracker:
    daily_costs: dict = field(default_factory=lambda: defaultdict(float))
    model_costs: dict = field(default_factory=lambda: defaultdict(float))
    total_tokens: dict = field(default_factory=lambda: defaultdict(int))
    
    def record(self, model: str, input_tokens: int, output_tokens: int, cache_read_tokens: int = 0):
        costs = COST_PER_1M.get(model, {"input": 3.0, "output": 15.0})
        
        # 캐시 읽기 비용 (일반의 10%)
        cache_cost = cache_read_tokens / 1_000_000 * costs["input"] * 0.1
        input_cost = input_tokens / 1_000_000 * costs["input"]
        output_cost = output_tokens / 1_000_000 * costs["output"]
        total = input_cost + output_cost + cache_cost
        
        today = datetime.now().strftime("%Y-%m-%d")
        self.daily_costs[today] += total
        self.model_costs[model] += total
        self.total_tokens[model] += input_tokens + output_tokens
    
    def report(self):
        print("=== 비용 리포트 ===")
        print(f"오늘 총 비용: ${self.daily_costs.get(datetime.now().strftime('%Y-%m-%d'), 0):.4f}")
        print("\n모델별 누적 비용:")
        for model, cost in sorted(self.model_costs.items(), key=lambda x: -x[1]):
            tokens = self.total_tokens[model]
            print(f"  {model}: ${cost:.4f} ({tokens:,} tokens)")

# 전역 트래커
cost_tracker = CostTracker()

마무리

LLM 비용 최적화의 핵심은 "더 적게 쓰는 것"이 아니라 "더 효율적으로 쓰는 것"이다. 프롬프트 캐싱으로 반복 비용을 제거하고, 배치 API로 비실시간 작업을 묶고, 모델 라우팅으로 복잡도에 맞는 모델을 선택하면 품질 저하 없이 비용을 크게 줄일 수 있다.

먼저 모니터링부터 시작하라. 어디서 비용이 나오는지 알아야 무엇을 최적화할지 보인다.