Observability, Cost, and Latency Monitoring in Production

Why "we tested and released it" isn't enough

You measured your golden set on deploy day. nDCG@10 = 0.84. You shipped.

Six weeks later, a large customer migrated to a new documentation system. Your retrieval index wasn't updated. Quality silently dropped to 0.71. Users complained, but the issue took three days to diagnose because nobody was monitoring.

The core insight: A golden-set score is a point-in-time measurement. Production is a moving target. Documents change, user behavior shifts, upstream dependencies break. Monitoring is how you catch these changes before they become outages.

This lesson closes the course by connecting evaluation (lab metrics) to monitoring (production signals), and showing you how to build an operating loop that keeps your system healthy.

Four signals worth alerting on

Signal 1: Latency (p50, p95, p99)

Why it matters: Users abandon slow systems. A 500ms response becomes a 2-second response (2 extra hops in the retrieval index, or a model timeout?), and your NPS crashes.

What to measure:

• p50 (median): Day-to-day baseline
• p95: SLA target ("99% of requests under 800ms")
• p99: Tail degradation (catch rare but awful cases)

Typical thresholds:

LATENCY_THRESHOLDS = {
    "p50_ms": 400,     # alert if p50 > 400
    "p95_ms": 800,     # alert if p95 > 800
    "p99_ms": 1500,    # alert if p99 > 1500
}

def check_latency_alert(latencies_ms: List[float]) -> dict:
    """
    Check if latency has regressed.
    """
    import numpy as np

    p50 = np.percentile(latencies_ms, 50)
    p95 = np.percentile(latencies_ms, 95)
    p99 = np.percentile(latencies_ms, 99)

    alerts = []
    if p50 > LATENCY_THRESHOLDS["p50_ms"]:
        alerts.append(f"p50 latency {p50:.0f}ms > {LATENCY_THRESHOLDS['p50_ms']}ms")
    if p95 > LATENCY_THRESHOLDS["p95_ms"]:
        alerts.append(f"p95 latency {p95:.0f}ms > {LATENCY_THRESHOLDS['p95_ms']}ms")
    if p99 > LATENCY_THRESHOLDS["p99_ms"]:
        alerts.append(f"p99 latency {p99:.0f}ms > {LATENCY_THRESHOLDS['p99_ms']}ms")

    return {"alerts": alerts, "p50": p50, "p95": p95, "p99": p99}

Common causes of latency regression:

• Vector index larger than memory; slow disk I/O
• Embedding model inference slower (stale GPU, cold start)
• More chunks retrieved (chunking strategy changed? index grew?)
• Network hops added (cross-region latency)

Signal 2: Cost (per request and per day)

Why it matters: LLM APIs charge per token. A prompt change that adds 100 tokens per call × 10,000 daily requests = +1M tokens/day = thousands of dollars/month.

What to measure:

• Cost per request: Sum input + output tokens × model pricing
• Cost per day: Aggregate cost across all requests
• Cost per feature: Which features or user cohorts are expensive?

MODEL_COSTS = {
    "claude-opus-4-1-20250805": {
        "input_tokens_per_1m": 3,  # $3 per 1M input tokens
        "output_tokens_per_1m": 15  # $15 per 1M output tokens
    },
    "claude-3-5-sonnet-20241022": {
        "input_tokens_per_1m": 3,
        "output_tokens_per_1m": 15
    }
}

def compute_request_cost(model_id: str, input_tokens: int, output_tokens: int) -> float:
    """
    Compute cost of a single request.
    """
    prices = MODEL_COSTS[model_id]
    input_cost = input_tokens * prices["input_tokens_per_1m"] / 1_000_000
    output_cost = output_tokens * prices["output_tokens_per_1m"] / 1_000_000
    return input_cost + output_cost

def aggregate_daily_cost(requests: List[dict]) -> dict:
    """
    Args:
        requests: [{"model": "claude-...", "input_tokens": 200, "output_tokens": 50}, ...]

    Returns:
        dict: {"total_cost": 12.45, "request_count": 1000, "avg_cost_per_request": 0.01245}
    """
    total_cost = sum(
        compute_request_cost(req["model"], req["input_tokens"], req["output_tokens"])
        for req in requests
    )

    avg_cost = total_cost / len(requests) if requests else 0

    return {
        "total_cost": total_cost,
        "request_count": len(requests),
        "avg_cost_per_request": avg_cost
    }

COST_THRESHOLD_USD_PER_DAY = 500  # alert if daily cost > $500

daily_cost = aggregate_daily_cost(requests_from_today)
if daily_cost["total_cost"] > COST_THRESHOLD_USD_PER_DAY:
    print(f"ALERT: Daily cost ${daily_cost['total_cost']:.2f} > ${COST_THRESHOLD_USD_PER_DAY}")

Common cost surprises:

• Chunking changed (fewer, larger chunks) → more tokens per request
• System prompt expanded → fixed overhead per request
• Switched to a more expensive model variant

Signal 3: Quality drift (continuous sampling)

Why it matters: Documents update, upstream data sources change, user patterns shift. Your model's answer quality drifts slowly, and if you're not sampling, you won't notice until complaints pile up.

Strategy: Continuously sample live traffic, have an LLM judge evaluate a sample (5-10% of requests daily), and alert if the mean judge score drops.

import random
from datetime import datetime, timedelta

QUALITY_EVAL_SAMPLE_RATE = 0.05  # sample 5% of traffic for continuous eval
QUALITY_THRESHOLD_SCORE = 0.82   # alert if rolling mean < 0.82

def should_sample_for_quality_eval() -> bool:
    """
    Decide whether to evaluate this request's answer.
    """
    return random.random() < QUALITY_EVAL_SAMPLE_RATE

def evaluate_live_request(question: str, answer: str, context: str) -> float:
    """
    Run LLM judge on a sampled request.
    Returns: quality score (0-1)
    """
    # Use your calibrated judge (see lesson 21)
    from judge import judge_answer

    result = judge_answer(question, answer, context)
    return result["overall_score"] / 10.0  # normalize to 0-1

def check_quality_drift(
    scores_last_24_hours: List[float],
    threshold: float = QUALITY_THRESHOLD_SCORE
) -> dict:
    """
    Check for quality regression over time.
    """
    import numpy as np

    if not scores_last_24_hours:
        return {"status": "no_data", "sample_count": 0}

    mean_score = np.mean(scores_last_24_hours)
    std_dev = np.std(scores_last_24_hours)

    alert = mean_score < threshold

    return {
        "status": "alert" if alert else "ok",
        "mean_score": mean_score,
        "std_dev": std_dev,
        "sample_count": len(scores_last_24_hours),
        "threshold": threshold
    }

# In your request handler:
def handle_rag_request(question: str) -> dict:
    # ... retrieve and generate ...
    answer = generate(question, retrieved_context)

    # Opportunistically sample for evaluation
    if should_sample_for_quality_eval():
        quality_score = evaluate_live_request(question, answer, retrieved_context)
        # Store score for daily roll-up
        store_quality_score(quality_score)

    return answer

Why continuous sampling (not once-per-day):

• Bursty evaluation runs (daily batch job) can miss issues if they're brief
• Continuous sampling spreads the cost and catches drift faster
• Natural stratification: different types of requests get evaluated throughout the day

Signal 4: Error rate (timeouts, empty retrievals, crashes)

Why it matters: A system that returns an error half the time is broken, but if you're only monitoring the "happy path," you won't know.

Categories of errors:

from enum import Enum

class RAGErrorType(Enum):
    EMPTY_RETRIEVAL = "no_relevant_chunks_found"
    RETRIEVAL_TIMEOUT = "vector_search_exceeded_timeout"
    MODEL_TIMEOUT = "llm_inference_timeout"
    MALFORMED_RETRIEVAL = "chunks_missing_metadata"
    CRASH = "unhandled_exception"

ERROR_RATE_THRESHOLDS = {
    RAGErrorType.EMPTY_RETRIEVAL: 0.05,      # alert if > 5% of requests
    RAGErrorType.RETRIEVAL_TIMEOUT: 0.01,    # alert if > 1% of requests
    RAGErrorType.MODEL_TIMEOUT: 0.02,        # alert if > 2% of requests
    RAGErrorType.CRASH: 0.001,               # alert if > 0.1% of requests
}

def track_error(error_type: RAGErrorType):
    """
    Log an error occurrence.
    """
    # Increment counter for this error type
    pass

def check_error_rates(
    error_counts: dict,  # {"empty_retrieval": 50, "retrieval_timeout": 10, ...}
    total_requests: int
) -> dict:
    """
    Check if error rates exceed thresholds.
    """
    alerts = []

    for error_type, threshold in ERROR_RATE_THRESHOLDS.items():
        count = error_counts.get(error_type.value, 0)
        rate = count / total_requests if total_requests > 0 else 0

        if rate > threshold:
            alerts.append(f"{error_type.value}: {rate:.1%} > {threshold:.1%}")

    return {"alerts": alerts, "error_counts": error_counts, "total_requests": total_requests}

Common error sources:

• Vector index becomes corrupted or out of sync
• Upstream document sources are down
• LLM API rate limits or service degradation
• Network instability (intermittent failures)

Tracing and observability: debugging in production

When an alert fires, you need to debug. This requires logging the right data at request time.

import json
import logging
from datetime import datetime
from typing import List
import uuid

# Structured logging
logger = logging.getLogger("rag_system")

@dataclass
class RAGTrace:
    """
    Full trace of a single request, for post-hoc analysis.
    """
    request_id: str
    timestamp: str
    question: str
    retrieved_chunk_ids: List[str]  # crucial for debugging retrieval
    retrieved_chunk_scores: List[float]  # relevance scores
    prompt_tokens: int
    completion_tokens: int
    model_id: str
    generated_answer: str
    answer_tokens: int
    latency_ms: float
    quality_score: float = None  # filled in if sampled for evaluation

def log_rag_request(trace: RAGTrace):
    """
    Log a structured trace of the request.
    """
    logger.info(json.dumps({
        "request_id": trace.request_id,
        "timestamp": trace.timestamp,
        "question": trace.question,
        "retrieved_chunks": trace.retrieved_chunk_ids,
        "chunk_scores": trace.retrieved_chunk_scores,
        "prompt_tokens": trace.prompt_tokens,
        "answer_tokens": trace.answer_tokens,
        "total_tokens": trace.prompt_tokens + trace.answer_tokens,
        "model": trace.model_id,
        "latency_ms": trace.latency_ms,
        "quality_score": trace.quality_score
    }))

def retrieve_and_generate_with_tracing(question: str) -> tuple:
    """
    Full RAG pipeline with comprehensive logging.
    """
    request_id = str(uuid.uuid4())
    trace = RAGTrace(
        request_id=request_id,
        timestamp=datetime.utcnow().isoformat()
        question=question,
        retrieved_chunk_ids=[],
        retrieved_chunk_scores=[],
        prompt_tokens=0,
        completion_tokens=0,
        model_id="claude-opus-4-1-20250805",
        generated_answer="",
        answer_tokens=0,
        latency_ms=0
    )

    import time
    start_time = time.time()

    try:
        # RETRIEVAL PHASE
        retrieved = retrieve(question, top_k=10)
        trace.retrieved_chunk_ids = [c["id"] for c in retrieved]
        trace.retrieved_chunk_scores = [c["score"] for c in retrieved]

        # PROMPT PHASE (count tokens)
        system_prompt = build_system_prompt()
        user_prompt = f"Question: {question}\n\nContext: {format_context(retrieved)}"
        trace.prompt_tokens = count_tokens(system_prompt + user_prompt)

        # GENERATION PHASE
        from anthropic import Anthropic
        client = Anthropic()
        response = client.messages.create(
            model=trace.model_id,
            max_tokens=500,
            system=system_prompt,
            messages=[{"role": "user", "content": user_prompt}]
        )

        trace.generated_answer = response.content[0].text
        trace.completion_tokens = response.usage.output_tokens

        # SAMPLE FOR CONTINUOUS EVAL
        if random.random() < 0.05:
            trace.quality_score = evaluate_answer(question, trace.generated_answer, retrieved)

    finally:
        trace.latency_ms = (time.time() - start_time) * 1000
        log_rag_request(trace)

    return trace.generated_answer, trace

What logs to keep:

• ✓ Retrieved chunk IDs (can re-fetch and debug why they were retrieved)
• ✓ Prompt structure (see if the system prompt changed)
• ✓ Token counts (track cost)
• ✓ Latency breakdown (retrieve time vs. generate time)
• ✗ Full context (PII risk; store hash or sample only)
• ✗ Every intermediate step (too verbose; summarize)

Using logs to debug: When quality score drops, query logs:

SELECT request_id, question, retrieved_chunk_ids, quality_score
FROM rag_logs
WHERE timestamp > NOW() - INTERVAL 24 HOURS
  AND quality_score < 0.7  -- poor quality
ORDER BY quality_score ASC
LIMIT 20;

Then fetch full chunk content for those IDs and see: "Why was this chunk retrieved for this query?"

Synthesis: the complete evaluation loop

The six lessons in this course have built an evaluation and monitoring system. Here's how the pieces fit:

┌─────────────────────────────────────────────────────────────────┐
│ LAB EVALUATION (lessons 19-22)                                  │
│ - Golden set with labeled retrieval + answer quality            │
│ - Retrieval metrics: Recall@k, Precision@k, nDCG                │
│ - Generation metrics: faithfulness, relevance                   │
│ - LLM judge with calibration against human labels               │
│ - Human evaluation with inter-rater agreement                   │
└─────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────┐
│ OFFLINE REGRESSION TESTING (lesson 23)                          │
│ - Golden set runs on every code/prompt/model change             │
│ - Fail deploy if metrics drop > 2%                              │
│ - Catch obvious regressions before ship                         │
└─────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────┐
│ ONLINE A/B TESTING (lesson 23)                                  │
│ - Split traffic: 90% control, 10% treatment                     │
│ - Measure quality + business metrics over 1-2 weeks             │
│ - Achieve statistical significance before shipping              │
│ - Guard against novelty bias, multiple comparisons              │
└─────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────┐
│ PRODUCTION DEPLOYMENT                                           │
│ - Ship change to 100% of users                                  │
│ - But evaluation doesn't stop...                                │
└─────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────┐
│ PRODUCTION MONITORING (lesson 24 — this lesson)                 │
│ - Track 4 signals: latency, cost, quality drift, error rate     │
│ - Continuous sampling for quality (5% of traffic/day)           │
│ - Alert if any signal breaches threshold                        │
│ - Trace logs capture chunk IDs, prompts for debugging           │
└─────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────┐
│ FEEDBACK LOOP                                                   │
│ - If monitoring detects regression:                             │
│   → Debug using trace logs (which chunks? which queries?)       │
│   → Add failing cases to golden set                             │
│   → Return to LAB EVALUATION                                    │
│ - If no regressions detected:                                   │
│   → Establish new baselines                                     │
│   → Grow golden set with hardest real-world cases               │
│   → Prepare next improvement                                    │
└─────────────────────────────────────────────────────────────────┘

Key insight: Evaluation is not a one-time gate before shipping. It's a continuous loop:

1. Lab evaluation ensures the change is real and safe before users see it
2. A/B testing confirms the lab results translate to the real world
3. Monitoring catches the inevitable slow degradation over time
4. Feedback loop uses production issues to improve both the system and the evaluation criteria

Teams that skip any part of this loop suffer:

• No lab evaluation → ship anything, users suffer
• No A/B test → miss distribution shift, false positives
• No monitoring → slow drift goes undetected for weeks
• No feedback loop → repeat the same mistakes

Putting it into practice: a three-month improvement cycle

Month 1: Build the golden set

• Collect 100-200 representative queries
• Have humans label retrieval (expected chunks) and answers (quality)
• Measure inter-rater agreement (target: kappa > 0.7)
• Establish baseline metrics (nDCG, faithfulness, etc.)

Month 2: Set up offline regression testing

• Add pytest-style tests to your CI/CD
• Golden set runs on every commit
• Establish 2-5% regression thresholds
• Measure the baseline latency and cost

Month 3: Deploy monitoring

• Log structured traces (chunk IDs, prompts, tokens)
• Set up alerts for latency, cost, quality drift, errors
• Sample 5% of live traffic for continuous evaluation
• Run A/B tests for any significant changes

Ongoing:

• Monthly: grow golden set with 20-30 new hard cases from production
• Monthly: recalibrate LLM judge against fresh human labels
• Quarterly: review monitoring thresholds (do they still make sense?)

Common mistake

Thinking "we shipped it; evaluation is done." Production is where the real test happens. A system can be perfect in the lab and terrible in production when:

• Real users ask questions the golden set didn't cover
• Upstream data sources change
• User volumes spike, causing latency
• A competitor launches, changing user expectations

Set up monitoring from day one, not "after we see problems." The cost of monitoring is tiny compared to the cost of an undetected outage.

Also, don't let alerts become noise. If you have 100 alerts firing and developers ignore them, they're worse than useless. Start with one to three key metrics per system, make sure alerts are meaningful, and add new metrics only if they actionable (i.e., "p95 latency > 800ms" is actionable; "average token count > 150" is just data).

Closing: what you've learned

This course has taken you from vibes-based evaluation to a complete, production-grade evaluation system:

• Lesson 19: Understand four retrieval metrics and when each one matters
• Lesson 20: Score generation quality beyond yes/no hallucination; claim-level faithfulness
• Lesson 21: Design LLM judges, avoid bias, calibrate against humans
• Lesson 22: Build human evaluation rubrics, measure agreement, use humans to calibrate judges
• Lesson 23: A/B test RAG changes online, regression test offline, avoid statistical pitfalls
• Lesson 24 (this one): Monitor production, catch drift, close the loop back to evaluation

You now know how to build a RAG system that you can trust, improve systematically, and operate reliably in production. That's the difference between a demo and a real product.

The next step is to apply this to your own system. Start with the golden set, run the regression tests, and set up monitoring. The investment pays off the first time you catch a degradation that would have otherwise gone unnoticed for weeks.