LLM API 비용을 90% 줄이는 실전 최적화 전략

시나리오: 소규모 B2B SaaS, 일 활성 사용자 5,000명

사용 패턴:
- 사용자 1명 × 하루 10회 대화
- 대화 1회 = 입력 200 토큰 + 출력 300 토큰

일일 토큰 사용량:
5,000명 × 10회 × 500 토큰 = 25,000,000 토큰/일 (2,500만 토큰)

월 기준:
25,000,000 × 30 = 750,000,000 토큰/월 (7억 5천만 토큰)

비용 비교 (월):
- GPT-4o:          $2.50/1M input + $10/1M output
  → 입력 $37,500 + 출력 $45,000 = 월 $82,500
- GPT-4o-mini:     $0.15/1M input + $0.60/1M output
  → 입력 $2,250 + 출력 $2,700 = 월 $4,950
- 자체 호스팅 Llama: 서버 비용 ~$500-2,000/월

import anthropic

client = anthropic.Anthropic()

# 회사 전체 RAG 컨텍스트나 긴 시스템 프롬프트
COMPANY_KNOWLEDGE_BASE = """
[여기에 수천 토큰의 회사 문서, 제품 정보, 정책 등]
...이 내용이 매 요청마다 반복 전송되면 비용이 폭발한다.
"""

def chat_with_caching(user_message: str) -> dict:
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": COMPANY_KNOWLEDGE_BASE,
                "cache_control": {"type": "ephemeral"}  # 이 블록을 캐시!
            },
            {
                "type": "text",
                "text": "당신은 고객 지원 전문가입니다. 위의 회사 정보를 바탕으로 답변하세요."
                # 이 부분은 캐시하지 않음 (짧고 변경 가능)
            }
        ],
        messages=[{"role": "user", "content": user_message}]
    )

    # 사용량 추적
    usage = response.usage
    print(f"캐시 쓰기: {usage.cache_creation_input_tokens} tokens (1.25x 비용)")
    print(f"캐시 읽기: {usage.cache_read_input_tokens} tokens (0.1x 비용! 90% 절감)")
    print(f"일반 입력: {usage.input_tokens} tokens (1x 비용)")

    return {
        "content": response.content[0].text,
        "cache_saved": usage.cache_read_input_tokens > 0
    }

# 첫 번째 호출: 캐시 생성 (1.25x 비용)
result1 = chat_with_caching("환불 정책이 어떻게 되나요?")

# 두 번째 호출부터: 캐시 히트 (0.1x 비용!)
result2 = chat_with_caching("배송 기간은 얼마나 걸리나요?")
result3 = chat_with_caching("제품 보증 기간은?")

from openai import OpenAI
client = OpenAI()

# OpenAI는 1,024 토큰 이상의 프롬프트를 자동으로 캐싱
# 별도 설정 없이 50% 할인 자동 적용
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            # 1,024 토큰 이상이면 자동 캐싱 (50% 할인)
            "content": LONG_SYSTEM_PROMPT  # 2,000 토큰 이상 권장
        },
        {"role": "user", "content": "질문입니다"}
    ]
)

# 캐시 히트 확인
usage = response.usage
if hasattr(usage, 'prompt_tokens_details'):
    cached = usage.prompt_tokens_details.cached_tokens
    print(f"캐시된 토큰: {cached} (50% 할인 적용)")

from openai import OpenAI
from anthropic import Anthropic
import re

openai_client = OpenAI()
anthropic_client = Anthropic()

class ModelRouter:
    """요청 복잡도에 따라 최적 모델로 라우팅"""

    # 간단한 요청 패턴
    SIMPLE_PATTERNS = [
        r"^(what is|what are|define|who is|when was|where is)",
        r"^(번역|translate|어떻게 말해)",
        r"^(yes/no|맞나요|맞아요\?)",
    ]

    # 복잡한 처리가 필요한 패턴
    COMPLEX_PATTERNS = [
        r"(analyze|분석|compare|비교|design|설계)",
        r"(step.by.step|단계별|detailed|자세히)",
        r"(code|코드|implement|구현|architecture|아키텍처)",
        r"(explain why|왜|reason|이유|pros.cons|장단점)",
    ]

    def classify_query(self, query: str) -> str:
        """쿼리 복잡도 분류: 'simple', 'medium', 'complex'"""
        query_lower = query.lower()

        # 복잡한 요청
        if any(re.search(p, query_lower) for p in self.COMPLEX_PATTERNS):
            return "complex"

        # 단순한 요청
        if (len(query.split()) < 15 and
                any(re.search(p, query_lower) for p in self.SIMPLE_PATTERNS)):
            return "simple"

        # 길이 기반 추가 판단
        if len(query) > 500:
            return "complex"

        return "medium"

    def route(self, query: str, context: str = "") -> dict:
        """적절한 모델로 라우팅하여 요청 처리"""

        complexity = self.classify_query(query)

        if complexity == "simple":
            # GPT-4o-mini: 복잡하지 않은 요청에 최적, 60배 저렴
            response = openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": query}],
                max_tokens=200
            )
            model_used = "gpt-4o-mini"
            cost_multiplier = 1  # 기준

        elif complexity == "medium":
            response = openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": query}],
                max_tokens=500
            )
            model_used = "gpt-4o-mini"
            cost_multiplier = 1

        else:  # complex
            # 복잡한 추론: GPT-4o 또는 Claude Sonnet 사용
            response = openai_client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": query}]
            )
            model_used = "gpt-4o"
            cost_multiplier = 60  # GPT-4o-mini 대비 약 60배

        return {
            "answer": response.choices[0].message.content,
            "model": model_used,
            "complexity": complexity
        }


# 실제 사용 + 비용 분석
router = ModelRouter()

queries = [
    "파이썬이 뭐야?",                           # simple
    "마이크로서비스 아키텍처를 설계해줘",         # complex
    "오늘 날씨 어때?",                           # simple
    "RESTful API와 GraphQL을 상세히 비교해줘",   # complex
    "Python 3.11 출시일이 언제야?",             # simple
]

# 80%가 simple/medium, 20%가 complex라고 가정하면:
# 전체를 GPT-4o로 처리 시 = 100% × 60 = 비용 60
# 라우팅 적용 시 = 80% × 1 + 20% × 60 = 80 + 12 = 비용 92의 차이
# → 약 85% 비용 절감!

import hashlib
import json
import numpy as np
from openai import OpenAI
from datetime import datetime, timedelta

client = OpenAI()

class SemanticCache:
    """
    의미적으로 유사한 쿼리를 캐싱
    "RAG가 뭐야?"와 "RAG에 대해 설명해줘"는 같은 캐시 항목을 반환
    """

    def __init__(self, similarity_threshold: float = 0.95, ttl_hours: int = 24):
        self.cache = {}  # {query_hash: {embedding, response, created_at}}
        self.threshold = similarity_threshold
        self.ttl = timedelta(hours=ttl_hours)
        self.stats = {"hits": 0, "misses": 0, "saved_tokens": 0}

    def _get_embedding(self, text: str) -> list:
        response = client.embeddings.create(
            input=text,
            model="text-embedding-3-small",
            dimensions=256  # 빠른 캐시 조회를 위해 작은 차원 사용
        )
        return response.data[0].embedding

    def _cosine_similarity(self, a: list, b: list) -> float:
        a_np, b_np = np.array(a), np.array(b)
        return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))

    def get(self, query: str) -> tuple:
        """캐시 조회. 반환: (cached_response or None, similarity_score)"""
        query_embedding = self._get_embedding(query)

        best_similarity = 0
        best_response = None

        for key, entry in self.cache.items():
            # TTL 확인
            if datetime.now() - entry["created_at"] > self.ttl:
                continue

            similarity = self._cosine_similarity(query_embedding, entry["embedding"])
            if similarity > best_similarity:
                best_similarity = similarity
                best_response = entry["response"]

        if best_similarity >= self.threshold:
            self.stats["hits"] += 1
            return best_response, best_similarity

        self.stats["misses"] += 1
        return None, best_similarity

    def set(self, query: str, response: str, tokens_used: int) -> None:
        """캐시에 저장"""
        embedding = self._get_embedding(query)
        key = hashlib.md5(query.encode()).hexdigest()
        self.cache[key] = {
            "embedding": embedding,
            "response": response,
            "created_at": datetime.now(),
            "tokens": tokens_used
        }

    def chat(self, query: str) -> dict:
        """캐시를 활용한 채팅"""
        cached_response, similarity = self.get(query)

        if cached_response:
            return {
                "response": cached_response,
                "cache_hit": True,
                "similarity": similarity,
                "api_cost": 0  # 캐시 히트 = API 비용 없음!
            }

        # 캐시 미스: API 호출
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": query}]
        )
        answer = response.choices[0].message.content
        tokens = response.usage.total_tokens

        # 캐시에 저장
        self.set(query, answer, tokens)

        return {
            "response": answer,
            "cache_hit": False,
            "similarity": 0,
            "api_cost": tokens * 0.000001  # 대략적인 비용
        }


# 실제 사용 예시
cache = SemanticCache(similarity_threshold=0.92)

# FAQ 사전 워밍 (자주 묻는 질문 미리 캐싱)
faq_questions = [
    "환불 정책이 어떻게 되나요?",
    "배송 기간은 얼마나 걸리나요?",
    "회원 가입 방법이 뭐예요?",
]

for q in faq_questions:
    cache.chat(q)  # 첫 번째 호출: API 호출 후 캐싱

# 이후 유사한 질문은 캐시에서 반환
result = cache.chat("반품하려면 어떻게 해야 하나요?")  # "환불 정책"과 유사 → 캐시 히트
result2 = cache.chat("배송이 얼마나 걸려요?")  # "배송 기간"과 동의어 → 캐시 히트

from openai import OpenAI
import json
import tempfile

client = OpenAI()

def run_batch_analysis(texts: list) -> list:
    """
    대량의 텍스트 분석을 50% 저렴하게 처리
    - 24시간 내 처리 (비실시간)
    - 50% 할인 자동 적용
    적합한 태스크: 감성 분석, 분류, 요약, 번역
    """

    # 배치 요청 준비
    requests = []
    for i, text in enumerate(texts):
        requests.append({
            "custom_id": f"request-{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4o-mini",
                "messages": [
                    {
                        "role": "system",
                        "content": "다음 텍스트의 감성을 분석하세요. JSON으로 응답: {\"sentiment\": \"positive/negative/neutral\", \"confidence\": 0.0-1.0}"
                    },
                    {"role": "user", "content": text}
                ],
                "max_tokens": 50
            }
        })

    # 임시 파일로 배치 업로드
    with tempfile.NamedTemporaryFile(mode='w', suffix='.jsonl', delete=False) as f:
        for req in requests:
            f.write(json.dumps(req, ensure_ascii=False) + '\n')
        temp_path = f.name

    # 파일 업로드
    with open(temp_path, 'rb') as f:
        batch_file = client.files.create(file=f, purpose="batch")

    # 배치 제출 (24시간 내 처리, 50% 할인)
    batch = client.batches.create(
        input_file_id=batch_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h"
    )

    print(f"배치 ID: {batch.id}")
    print(f"상태: {batch.status}")
    print(f"비용: 실시간 대비 50% 절감")
    return batch.id

def get_batch_results(batch_id: str) -> list:
    """배치 완료 후 결과 조회"""
    batch = client.batches.retrieve(batch_id)

    if batch.status != "completed":
        print(f"아직 처리 중: {batch.status}")
        return []

    # 결과 파일 다운로드
    result_file = client.files.content(batch.output_file_id)
    results = []
    for line in result_file.text.strip().split('\n'):
        result = json.loads(line)
        response_body = result["response"]["body"]
        content = response_body["choices"][0]["message"]["content"]
        results.append({
            "id": result["custom_id"],
            "result": json.loads(content)
        })

    return results

# 사용 예시: 상품 리뷰 10,000개 감성 분석
reviews = ["정말 좋은 제품이에요!", "배송이 너무 늦었어요", ...] * 5000  # 1만 개
batch_id = run_batch_analysis(reviews)
# 24시간 후:
results = get_batch_results(batch_id)

from openai import OpenAI
from pydantic import BaseModel
from typing import Literal

client = OpenAI()

# 나쁜 예: 긴 산문 형태의 응답 요청
def analyze_sentiment_verbose(text: str) -> str:
    """비효율적: 불필요하게 긴 출력 생성"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"다음 텍스트의 감성을 분석해주세요: {text}"
        }]
    )
    # 출력 예시: "이 텍스트는 긍정적인 감성을 나타냅니다. 작성자가 제품에 만족했음을 알 수 있으며..."
    # 평균 100-200 토큰 출력
    return response.choices[0].message.content


# 좋은 예: 구조화된 간결한 출력
class SentimentResult(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    confidence: float
    key_phrase: str  # 최대 5단어

def analyze_sentiment_structured(text: str) -> SentimentResult:
    """효율적: 필요한 정보만 담은 구조화 출력"""
    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Analyze sentiment: {text}"
        }],
        response_format=SentimentResult
    )
    # 출력 예시: {"sentiment": "positive", "confidence": 0.92, "key_phrase": "정말 좋은 제품"}
    # 평균 20-30 토큰 출력 → 80% 절감!
    return response.choices[0].message.parsed


# max_tokens로 출력 길이 제한
def summarize_with_limit(text: str, max_words: int = 50) -> str:
    """출력 길이 명시적 제한"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "system",
            "content": f"요약은 반드시 {max_words}단어 이내로 작성하세요."
        }, {
            "role": "user",
            "content": f"다음을 요약해주세요: {text}"
        }],
        max_tokens=max_words * 2  # 한국어는 토큰/단어 비율 고려
    )
    return response.choices[0].message.content

import time
from collections import defaultdict
from datetime import datetime, date
from openai import OpenAI

client = OpenAI()

class CostTracker:
    """LLM API 비용 실시간 추적"""

    # 2025년 기준 가격 ($/1M 토큰)
    PRICING = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-3-5-sonnet-20241022": {"input": 3.00, "output": 15.00},
        "claude-3-haiku-20240307": {"input": 0.25, "output": 1.25},
    }

    def __init__(self, daily_budget_usd: float = 100.0):
        self.daily_budget = daily_budget_usd
        self.usage = defaultdict(lambda: {"input_tokens": 0, "output_tokens": 0, "cost": 0})
        self.daily_cost = defaultdict(float)

    def track(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """사용량 추적 및 비용 계산"""
        if model not in self.PRICING:
            return 0.0

        pricing = self.PRICING[model]
        cost = (input_tokens * pricing["input"] + output_tokens * pricing["output"]) / 1_000_000

        today = date.today().isoformat()
        self.usage[model]["input_tokens"] += input_tokens
        self.usage[model]["output_tokens"] += output_tokens
        self.usage[model]["cost"] += cost
        self.daily_cost[today] += cost

        # 예산 임계값 경고
        if self.daily_cost[today] > self.daily_budget * 0.8:
            print(f"경고: 일일 예산의 80% 도달! (${self.daily_cost[today]:.2f} / ${self.daily_budget:.2f})")

        if self.daily_cost[today] > self.daily_budget:
            raise Exception(f"일일 예산 초과! ${self.daily_cost[today]:.2f} > ${self.daily_budget:.2f}")

        return cost

    def report(self) -> None:
        """비용 보고서 출력"""
        print("\n=== LLM API 비용 보고서 ===")
        total_cost = 0
        for model, data in self.usage.items():
            print(f"\n모델: {model}")
            print(f"  입력 토큰: {data['input_tokens']:,}")
            print(f"  출력 토큰: {data['output_tokens']:,}")
            print(f"  비용: ${data['cost']:.4f}")
            total_cost += data['cost']
        print(f"\n총 비용: ${total_cost:.4f}")

        today = date.today().isoformat()
        print(f"오늘 비용: ${self.daily_cost[today]:.4f} / ${self.daily_budget:.2f} (예산)")


# 전역 트래커
tracker = CostTracker(daily_budget_usd=50.0)

def tracked_completion(model: str, messages: list, **kwargs) -> str:
    """비용 추적이 포함된 API 호출 래퍼"""
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        **kwargs
    )
    tracker.track(
        model=model,
        input_tokens=response.usage.prompt_tokens,
        output_tokens=response.usage.completion_tokens
    )
    return response.choices[0].message.content

기준: 월 100,000 요청, GPT-4o, 평균 1,000 토큰/요청

최적화 전 월 비용:
100,000 × 1,000 토큰 × ($2.50 + $10.00) / 1,000,000 = $1,250/월

전략 적용 후:
1. Prompt Caching (80% 요청에 적용, 85% 절감): -$850
2. 모델 라우팅 (70%를 mini로): -$175 추가
3. 시맨틱 캐싱 (40% 히트율): -$45 추가
4. Batch API (20%를 배치로): -$25 추가
5. 출력 최적화 (30% 감소): -$75 추가

최적화 후 월 비용: 약 $80/월
절감률: 94%