Monitoring RAG in Production

Monitoring means watching your RAG (Retrieval-Augmented Generation) app while real users use it. A RAG app can fail silently: the server stays up and returns answers, but the answers become wrong, slow or expensive. Good monitoring tells you that something changed, and tracing tells you why.

Overview: why RAG needs its own monitoring

A normal web service fails loudly. It throws errors or times out. A RAG pipeline usually fails quietly. The retriever may return weaker chunks after your documents change. The model may start to invent facts when no good context is found. Latency may double because someone raised k. None of these raise an exception.

So you monitor four layers:

• System health: error rate, timeouts, rate-limit errors from the model provider, latency per step.
• Cost: input and output tokens per request, embedding calls, and total spend. Prices change often, so check your provider’s pricing page for current values.
• Retrieval quality: how many documents came back, their similarity scores, which sources were used, and how often nothing useful was found (the “no-context rate”).
• Answer quality: user feedback (thumbs up or down), sampled automatic evaluation, and groundedness checks (is the answer supported by the retrieved text?).

Logs, metrics and traces

These three tools answer different questions.

A trace is a tree of steps (called runs) for one request. LangChain produces this tree naturally: each retriever, prompt, model and parser in an LCEL chain is a run. The tracing platform LangSmith records these runs for you when you switch it on.

Key concepts: what to record for every request

For each request, record enough to replay and judge it later:

• A request ID and timestamp, plus the version of your prompt, embedding model, index and LLM.
• The user question (redacted, see Common Mistakes).
• The retrieved chunks: source, chunk ID and score. Chunk text is useful but can be large or private, so you may store only IDs.
• The final prompt size in tokens, and the answer.
• Latency of each step and the total.
• Errors, retries and fallbacks.
• User feedback, linked by request ID.

Version tags matter a lot. When quality drops on Tuesday, you want to see that Monday’s deployment changed the chunk size.

Examples

Example 1: structured logging of every RAG request

This example runs offline. It uses a tiny keyword retriever and a fake generator, wrapped in RunnableLambda so they behave like normal LCEL components. The function builds one JSON-friendly record per request. In real code, you would send this record to your log system.

import json
import time
from langchain_core.documents import Document
from langchain_core.runnables import RunnableLambda

DOCS = [
    Document(page_content='Refunds are processed within 5 business days.', metadata={'source': 'refund.md'}),
    Document(page_content='Shipping takes 3 days to most cities.', metadata={'source': 'shipping.md'}),
    Document(page_content='Support is open on weekdays.', metadata={'source': 'support.md'}),
]

def retrieve(question):
    words = set(question.lower().split())
    scored = []
    for doc in DOCS:
        doc_words = set(doc.page_content.lower().replace('.', '').split())
        scored.append((len(words & doc_words), doc))
    scored.sort(key=lambda pair: -pair[0])
    return [doc for score, doc in scored[:2] if score > 0]

def generate(docs):
    if not docs:
        return 'I do not know.'
    return docs[0].page_content

retriever = RunnableLambda(retrieve)
generator = RunnableLambda(generate)

def answer_with_logging(question):
    start = time.perf_counter()
    docs = retriever.invoke(question)
    answer = generator.invoke(docs)
    record = {
        'question': question,
        'num_docs': len(docs),
        'sources': [d.metadata['source'] for d in docs],
        'no_context': len(docs) == 0,
        'answer': answer,
        'latency_ms': round((time.perf_counter() - start) * 1000, 2),
    }
    return record

for q in ['how long do refunds take', 'do you accept crypto payments']:
    record = answer_with_logging(q)
    print(json.dumps({k: v for k, v in record.items() if k != 'latency_ms'}))

Output:

{"question": "how long do refunds take", "num_docs": 1, "sources": ["refund.md"], "no_context": false, "answer": "Refunds are processed within 5 business days."}
{"question": "do you accept crypto payments", "num_docs": 0, "sources": [], "no_context": true, "answer": "I do not know."}

We hide latency_ms in the printout only because it changes on every run. The important field is no_context. The second question found nothing, so we can count it. A rising count of such questions means users ask about topics your knowledge base does not cover, or your retriever is failing.

Example 2: a rolling-window monitor with alerts

One bad request is noise. A bad trend needs action. This class keeps the last few events and raises alerts when the no-context rate or the mean latency crosses a limit. In production you would compute the same numbers in your metrics system (Prometheus, Datadog, CloudWatch and so on), but the logic is the same.

from collections import deque


class RagMonitor:
    def __init__(self, window=5, max_no_context_rate=0.4, max_mean_latency_ms=2000):
        self.window = window
        self.max_rate = max_no_context_rate
        self.max_latency = max_mean_latency_ms
        self.events = deque(maxlen=window)

    def record(self, no_context, latency_ms):
        self.events.append((no_context, latency_ms))

    def alerts(self):
        if len(self.events) < self.window:
            return []
        found = []
        rate = sum(1 for nc, _ in self.events if nc) / len(self.events)
        mean_latency = sum(lat for _, lat in self.events) / len(self.events)
        if rate > self.max_rate:
            found.append(f'no-context rate {rate:.0%} above limit {self.max_rate:.0%}')
        if mean_latency > self.max_latency:
            found.append(f'mean latency {mean_latency:.0f} ms above limit {self.max_latency} ms')
        return found


monitor = RagMonitor()
traffic = [(False, 800), (False, 900), (True, 700), (True, 1000), (False, 600), (True, 3500)]
for i, (no_context, latency) in enumerate(traffic, start=1):
    monitor.record(no_context, latency)
    if i >= 5:
        print(f'after {i} requests: {monitor.alerts()}')

Output:

after 5 requests: []
after 6 requests: ['no-context rate 60% above limit 40%']

After five requests, 2 of 5 had no context, which is exactly 40%, not above the limit. The sixth request pushes the oldest event out of the window. Now 3 of 5 have no context (60%), so the alert fires. The mean latency in that window is 1340 ms, which is under the 2000 ms limit, so only one alert appears. The deque(maxlen=...) drops old events automatically, which gives a rolling window.

Example 3: turn on LangSmith tracing and collect feedback

LangChain apps send traces to LangSmith when environment variables are set. You do not change your chain code. Set the variables in your deployment environment, never in source code.

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-langsmith-api-key"
export LANGSMITH_PROJECT="rag-prod"

You can then add tags and metadata to each call, so you can filter traces by version or user tier. You can also attach user feedback to a run. This example needs a LangSmith account and network access, so its output is illustrative: the traces appear in the LangSmith web app under the project name.

from langsmith import Client

# chain is your LCEL RAG chain, defined elsewhere
answer = chain.invoke(
    'How long do refunds take?',
    config={
        'tags': ['prod', 'rag-v3'],
        'metadata': {'user_tier': 'free', 'index_version': '2026-10-01'},
    },
)

client = Client()
client.create_feedback(
    run_id='RUN_ID_FROM_YOUR_TRACE',
    key='user_thumbs',
    score=1,
    comment='Helpful answer',
)

The tags and metadata keys inside config are standard LangChain run configuration. They are attached to the trace. The feedback call links a score to one specific run, so you can later open all traces with a low score and read what the retriever returned. Check the current LangSmith documentation for exact setup options, since platform features change.

Under the hood: how tracing sees your chain

1. You call chain.invoke(question, config=...).
2. LangChain creates a root run and passes a callback manager down through every step.
3. Each component (retriever, prompt, chat model, parser) starts a child run, and then reports its inputs and outputs when it ends.
4. A tracer, if enabled, sends these runs in the background to the tracing service, so your user does not wait for it.
5. You can also write your own callback handler (a subclass of BaseCallbackHandler) to push the same events into your own metrics system.

This is why tracing is cheap to adopt: the instrumentation lives in the framework, not in your business logic.

Monitoring quality over time

Sample and score online

You cannot read every conversation. Instead, sample a small share of traffic (for example 1–5%) and score it with an LLM-as-judge or a groundedness check, as you learned in the evaluation lessons. Always send sampled data through the same redaction step. Judges cost money and can be wrong, so use them to find trends, not to decide single cases.

Watch for drift

Drift means the world changed while your system stayed the same. Common causes:

• Query drift: users start asking about a new product.
• Data drift: documents are updated, deleted or ingested badly, so the index is stale.
• Model drift: your provider updates a model alias, and behaviour changes. Pin a specific model version if your provider offers it.

Close the loop

Every thumbs-down or no-context question is a free test case. Add the good ones to your evaluation dataset, fix the cause, and re-run the evaluation before you deploy. This turns monitoring into continuous improvement.

Common Mistakes

Mistake 1: logging raw user text with personal data

import logging
logger = logging.getLogger('rag')

def handle(question, user_email):
    logger.info('question=%s email=%s', question, user_email)

This is wrong because logs are copied to many systems and kept for a long time. Users often type emails, phone numbers or account IDs inside questions. You should not log identifiers you do not need. A safer version redacts the text before logging:

import re

EMAIL = re.compile(r'[\w.+-]+@[\w-]+\.[\w.]+')


def redact(text):
    return EMAIL.sub('[EMAIL]', text)


print(redact('Refund for jane.doe@example.com please'))

Output:

Refund for [EMAIL] please

Real systems also redact phone numbers, card numbers and names, often with a dedicated PII-detection library. Apply redaction before data reaches logs, traces and evaluation sets.

Mistake 2: watching only uptime and average latency

An average hides slow requests: if 95 requests take 1 s and 5 take 30 s, the mean looks acceptable but some users wait half a minute. Track percentiles (p50, p95, p99). More importantly, uptime says nothing about answer quality. A RAG app that answers “I do not know” to every question has 100% uptime. Always add retrieval and feedback metrics next to the system metrics.

Mistake 3: no version tags

If traces do not say which prompt, index and model produced them, you cannot compare before and after a change. Add version fields to metadata on every call.

Mistake 4: alerting on a single event

Alerting on one slow request or one no-context answer wakes people for noise. Use a window, as in Example 2, and a minimum number of events.

Best Practices

• Trace every request in production, and use tags and metadata for version, environment and user tier.
• Log structured JSON records, not free text, so you can query them.
• Track the no-context rate, the number of retrieved documents, and similarity scores as first-class metrics.
• Record tokens and estimated cost per request, and set a daily budget alert. Look up current prices on your provider’s pricing page.
• Measure latency per step (retrieve, rerank, generate) so you know what to optimise.
• Collect simple user feedback and link it to the request or run ID.
• Sample traffic for automatic scoring, and review the worst cases by hand every week.
• Redact personal data before storage, set retention limits, and restrict who can read traces.
• Handle failures gracefully: timeouts, retries with backoff, and a clear fallback message, and monitor how often each fires.
• Feed production failures back into your evaluation dataset.
• Keep API keys in environment variables or a secret manager, never in code or logs.

Practice Exercises

1. Add a score field. Change the keyword retriever in Example 1 to return the overlap score for each document. Add top_score to the record (use 0 when nothing is found). Hint: keep the scores from the sorted list instead of dropping them.
2. Add a p95 alert. Extend RagMonitor with a max_p95_latency_ms setting. Compute the 95th percentile of the window by sorting the latencies. Decide how to pick the index, and write a test with six events.
3. Write a redaction function for phone numbers. Replace sequences of 10 digits with [PHONE] using re. Test it on three strings, including one with no number. Hint: use \d{10} and think about whether longer digit strings should match.

Summary

• RAG apps fail silently, so you need monitoring for system health, cost, retrieval quality and answer quality.
• Metrics tell you something is wrong, logs describe one request, and traces show where it went wrong.
• LangChain’s callback system lets LangSmith (or your own handler) record every step of a chain without changing your logic.
• The no-context rate, retrieved document count and user feedback are the most useful RAG-specific signals.
• Use rolling windows and percentiles, not single events and averages, for alerts.
• Tag every run with versions, redact personal data, and feed production failures back into your evaluation set.