LLM Cost Optimization: Оптимизация расходов на LLM

Введение

Стоимость использования LLM может быстро вырасти при масштабировании. Оптимизация затрат — критическая задача для production систем.

---

Cost Breakdown

Cost Component         | % of Total | Optimization Potential
-----------------------|------------|----------------------
API/Cloud GPU          | 60-70%     | High (30-50%)
Bandwidth              | 10-15%     | Medium (20-40%)
Storage                | 5-10%      | Low (10-20%)
Engineering            | 15-20%     | Medium (15-25%)
Monitoring/Tools       | 5%         | Low (10-15%)

---

Token Optimization

Input token reduction

def optimize_input(messages: list[dict]) -> list[dict]:
    """Reduce input tokens without losing context"""
    optimized = []
    
    for msg in messages:
        if msg["role"] == "system":
            msg["content"] = truncate_to_tokens(msg["content"], max_tokens=256)
        elif msg["role"] == "user":
            msg["content"] = remove_filler_phrases(msg["content"])
        elif msg["role"] == "assistant":
            if len(optimized) > 4:
                pass  # summarize older messages
        optimized.append(msg)
    
    return optimized

def remove_filler_phrases(text: str) -> str:
    """Remove common filler phrases"""
    fillers = ["Please", "Could you", "I would like you to", "It would be great if"]
    result = text
    for filler in fillers:
        result = result.replace(f"{filler}, ", "")
    return result.strip()

Output token control

import tiktoken

def count_tokens(text: str, model: str = "gpt-4") -> int:
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

MODEL_COSTS = {
    "gpt-4o": {"input": 0.0025, "output": 0.010, "unit": "1K tokens"},
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006, "unit": "1K tokens"},
    "claude-sonnet-4": {"input": 0.003, "output": 0.015, "unit": "1K tokens"},
    "claude-haiku-3.5": {"input": 0.00025, "output": 0.00125, "unit": "1K tokens"},
    "gemini-pro": {"input": 0.0005, "output": 0.0015, "unit": "1K tokens"},
}

---

Caching Strategies

Prompt-level caching

import hashlib
import json
import time
from typing import Optional

class PromptCache:
    """Cache identical prompts to avoid re-computation"""
    
    def __init__(self, ttl: int = 3600):
        self.ttl = ttl
        self._cache = {}
    
    def _make_key(self, prompt: str, params: dict) -> str:
        key_data = json.dumps({
            "prompt": prompt,
            "params": {k: v for k, v in params.items() if k != "api_key"}
        }, sort_keys=True)
        return hashlib.sha256(key_data.encode()).hexdigest()
    
    def get(self, prompt: str, params: dict) -> Optional[str]:
        key = self._make_key(prompt, params)
        if key in self._cache:
            entry = self._cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                return entry["result"]
        return None
    
    def set(self, prompt: str, params: dict, result: str):
        key = self._make_key(prompt, params)
        self._cache[key] = {"result": result, "timestamp": time.time()}

# Usage
cache = PromptCache()

def generate(prompt: str, temperature: float = 0.7) -> str:
    cached = cache.get(prompt, {"temperature": temperature})
    if cached:
        return cached  # $0.00 cost
    result = call_llm(prompt, temperature)
    cache.set(prompt, {"temperature": temperature}, result)
    return result

Embedding cache

class EmbeddingCache:
    """Cache embeddings to avoid regenerating them"""
    
    def __init__(self, vector_store, ttl: int = 86400 * 7):
        self.vector_store = vector_store
        self.ttl = ttl
    
    def get_embedding(self, text: str) -> Optional[list]:
        results = self.vector_store.similarity_search(text, k=1, score_threshold=0.999)
        if results:
            return results[0].metadata.get("embedding")
        return None
    
    def store_embedding(self, text: str, embedding: list, doc_id: str):
        self.vector_store.upsert(ids=[doc_id], embeddings=[embedding], documents=[text])

# Cost savings: embedding 1M docs
# Without cache: 1M × $0.00002 = $20
# With cache: ~100K unique × $0.00002 = $2 (90% savings)

---

Model Selection Strategy

Right-sizing models

MODEL_TIER = {
    "fast": {
        "model": "gpt-4o-mini",
        "cost_per_1k": 0.00015,
        "use_for": ["Classification", "Simple Q&A", "Text extraction", "Routing"]
    },
    "balanced": {
        "model": "claude-haiku-3.5",
        "cost_per_1k": 0.00025,
        "use_for": ["Summarization", "Translation", "Entity extraction"]
    },
    "smart": {
        "model": "claude-sonnet-4",
        "cost_per_1k": 0.003,
        "use_for": ["Complex reasoning", "Code generation", "Analysis"]
    },
    "expert": {
        "model": "gpt-4o",
        "cost_per_1k": 0.0025,
        "use_for": ["Multi-modal", "Advanced reasoning", "Debugging"]
    },
}

def route_to_model(task_type: str, complexity: str) -> str:
    """Route tasks to appropriate model tier"""
    routing = {
        ("classification", "low"): "fast",
        ("classification", "high"): "balanced",
        ("reasoning", "low"): "balanced",
        ("reasoning", "high"): "expert",
        ("summarization", "any"): "fast",
        ("code", "any"): "smart",
    }
    return routing.get((task_type, complexity), "balanced")

Router pattern

class ModelRouter:
    """Route requests to cheapest capable model"""
    
    def __init__(self):
        self.classifier = load_classifier()
        self.models = {
            "fast": OpenAI(model="gpt-4o-mini"),
            "balanced": Anthropic(model="claude-haiku-3.5"),
            "smart": Anthropic(model="claude-sonnet-4"),
            "expert": OpenAI(model="gpt-4o"),
        }
    
    def invoke(self, prompt: str, required_capability: str) -> str:
        tier = self.classifier.predict(prompt, required_capability)
        model = self.models[tier]
        return model.invoke(prompt)
    
    def estimate_cost(self, prompt: str, capability: str) -> float:
        tier = self.classifier.predict(prompt, capability)
        tokens = estimate_tokens(prompt)
        return tokens * MODEL_TIER[tier]["cost_per_1k"] / 1000

# Example: routing saves 60% vs always using gpt-4o
# Without router: 1M requests × gpt-4o = $2,500
# With router: 600K fast + 300K balanced + 100K smart = $950

---

Batch Processing

Batch API usage

import asyncio
from typing import List

class BatchProcessor:
    """Process requests in batches for cost and rate limit efficiency"""
    
    def __init__(self, llm, batch_size: int = 10, delay: float = 1.0):
        self.llm = llm
        self.batch_size = batch_size
        self.delay = delay
    
    async def process_batch(self, prompts: List[str]) -> List[str]:
        """Process prompts in parallel batches"""
        results = []
        
        for i in range(0, len(prompts), self.batch_size):
            batch = prompts[i:i + self.batch_size]
            
            # Parallel processing
            tasks = [self.llm.ainvoke(p) for p in batch]
            responses = await asyncio.gather(*tasks, return_exceptions=True)
            
            for response in responses:
                if isinstance(response, Exception):
                    results.append(f"Error: {response}")
                else:
                    results.append(response.content)
            
            # Rate limit delay
            await asyncio.sleep(self.delay)
        
        return results
    
    def calculate_batch_savings(self, single_cost: float, batch_count: int) -> float:
        """Batch APIs often have 10-20% discount"""
        batch_discount = 0.15
        return single_cost * batch_discount

# OpenAI Batch API pricing
# Standard: $2.50/1M input tokens, $5/1M output tokens
# Batch: $1.25/1M input tokens, $2.50/1M output tokens (50% off)

---

Self-Correction and Retry Logic

Cost-aware retry

import random
from typing import Optional

class CostAwareRetry:
    """Smart retry with cost considerations"""
    
    def __init__(self, max_retries: int = 3, base_cost: float = 0.01):
        self.max_retries = max_retries
        self.base_cost = base_cost
        self._spent = 0.0
    
    async def call_with_retry(
        self,
        llm_func,
        prompt: str,
        budget: float = 0.05,
        fallback_func=None
    ) -> Optional[str]:
        for attempt in range(self.max_retries):
            try:
                result = await llm_func(prompt)
                self._validate_result(result)
                return result
            except Exception as e:
                cost = self._estimate_attempt_cost(prompt)
                self._spent += cost
                
                if self._spent >= budget:
                    break
                
                # Exponential backoff + jitter
                delay = (2 ** attempt) * 0.5 + random.uniform(0, 0.5)
                await asyncio.sleep(delay)
        
        # Fallback to cheaper model
        if fallback_func:
            return await fallback_func(prompt)
        
        return None
    
    def _estimate_attempt_cost(self, prompt: str) -> float:
        tokens = len(prompt.split()) / 4  # rough estimate
        return tokens * 0.00015 * 2  # input + output estimate

---

Local Model Deployment

When to go local

Scenario                    | API Cost/Mo | Local Cost/Mo | Savings
----------------------------|-------------|---------------|--------
100K requests/day           | $2,500      | $300 (GPU)    | 88%
500K requests/day           | $12,500     | $600 (2xGPU)  | 95%
1M+ requests/day            | $25,000     | $1,200 (4xGPU)| 95%
Burst traffic (spiky)       | $3,000      | $500 (base)   | 83%

Local model setup

# Ollama local deployment
# docker-compose.yml
"""
version: '3.8'
services:
  ollama:
    image: ollama/ollama
    ports:
      - "11434:11434"
    volumes:
      - ./models:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
"""

# Cost comparison (monthly, 1M requests)
COST_COMPARISON = {
    "gpt-4o-api": {"cost": 25000, "quality": 10, "latency_ms": 2000},
    "claude-sonnet-api": {"cost": 30000, "quality": 10, "latency_ms": 2500},
    "llama-3-70b-local": {"cost": 300, "quality": 7, "latency_ms": 500},
    "mistral-large-local": {"cost": 300, "quality": 6, "latency_ms": 400},
    "qwen-2.5-72b-local": {"cost": 300, "quality": 8, "latency_ms": 600},
}

---

Monitoring and Alerts

Cost tracking dashboard

import sqlite3
from datetime import datetime, timedelta

class CostTracker:
    """Track and alert on LLM costs"""
    
    def __init__(self, db_path: str = "costs.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_db()
    
    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS requests (
                id INTEGER PRIMARY KEY,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                model TEXT,
                input_tokens INTEGER,
                output_tokens INTEGER,
                cost REAL,
                success INTEGER,
                latency_ms INTEGER
            )
        """)
        self.conn.commit()
    
    def log_request(self, model: str, input_tokens: int, 
                    output_tokens: int, cost: float, 
                    success: bool, latency_ms: int):
        self.conn.execute(
            """INSERT INTO requests 
               (model, input_tokens, output_tokens, cost, success, latency_ms)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (model, input_tokens, output_tokens, cost, success, latency_ms)
        )
        self.conn.commit()
    
    def get_daily_cost(self, days: int = 7) -> list:
        cutoff = datetime.now() - timedelta(days=days)
        cursor = self.conn.execute("""
            SELECT DATE(timestamp), SUM(cost), COUNT(*)
            FROM requests
            WHERE timestamp > ?
            GROUP BY DATE(timestamp)
            ORDER BY DATE(timestamp)
        """, (cutoff.isoformat(),))
        return cursor.fetchall()
    
    def get_model_breakdown(self) -> list:
        cursor = self.conn.execute("""
            SELECT model, SUM(cost), COUNT(*), AVG(latency_ms)
            FROM requests
            GROUP BY model
        """)
        return cursor.fetchall()
    
    def check_budget(self, daily_budget: float) -> dict:
        today = datetime.now().strftime("%Y-%m-%d")
        cursor = self.conn.execute("""
            SELECT SUM(cost) FROM requests
            WHERE DATE(timestamp) = ?
        """, (today,))
        spent = cursor.fetchone()[0] or 0
        return {
            "daily_budget": daily_budget,
            "spent_today": spent,
            "remaining": daily_budget - spent,
            "utilization": spent / daily_budget * 100
        }

---

Optimization Checklist

✅ Token optimization
   - Truncate system prompts
   - Remove filler phrases
   - Use concise language

✅ Caching
   - Prompt-level cache (identical requests)
   - Embedding cache (near-duplicate documents)
   - Response cache (RAG results)

✅ Model selection
   - Route simple tasks to cheap models
   - Use classifier for auto-routing
   - Reserve expensive models for hard tasks

✅ Batch processing
   - Use batch APIs for non-urgent tasks
   - Parallelize independent requests
   - Schedule off-peak processing

✅ Local deployment
   - Move high-volume patterns local
   - Use smaller models for simple tasks
   - Hybrid: local + API fallback

✅ Monitoring
   - Track cost per model per day
   - Set budget alerts at 80%
   - Monitor cost per task type

✅ Architecture
   - Implement graceful degradation
   - Use streaming for long responses
   - Compress responses when possible

---

Итоги

Оптимизация затрат на LLM — это комбинация: token optimization, caching, right-sizing моделей, batch processing и мониторинга. Начните с tracking — вы не оптимизируете то, что не измеряете.