4. Errors: Rate, Types, Retries

Capsule description

Your system returns an error. What kind of error? In traditional software, the answer is direct: a 500 is a server error, a 429 is rate limiting, a 404 is resource not found. In AI systems, the answer is much more complex: a provider timeout and a hallucination are both errors, but they're fundamentally different problems, with different causes, different solutions, and different metrics.

A timeout is a technical error: the infrastructure failed, the provider didn't respond, your API key expired. You detect it with HTTP status codes. You fix it with retries, circuit breakers, and fallbacks. A hallucination is a functional error: the infrastructure worked perfectly (status 200), the model responded fast, the tokens were billed — but the response is incorrect. You don't detect it with status codes. You don't fix it with retries (the retry may generate another hallucination).

If you measure both with the same metric ("error rate"), you lose the most important information: what kind of problem you have. A system with 0.1% technical errors and 8% functional errors has an "error rate" of 8.1%. But the action is completely different for each type. This capsule teaches you to classify, measure, and act on each error category separately.

Technical Errors

What they are

Technical errors are infrastructure, network, or API failures that prevent completing the request. The system couldn't generate a response. These errors are detectable with HTTP status codes or exceptions.

Main types

Technical error         Status/Exception         Typical cause
───────────────────────────────────────────────────────────────────
Timeout                 408 / TimeoutError       Slow provider, network
Rate limit              429                      Too many requests/minute
API down                500-503                  Provider having problems
Auth error              401 / 403                Invalid/expired API key
Bad request             400                      Malformed prompt, tokens exceeded
Connection error        ConnectionError          Network, DNS, firewall
Model not found         404                      Deprecated/nonexistent model
Context length exceeded 400                      Prompt + max_tokens > model limit
───────────────────────────────────────────────────────────────────

Code: Capturing technical errors

import time
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Optional
from openai import OpenAI, APIError, APITimeoutError, RateLimitError, APIConnectionError
from dotenv import load_dotenv

load_dotenv()
client = OpenAI()


class ErrorCategory(Enum):
    NONE = "none"
    TIMEOUT = "timeout"
    RATE_LIMIT = "rate_limit"
    API_ERROR = "api_error"
    AUTH_ERROR = "auth_error"
    CONNECTION_ERROR = "connection_error"
    CONTEXT_LENGTH = "context_length"
    BAD_REQUEST = "bad_request"
    UNKNOWN_TECHNICAL = "unknown_technical"
    HALLUCINATION = "hallucination"
    FORMAT_VIOLATION = "format_violation"
    OFF_TOPIC = "off_topic"
    EMPTY_RESPONSE = "empty_response"
    UNKNOWN_FUNCTIONAL = "unknown_functional"


@dataclass
class ErrorRecord:
    timestamp: str
    endpoint: str
    category: ErrorCategory
    is_technical: bool
    is_functional: bool
    message: str
    status_code: Optional[int] = None
    model: Optional[str] = None
    prompt_preview: Optional[str] = None
    output_preview: Optional[str] = None
    retry_count: int = 0
    cost_usd: float = 0.0


def classify_technical_error(exception: Exception) -> ErrorCategory:
    """Classifies a technical error based on the exception."""
    if isinstance(exception, APITimeoutError):
        return ErrorCategory.TIMEOUT
    elif isinstance(exception, RateLimitError):
        return ErrorCategory.RATE_LIMIT
    elif isinstance(exception, APIConnectionError):
        return ErrorCategory.CONNECTION_ERROR
    elif isinstance(exception, APIError):
        if hasattr(exception, 'status_code'):
            if exception.status_code == 401 or exception.status_code == 403:
                return ErrorCategory.AUTH_ERROR
            elif exception.status_code == 400:
                msg = str(exception).lower()
                if "context_length" in msg or "maximum context" in msg or "token" in msg:
                    return ErrorCategory.CONTEXT_LENGTH
                return ErrorCategory.BAD_REQUEST
            elif exception.status_code >= 500:
                return ErrorCategory.API_ERROR
        return ErrorCategory.API_ERROR
    else:
        return ErrorCategory.UNKNOWN_TECHNICAL


def call_llm_with_error_tracking(
    prompt: str,
    model: str = "gpt-4o-mini",
    endpoint: str = "/chat",
    max_tokens: int = 300,
) -> tuple[Optional[str], Optional[ErrorRecord]]:
    """Calls the LLM capturing and classifying technical errors."""
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=max_tokens,
            timeout=30,
        )
        output = response.choices[0].message.content
        return output, None

    except Exception as e:
        category = classify_technical_error(e)
        status_code = getattr(e, 'status_code', None)

        error_record = ErrorRecord(
            timestamp=datetime.now().isoformat(),
            endpoint=endpoint,
            category=category,
            is_technical=True,
            is_functional=False,
            message=str(e)[:200],
            status_code=status_code,
            model=model,
            prompt_preview=prompt[:100],
        )

        return None, error_record


output, error = call_llm_with_error_tracking("What is monitoring?")
if error:
    print(f"Error: {error.category.value} — {error.message}")
else:
    print(f"OK: {output[:80]}...")

Functional Errors

What they are

Functional errors occur when the system responds successfully (status 200, tokens billed, normal latency) but the output is incorrect, useless, or dangerous. They're invisible to traditional monitoring.

Main types

Functional error         Status    How to detect
───────────────────────────────────────────────────────────────────
Hallucination            200       Output contains made-up data
Format violation         200       JSON was requested, responds with free text
Off-topic response       200       Response unrelated to the question
Empty/minimal response   200       Empty response or "I can't help you"
Contradicts source       200       Output contradicts the RAG context
Harmful content          200       Inappropriate, biased, or dangerous content
Incomplete answer        200       Truncated or partial response
Language mismatch        200       Asked in English, responds in Spanish
───────────────────────────────────────────────────────────────────

Code: Detecting functional errors with heuristics

import re
from dataclasses import dataclass


@dataclass
class FunctionalCheck:
    check_name: str
    passed: bool
    detail: str
    severity: str  # "low", "medium", "high"


def check_hallucination_signals(output: str) -> FunctionalCheck:
    """Detects common hallucination signals."""
    hallucination_phrases = [
        "i don't have information",
        "i can't confirm",
        "based on my knowledge",
        "as far as i know",
        "i think",
        "it could be",
        "according to my data",
        "i'm not sure but",
    ]

    found = [p for p in hallucination_phrases if p in output.lower()]

    if found:
        return FunctionalCheck(
            check_name="hallucination_signals",
            passed=False,
            detail=f"Uncertainty signals detected: {found}",
            severity="medium",
        )
    return FunctionalCheck(
        check_name="hallucination_signals",
        passed=True,
        detail="No hallucination signals detected",
        severity="low",
    )


def check_format_compliance(output: str, expected_format: str = None) -> FunctionalCheck:
    """Verifies that the output complies with the expected format."""
    if expected_format == "json":
        try:
            import json
            json.loads(output)
            return FunctionalCheck("format_compliance", True, "Valid JSON", "low")
        except json.JSONDecodeError as e:
            return FunctionalCheck(
                "format_compliance", False,
                f"Expected JSON, got invalid: {str(e)[:80]}", "high",
            )
    elif expected_format == "list":
        lines = [l.strip() for l in output.strip().split("\n") if l.strip()]
        has_list = any(
            re.match(r'^[\d]+[.)\-]|^[-*•]', line) for line in lines
        )
        if not has_list:
            return FunctionalCheck(
                "format_compliance", False,
                "Expected list format, got free text", "medium",
            )
    return FunctionalCheck("format_compliance", True, "Format OK", "low")


def check_response_length(
    output: str,
    min_length: int = 10,
    max_length: int = 5000,
) -> FunctionalCheck:
    """Verifies that the response has a reasonable length."""
    length = len(output.strip())

    if length < min_length:
        return FunctionalCheck(
            "response_length", False,
            f"Response too short: {length} chars (min: {min_length})", "high",
        )
    if length > max_length:
        return FunctionalCheck(
            "response_length", False,
            f"Response too long: {length} chars (max: {max_length})", "medium",
        )
    return FunctionalCheck(
        "response_length", True,
        f"Length OK: {length} chars", "low",
    )


def check_language(output: str, expected_lang: str = "en") -> FunctionalCheck:
    """Simple heuristic to verify the response is in the expected language."""
    spanish_indicators = ["el", "la", "los", "las", "de", "en", "que", "por", "para", "con"]
    english_indicators = ["the", "is", "are", "was", "were", "have", "has", "for", "with", "that"]

    words = output.lower().split()[:50]

    es_count = sum(1 for w in words if w in spanish_indicators)
    en_count = sum(1 for w in words if w in english_indicators)

    if expected_lang == "en" and es_count > en_count and es_count > 5:
        return FunctionalCheck(
            "language", False,
            f"Expected English, detected Spanish (en:{en_count}, es:{es_count})",
            "medium",
        )
    return FunctionalCheck("language", True, "Language OK", "low")


def classify_functional_errors(
    output: str,
    expected_format: str = None,
    expected_lang: str = "en",
    min_length: int = 10,
) -> list[FunctionalCheck]:
    """Runs all functional checks on an output."""
    checks = [
        check_hallucination_signals(output),
        check_format_compliance(output, expected_format),
        check_response_length(output, min_length=min_length),
        check_language(output, expected_lang),
    ]
    return checks


output_good = "Latency in AI systems is measured in three dimensions: TTFT, TTI, and end-to-end."
output_hallucination = "According to my data, I think the latency is measured with the XYZ-42 protocol."
output_wrong_format = "Here are the results: the latency is high and the cost is low."
output_wrong_lang = "La latencia en sistemas AI se mide usando tres dimensiones clave."

test_cases = [
    ("Good response", output_good, None),
    ("Hallucination signals", output_hallucination, None),
    ("Wrong format (expected JSON)", output_wrong_format, "json"),
    ("Wrong language", output_wrong_lang, None),
]

for name, output, fmt in test_cases:
    print(f"\n--- {name} ---")
    checks = classify_functional_errors(output, expected_format=fmt)
    for c in checks:
        status = "PASS" if c.passed else "FAIL"
        print(f"  [{status}] {c.check_name}: {c.detail}")

ErrorTracker: Complete Class

from collections import defaultdict
from datetime import datetime
from dataclasses import dataclass, field
from typing import Optional


class ErrorTracker:
    """Tracks technical and functional errors separately."""

    def __init__(self):
        self.technical_errors: list[ErrorRecord] = []
        self.functional_errors: list[ErrorRecord] = []
        self.total_requests: int = 0

    def record_request(
        self,
        endpoint: str,
        model: str,
        technical_error: Optional[ErrorRecord] = None,
        functional_checks: Optional[list[FunctionalCheck]] = None,
        output: str = "",
        cost_usd: float = 0.0,
        prompt: str = "",
    ):
        self.total_requests += 1

        if technical_error:
            self.technical_errors.append(technical_error)
            return

        if functional_checks:
            failures = [c for c in functional_checks if not c.passed]
            if failures:
                worst = max(failures, key=lambda c: {"low": 0, "medium": 1, "high": 2}[c.severity])

                category_map = {
                    "hallucination_signals": ErrorCategory.HALLUCINATION,
                    "format_compliance": ErrorCategory.FORMAT_VIOLATION,
                    "language": ErrorCategory.OFF_TOPIC,
                    "response_length": ErrorCategory.EMPTY_RESPONSE,
                }
                category = category_map.get(worst.check_name, ErrorCategory.UNKNOWN_FUNCTIONAL)

                error = ErrorRecord(
                    timestamp=datetime.now().isoformat(),
                    endpoint=endpoint,
                    category=category,
                    is_technical=False,
                    is_functional=True,
                    message=worst.detail,
                    model=model,
                    prompt_preview=prompt[:100],
                    output_preview=output[:200],
                    cost_usd=cost_usd,
                )
                self.functional_errors.append(error)

    def technical_error_rate(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return len(self.technical_errors) / self.total_requests

    def functional_error_rate(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return len(self.functional_errors) / self.total_requests

    def combined_error_rate(self) -> float:
        return self.technical_error_rate() + self.functional_error_rate()

    def errors_by_category(self) -> dict[str, int]:
        counts: dict[str, int] = defaultdict(int)
        for e in self.technical_errors:
            counts[e.category.value] += 1
        for e in self.functional_errors:
            counts[e.category.value] += 1
        return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True))

    def errors_by_endpoint(self) -> dict[str, dict]:
        endpoints: dict[str, dict] = defaultdict(
            lambda: {"technical": 0, "functional": 0, "total_requests": 0}
        )
        for e in self.technical_errors:
            endpoints[e.endpoint]["technical"] += 1
        for e in self.functional_errors:
            endpoints[e.endpoint]["functional"] += 1
        return dict(endpoints)

    def wasted_cost(self) -> float:
        return sum(e.cost_usd for e in self.functional_errors)

    def print_dashboard(self):
        print("=" * 60)
        print("ERROR DASHBOARD")
        print("=" * 60)
        print(f"Total requests:       {self.total_requests}")
        print(f"Technical errors:     {len(self.technical_errors)} "
              f"({self.technical_error_rate():.2%})")
        print(f"Functional errors:    {len(self.functional_errors)} "
              f"({self.functional_error_rate():.2%})")
        print(f"Combined error rate:  {self.combined_error_rate():.2%}")
        print(f"Wasted cost:          ${self.wasted_cost():.4f} "
              f"(functional errors that were billed)")
        print()

        print("By category:")
        for category, count in self.errors_by_category().items():
            pct = count / self.total_requests * 100 if self.total_requests > 0 else 0
            print(f"  {category:<25} {count:>4} ({pct:.1f}%)")
        print()

        by_ep = self.errors_by_endpoint()
        if by_ep:
            print("By endpoint:")
            for ep, data in by_ep.items():
                print(f"  {ep}: tech={data['technical']}, func={data['functional']}")

        print("=" * 60)

Using the ErrorTracker

import random

random.seed(42)

tracker = ErrorTracker()

for i in range(500):
    endpoint = random.choice(["/chat", "/chat", "/chat", "/analyze", "/summarize"])
    model = "gpt-4o-mini" if endpoint != "/analyze" else "gpt-4o"

    r = random.random()

    if r < 0.02:
        # 2% timeout
        error = ErrorRecord(
            timestamp=datetime.now().isoformat(),
            endpoint=endpoint, category=ErrorCategory.TIMEOUT,
            is_technical=True, is_functional=False,
            message="Request timed out after 30s", model=model,
        )
        tracker.record_request(endpoint, model, technical_error=error)

    elif r < 0.035:
        # 1.5% rate limit
        error = ErrorRecord(
            timestamp=datetime.now().isoformat(),
            endpoint=endpoint, category=ErrorCategory.RATE_LIMIT,
            is_technical=True, is_functional=False,
            message="Rate limit exceeded", model=model,
        )
        tracker.record_request(endpoint, model, technical_error=error)

    elif r < 0.085:
        # 5% functional errors
        error_type = random.choice(["hallucination", "format", "language", "empty"])

        checks = []
        if error_type == "hallucination":
            checks.append(FunctionalCheck(
                "hallucination_signals", False,
                "Uncertainty signals detected", "medium",
            ))
        elif error_type == "format":
            checks.append(FunctionalCheck(
                "format_compliance", False,
                "Expected JSON, got free text", "high",
            ))
        elif error_type == "language":
            checks.append(FunctionalCheck(
                "language", False,
                "Expected English, detected Spanish", "medium",
            ))
        else:
            checks.append(FunctionalCheck(
                "response_length", False,
                "Response too short: 3 chars", "high",
            ))

        cost = random.uniform(0.0001, 0.01)
        tracker.record_request(
            endpoint, model,
            functional_checks=checks,
            cost_usd=cost,
            prompt=f"Test prompt {i}",
            output="Some output",
        )
    else:
        # Success
        tracker.record_request(endpoint, model)

tracker.print_dashboard()

Retry Storms: How Retries Multiply Problems

The problem

When a request fails, the natural reaction is to retry. But in AI systems, retries have a cost that doesn't exist in traditional APIs:

• Multiplied cost: Each retry consumes tokens. 3 retries = 3x cost (although only successful ones bill completion tokens, failed ones bill if the timeout happens after the model started generating)
• Multiplied latency: 3 retries of 5 seconds = 15 seconds of total wait
• Cascade: If the problem is rate limiting, the retries worsen the rate limit

Code: Retry tracker

import time
import random
from dataclasses import dataclass
from typing import Optional


@dataclass
class RetryRecord:
    request_id: str
    endpoint: str
    model: str
    total_attempts: int
    successful: bool
    total_latency_ms: float
    total_cost_usd: float
    per_attempt: list[dict]


class RetryTracker:
    """Tracks retries and their impact on cost and latency."""

    def __init__(self, max_retries: int = 3, base_delay_ms: int = 1000):
        self.max_retries = max_retries
        self.base_delay_ms = base_delay_ms
        self.records: list[RetryRecord] = []

    def execute_with_retry(
        self,
        func,
        request_id: str,
        endpoint: str,
        model: str,
        cost_per_attempt: float = 0.001,
    ) -> tuple[Optional[str], RetryRecord]:
        attempts = []
        total_cost = 0.0
        start = time.perf_counter()

        for attempt in range(self.max_retries + 1):
            attempt_start = time.perf_counter()

            try:
                result = func()
                attempt_ms = (time.perf_counter() - attempt_start) * 1000
                total_cost += cost_per_attempt

                attempts.append({
                    "attempt": attempt + 1,
                    "status": "success",
                    "latency_ms": attempt_ms,
                    "cost_usd": cost_per_attempt,
                })

                record = RetryRecord(
                    request_id=request_id,
                    endpoint=endpoint,
                    model=model,
                    total_attempts=attempt + 1,
                    successful=True,
                    total_latency_ms=(time.perf_counter() - start) * 1000,
                    total_cost_usd=total_cost,
                    per_attempt=attempts,
                )
                self.records.append(record)
                return result, record

            except Exception as e:
                attempt_ms = (time.perf_counter() - attempt_start) * 1000
                total_cost += cost_per_attempt * 0.3

                attempts.append({
                    "attempt": attempt + 1,
                    "status": "failed",
                    "error": str(e)[:100],
                    "latency_ms": attempt_ms,
                    "cost_usd": cost_per_attempt * 0.3,
                })

                if attempt < self.max_retries:
                    delay = self.base_delay_ms * (2 ** attempt) / 1000
                    time.sleep(delay)

        total_ms = (time.perf_counter() - start) * 1000
        record = RetryRecord(
            request_id=request_id,
            endpoint=endpoint,
            model=model,
            total_attempts=self.max_retries + 1,
            successful=False,
            total_latency_ms=total_ms,
            total_cost_usd=total_cost,
            per_attempt=attempts,
        )
        self.records.append(record)
        return None, record

    def retry_summary(self) -> dict:
        if not self.records:
            return {}

        total = len(self.records)
        retried = [r for r in self.records if r.total_attempts > 1]
        failed = [r for r in self.records if not r.successful]
        extra_cost = sum(
            r.total_cost_usd - r.per_attempt[0]["cost_usd"]
            for r in retried
        )

        return {
            "total_requests": total,
            "retried": len(retried),
            "retry_rate": len(retried) / total,
            "failed_after_retries": len(failed),
            "extra_cost_from_retries": extra_cost,
            "avg_attempts": sum(r.total_attempts for r in self.records) / total,
        }

    def print_summary(self):
        s = self.retry_summary()
        if not s:
            print("No retry data.")
            return

        print("=" * 55)
        print("RETRY SUMMARY")
        print("=" * 55)
        print(f"Total requests:        {s['total_requests']}")
        print(f"Retried:               {s['retried']} ({s['retry_rate']:.1%})")
        print(f"Failed after retries:  {s['failed_after_retries']}")
        print(f"Avg attempts/request:  {s['avg_attempts']:.2f}")
        print(f"Extra cost (retries):  ${s['extra_cost_from_retries']:.4f}")
        print("=" * 55)


# Retry storm simulation
random.seed(42)
retry_tracker = RetryTracker(max_retries=3, base_delay_ms=100)

def flaky_llm_call():
    """Simulates a call that fails 40% of the time."""
    if random.random() < 0.4:
        raise TimeoutError("Request timed out")
    return "Response OK"

for i in range(20):
    result, record = retry_tracker.execute_with_retry(
        func=flaky_llm_call,
        request_id=f"req_{i}",
        endpoint="/chat",
        model="gpt-4o-mini",
        cost_per_attempt=0.001,
    )
    if record.total_attempts > 1:
        print(f"  req_{i}: {record.total_attempts} attempts, "
              f"{'OK' if record.successful else 'FAILED'}, "
              f"${record.total_cost_usd:.4f}, "
              f"{record.total_latency_ms:.0f}ms")

print()
retry_tracker.print_summary()

Comparison: Traditional vs AI Error Handling

Why separating matters

Scenario                     Combined      Correct diagnosis
                             error rate     
─────────────────────────────────────────────────────────────────
A: 3% timeouts, 0% functional   3%         "Infra problem — retry"
B: 0% technical, 3% hallucinations  3%     "Quality problem — review prompts"
C: 1% rate limits, 5% format    6%         "Two different problems"
─────────────────────────────────────────────────────────────────

The three scenarios have similar error rates, but the actions are completely different. If you mix them into a single metric, you lose the information you need to act.

Connection with the Project

How this connects with Metrics Instrumentation

In the module's project (capsule 08) you'll instrument complete error tracking:

ERROR INSTRUMENTATION — Checklist
=====================================
☐ Automatic classification: technical vs functional
☐ Error rate per category (timeout, rate_limit, hallucination, etc.)
☐ Error rate per endpoint
☐ Wasted cost tracking (cost of failed functional responses)
☐ Retry tracking with impact on cost and latency
☐ Configurable functional detection heuristics
☐ Report of top error categories

The ErrorTracker combines with the LatencyTracker (capsule 02) and the CostCalculator (capsule 03). Together they give you the complete picture: "this request took 8.3s (3 retries due to timeout), cost $0.009 (3x the normal cost), and in the end the response was a hallucination (functional error detected)". Three problems, three metrics, three different actions.

Troubleshooting

"My error rate is low but users complain a lot"

You're probably measuring only technical errors. A system with 0.5% timeouts but 8% hallucinations has a "0.5% error rate" according to traditional monitoring. Implement functional error detection — the user's complaint IS a functional error you're not counting.

"Retries fix my timeouts but the cost doubled"

Each retry is an additional request with a cost. If 5% of your requests need 3 retries, you're paying 15% extra in tokens. Implement the RetryTracker to quantify the cost of retries. Consider circuit breakers: after N consecutive failures, wait before trying, instead of retrying immediately.

"How do I detect hallucinations automatically?"

The heuristics in this capsule are a first level: uncertainty signals, incorrect format, anomalous length. For more sophisticated detection, capsule 05 (Quality) and module 6 cover LLM-as-judge — using a second LLM to evaluate the output of the first. It's not perfect, but it's significantly better than detecting nothing.

"Should I retry on functional errors?"

It depends on the type. A retry can resolve a hallucination (the model is non-deterministic — it may respond well on the second attempt). But a retry does NOT resolve a format problem if the prompt doesn't explicitly ask for a format. General rule: retry on functional errors only if the cause is probabilistic (hallucination), not if it's systematic (poorly designed prompt).

Exercises

Exercise 1: Classify errors from a log (Easy)

Given this log of 10 requests, classify each one as technical, functional, or success:

requests_log = [
    {"status": 200, "output": "The price is $29/month.", "expected_format": None},
    {"status": 429, "output": None, "expected_format": None},
    {"status": 200, "output": "Creo que el precio podría rondar los $15.", "expected_format": None},
    {"status": 500, "output": None, "expected_format": None},
    {"status": 200, "output": '{"price": 29, "currency": "USD"}', "expected_format": "json"},
    {"status": 200, "output": "", "expected_format": None},
    {"status": 200, "output": "I think it could be... I'm not sure but maybe $29.", "expected_format": None},
    {"status": 408, "output": None, "expected_format": None},
    {"status": 200, "output": "The premium plan costs $29 per month.", "expected_format": None},
    {"status": 200, "output": "¡Claro! El plan premium cuesta $29.", "expected_format": None},
]

def classify_request(req: dict) -> dict:
    status = req["status"]
    output = req.get("output", "")
    expected_fmt = req.get("expected_format")

    if status >= 400:
        category_map = {
            429: "rate_limit",
            500: "api_error",
            408: "timeout",
        }
        return {
            "type": "technical",
            "category": category_map.get(status, "unknown_technical"),
            "detail": f"HTTP {status}",
        }

    if not output or len(output.strip()) < 5:
        return {
            "type": "functional",
            "category": "empty_response",
            "detail": "Response empty or too short",
        }

    checks = classify_functional_errors(
        output,
        expected_format=expected_fmt,
        expected_lang="en",
    )

    failures = [c for c in checks if not c.passed]
    if failures:
        worst = max(failures, key=lambda c: {"low": 0, "medium": 1, "high": 2}[c.severity])
        return {
            "type": "functional",
            "category": worst.check_name,
            "detail": worst.detail,
        }

    return {"type": "success", "category": "ok", "detail": "All checks passed"}


requests_log = [
    {"status": 200, "output": "The price is $29/month.", "expected_format": None},
    {"status": 429, "output": None, "expected_format": None},
    {"status": 200, "output": "Creo que el precio podría rondar los $15.", "expected_format": None},
    {"status": 500, "output": None, "expected_format": None},
    {"status": 200, "output": '{"price": 29, "currency": "USD"}', "expected_format": "json"},
    {"status": 200, "output": "", "expected_format": None},
    {"status": 200, "output": "I think it could be... I'm not sure but maybe $29.",
     "expected_format": None},
    {"status": 408, "output": None, "expected_format": None},
    {"status": 200, "output": "The premium plan costs $29 per month.",
     "expected_format": None},
    {"status": 200, "output": "¡Claro! El plan premium cuesta $29.", "expected_format": None},
]

tech_count = 0
func_count = 0
success_count = 0

print(f"{'#':>3} {'Type':<12} {'Category':<25} {'Detail':<40}")
print("-" * 85)

for i, req in enumerate(requests_log, 1):
    result = classify_request(req)
    print(f"{i:>3} {result['type']:<12} {result['category']:<25} {result['detail']:<40}")

    if result["type"] == "technical":
        tech_count += 1
    elif result["type"] == "functional":
        func_count += 1
    else:
        success_count += 1

total = len(requests_log)
print(f"\nSummary: {success_count} success, {tech_count} technical ({tech_count/total:.0%}), "
      f"{func_count} functional ({func_count/total:.0%})")

Exercise 2: Implement a circuit breaker for technical errors (Medium)

Create a CircuitBreaker class that stops sending requests to a provider if more than N consecutive errors accumulate. The breaker has three states: CLOSED (normal), OPEN (blocking requests), and HALF_OPEN (testing whether the service recovered).

import time
from enum import Enum


class BreakerState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"


class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout_s: float = 30.0,
        half_open_max_calls: int = 2,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout_s = recovery_timeout_s
        self.half_open_max_calls = half_open_max_calls

        self.state = BreakerState.CLOSED
        self.consecutive_failures = 0
        self.last_failure_time: float = 0
        self.half_open_calls = 0
        self.half_open_successes = 0

        self.stats = {
            "total_calls": 0,
            "blocked_calls": 0,
            "successful_calls": 0,
            "failed_calls": 0,
            "state_changes": [],
        }

    def _change_state(self, new_state: BreakerState):
        old = self.state
        self.state = new_state
        self.stats["state_changes"].append({
            "from": old.value,
            "to": new_state.value,
            "time": time.time(),
        })

    def can_execute(self) -> bool:
        if self.state == BreakerState.CLOSED:
            return True

        if self.state == BreakerState.OPEN:
            elapsed = time.time() - self.last_failure_time
            if elapsed >= self.recovery_timeout_s:
                self._change_state(BreakerState.HALF_OPEN)
                self.half_open_calls = 0
                self.half_open_successes = 0
                return True
            return False

        if self.state == BreakerState.HALF_OPEN:
            return self.half_open_calls < self.half_open_max_calls

        return False

    def record_success(self):
        self.stats["total_calls"] += 1
        self.stats["successful_calls"] += 1
        self.consecutive_failures = 0

        if self.state == BreakerState.HALF_OPEN:
            self.half_open_calls += 1
            self.half_open_successes += 1
            if self.half_open_successes >= self.half_open_max_calls:
                self._change_state(BreakerState.CLOSED)

    def record_failure(self):
        self.stats["total_calls"] += 1
        self.stats["failed_calls"] += 1
        self.consecutive_failures += 1
        self.last_failure_time = time.time()

        if self.state == BreakerState.HALF_OPEN:
            self._change_state(BreakerState.OPEN)
        elif self.consecutive_failures >= self.failure_threshold:
            self._change_state(BreakerState.OPEN)

    def execute(self, func):
        if not self.can_execute():
            self.stats["blocked_calls"] += 1
            raise RuntimeError(
                f"Circuit breaker OPEN — blocked call. "
                f"State: {self.state.value}, "
                f"Consecutive failures: {self.consecutive_failures}"
            )

        try:
            result = func()
            self.record_success()
            return result
        except Exception as e:
            self.record_failure()
            raise


import random
random.seed(42)

cb = CircuitBreaker(failure_threshold=3, recovery_timeout_s=0.5)

def unreliable_api():
    if random.random() < 0.5:
        raise TimeoutError("API timeout")
    return "OK"

for i in range(20):
    try:
        result = cb.execute(unreliable_api)
        print(f"  Call {i+1}: SUCCESS ({cb.state.value})")
    except RuntimeError as e:
        print(f"  Call {i+1}: BLOCKED ({cb.state.value})")
    except TimeoutError:
        print(f"  Call {i+1}: FAILED  ({cb.state.value})")

    if cb.state == BreakerState.OPEN:
        time.sleep(0.6)

print(f"\nStats:")
print(f"  Total: {cb.stats['total_calls']}")
print(f"  Blocked: {cb.stats['blocked_calls']}")
print(f"  Success: {cb.stats['successful_calls']}")
print(f"  Failed: {cb.stats['failed_calls']}")
print(f"  State changes: {len(cb.stats['state_changes'])}")

Exercise 3: Calculate the cost of functional errors (Medium)

Given an ErrorTracker with data, calculate the "wasted cost" — the total cost of requests that were functionally incorrect. Generate a report showing how much money was spent on responses that were useless, broken down by type of functional error.

from collections import defaultdict


def wasted_cost_report(tracker: ErrorTracker) -> dict:
    total_wasted = tracker.wasted_cost()

    by_category = defaultdict(lambda: {"count": 0, "cost": 0.0})
    for e in tracker.functional_errors:
        cat = e.category.value
        by_category[cat]["count"] += 1
        by_category[cat]["cost"] += e.cost_usd

    by_endpoint = defaultdict(lambda: {"count": 0, "cost": 0.0})
    for e in tracker.functional_errors:
        by_endpoint[e.endpoint]["count"] += 1
        by_endpoint[e.endpoint]["cost"] += e.cost_usd

    return {
        "total_wasted_usd": total_wasted,
        "total_functional_errors": len(tracker.functional_errors),
        "by_category": dict(by_category),
        "by_endpoint": dict(by_endpoint),
    }


report = wasted_cost_report(tracker)

print("WASTED COST REPORT")
print("=" * 55)
print(f"Total wasted:          ${report['total_wasted_usd']:.4f}")
print(f"Functional errors:     {report['total_functional_errors']}")
print()

print("By error category:")
for cat, data in sorted(report["by_category"].items(),
                        key=lambda x: x[1]["cost"], reverse=True):
    print(f"  {cat:<25} {data['count']:>4} errors, "
          f"${data['cost']:.4f} wasted")

print()
print("By endpoint:")
for ep, data in sorted(report["by_endpoint"].items(),
                       key=lambda x: x[1]["cost"], reverse=True):
    print(f"  {ep:<20} {data['count']:>4} errors, "
          f"${data['cost']:.4f} wasted")
print("=" * 55)

if report["total_wasted_usd"] > 0:
    daily_projection = report["total_wasted_usd"] * (500 / tracker.total_requests) * 20
    print(f"\nProjection at 10K req/day:")
    print(f"  Daily waste: ${daily_projection:.2f}")
    print(f"  Monthly waste: ${daily_projection * 30:.2f}")

Exercise 4: Complete error handling pipeline (Hard)

Create an instrumented_llm_call function that combines everything: it attempts the LLM call, classifies technical errors if it fails, runs functional checks if it succeeds, records everything in an ErrorTracker, and returns a complete report of the request.

import time
import uuid
from typing import Optional


def instrumented_llm_call(
    prompt: str,
    model: str = "gpt-4o-mini",
    endpoint: str = "/chat",
    expected_format: str = None,
    expected_lang: str = "en",
    error_tracker: ErrorTracker = None,
    max_retries: int = 2,
) -> dict:
    """Instrumented call with complete error tracking."""
    request_id = str(uuid.uuid4())[:8]
    start = time.perf_counter()
    attempts = 0
    last_error = None

    for attempt in range(max_retries + 1):
        attempts += 1
        try:
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=300,
                timeout=30,
            )

            output = response.choices[0].message.content
            usage = response.usage
            latency = (time.perf_counter() - start) * 1000

            PRICING = {
                "gpt-4o": {"input": 2.50e-6, "output": 10.00e-6},
                "gpt-4o-mini": {"input": 0.15e-6, "output": 0.60e-6},
            }
            rates = PRICING.get(model, {"input": 1e-6, "output": 3e-6})
            cost = (usage.prompt_tokens * rates["input"]
                    + usage.completion_tokens * rates["output"])

            checks = classify_functional_errors(
                output,
                expected_format=expected_format,
                expected_lang=expected_lang,
            )
            failures = [c for c in checks if not c.passed]

            if error_tracker:
                error_tracker.record_request(
                    endpoint=endpoint,
                    model=model,
                    functional_checks=checks,
                    output=output,
                    cost_usd=cost,
                    prompt=prompt,
                )

            return {
                "request_id": request_id,
                "status": "success" if not failures else "functional_error",
                "output": output,
                "model": model,
                "tokens": {
                    "prompt": usage.prompt_tokens,
                    "completion": usage.completion_tokens,
                    "total": usage.total_tokens,
                },
                "cost_usd": cost,
                "latency_ms": latency,
                "attempts": attempts,
                "functional_checks": [
                    {"name": c.check_name, "passed": c.passed, "detail": c.detail}
                    for c in checks
                ],
            }

        except Exception as e:
            last_error = e
            if attempt < max_retries:
                time.sleep(0.5 * (2 ** attempt))
                continue

            latency = (time.perf_counter() - start) * 1000
            category = classify_technical_error(e)

            if error_tracker:
                error_record = ErrorRecord(
                    timestamp=datetime.now().isoformat(),
                    endpoint=endpoint,
                    category=category,
                    is_technical=True,
                    is_functional=False,
                    message=str(e)[:200],
                    model=model,
                    prompt_preview=prompt[:100],
                    retry_count=attempts - 1,
                )
                error_tracker.record_request(
                    endpoint=endpoint,
                    model=model,
                    technical_error=error_record,
                )

            return {
                "request_id": request_id,
                "status": "technical_error",
                "error": str(last_error)[:200],
                "category": category.value,
                "latency_ms": latency,
                "attempts": attempts,
            }


et = ErrorTracker()

result = instrumented_llm_call(
    prompt="What error metrics should I track in an AI system?",
    model="gpt-4o-mini",
    endpoint="/chat",
    error_tracker=et,
)

print(f"Status:    {result['status']}")
print(f"Attempts:  {result['attempts']}")
if result["status"] != "technical_error":
    print(f"Output:    {result['output'][:80]}...")
    print(f"Cost:      ${result['cost_usd']:.6f}")
    print(f"Latency:   {result['latency_ms']:.0f}ms")
    print(f"Checks:")
    for c in result["functional_checks"]:
        status = "PASS" if c["passed"] else "FAIL"
        print(f"  [{status}] {c['name']}: {c['detail']}")
else:
    print(f"Error:     {result['error']}")
    print(f"Category:  {result['category']}")

Summary

• Errors in AI aren't just HTTP 5xx. A timeout (technical) and a hallucination (functional) are different problems with different solutions. Measure them separately.
• Technical errors are infrastructure failures: timeouts, rate limits, API down, auth errors. They're detected with status codes and exceptions. They're fixed with retries, circuit breakers, and fallbacks.
• Functional errors are incorrect outputs with status 200: hallucinations, incorrect format, off-topic, wrong language. They're detected with heuristics over the output. They're fixed with better prompts, context, or models.
• Measure both separately. "3% error rate" isn't actionable. "0.5% technical + 2.5% functional" is: the technical is fixed with infra, the functional with prompts.
• Retries have a cost. Each retry consumes tokens. A system with 5% retries pays 5% extra in tokens, plus the additional latency. Track the cost of retries explicitly.
• Retry storms are real. Without a circuit breaker, a down provider generates thousands of retries that worsen the rate limit, cost tokens, and resolve nothing.
• The "wasted cost" is money you paid for functional errors. Tokens billed for hallucinations, incorrect format, empty responses. Quantifying that waste gives you optimization priorities.
• These metrics feed SLOs (capsule 06: "5% weekly error budget, broken down 1% technical + 4% functional"), dashboards (module 4), alerts (module 5: "alert if functional error rate exceeds 5%"), and debugging (module 7).

Additional Resources

1. OpenAI Error Codes — Official reference for OpenAI API errors
2. Anthropic Error Handling — Error handling in the Claude API
3. Microsoft — Circuit Breaker Pattern — The circuit breaker pattern explained by Microsoft
4. OpenAI Rate Limits — Understanding and handling rate limits
5. Google SRE Book — Handling Overload — Strategies for handling overload in services
6. Hamel Husain — Your AI Product Needs Evals — Why detecting functional errors is critical
7. tenacity — Python Retry Library — Robust library for implementing retries in Python
8. OpenTelemetry — Error Handling — How to instrument errors with the industry standard