LLM Production Patterns

Real-World Scenario

A SaaS product offers AI-powered document analysis. In the first week, costs are 3x the estimate because: every request re-sends the full 2,000-token system prompt, there’s no caching for identical queries, and max_tokens is set to 4096 regardless of task. After applying production patterns, costs drop 70% with no quality regression.

Retry with Exponential Backoff

import anthropic
import time
import random
import logging
from functools import wraps

logger = logging.getLogger(__name__)

def with_retry(
    max_retries: int = 3,
    base_delay:  float = 1.0,
    max_delay:   float = 60.0,
    jitter:      bool  = True,
):
    """Decorator: retry with exponential backoff on transient errors."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)

                except anthropic.RateLimitError as e:
                    wait = min(base_delay * (2 ** attempt), max_delay)
                    if jitter:
                        wait *= (0.5 + random.random() * 0.5)
                    logger.warning(f"Rate limit (attempt {attempt+1}). Waiting {wait:.1f}s")
                    last_error = e
                    time.sleep(wait)

                except anthropic.InternalServerError as e:
                    if attempt == max_retries:
                        raise
                    wait = min(base_delay * (2 ** attempt), max_delay)
                    logger.warning(f"Server error (attempt {attempt+1}). Waiting {wait:.1f}s")
                    last_error = e
                    time.sleep(wait)

                except anthropic.BadRequestError:
                    raise   # don't retry client errors

            raise last_error
        return wrapper
    return decorator


client = anthropic.Anthropic()

@with_retry(max_retries=3, base_delay=2.0)
def call_claude(messages: list[dict], system: str = "", max_tokens: int = 1024) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=max_tokens,
        system=system,
        messages=messages,
    )
    return response.content[0].text

Response Caching

import anthropic
import hashlib
import json
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path

client = anthropic.Anthropic()

class LLMCache:
    """SQLite-backed cache for LLM responses."""

    def __init__(self, db_path: str = "./llm_cache.db", ttl_hours: int = 24):
        self.conn = sqlite3.connect(db_path)
        self.ttl  = timedelta(hours=ttl_hours)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS cache (
                key       TEXT PRIMARY KEY,
                response  TEXT NOT NULL,
                created   TEXT NOT NULL,
                hits      INTEGER DEFAULT 0
            )
        """)
        self.conn.commit()

    def _key(self, model: str, messages: list, system: str, max_tokens: int) -> str:
        payload = json.dumps({
            "model": model, "messages": messages,
            "system": system, "max_tokens": max_tokens
        }, sort_keys=True)
        return hashlib.sha256(payload.encode()).hexdigest()

    def get(self, key: str) -> str | None:
        row = self.conn.execute(
            "SELECT response, created FROM cache WHERE key = ?", (key,)
        ).fetchone()
        if not row:
            return None
        age = datetime.now() - datetime.fromisoformat(row[1])
        if age > self.ttl:
            self.conn.execute("DELETE FROM cache WHERE key = ?", (key,))
            self.conn.commit()
            return None
        self.conn.execute("UPDATE cache SET hits = hits + 1 WHERE key = ?", (key,))
        self.conn.commit()
        return row[0]

    def set(self, key: str, response: str) -> None:
        now = datetime.now().isoformat()
        self.conn.execute(
            "INSERT OR REPLACE INTO cache (key, response, created) VALUES (?, ?, ?)",
            (key, response, now)
        )
        self.conn.commit()

    def stats(self) -> dict:
        row = self.conn.execute(
            "SELECT COUNT(*), SUM(hits), AVG(hits) FROM cache"
        ).fetchone()
        return {"entries": row[0], "total_hits": row[1] or 0, "avg_hits": round(row[2] or 0, 1)}


cache = LLMCache(ttl_hours=24)

def cached_call(
    messages: list[dict],
    system: str = "",
    model: str = "claude-haiku-4-5-20251001",
    max_tokens: int = 512,
) -> tuple[str, bool]:
    """Returns (response_text, was_cached)."""
    key = cache._key(model, messages, system, max_tokens)
    hit = cache.get(key)
    if hit:
        return hit, True

    response = client.messages.create(
        model=model, max_tokens=max_tokens, system=system, messages=messages
    )
    text = response.content[0].text
    cache.set(key, text)
    return text, False


# Test caching
question = [{"role": "user", "content": "What is 2 + 2?"}]
resp1, cached1 = cached_call(question)
resp2, cached2 = cached_call(question)
print(f"First call:  cached={cached1}")
print(f"Second call: cached={cached2}")
print(f"Cache stats: {cache.stats()}")

Cost Tracking

import anthropic
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

# Pricing per million tokens (as of 2025, subject to change)
PRICING = {
    "claude-haiku-4-5-20251001": {"input": 1.00,  "output": 5.00},
    "claude-sonnet-4-6":         {"input": 3.00,  "output": 15.00},
    "claude-opus-4-8":           {"input": 15.00, "output": 75.00},
}

@dataclass
class UsageTracker:
    calls:         int   = 0
    input_tokens:  int   = 0
    output_tokens: int   = 0
    total_cost:    float = 0.0
    errors:        int   = 0
    by_model:      dict  = field(default_factory=dict)

    def record(self, model: str, usage: Any) -> float:
        self.calls        += 1
        self.input_tokens  += usage.input_tokens
        self.output_tokens += usage.output_tokens

        prices = PRICING.get(model, {"input": 3.0, "output": 15.0})
        cost   = (
            usage.input_tokens  / 1_000_000 * prices["input"]  +
            usage.output_tokens / 1_000_000 * prices["output"]
        )
        self.total_cost += cost

        if model not in self.by_model:
            self.by_model[model] = {"calls": 0, "cost": 0.0, "tokens": 0}
        self.by_model[model]["calls"] += 1
        self.by_model[model]["cost"]  += cost
        self.by_model[model]["tokens"] += usage.input_tokens + usage.output_tokens
        return cost

    def report(self) -> dict:
        return {
            "calls":          self.calls,
            "input_tokens":   self.input_tokens,
            "output_tokens":  self.output_tokens,
            "total_cost_usd": round(self.total_cost, 4),
            "avg_cost_usd":   round(self.total_cost / max(self.calls, 1), 6),
            "by_model":       self.by_model,
        }


tracker = UsageTracker()
client  = anthropic.Anthropic()

def tracked_call(messages: list[dict], model: str = "claude-haiku-4-5-20251001",
                 max_tokens: int = 512, system: str = "") -> str:
    response = client.messages.create(
        model=model, max_tokens=max_tokens, system=system, messages=messages
    )
    cost = tracker.record(model, response.usage)
    return response.content[0].text


# Example: compare cost between models
short_task = [{"role": "user", "content": "What is the capital of France?"}]
tracked_call(short_task, model="claude-haiku-4-5-20251001",  max_tokens=50)
tracked_call(short_task, model="claude-sonnet-4-6",          max_tokens=50)

import json
print(json.dumps(tracker.report(), indent=2))

Choosing the Right Model and Token Budget

import anthropic

client = anthropic.Anthropic()

def route_to_model(task_type: str, content_length: int) -> tuple[str, int]:
    """
    Route requests to the cheapest model that can handle the task.
    Returns (model_name, max_tokens).
    """
    routing_rules = {
        "classification":    ("claude-haiku-4-5-20251001",  32),
        "extraction":        ("claude-haiku-4-5-20251001", 256),
        "summarization":     ("claude-haiku-4-5-20251001", 512),
        "qa_simple":         ("claude-haiku-4-5-20251001", 256),
        "qa_complex":        ("claude-sonnet-4-6",          512),
        "code_review":       ("claude-sonnet-4-6",         1024),
        "code_generation":   ("claude-sonnet-4-6",         2048),
        "complex_reasoning": ("claude-opus-4-8",           4096),
    }
    return routing_rules.get(task_type, ("claude-sonnet-4-6", 1024))


def smart_call(
    user_message: str,
    task_type: str,
    system: str = "",
) -> dict:
    model, max_tokens = route_to_model(task_type, len(user_message))

    response = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=system,
        messages=[{"role": "user", "content": user_message}],
    )

    prices = PRICING.get(model, {"input": 3.0, "output": 15.0})
    cost = (
        response.usage.input_tokens  / 1_000_000 * prices["input"] +
        response.usage.output_tokens / 1_000_000 * prices["output"]
    )

    return {
        "response":       response.content[0].text,
        "model":          model,
        "input_tokens":   response.usage.input_tokens,
        "output_tokens":  response.usage.output_tokens,
        "cost_usd":       round(cost, 6),
    }


# Routing demo
tasks = [
    ("Is this review positive or negative? 'Great product, very happy!'", "classification"),
    ("Explain the trade-offs between microservices and monoliths.", "qa_complex"),
    ("Write a Python function to parse a CSV file.", "code_generation"),
]

for text, task_type in tasks:
    result = smart_call(text, task_type)
    print(f"Task: {task_type:<20}  Model: {result['model']:<35}  Cost: ${result['cost_usd']:.6f}")