Module 7: Use Cases

6. Production Patterns

Description

The pipelines from the previous capsules work. But "working" in a notebook isn't "working" in production. In production you have 1000 concurrent users, API bills that grow, rate limits that block requests, intermittent errors, and the need to monitor everything. In this capsule you implement the patterns that make multimodal systems scalable, economical, and reliable.

Why it matters: The difference between a prototype and a product is the infrastructure around the core. Caching avoids paying twice for the same analysis. Batching maximizes throughput. Rate limit management prevents 429 errors. Cost tracking avoids surprises on the bill. Logging and monitoring tell you when something fails before the user reports it.

Connection with the module: These patterns apply cross-cuttingly to all the pipelines: Document Q&A (capsule 02), Image Analysis (03), Video (04), Multi-Modal (05). The Use Case Selector (capsule 08) integrates these patterns into its implementation.

The 7 Production Patterns

#	Pattern	Problem it solves	Impact
1	Caching	Repeated calls to the same content	Reduces costs 30-80%
2	Batching	Many slow individual calls	Increases throughput 3-5x
3	Rate Limit Management	429 errors during traffic spikes	Eliminates rate errors
4	Cost Optimization	High API bill	Reduces costs 50-90%
5	Error Handling	Intermittent API failures	99.9% uptime
6	Logging and Monitoring	Not knowing what happens in production	Complete visibility
7	A/B Testing	Not knowing which provider is better	Data-driven decisions

Pattern 1: Caching

In-memory cache with content hash

import hashlib
import json
from functools import lru_cache

def content_hash(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def file_hash(path: str) -> str:
    with open(path, "rb") as f:
        return content_hash(f.read())

Persistent disk cache

from pathlib import Path
import time

class DiskCache:
    def __init__(self, cache_dir: str = "/tmp/multimodal_cache", ttl_hours: int = 24):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self.ttl_seconds = ttl_hours * 3600

    def get(self, key: str) -> str | None:
        cache_file = self.cache_dir / f"{key}.json"
        if not cache_file.exists():
            return None

        data = json.loads(cache_file.read_text())
        if time.time() - data["timestamp"] > self.ttl_seconds:
            cache_file.unlink()
            return None

        return data["value"]

    def set(self, key: str, value: str) -> None:
        cache_file = self.cache_dir / f"{key}.json"
        cache_file.write_text(json.dumps({
            "value": value,
            "timestamp": time.time()
        }))

    def stats(self) -> dict:
        files = list(self.cache_dir.glob("*.json"))
        total_size = sum(f.stat().st_size for f in files)
        return {
            "entries": len(files),
            "total_size_mb": round(total_size / (1024 * 1024), 2)
        }

Cache for transcriptions

from openai import OpenAI

client = OpenAI()
cache = DiskCache(cache_dir="/tmp/transcription_cache")

def transcribe_cached(audio_path: str) -> str:
    key = file_hash(audio_path)
    cached = cache.get(key)
    if cached:
        return cached

    with open(audio_path, "rb") as f:
        result = client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            response_format="text"
        )

    cache.set(key, result)
    return result

Cache for Vision

vision_cache = DiskCache(cache_dir="/tmp/vision_cache")

def describe_image_cached(image_path: str, prompt: str = "Describe this image.") -> str:
    key = hashlib.sha256(
        f"{file_hash(image_path)}:{prompt}".encode()
    ).hexdigest()

    cached = vision_cache.get(key)
    if cached:
        return cached

    import base64
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ]
        }],
        max_tokens=300
    )

    result = response.choices[0].message.content
    vision_cache.set(key, result)
    return result

Pattern 2: Batching

Embeddings in batch

def get_embeddings_batch(
    texts: list[str],
    model: str = "text-embedding-3-small",
    batch_size: int = 100
) -> list[list[float]]:
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = client.embeddings.create(model=model, input=batch)
        batch_embeddings = [item.embedding for item in response.data]
        all_embeddings.extend(batch_embeddings)

    return all_embeddings

Async image processing

import asyncio
from openai import AsyncOpenAI

async def process_images_async(
    image_paths: list[str],
    prompt: str,
    max_concurrent: int = 5
) -> list[dict]:
    aclient = AsyncOpenAI()
    semaphore = asyncio.Semaphore(max_concurrent)

    async def process_one(path: str) -> dict:
        async with semaphore:
            import base64
            with open(path, "rb") as f:
                b64 = base64.b64encode(f.read()).decode()

            try:
                response = await aclient.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{
                        "role": "user",
                        "content": [
                            {"type": "text", "text": prompt},
                            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
                        ]
                    }],
                    max_tokens=200
                )
                return {"path": path, "result": response.choices[0].message.content, "status": "success"}
            except Exception as e:
                return {"path": path, "result": None, "status": "error", "error": str(e)}

    tasks = [process_one(path) for path in image_paths]
    return await asyncio.gather(*tasks)

Pattern 3: Rate Limit Management

Retry with exponential backoff

import time
import random

def retry_with_backoff(
    func,
    *args,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    **kwargs
):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            error_str = str(e).lower()
            is_retryable = any(keyword in error_str for keyword in ["rate", "429", "timeout", "503", "overloaded"])

            if not is_retryable or attempt == max_retries - 1:
                raise

            delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
            time.sleep(delay)

Token bucket rate limiter

import threading

class TokenBucketRateLimiter:
    def __init__(self, requests_per_minute: int = 60):
        self.rate = requests_per_minute / 60.0
        self.tokens = requests_per_minute
        self.max_tokens = requests_per_minute
        self.last_refill = time.time()
        self.lock = threading.Lock()

    def acquire(self):
        with self.lock:
            now = time.time()
            elapsed = now - self.last_refill
            self.tokens = min(self.max_tokens, self.tokens + elapsed * self.rate)
            self.last_refill = now

            if self.tokens >= 1:
                self.tokens -= 1
                return True

            wait_time = (1 - self.tokens) / self.rate
            time.sleep(wait_time)
            self.tokens = 0
            self.last_refill = time.time()
            return True


rate_limiter = TokenBucketRateLimiter(requests_per_minute=50)

def rate_limited_call(func, *args, **kwargs):
    rate_limiter.acquire()
    return func(*args, **kwargs)

Circuit breaker

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure_time = 0
        self.state = "closed"

    def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
            else:
                raise Exception(f"Circuit breaker open. Retry in {self.reset_timeout - (time.time() - self.last_failure_time):.0f}s")

        try:
            result = func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "open"
            raise

Pattern 4: Cost Optimization

Cost tracker

PRICING = {
    "gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000},
    "gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
    "text-embedding-3-small": {"input": 0.02 / 1_000_000},
    "text-embedding-3-large": {"input": 0.13 / 1_000_000},
    "whisper-1": {"per_minute": 0.006},
    "tts-1": {"per_million_chars": 15.00},
    "dall-e-3": {"per_image_1024": 0.040},
}

class CostTracker:
    def __init__(self):
        self.calls: list[dict] = []

    def track_chat(self, model: str, input_tokens: int, output_tokens: int) -> float:
        pricing = PRICING.get(model, {})
        cost = (
            input_tokens * pricing.get("input", 0) +
            output_tokens * pricing.get("output", 0)
        )
        self._record(model, "chat", cost, {
            "input_tokens": input_tokens,
            "output_tokens": output_tokens
        })
        return cost

    def track_embedding(self, model: str, tokens: int) -> float:
        pricing = PRICING.get(model, {})
        cost = tokens * pricing.get("input", 0)
        self._record(model, "embedding", cost, {"tokens": tokens})
        return cost

    def track_whisper(self, duration_minutes: float) -> float:
        cost = duration_minutes * PRICING["whisper-1"]["per_minute"]
        self._record("whisper-1", "transcription", cost, {"minutes": duration_minutes})
        return cost

    def track_tts(self, characters: int) -> float:
        cost = (characters / 1_000_000) * PRICING["tts-1"]["per_million_chars"]
        self._record("tts-1", "tts", cost, {"characters": characters})
        return cost

    def track_dalle(self, count: int = 1) -> float:
        cost = count * PRICING["dall-e-3"]["per_image_1024"]
        self._record("dall-e-3", "image_generation", cost, {"images": count})
        return cost

    def _record(self, model: str, operation: str, cost: float, details: dict):
        self.calls.append({
            "model": model,
            "operation": operation,
            "cost": round(cost, 6),
            "details": details,
            "timestamp": time.time()
        })

    def summary(self) -> dict:
        total = sum(c["cost"] for c in self.calls)
        by_model = {}
        for c in self.calls:
            model = c["model"]
            by_model[model] = by_model.get(model, 0) + c["cost"]

        by_operation = {}
        for c in self.calls:
            op = c["operation"]
            by_operation[op] = by_operation.get(op, 0) + c["cost"]

        return {
            "total_cost": round(total, 4),
            "total_calls": len(self.calls),
            "by_model": {k: round(v, 4) for k, v in by_model.items()},
            "by_operation": {k: round(v, 4) for k, v in by_operation.items()}
        }

Optimization strategies

class CostOptimizer:
    def __init__(self, tracker: CostTracker):
        self.tracker = tracker

    def select_model(self, task_complexity: str) -> str:
        model_map = {
            "simple": "gpt-4o-mini",
            "medium": "gpt-4o-mini",
            "complex": "gpt-4o",
            "vision_simple": "gpt-4o-mini",
            "vision_complex": "gpt-4o",
        }
        return model_map.get(task_complexity, "gpt-4o-mini")

    def should_use_cache(self, estimated_cost: float) -> bool:
        return estimated_cost > 0.001

    def optimize_image_for_cost(self, image_path: str, task: str) -> dict:
        from PIL import Image
        img = Image.open(image_path)
        w, h = img.size
        img.close()

        if task in ("classify", "detect"):
            target = 512
        elif task in ("extract_text", "ocr"):
            target = 1024
        else:
            target = 768

        if max(w, h) > target:
            return {"resize_to": target, "estimated_savings": "50-75%"}

        return {"resize_to": None, "estimated_savings": "0%"}

Pattern 5: Error Handling

class MultiModalError(Exception):
    def __init__(self, message: str, error_type: str, retryable: bool = False):
        super().__init__(message)
        self.error_type = error_type
        self.retryable = retryable


def safe_api_call(func, *args, fallback=None, **kwargs):
    try:
        return func(*args, **kwargs)
    except Exception as e:
        error_str = str(e).lower()

        if "rate" in error_str or "429" in error_str:
            raise MultiModalError(str(e), "rate_limit", retryable=True)
        elif "timeout" in error_str:
            raise MultiModalError(str(e), "timeout", retryable=True)
        elif "invalid" in error_str or "400" in error_str:
            raise MultiModalError(str(e), "invalid_input", retryable=False)
        elif fallback is not None:
            return fallback
        else:
            raise


def with_fallback_provider(
    primary_fn,
    fallback_fn,
    *args,
    **kwargs
):
    try:
        return {"result": primary_fn(*args, **kwargs), "provider": "primary"}
    except Exception as primary_error:
        try:
            return {"result": fallback_fn(*args, **kwargs), "provider": "fallback"}
        except Exception as fallback_error:
            raise MultiModalError(
                f"Both providers failed. Primary: {primary_error}, Fallback: {fallback_error}",
                "all_providers_failed",
                retryable=False
            )

Pattern 6: Logging and Monitoring

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("multimodal")


class PipelineMonitor:
    def __init__(self):
        self.metrics: list[dict] = []

    def start_operation(self, operation: str, details: dict = None) -> dict:
        entry = {
            "operation": operation,
            "start_time": time.time(),
            "details": details or {},
            "status": "in_progress"
        }
        self.metrics.append(entry)
        logger.info(f"START: {operation} | {details}")
        return entry

    def end_operation(self, entry: dict, status: str = "success", result_info: dict = None):
        entry["end_time"] = time.time()
        entry["duration_ms"] = round((entry["end_time"] - entry["start_time"]) * 1000, 2)
        entry["status"] = status
        entry["result_info"] = result_info or {}
        logger.info(f"END: {entry['operation']} | {status} | {entry['duration_ms']}ms")

    def get_stats(self) -> dict:
        if not self.metrics:
            return {"total_operations": 0}

        completed = [m for m in self.metrics if "duration_ms" in m]
        durations = [m["duration_ms"] for m in completed]
        errors = sum(1 for m in completed if m["status"] == "error")

        return {
            "total_operations": len(self.metrics),
            "completed": len(completed),
            "errors": errors,
            "error_rate": round(errors / len(completed) * 100, 1) if completed else 0,
            "avg_duration_ms": round(sum(durations) / len(durations), 2) if durations else 0,
            "max_duration_ms": max(durations) if durations else 0,
            "min_duration_ms": min(durations) if durations else 0,
            "p95_duration_ms": round(sorted(durations)[int(len(durations) * 0.95)] if durations else 0, 2)
        }


monitor = PipelineMonitor()

Usage with a context manager

from contextlib import contextmanager

@contextmanager
def tracked_operation(operation: str, details: dict = None):
    entry = monitor.start_operation(operation, details)
    try:
        yield entry
        monitor.end_operation(entry, "success")
    except Exception as e:
        monitor.end_operation(entry, "error", {"error": str(e)})
        raise

Example:

with tracked_operation("image_classification", {"path": "product.jpg"}):
    result = classify_image(b64, categories)

Pattern 7: A/B Testing Between Providers

import random

class ProviderABTest:
    def __init__(self, providers: dict[str, callable], split: dict[str, float] = None):
        self.providers = providers
        self.split = split or {name: 1.0 / len(providers) for name in providers}
        self.results: list[dict] = []

    def call(self, *args, **kwargs) -> dict:
        provider_name = self._select_provider()
        provider_fn = self.providers[provider_name]

        start = time.time()
        try:
            result = provider_fn(*args, **kwargs)
            duration = time.time() - start

            entry = {
                "provider": provider_name,
                "status": "success",
                "duration_ms": round(duration * 1000, 2),
                "timestamp": time.time()
            }
            self.results.append(entry)

            return {"result": result, "provider": provider_name, "duration_ms": entry["duration_ms"]}

        except Exception as e:
            duration = time.time() - start
            self.results.append({
                "provider": provider_name,
                "status": "error",
                "error": str(e),
                "duration_ms": round(duration * 1000, 2),
                "timestamp": time.time()
            })
            raise

    def _select_provider(self) -> str:
        r = random.random()
        cumulative = 0
        for name, weight in self.split.items():
            cumulative += weight
            if r <= cumulative:
                return name
        return list(self.providers.keys())[-1]

    def report(self) -> dict:
        report = {}
        for provider_name in self.providers:
            entries = [r for r in self.results if r["provider"] == provider_name]
            successes = [r for r in entries if r["status"] == "success"]
            durations = [r["duration_ms"] for r in successes]

            report[provider_name] = {
                "total_calls": len(entries),
                "successes": len(successes),
                "errors": len(entries) - len(successes),
                "error_rate": round((len(entries) - len(successes)) / len(entries) * 100, 1) if entries else 0,
                "avg_duration_ms": round(sum(durations) / len(durations), 2) if durations else 0,
                "p95_duration_ms": round(sorted(durations)[int(len(durations) * 0.95)], 2) if durations else 0
            }

        return report

Usage:

import anthropic

def classify_openai(image_b64: str, categories: list[str]) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": f"Classify into: {', '.join(categories)}. Only the category."},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}}
            ]
        }],
        max_tokens=20, temperature=0
    )
    return response.choices[0].message.content.strip()

def classify_anthropic(image_b64: str, categories: list[str]) -> str:
    claude = anthropic.Anthropic()
    response = claude.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=20,
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": f"Classify into: {', '.join(categories)}. Only the category."},
                {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": image_b64}}
            ]
        }]
    )
    return response.content[0].text.strip()

ab_test = ProviderABTest(
    providers={"openai": classify_openai, "anthropic": classify_anthropic},
    split={"openai": 0.7, "anthropic": 0.3}
)

Troubleshooting

Problem 1: Stale cache (outdated data)

Symptom: The cache returns results from a previous version of the file.

Solution: Use the hash of the file's content (not the path) as the cache key. Already implemented in file_hash.

Problem 2: Memory leak with in-memory cache

Symptom: The process consumes more and more RAM.

Solution: Use DiskCache with TTL instead of lru_cache for large data. For lru_cache, configure an appropriate maxsize.

Problem 3: Race conditions in the rate limiter

Symptom: More requests than expected pass the rate limiter in multi-thread environments.

Solution: The TokenBucketRateLimiter already uses threading.Lock. For async environments, use asyncio.Semaphore.

Problem 4: Costs spike without warning

Symptom: The monthly bill is 5x higher than expected.

Solution:

class CostAlert:
    def __init__(self, tracker: CostTracker, daily_limit: float = 10.0):
        self.tracker = tracker
        self.daily_limit = daily_limit

    def check(self) -> dict:
        today_calls = [
            c for c in self.tracker.calls
            if c["timestamp"] > time.time() - 86400
        ]
        today_cost = sum(c["cost"] for c in today_calls)

        alert = today_cost > self.daily_limit * 0.8

        return {
            "today_cost": round(today_cost, 4),
            "daily_limit": self.daily_limit,
            "percentage_used": round(today_cost / self.daily_limit * 100, 1),
            "alert": alert
        }

Exercises

Exercise 1: Cache with Redis

Implement an equivalent of DiskCache using Redis, with configurable TTL and JSON serialization.

See solution

import redis

class RedisCache:
    def __init__(self, host: str = "localhost", port: int = 6379, ttl_hours: int = 24, prefix: str = "mm"):
        self.client = redis.Redis(host=host, port=port, decode_responses=True)
        self.ttl_seconds = ttl_hours * 3600
        self.prefix = prefix

    def get(self, key: str) -> str | None:
        return self.client.get(f"{self.prefix}:{key}")

    def set(self, key: str, value: str) -> None:
        self.client.setex(f"{self.prefix}:{key}", self.ttl_seconds, value)

    def stats(self) -> dict:
        keys = self.client.keys(f"{self.prefix}:*")
        total_size = sum(self.client.memory_usage(k) or 0 for k in keys)
        return {
            "entries": len(keys),
            "total_size_mb": round(total_size / (1024 * 1024), 2)
        }

Exercise 2: Cost dashboard

Create a function that generates a cost report grouped by day, model, and operation.

See solution

from datetime import datetime
from collections import defaultdict

def generate_cost_report(tracker: CostTracker) -> dict:
    by_day = defaultdict(float)
    by_day_model = defaultdict(lambda: defaultdict(float))

    for call in tracker.calls:
        day = datetime.fromtimestamp(call["timestamp"]).strftime("%Y-%m-%d")
        by_day[day] += call["cost"]
        by_day_model[day][call["model"]] += call["cost"]

    report = {
        "summary": tracker.summary(),
        "daily": {}
    }

    for day in sorted(by_day.keys()):
        report["daily"][day] = {
            "total": round(by_day[day], 4),
            "by_model": {k: round(v, 4) for k, v in by_day_model[day].items()}
        }

    return report

Exercise 3: Production middleware

Create a decorator that combines cache, rate limiting, retry, cost tracking, and logging for any API function.

See solution

import functools

def production_middleware(
    cache_instance: DiskCache = None,
    rate_limiter_instance: TokenBucketRateLimiter = None,
    cost_tracker: CostTracker = None,
    max_retries: int = 3
):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = None
            if cache_instance:
                key_data = f"{func.__name__}:{str(args)}:{str(kwargs)}"
                cache_key = hashlib.sha256(key_data.encode()).hexdigest()
                cached = cache_instance.get(cache_key)
                if cached:
                    logger.info(f"CACHE HIT: {func.__name__}")
                    return json.loads(cached)

            if rate_limiter_instance:
                rate_limiter_instance.acquire()

            entry = monitor.start_operation(func.__name__)

            try:
                result = retry_with_backoff(func, *args, max_retries=max_retries, **kwargs)

                if cache_instance and cache_key:
                    cache_instance.set(cache_key, json.dumps(result) if isinstance(result, (dict, list)) else result)

                monitor.end_operation(entry, "success")
                return result

            except Exception as e:
                monitor.end_operation(entry, "error", {"error": str(e)})
                raise

        return wrapper
    return decorator

Usage:

@production_middleware(
    cache_instance=DiskCache(),
    rate_limiter_instance=TokenBucketRateLimiter(requests_per_minute=50),
    cost_tracker=CostTracker()
)
def describe_image_production(image_path: str) -> str:
    return describe_image_cached(image_path)

Additional Resources

OpenAI Rate Limits — Limits per model
Redis Documentation — Distributed cache
OpenAI Pricing — Up-to-date prices
Circuit Breaker Pattern — Resilience pattern

← Previous5. Multi-Modality Combinations Next →7. Use Case Troubleshooting