Testing and Observability for Code RAG

Part 10 of the Code RAG series. Parts 1–9 built a pipeline that indexes correctly, retrieves relevantly, and rebuilds safely. This post covers verification: an offline recall harness you run before shipping any change, and production tracing that shows what’s actually happening at query time.

---

The two verification problems

Building a RAG pipeline well is not enough. You also need to know it’s working — and there are two distinct moments when it can fail silently:

Before you ship: a change to chunking, indexed text, embedding model, or retrieval parameters improved some queries and broke others. Without a test suite, you won’t know until a user reports a wrong answer.

After you ship: a query returns the wrong chunks. The LLM faithfully reasons over that wrong context and produces a confident but incorrect answer. Without chunk-level tracing, this looks identical to an LLM hallucination from the outside.

These are different failure modes with different tools. The first half of this post covers the offline eval harness — a pytest suite you run before every change. The second half covers production tracing.

---

Offline: building a recall@k eval harness

The metric is recall@3: for a given query, is the correct chunk anywhere in the top 3 results? Not rank 1 — rank 1 is too strict for a system that passes all three results to an LLM. If the right chunk is in the top 3, the model has the information it needs.

Recall@3 = (queries where correct chunk is in top 3) / (total queries)

Binary per test case. Either the chunk is there or it isn’t. No threshold guessing, no partial credit. Regressions are immediately visible.

Inspecting the corpus first

Before writing test cases, list what’s actually indexed:

results = collection.get(include=["documents", "metadatas"])
for chunk_id, doc, meta in zip(results["ids"], results["documents"], results["metadatas"]):
    print(f"{meta['file_path']}::{meta.get('function_name', '?')}  ({meta['start_line']}–{meta['end_line']})")

Test targets should come from this list, not from memory. The function you think is in the index might not be — or it might be split across two chunks.

Three query angles per function

For each target function, write queries from three angles:

Angle 1 — Exact identifier. The function name verbatim. The easiest case: the token appears literally in the indexed chunk.

Angle 2 — Paraphrase. What the function does in plain language, without the name. Tests whether the embedding model captures intent, not just identifier presence.

Angle 3 — Root cause / error context. The language a diagnosis agent uses when it arrives with a theory. No function name — just symptoms. From Part 2 we know this is the hardest case: code RAG has a vocabulary gap between what the function does and what it can fail with. If this angle passes, the indexed text enrichment from Part 4 is working.

The full test suite

import pytest
from app.services.rag import CodeRAGService

# Canonical functions — stable reference points for assertions
MOVE_REMOVE_S3  = "moveAndRemoveFileFromS3"   # routes/services/image.js
CLASSIFY_FIELDS = "classifyFields"             # app/constants/prankCheckerOpenAI.js
RUN_PRANK       = "runPrankChecker"            # app/constants/prankCheckerOpenAI.js


@pytest.fixture(scope="module")
def rag():
    svc = CodeRAGService()
    yield svc


def top3_functions(results):
    return {r.get("function_name", "") for r in results[:3]}


# --- moveAndRemoveFileFromS3 ---

def test_s3_exact_identifier(rag):
    results = rag.search("moveAndRemoveFileFromS3", n_results=3)
    assert MOVE_REMOVE_S3 in top3_functions(results)

def test_s3_paraphrase(rag):
    results = rag.search("copy S3 object to new key then delete the source", n_results=3)
    assert MOVE_REMOVE_S3 in top3_functions(results)

def test_s3_root_cause(rag):
    # Hardest: no function name, only error vocabulary
    # Passes only if indexed text includes enriched error context from Part 4
    results = rag.search("S3 operation missing key existence check before copy", n_results=3)
    assert MOVE_REMOVE_S3 in top3_functions(results)


# --- classifyFields ---

def test_classify_exact_identifier(rag):
    results = rag.search("classifyFields", n_results=3)
    assert CLASSIFY_FIELDS in top3_functions(results)

def test_classify_paraphrase(rag):
    results = rag.search("parse openai response as JSON object", n_results=3)
    assert CLASSIFY_FIELDS in top3_functions(results)

def test_classify_root_cause(rag):
    results = rag.search("LLM response not formatted as JSON causes parse error", n_results=3)
    assert CLASSIFY_FIELDS in top3_functions(results)


# --- Precision ---

def test_s3_query_does_not_return_classify(rag):
    results = rag.search("S3 copyObject deleteObject bucket key", n_results=3)
    assert CLASSIFY_FIELDS not in top3_functions(results)

def test_classify_query_does_not_return_s3(rag):
    results = rag.search("JSON parse error from language model", n_results=3)
    assert MOVE_REMOVE_S3 not in top3_functions(results)


# --- Noise floor ---

def test_unrelated_query_low_score(rag):
    results = rag.search("OAuth token refresh middleware", n_results=3, min_score=0.0)
    if results:
        assert results[0]["score"] < 0.55, f"Noise floor too high: {results[0]['score']}"

Baseline and regression use

Run this before and after every change to chunking, indexed text, or retrieval parameters:

pytest tests/test_code_rag_recall.py -v

test_s3_exact_identifier               PASSED
test_s3_paraphrase                     PASSED
test_s3_root_cause                     PASSED
test_classify_exact_identifier         PASSED
test_classify_paraphrase               PASSED
test_classify_root_cause               PASSED
test_s3_query_does_not_return_classify PASSED
test_classify_query_does_not_return_s3 PASSED
test_unrelated_query_low_score         PASSED

9 passed in 4.1s

A drop from 9/9 to 8/9 after a change tells you exactly which query angle broke. test_s3_root_cause is the most valuable — if it’s the one that keeps failing, the vocabulary enrichment from Part 4 needs more coverage of the error vocabulary, not just the function description.

The suite doesn’t need to be large. Nine test cases covering two functions and a noise floor check is enough to catch most regressions. Add cases as new functions become important to retrieval quality.

---

Online: production tracing

The eval harness tells you retrieval was working when you shipped. It can’t tell you what’s happening to a specific production query right now. That requires instrumentation.

The invisible failure

Production RAG systems fail in ways that look like LLM problems but are actually retrieval problems. The answer is confidently wrong not because the model hallucinated — it faithfully reasoned over the wrong context. It did exactly what it was supposed to do with what it was given. The model is not the bug.

Without end-to-end tracing, you cannot distinguish these two failure modes:

User:    "How do I reset my API key?"
System:  "You can reset your API key by navigating to Settings → Security
          and clicking Regenerate. Note that all existing integrations using
          the old key will stop working immediately."

[The settings page was redesigned 3 months ago. The old path no longer exists.
 The system returned a chunk from a deprecated help doc. The LLM did not hallucinate —
 it correctly summarized text that is now wrong.]

From the outside, this looks like a hallucination. With retrieval tracing, you open the trace, see that the top chunk came from docs/v1/api-keys.md last indexed 4 months ago, and the diagnosis takes 30 seconds.

The standard observability stack — traces, metrics, logs via OpenTelemetry — applies here, but a RAG pipeline has primitives that OTel’s generic span model does not capture natively. Similarity scores, chunk provenance, index versions, and reranking decisions need to be instrumented explicitly.

---

The span architecture

A complete RAG request should produce a trace with these spans nested under a single root:

rag_request (root)
  ├── embedding.query          (latency, model, input tokens)
  ├── retrieval.vector_search  (latency, num_results, top_k, filter applied)
  ├── retrieval.rerank         (latency, num_input, num_output, model)
  ├── prompt.assembly          (latency, total_tokens, num_chunks_used)
  └── llm.generate             (latency, model, input_tokens, output_tokens, stop_reason)

Each span carries structured attributes. The spans themselves are cheap to produce — you are adding metadata to operations you are already running. The cost is zero. The debugging value is high.

from opentelemetry import trace

tracer = trace.get_tracer("rag_service")

async def query(self, question: str) -> RAGResponse:
    with tracer.start_as_current_span("rag_request") as root:
        root.set_attribute("query", question)
        root.set_attribute("index.collection", self._collection_name)
        root.set_attribute("index.version", self._index_version)

        with tracer.start_as_current_span("embedding.query") as span:
            query_vector = await self._embed([question])
            span.set_attribute("embedding.model", EMBEDDING_MODEL)
            span.set_attribute("embedding.input_tokens", token_count(question))

        with tracer.start_as_current_span("retrieval.vector_search") as span:
            raw_results = self._collection.query(query_vector, n_results=TOP_K)
            span.set_attribute("retrieval.top_k", TOP_K)
            span.set_attribute("retrieval.num_results", len(raw_results))
            span.set_attribute("retrieval.min_score", self._min_score)
            for i, r in enumerate(raw_results):
                span.add_event("chunk_retrieved", {
                    "rank": i,
                    "score": r.score,
                    "doc_id": r.metadata["file_path"],
                    "start_line": r.metadata["start_line"],
                    "end_line": r.metadata["end_line"],
                    "chunk_id": r.metadata["chunk_id"],
                })

        with tracer.start_as_current_span("retrieval.rerank") as span:
            reranked = self._rerank(question, raw_results)
            span.set_attribute("rerank.num_input", len(raw_results))
            span.set_attribute("rerank.num_output", len(reranked))

        with tracer.start_as_current_span("prompt.assembly") as span:
            prompt = self._build_prompt(question, reranked)
            span.set_attribute("prompt.total_tokens", token_count(prompt))
            span.set_attribute("prompt.num_chunks", len(reranked))

        with tracer.start_as_current_span("llm.generate") as span:
            answer = await self._llm.generate(prompt)
            span.set_attribute("llm.model", LLM_MODEL)
            span.set_attribute("llm.input_tokens", answer.usage.input_tokens)
            span.set_attribute("llm.output_tokens", answer.usage.output_tokens)
            span.set_attribute("llm.stop_reason", answer.stop_reason)

        return RAGResponse(answer=answer.text, chunks=reranked)

The chunk_retrieved events inside retrieval.vector_search are what make bad answers debuggable. When you investigate a support ticket about a wrong answer, you open the trace, expand the retrieval span, and immediately see which chunks scored highest and where they came from. “The system retrieved three chunks from the deprecated v1 policy document” is an actionable finding. “The system returned a bad answer” is not.

---

Logging the “why”

A common question in production is not just “what was retrieved?” but “why did the system think this was relevant?” The similarity score alone does not answer this. A score of 0.82 might be genuinely relevant, or it might be a false positive from an embedding space where the query and an unrelated chunk happen to land nearby.

For the AllInterviews codebase (Part 2), we saw this directly: the query "classifyFields" returned a TypeError incident as rank 1 with score 0.235, not because it was about classifyFields, but because the embedding space placed error classification semantics close to field classification semantics. The score gave no indication the result was a false positive.

To address this, add a lightweight rationale step after reranking:

RATIONALE_PROMPT = """\
You will receive a search query and up to 5 retrieved context chunks.
For each chunk, explain in one sentence why it is or is not relevant to the query.
Be brief. This output is for debugging, not for the user.

Query: {query}

Chunks:
{chunks}
"""

async def _log_retrieval_rationale(self, query: str, chunks: list[Chunk], trace_id: str) -> None:
    prompt = RATIONALE_PROMPT.format(
        query=query,
        chunks="\n\n".join(f"[{i+1}] {c.document[:300]}" for i, c in enumerate(chunks))
    )
    rationale = await self._llm.generate(prompt, model="claude-haiku-4-5")
    logger.info(
        "retrieval_rationale",
        extra={"trace_id": trace_id, "query": query, "rationale": rationale.text}
    )

This is expensive if run per-request. Run it on a sampled basis instead — 1% of production traffic, plus 100% of user-flagged responses. The rationale becomes a structured log field queryable in your observability platform. Over time, patterns in the rationale (“chunk came from file X but query was about Y”) point directly at indexing or chunking problems.

---

Retrieval quality vs answer quality

The highest-value observability investment is closing the feedback loop: connecting what was retrieved to how good the final answer was.

Use a lightweight LLM-as-judge approach. After the main LLM generates an answer, send the answer, the retrieved context, and the original question to a smaller, cheaper model with a scoring rubric:

JUDGE_PROMPT = """\
You are evaluating a RAG system response. Score on two dimensions.

Question: {question}

Retrieved context:
{context}

System answer: {answer}

Score each dimension from 0.0 to 1.0:

faithfulness: Did the answer stay within what the retrieved context says?
  1.0 = every claim in the answer is supported by the context
  0.0 = the answer contradicts or ignores the context entirely

relevance: Did the answer address the question?
  1.0 = fully answers the question
  0.0 = does not address the question at all

Respond with JSON: {{"faithfulness": float, "relevance": float, "notes": "one-line explanation"}}
"""

async def _evaluate_response(
    self,
    question: str,
    chunks: list[Chunk],
    answer: str,
    trace_id: str,
) -> None:
    context = "\n\n".join(c.document for c in chunks)
    prompt = JUDGE_PROMPT.format(question=question, context=context, answer=answer)
    result = await self._llm.generate(prompt, model="claude-haiku-4-5")

    try:
        scores = json.loads(result.text)
        span = trace.get_current_span()
        span.set_attribute("eval.faithfulness", scores["faithfulness"])
        span.set_attribute("eval.relevance", scores["relevance"])
        logger.info("rag_eval", extra={
            "trace_id": trace_id,
            "faithfulness": scores["faithfulness"],
            "relevance": scores["relevance"],
            "notes": scores.get("notes"),
        })
    except (json.JSONDecodeError, KeyError):
        pass

This gives you a queryable dataset: “show me all requests where faithfulness score was below 0.7 in the last 7 days.”

Drilling into those traces, you will typically find one of three patterns:

Pattern	What you see in the trace	Root cause
Wrong document	Top chunks from stale or deleted file	Index drift or missing deletion
Right document, wrong section	Top chunks from adjacent function / heading	Chunking boundary problem
Correct chunks, wrong answer	High faithfulness but low relevance	Generation problem — not retrieval

Only traces with chunk-level attribution let you distinguish these cases. Without them, every bad answer looks identical from the outside.

---

Index version attribution

One failure mode deserves special mention: your index was updated, retrieval behavior changed, and answer quality dropped. Without index version attribution in your traces, you cannot correlate the quality drop to the update.

This is not a hypothetical. Every time you swap the alias to a new collection, change the chunking strategy, or upgrade the embedding model — retrieval behavior changes. Sometimes it gets better. Sometimes it regresses in ways you won’t notice until users complain.

The fix is to attach the index version to every retrieval span:

span.set_attribute("retrieval.index_version", self._index_version)
span.set_attribute("retrieval.index_collection", self._collection_name)
span.set_attribute("retrieval.index_updated_at", self._index_metadata["updated_at"])

In the alias-swap flow from Part 8, _index_version is the version string set when you call swap_index(new_version). Every request that hit the old collection and every request that hit the new collection is now labeled.

# Pseudocode — Langfuse / Datadog / Honeycomb query
traces
  .filter(eval.faithfulness < 0.7)
  .filter(timestamp > "2026-05-14T15:00:00Z")  # when the swap happened
  .group_by(retrieval.index_version)
  .count()

# If v1732456789 shows 3x more low-quality results than v1732456000,
# the swap introduced a regression. Roll back immediately.

This sounds obvious in retrospect. Almost nobody instruments it until they spend a painful post-incident trying to figure out why answer quality degraded on a Tuesday afternoon.

---

The Langfuse integration

For the agent-platform, Langfuse already traces every LLM call through the @trace_agent decorator in app/services/tracing.py. RAG retrieval sits outside the agent loop, so it doesn’t get traced automatically.

The integration point is RAGService.search() in app/services/rag.py. The search call happens inside FixGenerationAgent and IncidentResponseAgent before the LLM call — it’s the first operation of every agent run. Adding a retrieval span there connects the chunk provenance to the downstream LLM generation in a single trace.

# In RAGService.search():
from langfuse.decorators import observe

@observe(name="retrieval.vector_search")
async def search(self, query: str, n_results: int = 5, min_score: float = 0.0) -> list[SearchResult]:
    # ... existing implementation ...
    langfuse_context.update_current_observation(
        metadata={
            "collection": self._collection._name,
            "index_version": self._index_version,
            "num_results_before_filter": len(raw_results),
            "num_results_after_filter": len(results),
            "top_score": results[0].score if results else None,
        }
    )
    return results

Now every agent trace in Langfuse shows the retrieval step: what query was embedded, which chunks were returned with what scores, and which version of the index they came from.

---

Metrics to track

Beyond per-request traces, track these aggregates:

Metric	What it catches
retrieval.top_score p50/p95	Index drift — scores drop when chunking/model diverges from queries
retrieval.num_results_after_filter	min_score calibration — if frequently 0, threshold is too high
eval.faithfulness p50	Answer quality trend — drops after index rebuilds or model upgrades
retrieval.latency p99	Vector store performance under load
embedding.query.latency p99	OpenAI API latency — spikes propagate directly to user-facing latency

Alert on eval.faithfulness p50 dropping below your acceptable floor — for code search, 0.7 is a reasonable threshold. A drop below it means something in the pipeline changed that you need to investigate.

---

This post completes the ten-part Code RAG series. Starting from a line-count chunker that split functions across arbitrary windows, we built: structural diagnosis of two failure modes, function-boundary extraction, LLM-generated descriptions, hybrid search, cross-encoder re-ranking, a document registry, three-way decision logic for re-chunking vs re-embedding vs re-indexing, alias-based zero-downtime rebuilds, and end-to-end testing and observability. Every change was measured against real code from a production codebase.