Capsule 03: Monitoring and Observability for RAG Systems

Capsule description

Without reliable metrics, operating RAG in production means reacting late. This capsule defines what to measure and how to use those signals to act before the user notices degradation: Prometheus, Grafana, alerting with concrete thresholds, structured JSON logging, runbooks, Python decorators for metrics, FastAPI middleware for request tracking, and practical troubleshooting.

Estimated time: 25–35 minutes
Prerequisites: Having read capsules 01–02 (scaling, production strategies).

1. Minimal metrics for RAG

Before instrumenting, define what matters to you. For a typical RAG system you need these metric families:

Without them, you'll be debugging blind when something fails.

2. Prometheus: essential metrics

2.1 Query latency

Latency of the full path (embedding + retrieval + optional LLM) and of the isolated retrieval:

from prometheus_client import Histogram, Counter, Gauge

# Retrieval latency (ms)
retrieval_latency = Histogram(
    "rag_retrieval_latency_ms",
    "Latency of vector search retrieval in milliseconds",
    ["collection", "top_k"],
    buckets=[10, 25, 50, 100, 200, 500, 1000, 2500, 5000],
)

# Total request latency (includes LLM if applicable)
query_latency_total = Histogram(
    "rag_query_latency_ms",
    "Full RAG query latency in milliseconds",
    ["endpoint"],
    buckets=[50, 100, 200, 500, 1000, 2000, 5000, 10000],
)

2.2 Ingestion throughput

Documents and vectors ingested per unit of time:

ingestion_documents_total = Counter(
    "rag_ingestion_documents_total",
    "Total documents ingested",
    ["collection", "status"],
)
ingestion_duration_seconds = Histogram(
    "rag_ingestion_duration_seconds",
    "Time to ingest a batch of documents",
    ["collection"],
    buckets=[1, 5, 10, 30, 60, 120],
)

2.3 Index size

Index size and number of vectors:

index_vectors_total = Gauge(
    "rag_index_vectors_total",
    "Number of vectors in the index",
    ["collection"],
)
index_size_bytes = Gauge(
    "rag_index_size_bytes",
    "Approximate index size in bytes",
    ["collection"],
)

2.4 Error rate

Errors per endpoint and per type:

query_errors_total = Counter(
    "rag_query_errors_total",
    "Total query errors",
    ["endpoint", "error_type"],
)
query_total = Counter(
    "rag_queries_total",
    "Total queries",
    ["endpoint"],
)

2.5 Cache hit rate (if you use a cache)

cache_hits_total = Counter("rag_cache_hits_total", "Cache hits", ["collection"])
cache_misses_total = Counter("rag_cache_misses_total", "Cache misses", ["collection"])

3. Python: decorators for metric collection

Encapsulate the instrumentation in reusable decorators:

import time
from functools import wraps
from prometheus_client import Histogram

retrieval_latency = Histogram(
    "rag_retrieval_latency_ms",
    "Retrieval latency in ms",
    ["collection"],
    buckets=[10, 25, 50, 100, 200, 500, 1000],
)


def track_retrieval_latency(collection: str = "default"):
    def decorator(func):
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                result = await func(*args, **kwargs)
                return result
            finally:
                elapsed_ms = (time.perf_counter() - start) * 1000
                retrieval_latency.labels(collection=collection).observe(elapsed_ms)

        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                return func(*args, **kwargs)
            finally:
                elapsed_ms = (time.perf_counter() - start) * 1000
                retrieval_latency.labels(collection=collection).observe(elapsed_ms)

        if asyncio.iscoroutinefunction(func):
            return async_wrapper
        return sync_wrapper

    return decorator

Usage:

@track_retrieval_latency(collection="docs")
async def search(query: str, top_k: int = 5):
    return await chroma_collection.query(query_texts=[query], n_results=top_k)

4. FastAPI: middleware for request tracking

Middleware that measures latency and errors per route:

from fastapi import FastAPI, Request
from prometheus_client import Histogram, Counter
import time

request_latency = Histogram(
    "rag_http_request_latency_seconds",
    "HTTP request latency",
    ["method", "path", "status"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
)
request_total = Counter(
    "rag_http_requests_total",
    "Total HTTP requests",
    ["method", "path", "status"],
)


@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    path = request.scope.get("path", "unknown")
    method = request.method
    status = response.status_code
    request_latency.labels(method=method, path=path, status=str(status)).observe(elapsed)
    request_total.labels(method=method, path=path, status=str(status)).inc()
    return response

5. Grafana: dashboard configuration

5.1 Essential PromQL queries

# p95 retrieval latency (ms)
histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le, collection))

# p50, p95, p99
histogram_quantile(0.50, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))
histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))
histogram_quantile(0.99, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))

# Throughput queries/sec
sum(rate(rag_queries_total[1m]))

# Error rate (%)
100 * sum(rate(rag_query_errors_total[5m])) / sum(rate(rag_queries_total[5m]))

# Cache hit rate (%)
100 * sum(rate(rag_cache_hits_total[5m])) / (sum(rate(rag_cache_hits_total[5m])) + sum(rate(rag_cache_misses_total[5m])))

# Ingestion throughput docs/sec
sum(rate(rag_ingestion_documents_total[5m]))

5.2 Panel JSON for Grafana (skeleton)

Save this as rag-dashboard.json and import it into Grafana:

{
  "dashboard": {
    "title": "RAG Observability",
    "panels": [
      {
        "title": "Retrieval Latency p50/p95/p99",
        "type": "timeseries",
        "targets": [
          {"expr": "histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))", "legendFormat": "p95"},
          {"expr": "histogram_quantile(0.50, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))", "legendFormat": "p50"},
          {"expr": "histogram_quantile(0.99, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))", "legendFormat": "p99"}
        ],
        "fieldConfig": {"defaults": {"unit": "ms"}}
      },
      {
        "title": "Query Throughput",
        "type": "timeseries",
        "targets": [{"expr": "sum(rate(rag_queries_total[1m]))", "legendFormat": "queries/s"}]
      },
      {
        "title": "Error Rate (%)",
        "type": "timeseries",
        "targets": [
          {"expr": "100 * sum(rate(rag_query_errors_total[5m])) / sum(rate(rag_queries_total[5m]))", "legendFormat": "error %"}
        ]
      },
      {
        "title": "Index Vectors",
        "type": "stat",
        "targets": [{"expr": "rag_index_vectors_total", "legendFormat": "vectors"}]
      }
    ]
  }
}

5.3 Recommended layout

1. Row 1: Latency p50/p95/p99 (lines).
2. Row 2: Throughput queries/s and error rate (%).
3. Row 3: Index size, cache hit rate, ingestion throughput.
4. Row 4: Errors by type (table or breakdown).

6. Alerting: concrete thresholds

6.1 Prometheus rules (example)

groups:
  - name: rag_alerts
    rules:
      # Warning: p95 > 200ms for 5 min
      - alert: RAGHighLatencyWarning
        expr: histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le)) > 200
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "RAG retrieval p95 > 200ms"

      # Critical: p95 > 500ms for 5 min
      - alert: RAGHighLatencyCritical
        expr: histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le)) > 500
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "RAG retrieval p95 > 500ms - severe degradation"

      # Critical: error rate > 2% for 5 min
      - alert: RAGHighErrorRate
        expr: 100 * sum(rate(rag_query_errors_total[5m])) / sum(rate(rag_queries_total[5m])) > 2
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "RAG error rate > 2%"

      # Warning: cache hit rate drops >15% vs baseline (requires a baseline recording rule)
      - alert: RAGCacheHitRateDrop
        expr: (rag_cache_hit_rate - rag_cache_hit_rate_baseline) < -15
        for: 10m
        labels:
          severity: warning

6.2 Quick threshold table

7. Structured JSON logging (Python)

Structured logs make searching in Elasticsearch, Loki, or CloudWatch easier.

7.1 Configuration with structlog

import structlog
import logging
import json

def configure_structured_logging():
    structlog.configure(
        processors=[
            structlog.stdlib.filter_by_level,
            structlog.stdlib.add_logger_name,
            structlog.stdlib.add_log_level,
            structlog.stdlib.PositionalArgumentsFormatter(),
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.UnicodeDecoder(),
            structlog.processors.JSONRenderer(),
        ],
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

7.2 Usage in RAG endpoints

logger = structlog.get_logger()

async def rag_query_endpoint(query: str):
    log = logger.bind(
        endpoint="rag_query",
        query_id=str(uuid.uuid4()),
        query_length=len(query),
    )
    try:
        results = await retrieve_and_generate(query)
        log.info(
            "query_completed",
            latency_ms=results.get("latency_ms"),
            num_results=len(results.get("documents", [])),
        )
        return results
    except Exception as e:
        log.error("query_failed", error=str(e), error_type=type(e).__name__, exc_info=True)
        raise

7.3 Example JSON output

{
  "event": "query_completed",
  "endpoint": "rag_query",
  "query_id": "a1b2c3d4-...",
  "query_length": 42,
  "latency_ms": 156,
  "num_results": 5,
  "timestamp": "2026-03-13T14:32:01.123Z",
  "level": "info",
  "logger": "app.routes"
}

8. Runbook templates

8.1 Runbook: RAGHighLatencyCritical

Condition: retrieval p95 > 500 ms for 5 min.

Steps:

1. Confirm the alert: Check Grafana. Is it a real regression or a one-off spike?
2. Identify the component:
   • Slow embedding API? Check the latency of calls to OpenAI/Cohere.
   • Chroma/vector DB? Check the container's CPU, memory, disk.
   • Number of vectors? If index_vectors_total rose a lot, it may be reindexing.
3. Temporary mitigation:
   • Rate limit if there's saturation.
   • Increase the RAG service's replicas.
   • Enable/verify the cache for frequent queries.
4. Post-incident: Record in the incident doc, update thresholds if needed.

8.2 Runbook: RAGHighErrorRate

Condition: Error rate > 2% for 5 min.

Steps:

1. Confirm: Review rag_query_errors_total by error_type.
2. Classify the errors:
   • timeout → Check embedding/LLM timeouts, increase or scale.
   • connection_refused → Check the health of Chroma/Postgres.
   • validation_error → Check logs, possible malformed input.
3. Mitigation:
   • Circuit breaker if an external dependency is down.
   • Roll back a recent deploy if it correlates.
4. Communication: If it affects users, notify per the incident process.

8.3 Generic runbook (template)

## [ALERT_NAME]

**Condition:** [expression or description]
**Severity:** warning | critical

### 1. Confirmation
- [ ] Check dashboards
- [ ] Verify it's not a false positive

### 2. Diagnosis
- [ ] Identify the affected component
- [ ] Review recent logs and metrics

### 3. Mitigation
- [ ] Immediate action
- [ ] Short-term action

### 4. Resolution
- [ ] Root cause (if applicable)
- [ ] Preventive actions

### 5. Post-mortem
- [ ] Document in [link]

9. Observability troubleshooting

9.1 "We have lots of metrics but they aren't actionable"

Problem: Dashboards full of charts that nobody uses to decide.

Solution: Reduce to metrics tied to product objectives or SLA. For example: retrieval p95 for "the user doesn't wait more than 300 ms", error rate for "fewer than 1% of errors". Hide or archive the rest until you have runbooks for each alert.

9.2 "Constant alerts that nobody attends to"

Problem: Alert fatigue; the team ignores notifications.

Solution: Raise thresholds or widen windows (for example, for: 10m instead of 5m). Group related alerts. Only create alerts that have a defined runbook and an owner.

9.3 "We don't know if we improved"

Problem: Changes in code or indexes without an objective way to compare.

Solution: Define baselines (e.g. p95 in week X) and always compare against them. Use dashboards with overlays of previous periods. Document which baseline you use for each release.

9.4 "The percentiles don't match what I see in logs"

Problem: p95 in Prometheus differs from the "worst case" you see in logs.

Solution: Percentiles are over a time window (e.g. 5 min). A short spike may not show well in percentiles. Review the Histogram buckets and add more if needed (e.g. 200, 500, 1000 ms). Also use p99 to capture outliers.

9.5 "We don't have budget for Grafana Cloud / Datadog"

Problem: An expensive commercial stack for a small team.

Solution: Prometheus + Grafana self-hosted + Alertmanager are free. For logs, Loki (Grafana) or basic Elasticsearch. Start with the minimum (Prometheus + Grafana in Docker Compose) and scale when you need to.

10. Exercises

Exercise 1: Implement a metrics decorator

Goal: Create a @track_latency(metric_name="my_metric") decorator that records latency in a Prometheus Histogram.

from functools import wraps
import time
from prometheus_client import Histogram

def track_latency(metric_name: str = "custom_latency", buckets=None):
    buckets = buckets or [0.01, 0.05, 0.1, 0.25, 0.5, 1.0]
    hist = Histogram(metric_name + "_seconds", "Latency", buckets=buckets)

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                return func(*args, **kwargs)
            finally:
                hist.observe(time.perf_counter() - start)
        return wrapper
    return decorator

# Usage:
@track_latency("my_retrieval")
def my_search():
    ...

Exercise 2: Per-route latency middleware

Goal: Add FastAPI middleware that records latency per route in a Histogram with path and method labels.

from fastapi import Request
from prometheus_client import Histogram
import time

latency = Histogram("http_latency", "HTTP latency", ["method", "path"], buckets=[0.05, 0.1, 0.25, 0.5, 1.0])

@app.middleware("http")
async def add_latency(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    latency.labels(method=request.method, path=request.url.path).observe(time.perf_counter() - start)
    return response

Exercise 3: PromQL for error rate

Goal: Write a PromQL expression that calculates the error rate (%) over the last 5 minutes, assuming rag_queries_total and rag_query_errors_total.

100 * sum(rate(rag_query_errors_total[5m])) / sum(rate(rag_queries_total[5m]))

100 * sum(rate(rag_query_errors_total[5m])) / (sum(rate(rag_queries_total[5m])) or vector(1))

Exercise 4: Structured log with structlog

Goal: Configure structlog so that each log includes request_id, endpoint, and outputs as JSON.

import structlog

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger()

# In your handler:
log = log.bind(request_id=request_id, endpoint="/rag/query")
log.info("request_started", query=query[:50])
# ... later ...
log.info("request_completed", latency_ms=123)

Exercise 5: Alert rule for slow ingestion

Goal: Create a rule that alerts if the ingestion queue has more than 1000 pending documents for 10 minutes. Assume a rag_ingestion_queue_size metric.

- alert: RAGIngestionBacklog
  expr: rag_ingestion_queue_size > 1000
  for: 10m
  labels:
    severity: warning
  annotations:
    summary: "RAG ingestion backlog > 1000 documents"
    description: "Queue has been above 1000 for 10 minutes. Check ingestion workers and embedding API."

Exercise 6: Runbook for "Index size grows too fast"

Goal: Write the steps of a runbook for when the rag_index_size_bytes metric grows more than 20% in 24 hours.

11. Summary

• Define minimal metrics: latency (p50/p95/p99), throughput, error rate, index size, cache hit rate.
• Instrument with Prometheus (Histogram, Counter, Gauge) and expose them at /metrics.
• Use Python decorators to capture the latency of critical functions (retrieval, embedding).
• Use FastAPI middleware for request tracking and per-route latency.
• Configure Grafana with PromQL for percentiles, throughput, and error rate.
• Define alerts with concrete thresholds: p95 > 200 ms warning, > 500 ms critical; error rate > 2% critical.
• Structured JSON logging (structlog) for traceability and searching in log systems.
• Write runbooks for each alert: confirmation, diagnosis, mitigation, and post-mortem.
• Avoid alert fatigue: only alerts with a defined runbook and owner.

12. Additional resources

• Prometheus - Overview
• Grafana Documentation
• prometheus_client (Python) - GitHub
• Structlog - Structured Logging for Python
• PromQL Tutorial
• Grafana Alerting
• The Four Golden Signals (Google SRE)
• OpenTelemetry for Python

Estimated time: 25–35 minutes
Next: 04-backup-disaster-recovery.md