7. Use Case Troubleshooting

Description

The previous capsules included troubleshooting sections specific to each pattern. This capsule is the centralized reference: a catalog of the most common problems when building multimodal systems in production, with diagnostic tools, resolution patterns, and monitoring techniques. When something fails in your pipeline, start here.

Why it matters: In production, errors aren't "the API doesn't work" — they're subtle combinations: a scanned PDF that breaks extraction, a video where all the frames are black, a multi-modal pipeline that exceeds the context window, a rate limit that only appears during peak hours. Diagnosing these problems requires tools and methodology, not guesswork.

Connection with the module: This capsule complements capsule 06 (Production Patterns) with the diagnosis and resolution aspect. The Use Case Selector (capsule 08) needs to gracefully handle all the errors cataloged here.

Diagnostic Methodology

When a pipeline fails, follow this order:

1. IDENTIFY   → Which error? Where in the pipeline?
2. REPRODUCE  → Is it consistent or intermittent?
3. ISOLATE    → Is it the input, the API, or the processing?
4. RESOLVE    → Apply fix + verify
5. PREVENT    → Add validation/test so it doesn't recur

Diagnostic tool

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
        }

Usage:

diag = PipelineDiagnostics()

pages = diag.run_step("extract_text", extract_text_from_pdf, "document.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, "question")
answer = diag.run_step("generation", generate_answer, "question", relevant)

print(diag.report())

Problem 1: Scanned PDF With No Extractable Text

Context: Document Q&A (capsule 02)

Symptom: extract_text_from_pdf returns empty pages or pages with very little text. The RAG finds nothing relevant.

Diagnosis:

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

Solution:

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": "Extract ALL the visible text on this page. Keep the structure, headings, and formatting."
                    },
                    {
                        "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

Problem 2: Rate Limit (429) During Traffic Spikes

Context: All pipelines

Symptom: Error 429 Too Many Requests during high-traffic hours. The pipeline works in tests but fails in production.

Diagnosis:

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": "Implement a token bucket rate limiter (see capsule 06)"
    }

Solution: Combine 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)

Problem 3: Variable Quality in Vision

Context: Image Analysis (capsule 03), Video Frames (capsule 04)

Symptom: Image classification or description is inconsistent. The same image generates different answers between calls.

Diagnosis:

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": "Use temperature=0, a more restrictive prompt, response_format json" if unique > 1 else "Consistency OK"
    }

Solution:

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"Classify this image into EXACTLY ONE of: {', '.join(categories)}.\n"
                            "Reply ONLY with the exact name of the category."
                        )
                    },
                    {"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
    }

Problem 4: Video With Black or Corrupt Frames

Context: Video Frames (capsule 04)

Symptom: The extracted frames are all black, or some frames don't read correctly.

Diagnosis:

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
    }

Solution:

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

Problem 5: Corrupt Documents or Unsupported Formats

Context: Document Q&A (capsule 02), Combinations (capsule 05)

Symptom: PyMuPDF raises an exception when opening the file. The pipeline breaks with no clear feedback.

Diagnosis and solution:

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"Doesn't exist: {path}"}

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

    if p.stat().st_size == 0:
        return {"error": "empty_file", "message": "Empty file"}

    if p.stat().st_size > 100 * 1024 * 1024:
        return {"error": "file_too_large", "message": f"File too large: {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"Can't open: {e}"}

Problem 6: Context Window Exceeded in Multi-Modal Pipelines

Context: Multi-Modality Combinations (capsule 05)

Symptom: Error maximum context length exceeded when combining a long transcript + long document + image analysis.

Diagnosis:

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": "Truncate the largest inputs or use prior summarization" if total > model_limit * 0.9 else "OK"
    }

Solution:

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

Problem 7: Unexpected Costs

Context: All pipelines

Symptom: The monthly bill is much higher than expected.

Diagnosis:

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 represents {gpt4o_cost/summary['total_cost']*100:.0f}% of the cost. "
                "Consider migrating simple tasks to gpt-4o-mini."
            )

    if summary.get("by_operation", {}).get("embedding", 0) > summary["total_cost"] * 0.2:
        recommendations.append("Embeddings represent >20% of the cost. Implement an embeddings cache.")

    if summary["total_calls"] > 1000:
        recommendations.append("High call volume. Consider the batch API to reduce overhead.")

    return recommendations

Problem 8: High Latency in Pipelines

Context: All pipelines

Symptom: The pipeline takes 20+ seconds to respond.

Diagnosis:

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": [
            "Parallelize independent steps with asyncio.gather",
            "Cache intermediate results (embeddings, transcriptions)",
            "Use gpt-4o-mini for non-critical steps",
            "Reduce image resolution before sending to 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"High error rate: {health['last_hour']['error_rate']}%"
            })

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

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

        if health.get("last_hour", {}).get("total_cost", 0) > 5.0:
            active_alerts.append({
                "severity": "warning",
                "message": f"High cost last hour: ${health['last_hour']['total_cost']:.2f}"
            })

        return active_alerts

Quick Reference: Errors and Solutions

Exercises

Exercise 1: Health check endpoint

Create a function that checks connectivity with OpenAI (chat, embeddings, whisper) and reports the status of each service.

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}

Exercise 2: Pipeline with automatic diagnosis

Modify a Document Q&A pipeline so it automatically diagnoses and reports problems when it fails.

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
        }

Exercise 3: Real-time rate limit monitor

Create a system that logs each API call and alerts when it approaches the 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)

Exercise 4: Test suite for a multimodal pipeline

Create a function that runs basic tests of each pipeline component and reports what works and what doesn't.

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

    try:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "Reply 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
    }

Additional Resources

1. OpenAI Error Codes — Error reference
2. OpenAI Rate Limits — Limits per model and tier
3. Structured Logging Best Practices — Structured logging in Python
4. Circuit Breaker Pattern — Resilience pattern