7. Troubleshooting Casos de Uso

Descripción

Las cápsulas anteriores incluyeron secciones de troubleshooting específicas para cada patrón. Esta cápsula es la referencia centralizada: un catálogo de los problemas más comunes al construir sistemas multimodales en producción, con herramientas de diagnóstico, patrones de resolución, y técnicas de monitoreo. Cuando algo falla en tu pipeline, empieza aquí.

Por qué importa: En producción, los errores no son "la API no funciona" — son combinaciones sutiles: un PDF escaneado que rompe la extracción, un video donde los frames son todos negros, un pipeline multi-modal que excede el context window, un rate limit que aparece solo en horas pico. Diagnosticar estos problemas requiere herramientas y metodología, no adivinanzas.

Conexión con el módulo: Esta cápsula complementa la cápsula 06 (Patrones de Producción) con el aspecto de diagnóstico y resolución. El Use Case Selector (cápsula 08) necesita manejar gracefully todos los errores catalogados aquí.

Metodología de Diagnóstico

Cuando un pipeline falla, sigue este orden:

1. IDENTIFICAR  → ¿Qué error? ¿Dónde en el pipeline?
2. REPRODUCIR   → ¿Es consistente o intermitente?
3. AISLAR       → ¿Es el input, la API, o el procesamiento?
4. RESOLVER     → Aplicar fix + verificar
5. PREVENIR     → Agregar validación/test para que no recurra

Herramienta de diagnóstico

import traceback
import time
from pathlib import Path

class PipelineDiagnostics:
    def __init__(self):
        self.steps: list[dict] = []

    def run_step(self, name: str, func, *args, **kwargs) -> dict:
        step = {
            "name": name,
            "start_time": time.time(),
            "status": "running"
        }

        try:
            result = func(*args, **kwargs)
            step["status"] = "success"
            step["duration_ms"] = round((time.time() - step["start_time"]) * 1000, 2)
            step["result_type"] = type(result).__name__
            step["result_size"] = len(str(result)) if result else 0
            self.steps.append(step)
            return result

        except Exception as e:
            step["status"] = "error"
            step["duration_ms"] = round((time.time() - step["start_time"]) * 1000, 2)
            step["error_type"] = type(e).__name__
            step["error_message"] = str(e)
            step["traceback"] = traceback.format_exc()
            self.steps.append(step)
            raise

    def report(self) -> dict:
        total_duration = sum(s.get("duration_ms", 0) for s in self.steps)
        failed_steps = [s for s in self.steps if s["status"] == "error"]

        return {
            "total_steps": len(self.steps),
            "successful": len(self.steps) - len(failed_steps),
            "failed": len(failed_steps),
            "total_duration_ms": round(total_duration, 2),
            "steps": self.steps,
            "first_error": failed_steps[0] if failed_steps else None
        }

Uso:

diag = PipelineDiagnostics()

pages = diag.run_step("extract_text", extract_text_from_pdf, "documento.pdf")
chunks = diag.run_step("chunking", chunk_pages, pages)
collection = diag.run_step("indexing", create_document_index, chunks)
relevant = diag.run_step("retrieval", retrieve_relevant_chunks, collection, "pregunta")
answer = diag.run_step("generation", generate_answer, "pregunta", relevant)

print(diag.report())

Problema 1: PDF Escaneado Sin Texto Extraíble

Contexto: Document Q&A (cápsula 02)

Síntoma: extract_text_from_pdf retorna páginas vacías o con muy poco texto. El RAG no encuentra nada relevante.

Diagnóstico:

def diagnose_pdf(pdf_path: str) -> dict:
    import fitz
    doc = fitz.open(pdf_path)
    diagnosis = {
        "total_pages": len(doc),
        "pages_with_text": 0,
        "pages_empty": 0,
        "total_chars": 0,
        "total_images": 0,
        "likely_scanned": False
    }

    for page_num in range(len(doc)):
        page = doc[page_num]
        text = page.get_text().strip()
        images = page.get_images(full=True)

        if len(text) > 50:
            diagnosis["pages_with_text"] += 1
        else:
            diagnosis["pages_empty"] += 1

        diagnosis["total_chars"] += len(text)
        diagnosis["total_images"] += len(images)

    doc.close()

    diagnosis["likely_scanned"] = (
        diagnosis["pages_empty"] > diagnosis["pages_with_text"] and
        diagnosis["total_images"] > 0
    )

    diagnosis["avg_chars_per_page"] = round(
        diagnosis["total_chars"] / diagnosis["total_pages"], 1
    ) if diagnosis["total_pages"] > 0 else 0

    return diagnosis

Solución:

import base64
import fitz
from openai import OpenAI

client = OpenAI()

def extract_scanned_pdf_with_vision(pdf_path: str) -> list[dict]:
    doc = fitz.open(pdf_path)
    pages = []

    for page_num in range(len(doc)):
        page = doc[page_num]
        pix = page.get_pixmap(dpi=200)
        img_bytes = pix.tobytes("png")
        b64 = base64.b64encode(img_bytes).decode()

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": "Extrae TODO el texto visible en esta página. Mantén la estructura, encabezados, y formato."
                    },
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/png;base64,{b64}"}
                    }
                ]
            }],
            max_tokens=2000
        )

        pages.append({
            "page": page_num + 1,
            "text": response.choices[0].message.content,
            "method": "vision_ocr"
        })

    doc.close()
    return pages

Problema 2: Rate Limit (429) en Picos de Tráfico

Contexto: Todos los pipelines

Síntoma: Error 429 Too Many Requests durante horas de alto tráfico. El pipeline funciona en pruebas pero falla en producción.

Diagnóstico:

def diagnose_rate_limits(error_log: list[dict]) -> dict:
    rate_errors = [e for e in error_log if "429" in str(e.get("error", ""))]

    if not rate_errors:
        return {"rate_limit_issues": False}

    timestamps = [e["timestamp"] for e in rate_errors]
    intervals = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps)-1)]

    return {
        "rate_limit_issues": True,
        "total_429_errors": len(rate_errors),
        "avg_interval_between_errors": round(sum(intervals) / len(intervals), 2) if intervals else 0,
        "suggestion": "Implementar rate limiter con token bucket (ver cápsula 06)"
    }

Solución: Combinar rate limiter + retry + queue:

import time
from collections import deque

class RequestQueue:
    def __init__(self, max_per_minute: int = 50):
        self.max_per_minute = max_per_minute
        self.timestamps: deque = deque()

    def wait_if_needed(self):
        now = time.time()

        while self.timestamps and now - self.timestamps[0] > 60:
            self.timestamps.popleft()

        if len(self.timestamps) >= self.max_per_minute:
            wait_time = 60 - (now - self.timestamps[0])
            if wait_time > 0:
                time.sleep(wait_time)

        self.timestamps.append(time.time())

    def execute(self, func, *args, **kwargs):
        self.wait_if_needed()
        return func(*args, **kwargs)

Problema 3: Calidad Variable en Vision

Contexto: Image Analysis (cápsula 03), Video Frames (cápsula 04)

Síntoma: La clasificación o descripción de imágenes es inconsistente. La misma imagen genera respuestas diferentes entre llamadas.

Diagnóstico:

def diagnose_vision_consistency(image_path: str, prompt: str, runs: int = 5) -> dict:
    results = []
    for _ in range(runs):
        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=100,
            temperature=0
        )
        results.append(response.choices[0].message.content.strip())

    unique = len(set(results))
    return {
        "total_runs": runs,
        "unique_responses": unique,
        "consistency": round((1 - (unique - 1) / runs) * 100, 1),
        "responses": results,
        "suggestion": "Usar temperature=0, prompt más restrictivo, response_format json" if unique > 1 else "Consistencia OK"
    }

Solución:

def robust_classify(image_b64: str, categories: list[str], runs: int = 3) -> dict:
    results = []
    for _ in range(runs):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": (
                            f"Clasifica esta imagen en EXACTAMENTE UNA de: {', '.join(categories)}.\n"
                            "Responde SOLO con el nombre exacto de la categoría."
                        )
                    },
                    {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}}
                ]
            }],
            max_tokens=20,
            temperature=0
        )
        results.append(response.choices[0].message.content.strip())

    from collections import Counter
    most_common = Counter(results).most_common(1)[0]

    return {
        "category": most_common[0],
        "confidence": most_common[1] / len(results),
        "all_results": results
    }

Problema 4: Video con Frames Negros o Corruptos

Contexto: Video Frames (cápsula 04)

Síntoma: Los frames extraídos son todos negros, o algunos frames no se leen correctamente.

Diagnóstico:

import cv2
import numpy as np

def diagnose_video_frames(frames: list[dict]) -> dict:
    issues = []
    valid_frames = []

    for frame in frames:
        img = cv2.imread(frame["path"])
        if img is None:
            issues.append({"frame": frame["path"], "issue": "cannot_read"})
            continue

        mean_brightness = img.mean()
        if mean_brightness < 5:
            issues.append({"frame": frame["path"], "issue": "too_dark", "brightness": round(mean_brightness, 2)})
        elif mean_brightness > 250:
            issues.append({"frame": frame["path"], "issue": "too_bright", "brightness": round(mean_brightness, 2)})
        else:
            valid_frames.append(frame)

    return {
        "total_frames": len(frames),
        "valid_frames": len(valid_frames),
        "issues": issues,
        "valid_list": valid_frames
    }

Solución:

def extract_frames_with_validation(
    video_path: str,
    interval_seconds: float = 5.0,
    min_brightness: float = 10.0,
    max_brightness: float = 245.0,
    max_frames: int = 30
) -> list[dict]:
    import os
    os.makedirs("/tmp/validated_frames", exist_ok=True)

    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    interval_frames = int(fps * interval_seconds)

    frames = []
    frame_id = 0
    skipped = 0

    while True:
        ret, frame = cap.read()
        if not ret:
            break

        if frame_id % interval_frames == 0:
            mean_brightness = frame.mean()
            if min_brightness <= mean_brightness <= max_brightness:
                gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
                sharpness = cv2.Laplacian(gray, cv2.CV_64F).var()

                if sharpness > 50:
                    path = f"/tmp/validated_frames/frame_{frame_id:06d}.jpg"
                    cv2.imwrite(path, frame)
                    frames.append({
                        "path": path,
                        "frame_id": frame_id,
                        "timestamp": round(frame_id / fps, 2),
                        "brightness": round(mean_brightness, 2),
                        "sharpness": round(sharpness, 2)
                    })
                else:
                    skipped += 1
            else:
                skipped += 1

            if len(frames) >= max_frames:
                break

        frame_id += 1

    cap.release()
    return frames

Problema 5: Documentos Corruptos o Formatos No Soportados

Contexto: Document Q&A (cápsula 02), Combinaciones (cápsula 05)

Síntoma: PyMuPDF lanza excepción al abrir el archivo. El pipeline se rompe sin feedback claro.

Diagnóstico y solución:

def safe_extract_document(path: str) -> dict:
    from pathlib import Path
    p = Path(path)

    if not p.exists():
        return {"error": "file_not_found", "message": f"No existe: {path}"}

    if p.suffix.lower() not in {".pdf", ".png", ".jpg", ".jpeg", ".tiff"}:
        return {"error": "unsupported_format", "message": f"Formato no soportado: {p.suffix}"}

    if p.stat().st_size == 0:
        return {"error": "empty_file", "message": "Archivo vacío"}

    if p.stat().st_size > 100 * 1024 * 1024:
        return {"error": "file_too_large", "message": f"Archivo muy grande: {p.stat().st_size / (1024*1024):.1f}MB"}

    try:
        import fitz
        doc = fitz.open(path)
        page_count = len(doc)
        doc.close()
        return {"status": "ok", "pages": page_count}
    except Exception as e:
        return {"error": "corrupt_file", "message": f"No se puede abrir: {e}"}

Problema 6: Context Window Excedido en Pipelines Multi-Modal

Contexto: Combinaciones Multi-Modalidad (cápsula 05)

Síntoma: Error maximum context length exceeded al combinar transcripción larga + documento largo + análisis de imagen.

Diagnóstico:

def estimate_tokens(text: str) -> int:
    return len(text) // 4

def diagnose_context_overflow(inputs: dict[str, str], model: str = "gpt-4o") -> dict:
    limits = {
        "gpt-4o": 128_000,
        "gpt-4o-mini": 128_000,
    }

    model_limit = limits.get(model, 128_000)

    token_estimates = {}
    total = 0
    for key, text in inputs.items():
        tokens = estimate_tokens(text)
        token_estimates[key] = tokens
        total += tokens

    return {
        "total_estimated_tokens": total,
        "model_limit": model_limit,
        "fits": total < model_limit * 0.9,
        "overflow_by": max(0, total - int(model_limit * 0.9)),
        "by_input": token_estimates,
        "suggestion": "Truncar inputs más grandes o usar summarization previa" if total > model_limit * 0.9 else "OK"
    }

Solución:

def smart_truncate(inputs: dict[str, str], max_total_tokens: int = 100_000) -> dict[str, str]:
    token_counts = {k: estimate_tokens(v) for k, v in inputs.items()}
    total = sum(token_counts.values())

    if total <= max_total_tokens:
        return inputs

    ratio = max_total_tokens / total
    truncated = {}

    for key, text in inputs.items():
        allowed_chars = int(len(text) * ratio)
        truncated[key] = text[:allowed_chars]

    return truncated

Problema 7: Costos Inesperados

Contexto: Todos los pipelines

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

Diagnóstico:

def diagnose_cost_issues(tracker) -> dict:
    summary = tracker.summary()
    calls = tracker.calls

    expensive_calls = sorted(calls, key=lambda c: c["cost"], reverse=True)[:10]

    high_cost_models = {
        k: v for k, v in summary["by_model"].items()
        if v > summary["total_cost"] * 0.3
    }

    return {
        "total_cost": summary["total_cost"],
        "total_calls": summary["total_calls"],
        "avg_cost_per_call": round(summary["total_cost"] / summary["total_calls"], 6) if summary["total_calls"] > 0 else 0,
        "most_expensive_calls": expensive_calls[:5],
        "high_cost_models": high_cost_models,
        "recommendations": generate_cost_recommendations(summary)
    }


def generate_cost_recommendations(summary: dict) -> list[str]:
    recommendations = []

    if "gpt-4o" in summary.get("by_model", {}):
        gpt4o_cost = summary["by_model"]["gpt-4o"]
        if gpt4o_cost > summary["total_cost"] * 0.5:
            recommendations.append(
                f"GPT-4o representa {gpt4o_cost/summary['total_cost']*100:.0f}% del costo. "
                "Evalúa migrar tareas simples a gpt-4o-mini."
            )

    if summary.get("by_operation", {}).get("embedding", 0) > summary["total_cost"] * 0.2:
        recommendations.append("Embeddings representan >20% del costo. Implementa cache de embeddings.")

    if summary["total_calls"] > 1000:
        recommendations.append("Alto volumen de llamadas. Evalúa batch API para reducir overhead.")

    return recommendations

Problema 8: Latencia Alta en Pipelines

Contexto: Todos los pipelines

Síntoma: El pipeline tarda 20+ segundos en responder.

Diagnóstico:

def diagnose_latency(diagnostics: PipelineDiagnostics) -> dict:
    report = diagnostics.report()
    steps = report["steps"]

    bottleneck = max(steps, key=lambda s: s.get("duration_ms", 0)) if steps else None

    return {
        "total_duration_ms": report["total_duration_ms"],
        "bottleneck": {
            "step": bottleneck["name"] if bottleneck else None,
            "duration_ms": bottleneck.get("duration_ms", 0) if bottleneck else 0,
            "percentage": round(
                bottleneck.get("duration_ms", 0) / report["total_duration_ms"] * 100, 1
            ) if bottleneck and report["total_duration_ms"] > 0 else 0
        },
        "steps_breakdown": [
            {"name": s["name"], "duration_ms": s.get("duration_ms", 0)}
            for s in steps
        ],
        "recommendations": [
            "Paralelizar pasos independientes con asyncio.gather",
            "Cachear resultados intermedios (embeddings, transcripciones)",
            "Usar gpt-4o-mini para pasos no críticos",
            "Reducir resolución de imágenes antes de enviar a Vision"
        ]
    }

Performance Monitoring Dashboard

class PerformanceDashboard:
    def __init__(self):
        self.operations: list[dict] = []

    def record(self, operation: str, duration_ms: float, status: str, cost: float = 0):
        self.operations.append({
            "operation": operation,
            "duration_ms": duration_ms,
            "status": status,
            "cost": cost,
            "timestamp": time.time()
        })

    def health_check(self) -> dict:
        if not self.operations:
            return {"status": "no_data"}

        last_hour = [o for o in self.operations if o["timestamp"] > time.time() - 3600]

        if not last_hour:
            return {"status": "idle", "message": "No operations in last hour"}

        error_rate = sum(1 for o in last_hour if o["status"] == "error") / len(last_hour) * 100
        avg_latency = sum(o["duration_ms"] for o in last_hour) / len(last_hour)
        total_cost = sum(o["cost"] for o in last_hour)

        status = "healthy"
        if error_rate > 10:
            status = "degraded"
        if error_rate > 50:
            status = "critical"
        if avg_latency > 10000:
            status = "slow"

        return {
            "status": status,
            "last_hour": {
                "operations": len(last_hour),
                "error_rate": round(error_rate, 1),
                "avg_latency_ms": round(avg_latency, 2),
                "total_cost": round(total_cost, 4)
            }
        }

    def alerts(self) -> list[dict]:
        active_alerts = []
        health = self.health_check()

        if health.get("status") == "degraded":
            active_alerts.append({
                "severity": "warning",
                "message": f"Error rate alto: {health['last_hour']['error_rate']}%"
            })

        if health.get("status") == "critical":
            active_alerts.append({
                "severity": "critical",
                "message": f"Error rate crítico: {health['last_hour']['error_rate']}%"
            })

        if health.get("last_hour", {}).get("avg_latency_ms", 0) > 10000:
            active_alerts.append({
                "severity": "warning",
                "message": f"Latencia alta: {health['last_hour']['avg_latency_ms']}ms promedio"
            })

        if health.get("last_hour", {}).get("total_cost", 0) > 5.0:
            active_alerts.append({
                "severity": "warning",
                "message": f"Costo alto última hora: ${health['last_hour']['total_cost']:.2f}"
            })

        return active_alerts

Quick Reference: Errores y Soluciones

Ejercicios

Ejercicio 1: Health check endpoint

Crea una función que verifique conectividad con OpenAI (chat, embeddings, whisper) y reporte el estado de cada servicio.

def comprehensive_health_check() -> dict:
    status = {}

    try:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "ping"}],
            max_tokens=5
        )
        status["chat"] = {"status": "ok", "model": "gpt-4o-mini"}
    except Exception as e:
        status["chat"] = {"status": "error", "error": str(e)}

    try:
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=["test"]
        )
        status["embeddings"] = {"status": "ok", "model": "text-embedding-3-small"}
    except Exception as e:
        status["embeddings"] = {"status": "error", "error": str(e)}

    try:
        import tempfile, wave, struct
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
            with wave.open(tmp.name, "w") as wav:
                wav.setnchannels(1)
                wav.setsampwidth(2)
                wav.setframerate(16000)
                wav.writeframes(struct.pack("<" + "h" * 16000, *([0] * 16000)))
            with open(tmp.name, "rb") as f:
                client.audio.transcriptions.create(model="whisper-1", file=f)
        status["whisper"] = {"status": "ok"}
    except Exception as e:
        status["whisper"] = {"status": "error", "error": str(e)}

    all_ok = all(s["status"] == "ok" for s in status.values())
    return {"overall": "healthy" if all_ok else "degraded", "services": status}

Ejercicio 2: Pipeline con diagnóstico automático

Modifica un pipeline de Document Q&A para que automáticamente diagnostique y reporte problemas cuando falla.

def document_qa_with_diagnostics(pdf_path: str, question: str) -> dict:
    diag = PipelineDiagnostics()

    try:
        pdf_check = diag.run_step("validate_pdf", safe_extract_document, pdf_path)
        if "error" in pdf_check:
            return {"error": pdf_check["error"], "diagnostics": diag.report()}

        pages = diag.run_step("extract_text", extract_text_from_pdf, pdf_path)

        total_text = sum(len(p["text"]) for p in pages)
        if total_text < 100:
            pages = diag.run_step("ocr_fallback", extract_scanned_pdf_with_vision, pdf_path)

        chunks = diag.run_step("chunking", chunk_pages, pages)
        collection = diag.run_step("indexing", create_document_index, chunks)
        relevant = diag.run_step("retrieval", retrieve_relevant_chunks, collection, question)
        answer = diag.run_step("generation", generate_answer, question, relevant)

        answer["diagnostics"] = diag.report()
        return answer

    except Exception as e:
        report = diag.report()
        return {
            "error": str(e),
            "failed_step": report["first_error"]["name"] if report["first_error"] else "unknown",
            "diagnostics": report
        }

Ejercicio 3: Monitor de rate limits en tiempo real

Crea un sistema que registre cada llamada a la API y alerte cuando se acerque al rate limit.

from collections import deque

class RateLimitMonitor:
    def __init__(self, limit_per_minute: int = 60, alert_threshold: float = 0.8):
        self.limit = limit_per_minute
        self.threshold = alert_threshold
        self.requests: deque = deque()

    def record_request(self) -> dict:
        now = time.time()
        self.requests.append(now)

        while self.requests and now - self.requests[0] > 60:
            self.requests.popleft()

        current_rate = len(self.requests)
        usage = current_rate / self.limit

        result = {
            "current_rpm": current_rate,
            "limit_rpm": self.limit,
            "usage_percent": round(usage * 100, 1)
        }

        if usage >= self.threshold:
            result["alert"] = True
            result["message"] = f"Rate limit usage at {usage*100:.0f}%. Slow down."
            result["recommended_delay"] = round(60 / self.limit, 2)
        else:
            result["alert"] = False

        return result

    def safe_execute(self, func, *args, **kwargs):
        status = self.record_request()
        if status["alert"]:
            delay = status["recommended_delay"]
            time.sleep(delay)
        return func(*args, **kwargs)

Ejercicio 4: Test suite para pipeline multimodal

Crea una función que ejecute tests básicos de cada componente del pipeline y reporte qué funciona y qué no.

def run_pipeline_tests() -> dict:
    tests = {}

    try:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "Responde OK"}],
            max_tokens=5
        )
        tests["llm_basic"] = {"status": "pass", "response": response.choices[0].message.content}
    except Exception as e:
        tests["llm_basic"] = {"status": "fail", "error": str(e)}

    try:
        emb = client.embeddings.create(model="text-embedding-3-small", input=["test"])
        tests["embeddings"] = {"status": "pass", "dimensions": len(emb.data[0].embedding)}
    except Exception as e:
        tests["embeddings"] = {"status": "fail", "error": str(e)}

    try:
        import fitz
        doc = fitz.open()
        page = doc.new_page()
        page.insert_text((50, 50), "Test content")
        doc.save("/tmp/test_doc.pdf")
        doc.close()
        doc2 = fitz.open("/tmp/test_doc.pdf")
        text = doc2[0].get_text()
        doc2.close()
        tests["pdf_extraction"] = {"status": "pass" if "Test" in text else "fail"}
    except Exception as e:
        tests["pdf_extraction"] = {"status": "fail", "error": str(e)}

    try:
        import chromadb
        c = chromadb.Client()
        col = c.create_collection("test_collection")
        col.add(documents=["test"], ids=["1"])
        r = col.query(query_texts=["test"], n_results=1)
        c.delete_collection("test_collection")
        tests["chromadb"] = {"status": "pass"}
    except Exception as e:
        tests["chromadb"] = {"status": "fail", "error": str(e)}

    passed = sum(1 for t in tests.values() if t["status"] == "pass")
    total = len(tests)

    return {
        "passed": passed,
        "failed": total - passed,
        "total": total,
        "tests": tests
    }

Recursos Adicionales

1. OpenAI Error Codes — Referencia de errores
2. OpenAI Rate Limits — Límites por modelo y tier
3. Structured Logging Best Practices — Logging estructurado en Python
4. Circuit Breaker Pattern — Patrón de resiliencia