Monitoring & Observability

You cannot improve what you do not measure. A RAG system in production degrades silently: retrieval quality drifts as the document corpus changes, latency creeps up as the index grows, and empty result rates spike when users ask questions your documents do not cover. Without monitoring, you find out about these problems from user complaints. With monitoring, you see the trend forming and fix it before anyone notices.

What you'll learn

• Track retrieval quality metrics: result count, empty result rate, top similarity scores, and latency percentiles
• Log every query with its top-3 results for offline evaluation and debugging
• Set up alerts on sustained deviations from baseline, not single anomalies

What to monitor

Metric	What it tells you	Healthy range	Alert when
P50/P95/P99 latency	How fast retrieval is	P95 < 100ms	P95 > 200ms sustained
Empty result rate	How often users get nothing back	< 5%	> 10% sustained for 1 hour
Top-1 similarity score	Whether the best result is actually relevant	> 0.75	P50 < 0.60 sustained
Result count distribution	Whether queries return enough context	5-10 results median	P50 < 3
Embedding API error rate	Provider health	< 0.1%	> 1%
Cache hit rate	Caching effectiveness	> 60%	< 40%
Index size and growth rate	Storage scaling needs	Steady growth	Sudden spikes or drops

Build it

Structured logging for every query

Effective observability begins with structured data. Wrap your search execution in a logging layer that records latency, result counts, and similarity scores as JSON-parsable fields.

import time
import structlog

logger = structlog.get_logger()

def monitored_search(query, top_k=10):
    start = time.time()
    embedding = generate_embedding(query)
    results = vector_search(embedding, top_k=top_k)
    elapsed_ms = (time.time() - start) * 1000

    logger.info("rag_search",
        query_preview=query[:100],
        result_count=len(results),
        empty_result=(len(results) == 0),
        top_similarity=round(results[0]["similarity"], 3) if results else 0,
        top3_similarities=[round(r["similarity"], 3) for r in results[:3]],
        latency_ms=round(elapsed_ms, 1),
        cache_hit=False,
    )

    if len(results) == 0:
        logger.warning("rag_empty_result", query=query[:200])
    elif results[0]["similarity"] < 0.70:
        logger.warning("rag_low_similarity", 
            query=query[:100], 
            similarity=results[0]["similarity"])

    return results

Baseline calculation

Raw metrics mean little without context. Calculate historical baselines by querying recent logs to establish normal operational thresholds for your system.

def compute_baselines(days=7):
    """Compute baseline metrics from the last N days of logs."""
    # In production, query your log aggregation system (Datadog, Grafana, etc.)
    # This is a simplified example using a local store
    cursor.execute("""
        SELECT 
            PERCENTILE_CONT(0.50) WITHIN GROUP (ORDER BY latency_ms) as p50_latency,
            PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms) as p95_latency,
            AVG(CASE WHEN empty_result THEN 1.0 ELSE 0.0 END) as empty_rate,
            AVG(top_similarity) as avg_similarity
        FROM rag_metrics
        WHERE timestamp > NOW() - INTERVAL '%s days'
    """, (days,))
    return cursor.fetchone()

Alerting rules

Configure programmatic alerts based on deviations from the established baselines. This logic identifies concerning trends, like spiking latency or dropping similarity scores, before users complain.

def check_alerts(current_metrics, baselines):
    alerts = []

    if current_metrics["p95_latency_ms"] > baselines["p95_latency"] * 1.5:
        alerts.append(f"P95 latency {current_metrics['p95_latency_ms']}ms exceeds 1.5x baseline {baselines['p95_latency']}ms")

    if current_metrics["empty_rate"] > 0.10:
        alerts.append(f"Empty result rate {current_metrics['empty_rate']:.1%} exceeds 10% threshold")

    if current_metrics["avg_similarity"] < baselines["avg_similarity"] * 0.8:
        alerts.append(f"Average similarity dropped to {current_metrics['avg_similarity']:.3f} from baseline {baselines['avg_similarity']:.3f}")

    return alerts

What goes wrong

Mistake	How you notice it	The fix
Not logging raw queries and results	Cannot debug why a specific query failed	Log every query with its top-3 results. Storage is cheap. Debugging without data is expensive
No baseline for "normal"	Do not know if 50ms latency is good or bad	Establish baselines in the first week. Track deviations from baseline, not absolute numbers
Alert fatigue from noisy thresholds	"Low similarity" alert fires 50 times a day, gets ignored	Use percentiles and sustained deviations. Alert on the trend, not individual spikes
Monitoring adds significant latency	Logging doubles query time	Use async logging. Write logs to a buffer, not synchronously per query

Confirm it worked

Run a sample query through the monitored search function. Verify that the structured log output captures all required dimensions for offline evaluation.

# Run a query and verify it's logged
results = monitored_search("return policy")
# Check your log output for the structured log entry
# Should contain: query_preview, result_count, latency_ms, top_similarity

Next: Cost Optimization