6. Patrones de Producción

Descripción

Los pipelines de las cápsulas anteriores funcionan. Pero "funcionar" en un notebook no es "funcionar" en producción. En producción tienes 1000 usuarios concurrentes, facturas de API que crecen, rate limits que bloquean requests, errores intermitentes, y la necesidad de monitorear todo. En esta cápsula implementas los patrones que hacen sistemas multimodales escalables, económicos, y confiables.

Por qué importa: La diferencia entre un prototipo y un producto es la infraestructura alrededor del core. Caching evita pagar dos veces por el mismo análisis. Batching maximiza throughput. Rate limit management previene errores 429. Cost tracking evita sorpresas en la factura. Logging y monitoring te dicen cuándo algo falla antes de que el usuario lo reporte.

Conexión con el módulo: Estos patrones se aplican transversalmente a todos los pipelines: Document Q&A (cápsula 02), Image Analysis (03), Video (04), Multi-Modal (05). El Use Case Selector (cápsula 08) integra estos patrones en su implementación.

Los 7 Patrones de Producción

Patrón 1: Caching

Cache en memoria con hash de contenido

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())

Cache en disco persistente

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 para transcripciones

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 para Vision

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

def describe_image_cached(image_path: str, prompt: str = "Describe esta imagen.") -> 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

Patrón 2: Batching

Embeddings en 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

Procesamiento async de imágenes

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)

Patrón 3: Rate Limit Management

Retry con 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

Patrón 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()}
        }

Estrategias de optimización

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%"}

Patrón 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
            )

Patrón 6: Logging y 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()

Uso con 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

Ejemplo:

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

Patrón 7: A/B Testing entre Proveedores

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

Uso:

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"Clasifica en: {', '.join(categories)}. Solo la categoría."},
                {"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"Clasifica en: {', '.join(categories)}. Solo la categoría."},
                {"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

Problema 1: Cache stale (datos desactualizados)

Síntoma: El cache devuelve resultados de una versión anterior del archivo.

Solución: Usar hash del contenido del archivo (no del path) como clave de cache. Ya implementado en file_hash.

Problema 2: Memory leak con cache en memoria

Síntoma: El proceso consume cada vez más RAM.

Solución: Usar DiskCache con TTL en lugar de lru_cache para datos grandes. Para lru_cache, configurar maxsize apropiado.

Problema 3: Race conditions en rate limiter

Síntoma: Más requests de los esperados pasan el rate limiter en entornos multi-thread.

Solución: El TokenBucketRateLimiter ya usa threading.Lock. Para entornos async, usar asyncio.Semaphore.

Problema 4: Costos disparan sin aviso

Síntoma: La factura mensual es 5x mayor de lo esperado.

Solución:

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
        }

Ejercicios

Ejercicio 1: Cache con Redis

Implementa DiskCache equivalente usando Redis, con TTL configurable y serialización JSON.

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)
        }

Ejercicio 2: Dashboard de costos

Crea una función que genere un reporte de costos agrupado por día, modelo, y operación.

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

Ejercicio 3: Middleware de producción

Crea un decorador que combine cache, rate limiting, retry, cost tracking, y logging para cualquier función de API.

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

@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)

Recursos Adicionales

1. OpenAI Rate Limits — Límites por modelo
2. Redis Documentation — Cache distribuido
3. OpenAI Pricing — Precios actualizados
4. Circuit Breaker Pattern — Patrón de resiliencia