캐싱, 속도 제한과 비용 최적화(Caching, Rate Limiting & Cost Optimization)

대부분의 AI 스타트업은 나쁜 모델 때문에 죽지 않습니다. 나쁜 단위 경제성(unit economics) 때문에 죽습니다. GPT-4o 호출 한 번은 1센트보다 훨씬 적게 듭니다. 하지만 사용자 10,000명이 하루 10번씩 호출하면, 한 달러를 벌기도 전에 입력 토큰(input token) 비용만 하루 250달러가 듭니다. 살아남는 회사는 모든 API 호출을 함수 호출(function call)이 아니라 금융 거래로 다루는 회사입니다.

유형: Build 언어: Python 선수 지식: Phase 11 Lesson 09(Function Calling) 예상 시간: 약 45분 관련: Phase 11 · 15(프롬프트 캐싱)는 제공자 계층(provider-layer) 프롬프트 캐싱을 다룹니다. 이 lesson은 애플리케이션 계층(application-layer) 캐싱, 즉 의미 기반 캐시(semantic cache), 정확 해시 캐시(exact hash cache), 모델 라우팅(model routing)을 다룹니다. Lesson 15는 Anthropic cache_control, OpenAI 자동 캐싱, Gemini CachedContent 같은 제공자 계층 프롬프트 캐싱을 다룹니다. 두 계층을 함께 사용하면 비용을 50-95% 줄일 수 있습니다.

학습 목표

• 새 API 호출을 만들지 않고 반복되거나 유사한 질의를 캐시에서 제공하는 의미 기반 캐싱(semantic caching)을 구현합니다.
• 제공자(provider)별 요청당 비용을 계산하고, 토큰을 인식하는 속도 제한(rate limiting)과 예산 알림(budget alert)을 구현합니다.
• 프롬프트 압축(prompt compression), 모델 라우팅(비싼 모델과 저렴한 모델), 응답 캐싱(response caching)을 포함한 비용 최적화 계층(cost optimization layer)을 만듭니다.
• 정확 매칭(exact match), 의미 유사도(semantic similarity), 접두사 캐싱(prefix caching)을 질의 유형에 따라 조합하는 계층형 캐싱 전략(tiered caching strategy)을 설계합니다.

문제

RAG 챗봇을 만들었습니다. 멋지게 동작합니다. 사용자도 좋아합니다.

그러다 청구서가 도착합니다.

GPT-5는 입력 100만 토큰당 5달러, 출력 100만 토큰당 15달러입니다. Claude Opus 4.7은 입력 15달러 / 출력 75달러입니다. Gemini 3 Pro는 입력 1.25달러 / 출력 5달러입니다. GPT-5-mini는 0.25달러 / 2달러입니다. 아래 가격은 설명을 위한 예시입니다. 실제 운영에서는 항상 제공자의 최신 가격 페이지를 확인해야 합니다.

스타트업을 죽이는 계산은 다음과 같습니다.

• 일간 활성 사용자(DAU) 10,000명
• 사용자당 하루 10개 질의
• 질의당 입력 토큰 1,000개(시스템 프롬프트 + 컨텍스트 + 사용자 메시지)
• 응답당 출력 토큰 500개

일간 입력 비용: 10,000 x 10 x 1,000 / 1,000,000 x $2.50 = $250/day 일간 출력 비용: 10,000 x 10 x 500 / 1,000,000 x $10.00 = $500/day 월간 합계: $22,500/month

이것은 LLM 비용만 계산한 것입니다. 여기에 임베딩(embedding), 벡터 데이터베이스(vector database) 호스팅, 인프라 비용이 더해집니다. 챗봇 하나에 월 30,000달러가 보일 수 있습니다.

더 가혹한 부분은 질의의 40-60%가 거의 중복(near-duplicate)이라는 점입니다. 사용자는 같은 질문을 조금씩 다른 말로 묻습니다. 모든 요청에서 동일한 시스템 프롬프트가 매번 과금됩니다. 같은 주제에 대해 묻는 사용자에게 RAG가 가져오는 컨텍스트 문서도 반복됩니다.

즉, 중복 계산에 정가를 내고 있는 것입니다.

개념

LLM 호출 비용 구조(The Cost Anatomy of an LLM Call)

모든 API 호출에는 다섯 가지 비용 구성요소가 있습니다.

graph LR
    A[User Query] --> B[System Prompt<br/>500-2000 tokens]
    A --> C[Retrieved Context<br/>500-4000 tokens]
    A --> D[User Message<br/>50-500 tokens]
    B --> E[Input Cost<br/>$2.50/1M tokens]
    C --> E
    D --> E
    E --> F[Model Processing]
    F --> G[Output Cost<br/>$10.00/1M tokens]

시스템 프롬프트는 조용한 살인자입니다. 모든 요청에 1,500토큰짜리 시스템 프롬프트를 보내면, 이 접두사(prefix)만으로도 100만 요청당 3.75달러가 듭니다. 하루 100,000 요청이면 하루 375달러, 한 달 11,250달러입니다. 바뀌지 않는 텍스트에 드는 비용입니다.

제공자 캐싱: 내장 할인(Provider Caching: Built-in Discounts)

2026년 기준 주요 제공자 세 곳은 모두 제공자 측 프롬프트 캐싱(provider-side prompt caching)을 제공합니다. 하지만 동작 방식은 다릅니다. 자세한 내용은 Phase 11 · 15를 참고합니다.

Anthropic 방식은 명시적입니다. 프롬프트 일부에 cache_control: {"type": "ephemeral"}을 표시합니다. 첫 요청은 25% 쓰기 프리미엄(write premium)을 냅니다. 이후 같은 접두사를 가진 요청은 90% 할인을 받습니다. 일반적으로 $0.005가 드는 2,000토큰 시스템 프롬프트는 캐시 적중 시 $0.000625가 됩니다. 100,000 요청이면 하루 $437.50를 절감합니다.

OpenAI 방식은 자동입니다. 이전 요청과 프롬프트 접두사가 일치하면 50% 할인을 받습니다. 마커가 필요 없습니다. 대신 할인 폭이 작고 제어권이 적지만 구현 노력이 없습니다.

의미 기반 캐싱: 직접 만드는 계층(Semantic Caching: Your Custom Layer)

제공자 캐싱은 동일한 접두사에만 동작합니다. 의미 기반 캐싱은 더 어려운 경우를 처리합니다. 문자열은 다르지만 의미가 같은 질의입니다.

What is the return policy?와 How do I return an item?은 문자열은 다르지만 의도(intent)는 같습니다. 의미 기반 캐시는 두 질의를 임베딩하고, 코사인 유사도(cosine similarity)를 계산한 뒤, 유사도가 임계값(threshold)을 넘으면 보통 0.92-0.95 수준에서 캐시된 응답을 반환합니다.

flowchart TD
    A[User Query] --> B[Embed Query]
    B --> C{Similar query<br/>in cache?}
    C -->|sim > 0.95| D[Return Cached Response]
    C -->|sim < 0.95| E[Call LLM API]
    E --> F[Cache Response<br/>with Embedding]
    F --> G[Return Response]
    D --> G

임베딩 비용은 거의 무시할 수 있습니다. OpenAI text-embedding-3-small은 100만 토큰당 $0.02입니다. 캐시를 확인하는 비용은 전체 LLM 호출에 비하면 거의 없습니다.

정확 캐싱: 해시와 매칭(Exact Caching: Hash and Match)

결정적 호출(deterministic call), 즉 temperature=0, 같은 모델, 같은 프롬프트라면 정확 캐싱이 더 단순하고 빠릅니다. 전체 프롬프트를 해시(hash)하고, 캐시를 확인한 뒤, 있으면 반환합니다.

이 방식은 다음에 완벽하게 맞습니다.

• 시스템 프롬프트 + 고정 컨텍스트 + 동일한 사용자 질의
• 동일한 도구 정의(tool definitions)를 사용하는 함수 호출(function calling)
• 같은 문서를 여러 번 처리하는 배치 처리(batch processing)

속도 제한: 예산 보호(Rate Limiting: Protecting Your Budget)

속도 제한은 단순히 공정성을 위한 것이 아닙니다. 생존을 위한 것입니다.

토큰 버킷 알고리즘(token bucket algorithm): 각 사용자는 N개의 토큰이 들어 있는 버킷(bucket)을 갖고, 버킷은 초당 R개 속도로 다시 채워집니다. 요청은 버킷에서 토큰을 소비합니다. 버킷이 비어 있으면 요청은 거부됩니다. 이 방식은 한 번에 버킷 전체를 사용하는 순간 폭증(burst)을 허용하면서도 평균 속도를 강제합니다.

사용자별 할당량(per-user quotas): 사용자 계층(tier)별로 일간/월간 토큰 한도를 설정합니다.

모델 라우팅: 작업에 맞는 모델(Model Routing: Right Model for the Right Job)

모든 질의가 GPT-4o를 필요로 하지는 않습니다.

What time does the store close?는 출력 100만 토큰당 10달러인 모델을 필요로 하지 않습니다. 출력 100만 토큰당 $0.60인 GPT-4o-mini가 완벽히 처리할 수 있습니다. 출력 100만 토큰당 $1.25인 Claude Haiku도 처리할 수 있습니다. 단순한 분류기(classifier)가 쉬운 질의는 저렴한 모델로, 복잡한 질의는 비싼 모델로 보냅니다.

flowchart TD
    A[User Query] --> B[Complexity Classifier]
    B -->|단순: lookup, FAQ| C[GPT-4o-mini<br/>$0.15/$0.60 per 1M]
    B -->|중간: analysis, summary| D[Claude Sonnet<br/>$3.00/$15.00 per 1M]
    B -->|복잡: reasoning, code| E[GPT-4o / Claude Opus<br/>$2.50/$10.00+]

잘 조정한 라우터(router)는 모델 비용만 40-70% 절감합니다.

비용 추적: 돈이 어디로 가는지 알기(Cost Tracking: Know Where the Money Goes)

측정하지 않는 것은 최적화할 수 없습니다. 모든 API 호출에 다음을 기록합니다.

• 타임스탬프(timestamp)
• 모델 이름
• 입력 토큰 수
• 출력 토큰 수
• 지연 시간(latency, ms)
• 계산된 비용($)
• 사용자 ID
• 캐시 적중/미스(hit/miss)
• 요청 카테고리

이 데이터는 어떤 기능이 비싼지, 어떤 사용자가 많이 소비하는지, 캐싱이 가장 큰 효과를 내는 위치가 어디인지 보여줍니다.

배치 처리: 대량 할인(Batching: Bulk Discounts)

OpenAI Batch API는 요청을 비동기로 처리하며 50% 할인을 제공합니다. 최대 50,000개 요청을 배치로 제출하면 24시간 안에 결과가 돌아옵니다.

배치 처리는 다음에 사용합니다.

• 야간 문서 처리
• 대량 분류(bulk classification)
• 평가 실행(evaluation runs)
• 데이터 보강(data enrichment) 파이프라인

다음 같은 경우에는 사용하지 않습니다. 실시간 사용자 대상(user-facing) 질의가 여기에 해당하며, 이때는 지연 시간이 중요합니다.

예산 알림과 회로 차단기(Budget Alerts and Circuit Breakers)

회로 차단기(circuit breaker)는 한도에 도달하면 지출을 멈춥니다. 회로 차단기가 없으면 버그나 악용(abuse)이 몇 시간 만에 월간 예산을 태울 수 있습니다.

세 가지 임계값을 설정합니다.

1. 경고(Warning) (예산 70%): 알림을 보냅니다.
2. 스로틀(Throttle) (예산 85%): 더 저렴한 모델만 사용하도록 전환합니다.
3. 중지(Stop) (예산 95%): 새 요청을 거부하고 캐시된 응답만 반환합니다.

최적화 스택(The Optimization Stack)

다음 기법을 순서대로 적용합니다. 각 계층은 이전 계층 위에서 복합 효과를 냅니다.

1-5 계층을 적용한 RAG 앱은 보통 비용이 월 $22,500에서 $4,000-6,000 수준으로 내려갑니다. 이것이 런웨이(runway)를 태우는 것과 사업을 만드는 것의 차이입니다.

실제 절감: 전과 후(Real Savings: Before and After)

다음은 10,000 DAU를 처리하는 RAG 챗봇의 실제 세부 내역(breakdown)입니다.

의미 기반 캐싱을 위한 임베딩 비용($180/month)은 캐시 적중이 시작된 첫 한 시간 안에 스스로를 회수합니다.

직접 만들기

Step 1: 비용 계산기(Cost Calculator)

주요 모델의 현재 가격을 알고 있는 토큰 비용 계산기를 만듭니다.

import hashlib
import time
import json
import math
from dataclasses import dataclass, field


MODEL_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00, "cached_input": 1.25},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60, "cached_input": 0.075},
    "gpt-4.1": {"input": 2.00, "output": 8.00, "cached_input": 0.50},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60, "cached_input": 0.10},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40, "cached_input": 0.025},
    "o3": {"input": 2.00, "output": 8.00, "cached_input": 0.50},
    "o3-mini": {"input": 1.10, "output": 4.40, "cached_input": 0.55},
    "o4-mini": {"input": 1.10, "output": 4.40, "cached_input": 0.275},
    "claude-opus-4": {"input": 15.00, "output": 75.00, "cached_input": 1.50},
    "claude-sonnet-4": {"input": 3.00, "output": 15.00, "cached_input": 0.30},
    "claude-haiku-3.5": {"input": 0.80, "output": 4.00, "cached_input": 0.08},
    "gemini-2.5-pro": {"input": 1.25, "output": 10.00, "cached_input": 0.3125},
    "gemini-2.5-flash": {"input": 0.15, "output": 0.60, "cached_input": 0.0375},
}


def calculate_cost(model, input_tokens, output_tokens, cached_input_tokens=0):
    if model not in MODEL_PRICING:
        return {"error": f"알 수 없는 모델입니다: {model}"}
    pricing = MODEL_PRICING[model]
    non_cached = input_tokens - cached_input_tokens
    input_cost = (non_cached / 1_000_000) * pricing["input"]
    cached_cost = (cached_input_tokens / 1_000_000) * pricing["cached_input"]
    output_cost = (output_tokens / 1_000_000) * pricing["output"]
    total = input_cost + cached_cost + output_cost
    return {
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "cached_input_tokens": cached_input_tokens,
        "input_cost": round(input_cost, 6),
        "cached_input_cost": round(cached_cost, 6),
        "output_cost": round(output_cost, 6),
        "total_cost": round(total, 6),
    }

Step 2: 정확 캐시(Exact Cache)

전체 프롬프트를 해시하고 동일한 요청에는 캐시된 응답을 반환합니다.

class ExactCache:
    def __init__(self, max_size=1000, ttl_seconds=3600):
        self.cache = {}
        self.max_size = max_size
        self.ttl = ttl_seconds
        self.hits = 0
        self.misses = 0

    def _hash(self, model, messages, temperature):
        key_data = json.dumps({"model": model, "messages": messages, "temperature": temperature}, sort_keys=True)
        return hashlib.sha256(key_data.encode()).hexdigest()

    def get(self, model, messages, temperature=0.0):
        if temperature > 0:
            self.misses += 1
            return None
        key = self._hash(model, messages, temperature)
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                self.hits += 1
                entry["access_count"] += 1
                return entry["response"]
            del self.cache[key]
        self.misses += 1
        return None

    def put(self, model, messages, temperature, response):
        if temperature > 0:
            return
        if len(self.cache) >= self.max_size:
            oldest_key = min(self.cache, key=lambda k: self.cache[k]["timestamp"])
            del self.cache[oldest_key]
        key = self._hash(model, messages, temperature)
        self.cache[key] = {
            "response": response,
            "timestamp": time.time(),
            "access_count": 1,
        }

    def stats(self):
        total = self.hits + self.misses
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": round(self.hits / total, 4) if total > 0 else 0,
            "cache_size": len(self.cache),
        }

Step 3: 의미 기반 캐시(Semantic Cache)

질의를 임베딩하고, 유사도가 임계값을 넘으면 캐시된 응답을 반환합니다.

def simple_embed(text):
    words = text.lower().split()
    vocab = {}
    for w in words:
        vocab[w] = vocab.get(w, 0) + 1
    norm = math.sqrt(sum(v * v for v in vocab.values()))
    if norm == 0:
        return {}
    return {k: v / norm for k, v in vocab.items()}


def cosine_similarity(a, b):
    if not a or not b:
        return 0.0
    all_keys = set(a) | set(b)
    dot = sum(a.get(k, 0) * b.get(k, 0) for k in all_keys)
    return dot


class SemanticCache:
    def __init__(self, similarity_threshold=0.85, max_size=500, ttl_seconds=3600):
        self.entries = []
        self.threshold = similarity_threshold
        self.max_size = max_size
        self.ttl = ttl_seconds
        self.hits = 0
        self.misses = 0

    def get(self, query):
        query_embedding = simple_embed(query)
        now = time.time()
        best_match = None
        best_sim = 0.0
        for entry in self.entries:
            if now - entry["timestamp"] > self.ttl:
                continue
            sim = cosine_similarity(query_embedding, entry["embedding"])
            if sim > best_sim:
                best_sim = sim
                best_match = entry
        if best_match and best_sim >= self.threshold:
            self.hits += 1
            best_match["access_count"] += 1
            return {"response": best_match["response"], "similarity": round(best_sim, 4), "original_query": best_match["query"]}
        self.misses += 1
        return None

    def put(self, query, response):
        if len(self.entries) >= self.max_size:
            self.entries.sort(key=lambda e: e["timestamp"])
            self.entries.pop(0)
        self.entries.append({
            "query": query,
            "embedding": simple_embed(query),
            "response": response,
            "timestamp": time.time(),
            "access_count": 1,
        })

    def stats(self):
        total = self.hits + self.misses
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": round(self.hits / total, 4) if total > 0 else 0,
            "cache_size": len(self.entries),
        }

Step 4: 속도 제한기(Rate Limiter)

사용자별 할당량을 갖춘 토큰 버킷 속도 제한기(token bucket rate limiter)를 만듭니다.

class TokenBucketRateLimiter:
    def __init__(self):
        self.buckets = {}
        self.tiers = {
            "free": {"capacity": 50_000, "refill_rate": 500, "max_requests_per_min": 10},
            "pro": {"capacity": 500_000, "refill_rate": 5_000, "max_requests_per_min": 60},
            "enterprise": {"capacity": 5_000_000, "refill_rate": 50_000, "max_requests_per_min": 300},
        }

    def _get_bucket(self, user_id, tier="free"):
        if user_id not in self.buckets:
            tier_config = self.tiers.get(tier, self.tiers["free"])
            self.buckets[user_id] = {
                "tokens": tier_config["capacity"],
                "capacity": tier_config["capacity"],
                "refill_rate": tier_config["refill_rate"],
                "last_refill": time.time(),
                "request_timestamps": [],
                "max_rpm": tier_config["max_requests_per_min"],
                "tier": tier,
                "total_tokens_used": 0,
            }
        return self.buckets[user_id]

    def _refill(self, bucket):
        now = time.time()
        elapsed = now - bucket["last_refill"]
        refill = int(elapsed * bucket["refill_rate"])
        if refill > 0:
            bucket["tokens"] = min(bucket["capacity"], bucket["tokens"] + refill)
            bucket["last_refill"] = now

    def check(self, user_id, tokens_needed, tier="free"):
        bucket = self._get_bucket(user_id, tier)
        self._refill(bucket)
        now = time.time()
        bucket["request_timestamps"] = [t for t in bucket["request_timestamps"] if now - t < 60]
        if len(bucket["request_timestamps"]) >= bucket["max_rpm"]:
            return {"allowed": False, "reason": "rate_limit", "retry_after_seconds": 60 - (now - bucket["request_timestamps"][0])}
        if bucket["tokens"] < tokens_needed:
            deficit = tokens_needed - bucket["tokens"]
            wait = deficit / bucket["refill_rate"]
            return {"allowed": False, "reason": "token_limit", "tokens_available": bucket["tokens"], "retry_after_seconds": round(wait, 1)}
        return {"allowed": True, "tokens_available": bucket["tokens"]}

    def consume(self, user_id, tokens_used, tier="free"):
        bucket = self._get_bucket(user_id, tier)
        bucket["tokens"] -= tokens_used
        bucket["request_timestamps"].append(time.time())
        bucket["total_tokens_used"] += tokens_used

    def get_usage(self, user_id):
        if user_id not in self.buckets:
            return {"error": "사용자를 찾을 수 없습니다"}
        b = self.buckets[user_id]
        return {
            "user_id": user_id,
            "tier": b["tier"],
            "tokens_remaining": b["tokens"],
            "capacity": b["capacity"],
            "total_tokens_used": b["total_tokens_used"],
            "utilization": round(b["total_tokens_used"] / b["capacity"], 4) if b["capacity"] else 0,
        }

Step 5: 비용 추적기(Cost Tracker)

모든 호출을 기록하고 누적 합계를 계산합니다.

class CostTracker:
    def __init__(self, monthly_budget=1000.0):
        self.logs = []
        self.monthly_budget = monthly_budget
        self.alerts = []

    def log_call(self, model, input_tokens, output_tokens, cached_input_tokens=0, latency_ms=0, user_id="anonymous", cache_status="miss"):
        cost = calculate_cost(model, input_tokens, output_tokens, cached_input_tokens)
        entry = {
            "timestamp": time.time(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cached_input_tokens": cached_input_tokens,
            "latency_ms": latency_ms,
            "cost": cost["total_cost"],
            "user_id": user_id,
            "cache_status": cache_status,
        }
        self.logs.append(entry)
        self._check_budget()
        return entry

    def _check_budget(self):
        total = self.total_cost()
        pct = total / self.monthly_budget if self.monthly_budget > 0 else 0
        if pct >= 0.95 and not any(a["level"] == "stop" for a in self.alerts):
            self.alerts.append({"level": "stop", "message": f"예산 95% 사용: ${total:.2f}/${self.monthly_budget:.2f}", "timestamp": time.time()})
        elif pct >= 0.85 and not any(a["level"] == "throttle" for a in self.alerts):
            self.alerts.append({"level": "throttle", "message": f"예산 85% 사용: ${total:.2f}/${self.monthly_budget:.2f}", "timestamp": time.time()})
        elif pct >= 0.70 and not any(a["level"] == "warning" for a in self.alerts):
            self.alerts.append({"level": "warning", "message": f"예산 70% 사용: ${total:.2f}/${self.monthly_budget:.2f}", "timestamp": time.time()})

    def total_cost(self):
        return round(sum(e["cost"] for e in self.logs), 6)

    def cost_by_model(self):
        by_model = {}
        for e in self.logs:
            m = e["model"]
            if m not in by_model:
                by_model[m] = {"calls": 0, "cost": 0, "input_tokens": 0, "output_tokens": 0}
            by_model[m]["calls"] += 1
            by_model[m]["cost"] = round(by_model[m]["cost"] + e["cost"], 6)
            by_model[m]["input_tokens"] += e["input_tokens"]
            by_model[m]["output_tokens"] += e["output_tokens"]
        return by_model

    def cache_savings(self):
        cache_hits = [e for e in self.logs if e["cache_status"] == "hit"]
        if not cache_hits:
            return {"saved": 0, "cache_hits": 0}
        saved = 0
        for e in cache_hits:
            full_cost = calculate_cost(e["model"], e["input_tokens"], e["output_tokens"])
            saved += full_cost["total_cost"]
        return {"saved": round(saved, 4), "cache_hits": len(cache_hits)}

    def summary(self):
        if not self.logs:
            return {"total_calls": 0, "total_cost": 0}
        total_latency = sum(e["latency_ms"] for e in self.logs)
        cache_hits = sum(1 for e in self.logs if e["cache_status"] == "hit")
        return {
            "total_calls": len(self.logs),
            "total_cost": self.total_cost(),
            "avg_cost_per_call": round(self.total_cost() / len(self.logs), 6),
            "avg_latency_ms": round(total_latency / len(self.logs), 1),
            "cache_hit_rate": round(cache_hits / len(self.logs), 4),
            "cost_by_model": self.cost_by_model(),
            "cache_savings": self.cache_savings(),
            "budget_remaining": round(self.monthly_budget - self.total_cost(), 2),
            "budget_utilization": round(self.total_cost() / self.monthly_budget, 4) if self.monthly_budget > 0 else 0,
            "alerts": self.alerts,
        }

Step 6: 모델 라우터(Model Router)

각 질의를 처리할 수 있는 가장 저렴한 모델로 라우팅합니다.

SIMPLE_KEYWORDS = ["what time", "hours", "address", "phone", "price", "return policy", "hello", "hi", "thanks", "yes", "no"]
COMPLEX_KEYWORDS = ["analyze", "compare", "explain why", "write code", "debug", "architect", "design", "trade-off", "evaluate"]


def classify_complexity(query):
    q = query.lower()
    if len(q.split()) <= 5 or any(kw in q for kw in SIMPLE_KEYWORDS):
        return "simple"
    if any(kw in q for kw in COMPLEX_KEYWORDS):
        return "complex"
    return "medium"


def route_model(query, tier="pro"):
    complexity = classify_complexity(query)
    routing_table = {
        "simple": {"free": "gpt-4.1-nano", "pro": "gpt-4o-mini", "enterprise": "gpt-4o-mini"},
        "medium": {"free": "gpt-4o-mini", "pro": "claude-sonnet-4", "enterprise": "claude-sonnet-4"},
        "complex": {"free": "gpt-4o-mini", "pro": "gpt-4o", "enterprise": "claude-opus-4"},
    }
    model = routing_table[complexity].get(tier, "gpt-4o-mini")
    return {"query": query, "complexity": complexity, "model": model, "tier": tier}

Step 7: 데모 실행하기

def simulate_llm_call(model, query):
    input_tokens = len(query.split()) * 4 + 500
    output_tokens = 150 + (len(query.split()) * 2)
    latency = 200 + (output_tokens * 2)
    return {
        "model": model,
        "response": f"[{model}의 시뮬레이션 응답: {query[:50]}...]",
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "latency_ms": latency,
    }


def run_demo():
    print("=" * 60)
    print("  캐싱, 속도 제한과 비용 최적화 데모")
    print("=" * 60)

    print("\n--- 모델 가격 ---")
    for model, pricing in list(MODEL_PRICING.items())[:6]:
        cost_1k = calculate_cost(model, 1000, 500)
        print(f"  {model}: 입력 1K + 출력 500 기준 ${cost_1k['total_cost']:.6f}")

    print("\n--- 비용 비교: 요청 10만 건 ---")
    for model in ["gpt-4o", "gpt-4o-mini", "claude-sonnet-4", "claude-haiku-3.5"]:
        cost = calculate_cost(model, 1000 * 100_000, 500 * 100_000)
        print(f"  {model}: ${cost['total_cost']:.2f}")

    print("\n--- Anthropic 캐시 절감 효과 ---")
    no_cache = calculate_cost("claude-sonnet-4", 2000, 500, 0)
    with_cache = calculate_cost("claude-sonnet-4", 2000, 500, 1500)
    saving = no_cache["total_cost"] - with_cache["total_cost"]
    print(f"  캐시 없음: ${no_cache['total_cost']:.6f}")
    print(f"  캐시된 토큰 1500개 사용: ${with_cache['total_cost']:.6f}")
    print(f"  호출당 절감액: ${saving:.6f} ({saving/no_cache['total_cost']*100:.1f}%)")

    exact_cache = ExactCache(max_size=100, ttl_seconds=300)
    semantic_cache = SemanticCache(similarity_threshold=0.75, max_size=100)
    rate_limiter = TokenBucketRateLimiter()
    tracker = CostTracker(monthly_budget=100.0)

    print("\n--- 정확 캐시 ---")
    messages_1 = [{"role": "user", "content": "What is the return policy?"}]
    result = exact_cache.get("gpt-4o-mini", messages_1, 0.0)
    print(f"  첫 번째 조회: {'HIT' if result else 'MISS'}")
    exact_cache.put("gpt-4o-mini", messages_1, 0.0, "You can return items within 30 days.")
    result = exact_cache.get("gpt-4o-mini", messages_1, 0.0)
    print(f"  두 번째 조회: {'HIT' if result else 'MISS'} -> {result}")
    result = exact_cache.get("gpt-4o-mini", messages_1, 0.7)
    print(f"  temp=0.7: {'HIT' if result else 'MISS (비결정적 호출이므로 캐시 건너뜀)'}")
    print(f"  통계: {exact_cache.stats()}")

    print("\n--- 의미 기반 캐시 ---")
    test_queries = [
        ("What is the return policy?", "Items can be returned within 30 days with receipt."),
        ("How do I return an item?", None),
        ("What are your store hours?", "We are open 9am-9pm Monday through Saturday."),
        ("When does the store open?", None),
        ("Tell me about quantum computing", "Quantum computers use qubits..."),
        ("Explain quantum mechanics", None),
    ]
    for query, response in test_queries:
        cached = semantic_cache.get(query)
        if cached:
            print(f"  '{query[:40]}' -> CACHE HIT (sim={cached['similarity']}, original='{cached['original_query'][:40]}')")
        elif response:
            semantic_cache.put(query, response)
            print(f"  '{query[:40]}' -> MISS (저장함)")
        else:
            print(f"  '{query[:40]}' -> MISS (매칭 없음)")
    print(f"  통계: {semantic_cache.stats()}")

    print("\n--- 속도 제한 ---")
    for i in range(12):
        check = rate_limiter.check("user_1", 1000, "free")
        if check["allowed"]:
            rate_limiter.consume("user_1", 1000, "free")
        status = "OK" if check["allowed"] else f"BLOCKED ({check['reason']})"
        if i < 5 or not check["allowed"]:
            print(f"  요청 {i+1}: {status}")
    print(f"  사용량: {rate_limiter.get_usage('user_1')}")

    print("\n--- 모델 라우팅 ---")
    routing_queries = [
        "What time do you close?",
        "Summarize this quarterly earnings report",
        "Analyze the trade-offs between microservices and monoliths",
        "Hello",
        "Write code for a binary search tree with deletion",
    ]
    for q in routing_queries:
        route = route_model(q, "pro")
        print(f"  '{q[:50]}' -> {route['model']} ({route['complexity']})")

    print("\n--- 전체 파이프라인: 최적화 전과 후 ---")
    queries = [
        "What is the return policy?",
        "How do I return something?",
        "What are your hours?",
        "When do you open?",
        "Explain the difference between TCP and UDP",
        "Compare TCP vs UDP protocols",
        "Hello",
        "What is your phone number?",
        "Write a Python function to sort a list",
        "Analyze the pros and cons of serverless architecture",
    ]

    print("\n  [최적화 전: 캐시 없음, 단일 모델(gpt-4o)]")
    tracker_before = CostTracker(monthly_budget=1000.0)
    for q in queries:
        result = simulate_llm_call("gpt-4o", q)
        tracker_before.log_call("gpt-4o", result["input_tokens"], result["output_tokens"], latency_ms=result["latency_ms"], cache_status="miss")
    before = tracker_before.summary()
    print(f"  총비용: ${before['total_cost']:.6f}")
    print(f"  호출당 평균 비용: ${before['avg_cost_per_call']:.6f}")
    print(f"  평균 지연 시간: {before['avg_latency_ms']}ms")

    print("\n  [최적화 후: 캐싱 + 라우팅 + 속도 제한]")
    exact_c = ExactCache()
    semantic_c = SemanticCache(similarity_threshold=0.75)
    tracker_after = CostTracker(monthly_budget=1000.0)

    for q in queries:
        messages = [{"role": "user", "content": q}]
        cached = exact_c.get("gpt-4o", messages, 0.0)
        if cached:
            tracker_after.log_call("gpt-4o-mini", 0, 0, latency_ms=5, cache_status="hit")
            continue
        sem_cached = semantic_c.get(q)
        if sem_cached:
            tracker_after.log_call("gpt-4o-mini", 0, 0, latency_ms=15, cache_status="hit")
            continue
        route = route_model(q)
        result = simulate_llm_call(route["model"], q)
        tracker_after.log_call(route["model"], result["input_tokens"], result["output_tokens"], latency_ms=result["latency_ms"], cache_status="miss")
        exact_c.put(route["model"], messages, 0.0, result["response"])
        semantic_c.put(q, result["response"])

    after = tracker_after.summary()
    print(f"  총비용: ${after['total_cost']:.6f}")
    print(f"  호출당 평균 비용: ${after['avg_cost_per_call']:.6f}")
    print(f"  평균 지연 시간: {after['avg_latency_ms']}ms")
    print(f"  캐시 적중률: {after['cache_hit_rate']:.0%}")

    if before["total_cost"] > 0:
        savings_pct = (1 - after["total_cost"] / before["total_cost"]) * 100
        print(f"\n  절감 효과: 비용 {savings_pct:.1f}% 감소")
        print(f"  지연 시간 개선: {(1 - after['avg_latency_ms'] / before['avg_latency_ms']) * 100:.1f}% 빨라짐")

    print("\n--- 예산 알림 데모 ---")
    alert_tracker = CostTracker(monthly_budget=0.01)
    for i in range(5):
        alert_tracker.log_call("gpt-4o", 5000, 2000, latency_ms=500)
    print(f"  총 지출: ${alert_tracker.total_cost():.6f} / ${alert_tracker.monthly_budget}")
    for alert in alert_tracker.alerts:
        print(f"  ALERT [{alert['level'].upper()}]: {alert['message']}")

    print("\n--- 모델별 비용 분석 ---")
    multi_tracker = CostTracker(monthly_budget=500.0)
    for _ in range(50):
        multi_tracker.log_call("gpt-4o-mini", 800, 200, latency_ms=150)
    for _ in range(30):
        multi_tracker.log_call("claude-sonnet-4", 1500, 500, latency_ms=400)
    for _ in range(10):
        multi_tracker.log_call("gpt-4o", 2000, 800, latency_ms=600)
    for _ in range(10):
        multi_tracker.log_call("claude-opus-4", 3000, 1000, latency_ms=1200)
    breakdown = multi_tracker.cost_by_model()
    for model, data in sorted(breakdown.items(), key=lambda x: x[1]["cost"], reverse=True):
        print(f"  {model}: {data['calls']}회 호출, ${data['cost']:.6f}, 입력 {data['input_tokens']:,} / 출력 {data['output_tokens']:,}")
    print(f"  합계: ${multi_tracker.total_cost():.6f}")

    print("\n" + "=" * 60)
    print("  데모 완료.")
    print("=" * 60)


if __name__ == "__main__":
    run_demo()

사용해보기

Anthropic 프롬프트 캐싱(Anthropic Prompt Caching)

# import anthropic
#
# client = anthropic.Anthropic()
#
# response = client.messages.create(
#     model="claude-sonnet-4-20250514",
#     max_tokens=1024,
#     system=[
#         {
#             "type": "text",
#             "text": "당신은 Acme Corp의 친절한 고객 지원 상담원입니다...",
#             "cache_control": {"type": "ephemeral"},
#         }
#     ],
#     messages=[{"role": "user", "content": "반품 정책이 어떻게 되나요?"}],
# )
#
# print(f"입력 토큰: {response.usage.input_tokens}")
# print(f"캐시 생성 토큰: {response.usage.cache_creation_input_tokens}")
# print(f"캐시 읽기 토큰: {response.usage.cache_read_input_tokens}")

첫 호출은 캐시에 씁니다(25% 프리미엄). 같은 시스템 프롬프트 접두사를 가진 이후 모든 호출은 캐시에서 읽습니다(90% 할인). 캐시는 5분 동안 유지되고, 적중할 때마다 타이머가 초기화됩니다.

OpenAI 자동 캐싱(OpenAI Automatic Caching)

# from openai import OpenAI
#
# client = OpenAI()
#
# response = client.chat.completions.create(
#     model="gpt-4o",
#     messages=[
#         {"role": "system", "content": "당신은 친절한 고객 지원 상담원입니다..."},
#         {"role": "user", "content": "반품 정책이 어떻게 되나요?"},
#     ],
# )
#
# print(f"프롬프트 토큰: {response.usage.prompt_tokens}")
# print(f"캐시된 토큰: {response.usage.prompt_tokens_details.cached_tokens}")
# print(f"완성 토큰: {response.usage.completion_tokens}")

OpenAI는 자동으로 캐싱합니다. 최근 요청과 일치하는 1,024토큰 이상의 프롬프트 접두사는 50% 할인을 받습니다. 코드 변경은 필요 없습니다. 응답의 prompt_tokens_details.cached_tokens를 확인해 동작 여부를 검증하면 됩니다.

OpenAI Batch API

# import json
# from openai import OpenAI
#
# client = OpenAI()
#
# requests = []
# for i, query in enumerate(queries):
#     requests.append({
#         "custom_id": f"request-{i}",
#         "method": "POST",
#         "url": "/v1/chat/completions",
#         "body": {
#             "model": "gpt-4o-mini",
#             "messages": [{"role": "user", "content": query}],
#         },
#     })
#
# with open("batch_input.jsonl", "w") as f:
#     for r in requests:
#         f.write(json.dumps(r) + "\n")
#
# batch_file = client.files.create(file=open("batch_input.jsonl", "rb"), purpose="batch")
# batch = client.batches.create(input_file_id=batch_file.id, endpoint="/v1/chat/completions", completion_window="24h")
# print(f"Batch ID: {batch.id}, 상태: {batch.status}")

Batch API는 모든 토큰에 일괄 50% 할인을 제공합니다. 결과는 24시간 안에 도착합니다. 평가, 데이터 라벨링(data labeling), 대량 요약(bulk summarization)처럼 실시간이 아닌 작업에 적합합니다.

Redis를 사용한 프로덕션 의미 기반 캐시(Production Semantic Cache with Redis)

# import redis
# import numpy as np
# from openai import OpenAI
#
# r = redis.Redis()
# client = OpenAI()
#
# def get_embedding(text):
#     response = client.embeddings.create(model="text-embedding-3-small", input=text)
#     return response.data[0].embedding
#
# def semantic_cache_lookup(query, threshold=0.95):
#     query_emb = np.array(get_embedding(query))
#     keys = r.keys("cache:emb:*")
#     best_sim, best_key = 0, None
#     for key in keys:
#         stored_emb = np.frombuffer(r.get(key), dtype=np.float32)
#         sim = np.dot(query_emb, stored_emb) / (np.linalg.norm(query_emb) * np.linalg.norm(stored_emb))
#         if sim > best_sim:
#             best_sim, best_key = sim, key
#     if best_sim >= threshold and best_key:
#         response_key = best_key.decode().replace("cache:emb:", "cache:resp:")
#         return r.get(response_key).decode()
#     return None

프로덕션에서는 선형 스캔(linear scan)을 벡터 인덱스(vector index)로 바꿉니다. Redis Vector Search, Pinecone, pgvector를 사용할 수 있습니다. 선형 스캔은 항목이 1,000개 미만일 때는 괜찮습니다. 그 이상에서는 O(log n) 조회를 위해 ANN(approximate nearest neighbor)을 사용합니다.

산출물 만들기

이 lesson은 outputs/prompt-cost-optimizer.md를 만듭니다. 이는 LLM 애플리케이션을 분석하고 예상 절감액과 함께 구체적인 비용 최적화를 추천하는 재사용 가능한 프롬프트입니다.

또한 outputs/skill-cost-patterns.md를 만듭니다. 이는 사용 사례에 맞는 캐싱 전략, 속도 제한 설정, 모델 라우팅 규칙을 선택하기 위한 의사결정 프레임워크입니다.

연습문제

1. 의미 기반 캐시에 LRU 퇴출(LRU eviction) 구현하기. 가장 오래된 항목을 먼저 제거하는 방식(oldest-first eviction)을 LRU(Least Recently Used) 방식으로 바꿉니다. 각 항목의 마지막 접근 시간을 추적하고, 캐시가 가득 찼을 때 가장 오래 접근되지 않은 항목을 제거합니다. 100개 질의에서 두 전략의 적중률을 비교합니다.

2. 비용 예측 도구 만들기. API 호출 로그(CostTracker 로그)를 입력받아 최근 7일 평균을 기준으로 월간 비용을 예측합니다. 평일/주말 패턴을 반영합니다. 예상 월간 비용이 예산을 20% 이상 초과하면 알림을 트리거합니다.

3. 계층형 의미 기반 캐싱 구현하기. 두 가지 유사도 임계값을 사용합니다. 0.98은 고신뢰 적중(high-confidence hit)으로 즉시 반환하고, 0.90은 중간 신뢰 적중(medium-confidence hit)으로 "이전의 유사한 질문을 기반으로..." 같은 고지와 함께 반환합니다. 각 적중이 어떤 계층(tier)에서 왔는지 추적하고 사용자 만족도 차이를 측정합니다.

4. 모델 라우팅 분류기 만들기. 키워드 기반 분류기를 임베딩 기반 분류기로 바꿉니다. 라벨(simple/medium/complex)이 붙은 질의 50개를 임베딩한 뒤, 새 질의는 가장 가까운 라벨 예시를 찾아 분류합니다. 20개 질의 테스트 세트에서 분류 정확도를 측정합니다.

5. 성능 저하 단계가 있는 회로 차단기 구현하기. 예산 70%에서는 경고를 기록합니다. 85%에서는 모든 라우팅을 가장 저렴한 모델(gpt-4o-mini)로 자동 전환합니다. 95%에서는 캐시된 응답만 제공하고 새 질의는 거부합니다. $1.00 예산으로 1,000개 요청을 시뮬레이션해 각 임계값이 정확히 트리거되는지 검증합니다.

핵심 용어

더 읽을거리

• Anthropic Prompt Caching Guide - Anthropic의 명시적 cache_control 마커, 가격, 캐시 수명 동작에 대한 공식 문서입니다.
• OpenAI Prompt Caching - OpenAI의 자동 캐싱, usage 필드로 캐시 적중을 확인하는 방법, 최소 접두사 길이를 설명합니다.
• OpenAI Batch API - 비동기 처리 50% 할인, JSONL 형식, 24시간 완료 창, 50K 요청 한도를 설명합니다.
• GPTCache - 여러 임베딩 백엔드, 벡터 저장소, 퇴출(eviction) 정책을 지원하는 오픈소스 의미 기반 캐싱 라이브러리입니다.
• Martian Model Router - 각 질의를 처리할 수 있는 가장 저렴한 모델을 자동 선택하는 프로덕션 모델 라우팅입니다.
• Not Diamond - 제공자 전반에서 비용/품질 트레이드오프(tradeoff)를 최적화하기 위해 트래픽 패턴에서 학습하는 ML 기반 모델 라우터입니다.
• Helicone - 프록시 계층(proxy layer)으로 비용 추적, 캐싱, 속도 제한, 예산 알림을 제공하는 LLM 관측 가능성 플랫폼입니다.
• Dean & Barroso, "The Tail at Scale" (CACM 2013) - 지연 시간, 처리량, TTFT/TPOT 백분위, 중복 송신 요청(hedged request)을 다룹니다. "P95를 만족하는 가장 저렴한 모델을 고른다"는 비용 모델의 기반입니다.
• Kwon et al., "Efficient Memory Management for Large Language Model Serving with PagedAttention" (SOSP 2023) - vLLM 논문입니다. paged KV-cache와 연속 배치(continuous batching)가 단순 서버(naive server)보다 처리량을 24배 높이는 이유를 설명합니다. "캐싱과 비용" 아래의 인프라 계층입니다.
• Dao et al., "FlashAttention-2: Faster Attention with Better Parallelism and Work Partitioning" (ICLR 2024) - 프롬프트 캐싱과 직교하는 커널(kernel) 수준 비용 절감입니다. 전체 비용 곡선을 이해하려면 추측 디코딩(speculative decoding), GQA와 함께 읽으면 좋습니다.