RAG Observability: The Dashboard That Catches Problems Before Your Users

TL;DR

Traditional monitoring (uptime, HTTP latency) is insufficient for a production RAG. You need to track retrieval relevance, answer quality, hallucination rate, and user satisfaction. This guide compares RAG observability tools (LangSmith, Langfuse, Phoenix/Arize, W&B), shows how to trace each pipeline step, and configure alerts that catch problems before your users do.

Why RAG Monitoring Is Different

Classic Metrics Are Not Enough

A RAG can return a 200 status with 500ms latency and still give a catastrophic answer. Here is what classic API monitoring misses:

Result: classic monitoring detects only 2 out of 8 problems. The other 6 require RAG-specific observability.

The 4 Pillars of RAG Observability

┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐
│  Retrieval   │  │  Generation  │  │  Latency     │  │  User        │
│  Quality     │  │  Quality     │  │  & Costs     │  │  Satisfaction│
├─────────────┤  ├─────────────┤  ├─────────────┤  ├─────────────┤
│ Relevance    │  │ Faithfulness │  │ P50/P95/P99  │  │ Thumbs up/  │
│ Recall       │  │ Hallucination│  │ Token usage  │  │ down         │
│ MRR/NDCG     │  │ Completeness │  │ Cost/query   │  │ Reformulation│
│ Empty results│  │ Toxicity     │  │ Cache hit    │  │ Escalation   │
└─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘

Essential Metrics to Track

Retrieval Metrics

DEVELOPERpython
from dataclasses import dataclass
from typing import Optional

@dataclass
class RetrievalMetrics:
    """Retrieval quality metrics."""

    # Number of documents retrieved
    num_docs_retrieved: int
    # Average relevance score of documents
    avg_relevance_score: float
    # Is the top document relevant?
    top_doc_relevant: bool
    # Empty results rate
    empty_results: bool
    # Search time (ms)
    retrieval_latency_ms: float
    # Document sources (to detect bias)
    doc_sources: list[str]

    @property
    def is_healthy(self) -> bool:
        return (
            not self.empty_results
            and self.avg_relevance_score > 0.7
            and self.retrieval_latency_ms < 500
        )

Generation Metrics

DEVELOPERpython
@dataclass
class GenerationMetrics:
    """Generation quality metrics."""

    # Faithfulness: is the response faithful to documents?
    faithfulness_score: float  # 0.0 - 1.0
    # Does the response contain hallucinations?
    hallucination_detected: bool
    # Response completeness
    completeness_score: float  # 0.0 - 1.0
    # Tokens used (input + output)
    input_tokens: int
    output_tokens: int
    # Generation time (ms)
    generation_latency_ms: float
    # Estimated cost ($)
    estimated_cost_usd: float

    @property
    def is_healthy(self) -> bool:
        return (
            self.faithfulness_score > 0.8
            and not self.hallucination_detected
            and self.generation_latency_ms < 3000
        )

User Metrics

DEVELOPERpython
@dataclass
class UserMetrics:
    """User satisfaction metrics."""

    # Explicit feedback (thumbs up/down)
    user_feedback: Optional[str]  # "positive" | "negative" | None
    # Did the user reformulate their question?
    query_reformulated: bool
    # Did the user escalate to a human?
    escalated_to_human: bool
    # Session duration (seconds)
    session_duration_seconds: float
    # Number of messages in conversation
    message_count: int

Summary Dashboard

Tracing the RAG Pipeline

Tracing Architecture

Each RAG request should be traced step by step:

DEVELOPERpython
import time
import uuid
from contextlib import contextmanager

class RAGTracer:
    """Trace each step of the RAG pipeline."""

    def __init__(self, trace_backend):
        self.backend = trace_backend

    @contextmanager
    def trace_request(self, user_id: str, query: str):
        trace_id = str(uuid.uuid4())
        trace = {
            "trace_id": trace_id,
            "user_id": user_id,
            "query": query,
            "started_at": time.time(),
            "steps": [],
        }

        yield trace

        trace["total_duration_ms"] = (
            (time.time() - trace["started_at"]) * 1000
        )
        self.backend.save_trace(trace)

    @contextmanager
    def trace_step(self, trace: dict, step_name: str):
        step = {
            "name": step_name,
            "started_at": time.time(),
            "metadata": {},
        }

        yield step

        step["duration_ms"] = (time.time() - step["started_at"]) * 1000
        trace["steps"].append(step)


# Usage in the pipeline
tracer = RAGTracer(backend=langfuse_backend)

async def process_rag_query(user_id: str, query: str):
    with tracer.trace_request(user_id, query) as trace:

        # Step 1: Query embedding
        with tracer.trace_step(trace, "query_embedding") as step:
            embedding = await embed_query(query)
            step["metadata"]["model"] = "text-embedding-3-small"
            step["metadata"]["dimensions"] = len(embedding)

        # Step 2: Vector search
        with tracer.trace_step(trace, "vector_search") as step:
            docs = await search_vectors(embedding, top_k=5)
            step["metadata"]["num_results"] = len(docs)
            step["metadata"]["avg_score"] = avg_score(docs)

        # Step 3: Reranking
        with tracer.trace_step(trace, "reranking") as step:
            ranked_docs = await rerank(query, docs)
            step["metadata"]["top_score"] = ranked_docs[0].score

        # Step 4: LLM generation
        with tracer.trace_step(trace, "llm_generation") as step:
            response = await generate(query, ranked_docs)
            step["metadata"]["model"] = "gpt-4o"
            step["metadata"]["input_tokens"] = response.usage.input
            step["metadata"]["output_tokens"] = response.usage.output

        return response.text

Trace Visualization

Trace: abc-123 | Total duration: 2340ms | Status: OK
├─ query_embedding     [45ms]  model=text-embedding-3-small
├─ vector_search       [120ms] results=5, avg_score=0.82
├─ reranking           [380ms] model=cohere-rerank-v3.5, top=0.94
├─ llm_generation      [1780ms] model=gpt-4o, tokens=1250/340
│   ├─ input_tokens:  1250
│   ├─ output_tokens: 340
│   └─ cost: $0.023
└─ total_cost: $0.028

Observability Tools Comparison

Detailed Comparison Table

Recommendation by Use Case

Implementation with Langfuse

Setup and Instrumentation

DEVELOPERpython
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

# Initialization
langfuse = Langfuse(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="https://cloud.langfuse.com"  # or self-hosted
)

@observe()
async def rag_pipeline(query: str, user_id: str) -> str:
    """Complete RAG pipeline with Langfuse tracing."""

    # Trace the embedding
    langfuse_context.update_current_observation(
        name="query_embedding",
        metadata={"model": "text-embedding-3-small"}
    )
    embedding = await embed_query(query)

    # Trace the search
    with langfuse_context.observe(name="vector_search") as span:
        docs = await search_vectors(embedding, top_k=5)
        span.update(
            metadata={
                "num_results": len(docs),
                "avg_score": sum(d.score for d in docs) / len(docs)
            }
        )

    # Trace the generation
    with langfuse_context.observe(
        name="llm_generation",
        model="gpt-4o"
    ) as generation:
        response = await generate_response(query, docs)
        generation.update(
            usage={
                "input": response.usage.prompt_tokens,
                "output": response.usage.completion_tokens,
            },
            metadata={"temperature": 0.1}
        )

    # Automatic scoring
    langfuse_context.score_current_trace(
        name="relevance",
        value=compute_relevance(query, docs),
        comment="Automatic relevance score"
    )

    return response.text


@observe()
async def embed_query(query: str) -> list[float]:
    """Embedding with automatic tracing."""
    result = await openai.embeddings.create(
        model="text-embedding-3-small",
        input=query
    )
    return result.data[0].embedding

Automatic Evaluations with Langfuse

DEVELOPERpython
from langfuse import Langfuse

langfuse = Langfuse()

# Faithfulness evaluation
async def evaluate_faithfulness(
    trace_id: str,
    query: str,
    response: str,
    documents: list[str]
):
    """Evaluate whether the response is faithful to documents."""
    eval_prompt = f"""
    Question: {query}
    Documents: {documents}
    Response: {response}

    Is the response fully supported by the documents?
    Score from 0.0 (total hallucination) to 1.0 (perfectly faithful).
    Respond only with the numeric score.
    """

    result = await openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": eval_prompt}],
        temperature=0
    )

    score = float(result.choices[0].message.content.strip())

    langfuse.score(
        trace_id=trace_id,
        name="faithfulness",
        value=score,
        comment="Auto-eval by GPT-4o-mini"
    )

    return score

Implementation with LangSmith

Setup and Tracing

DEVELOPERpython
import os
from langsmith import traceable
from langsmith.run_helpers import get_current_run_tree

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "lsv2_..."
os.environ["LANGCHAIN_PROJECT"] = "rag-production"

@traceable(name="rag_pipeline")
async def rag_pipeline(query: str, user_id: str) -> str:
    """RAG pipeline with LangSmith tracing."""

    embedding = await embed_query(query)
    docs = await search_vectors(embedding)
    response = await generate_response(query, docs)

    # Attach metadata
    run = get_current_run_tree()
    if run:
        run.extra["metadata"] = {
            "user_id": user_id,
            "num_docs": len(docs),
            "model": "gpt-4o",
        }

    return response


@traceable(name="vector_search")
async def search_vectors(embedding: list[float]) -> list:
    """Vector search with tracing."""
    results = await qdrant.search(
        collection="documents",
        query_vector=embedding,
        limit=5
    )
    return results


# Evaluation with LangSmith
from langsmith.evaluation import evaluate

results = evaluate(
    rag_pipeline,
    data="rag-test-dataset",
    evaluators=[
        "relevance",
        "faithfulness",
        "helpfulness",
    ],
    experiment_prefix="v2.1-gpt4o",
)

Setting Up Alerts

Critical Alerts to Configure

DEVELOPERpython
class RAGAlertManager:
    """Alert manager for RAG monitoring."""

    def __init__(self, notification_backend):
        self.backend = notification_backend
        self.thresholds = {
            "hallucination_rate": 0.15,      # Max 15%
            "empty_results_rate": 0.10,      # Max 10%
            "p95_latency_ms": 5000,          # Max 5s
            "avg_relevance_score": 0.5,      # Min 0.5
            "error_rate": 0.05,              # Max 5%
            "cost_per_query_usd": 0.15,      # Max $0.15
            "negative_feedback_rate": 0.30,  # Max 30%
        }

    async def check_metrics(self, window_minutes: int = 60):
        metrics = await self.get_aggregated_metrics(window_minutes)

        alerts = []

        if metrics["hallucination_rate"] > self.thresholds["hallucination_rate"]:
            alerts.append({
                "severity": "critical",
                "metric": "hallucination_rate",
                "value": metrics["hallucination_rate"],
                "message": (
                    f"Hallucination rate at "
                    f"{metrics['hallucination_rate']:.1%} "
                    f"(threshold: {self.thresholds['hallucination_rate']:.1%})"
                ),
                "action": "Check prompt and recent documents"
            })

        if metrics["avg_relevance_score"] < self.thresholds["avg_relevance_score"]:
            alerts.append({
                "severity": "warning",
                "metric": "avg_relevance_score",
                "value": metrics["avg_relevance_score"],
                "message": (
                    f"Relevance score at "
                    f"{metrics['avg_relevance_score']:.2f} "
                    f"(threshold: {self.thresholds['avg_relevance_score']:.2f})"
                ),
                "action": "Check embeddings and vector database"
            })

        if metrics["p95_latency_ms"] > self.thresholds["p95_latency_ms"]:
            alerts.append({
                "severity": "warning",
                "metric": "p95_latency_ms",
                "value": metrics["p95_latency_ms"],
                "message": (
                    f"P95 latency at {metrics['p95_latency_ms']}ms "
                    f"(threshold: {self.thresholds['p95_latency_ms']}ms)"
                ),
                "action": "Check cache and LLM performance"
            })

        for alert in alerts:
            await self.backend.send_alert(alert)

        return alerts

Slack/Discord Integration for Alerts

DEVELOPERpython
import httpx

class SlackAlertBackend:
    """Send RAG alerts to Slack."""

    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    async def send_alert(self, alert: dict):
        severity_labels = {
            "critical": "CRITICAL",
            "warning": "WARNING",
            "info": "INFO"
        }

        payload = {
            "blocks": [
                {
                    "type": "header",
                    "text": {
                        "type": "plain_text",
                        "text": (
                            f"[{severity_labels[alert['severity']]}] "
                            f"RAG Alert: {alert['metric']}"
                        )
                    }
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": (
                            f"*Message:* {alert['message']}\n"
                            f"*Action:* {alert['action']}"
                        )
                    }
                }
            ]
        }

        async with httpx.AsyncClient() as client:
            await client.post(self.webhook_url, json=payload)

Debugging with Traces

Identifying the Source of a Problem

When a user reports a bad response, traces enable quick diagnosis:

DEVELOPERpython
async def debug_bad_response(trace_id: str):
    """Analyze a trace to identify the problem."""
    trace = await langfuse.get_trace(trace_id)

    report = []

    # 1. Check retrieval
    search_step = find_step(trace, "vector_search")
    if search_step["metadata"]["num_results"] == 0:
        report.append("PROBLEM: No documents found")
    elif search_step["metadata"]["avg_score"] < 0.5:
        report.append("PROBLEM: Low relevance documents")

    # 2. Check reranking
    rerank_step = find_step(trace, "reranking")
    if rerank_step and rerank_step["metadata"]["top_score"] < 0.3:
        report.append("PROBLEM: Reranking did not improve results")

    # 3. Check generation
    gen_step = find_step(trace, "llm_generation")
    if gen_step["metadata"]["output_tokens"] < 20:
        report.append("PROBLEM: Response too short")
    if gen_step["duration_ms"] > 5000:
        report.append("WARNING: Very slow generation")

    # 4. Check scores
    scores = trace.get("scores", {})
    if scores.get("faithfulness", 1.0) < 0.5:
        report.append("PROBLEM: Hallucination detected")

    return report

Best Practices

Production Checklist

Phase 1: Instrumentation (Day 1)
  - Trace each pipeline step
  - Log basic metrics (latency, tokens, costs)
  - Capture user feedback

Phase 2: Evaluations (Week 1)
  - Set up automatic faithfulness evaluation
  - Create a test dataset (50+ question/answer pairs)
  - Establish metric baselines

Phase 3: Alerts (Week 2)
  - Configure critical alerts (hallucinations, errors)
  - Integrate Slack/Discord
  - Define response runbooks

Phase 4: Continuous Optimization (Month 1+)
  - A/B test prompts
  - Analyze problematic queries
  - Data-driven continuous improvement

Going Further

• RAG Evaluation and Metrics: the parent guide on evaluation
• Reduce RAG Latency: performance optimization
• Hallucination Detection: specific techniques
• RAG Guardrails: security and prompt injections
• Smart RAG Caching: cost optimization

FAQ

Which observability tool should I start with?

If you use LangChain, LangSmith is the natural choice with its native integration. If you have GDPR constraints or want self-hosted, Langfuse is excellent and open source. For a zero budget, Phoenix (Arize) offers a powerful local dashboard. Our recommendation for European companies: Langfuse self-hosted for full data control.

How many traces should I retain?

In production, keep at least 30 days of traces to detect trends. For debugging, the last 7 days are usually sufficient. On Langfuse self-hosted, retention is unlimited (limited only by your storage). On free LangSmith, you are limited to 14 days and 5,000 traces/month.

How do I measure quality without human evaluation?

Use LLM-as-Judge: a lightweight model (GPT-4o-mini) evaluates each response on faithfulness, relevance, and completeness. This approach costs about $0.002 per evaluation and correlates at 85% with human evaluation according to recent benchmarks. Combine with implicit feedback (reformulations, escalations) for a complete view.

How often should I check my dashboard?

Daily during the first month, then weekly once alerts are configured. Automatic alerts should cover critical scenarios. Schedule a thorough monthly review to analyze trends and identify structural improvements.

Does Ailog offer a built-in observability dashboard?

Yes. Ailog includes an observability dashboard that displays response quality, retrieval metrics, costs, and user feedback in real time. You can configure custom alerts and export data for in-depth analysis. Hosting in France ensures GDPR compliance for your observability data.