A RAG pipeline connects document search to a model answer. It searches for passages, combines selected text with the question, asks the model to answer from that context and turns returned source IDs into links.
When you need this. An answer is wrong or missing, and you need to find which part of that sequence went wrong. Did search miss the useful passage? Did the request include it? Did the model use it correctly? The final answer alone cannot tell you.
This lab adds a request trace, a record of those operations, to the application from How RAG works. We will follow the existing Python documentation question through each operation. Retrieval and context assembly execute locally; saved model replies are replayed only for their matching requests. No new answer is generated in this activity.
Inspect the ranked passages before the answer
A request trace is an ordered record of operations and their inputs or results. Here it is a local event list, not a distributed tracing system. Each event gets the next sequence number when that operation finishes or a branch is taken.
The lab reuses build_index, prompt, Reply and check_reply from the first lesson. InMemoryVectorStore.similarity_search_with_score retrieves two passages using the saved real embedding vectors. The trace records their rank, score, text, source ID and link.
[
{
"rank": 1,
"source_id": "path-read",
"score": 0.6381110754937956
},
{
"rank": 2,
"source_id": "path-write",
"score": 0.6157764004196788
}
]The read-text passage ranks above the write-text passage for this question. Both are selected, and the saved answer names the write-text source. A similarity score describes this vector comparison; it is not a probability that a passage answers the question. This small example shows why inspecting both ranking and the final source matters. No reranker runs here.
A source ID names an original document. A chunk ID names a passage using its source ID and a hash of its text and URL. The trace keeps both, so a source link and the exact selected text do not get confused. Later ingestion work can preserve finer source locations; this corpus has short original API summaries.
See exactly what context enters the request
Context assembly creates the message content sent toward generation. The reused prompt function serializes the question and selected passage texts with their source IDs. It adds a system instruction to answer from those passages or say the information is missing. The trace saves the complete application-level message list before calling the adapter.
That instruction is a rule for the model, not an access-control system. The code also checks the returned structure and source membership. These checks do not decide whether every sentence follows from the passages.
Empty context. A second run uses a genuinely empty InMemoryVectorStore. Retrieval returns no passages, so the application emits model_skipped and constructs a not_in_docs result with no source links. The test verifies that the adapter was not called. Add documents or inspect the retrieval input before spending a model request on an empty context.
An unsupported question is different from an empty retrieval result. The backup-retention question retrieves Python passages, but they do not state a service policy. Its saved model reply says not_in_docs and has no source links. The trace includes a returned replayed reply in that case. A nonempty result list does not show that the question is answerable.
Replay only the request that produced the saved reply
The first lesson replayed replies in question order. An iterator is enough to display that fixed recording, but it could return the wrong saved reply after a question or prompt changes. This lesson replaces that shortcut with BoundReplay.
A hash is a compact fingerprint used here to compare serialized inputs. The adapter pairs each saved question with its complete saved message list and reply. It computes a content hash to find the entry, then checks the exact messages and question before returning a copy. Repeating or reordering the supported questions returns their matching replies. Changing the policy, passage text or context order refuses replay.
Changed replay prompt. The recorded controlled probe appends a policy sentence before asking the adapter for a reply. It raises ReplayRequestMismatch. Restore the original request to replay it, or make and save a new model run for the changed request. Do not present an old answer as output from a prompt that was never executed.
Replay matches the application-level messages. Provider serialization and backend options are outside that comparison.
"""Inspect actual replay transitions, without inventing model spans or timings."""
from __future__ import annotations
import copy
import hashlib
import json
from pathlib import Path
from .concepts import ROOT, Reply, SavedEmbeddings, build_index, check_reply, chunks, prompt
from langchain_core.vectorstores import InMemoryVectorStore
def digest(value):
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest()
def chunk_id(doc):
return doc.metadata["id"] + ":" + digest({"text": doc.page_content, "url": doc.metadata["url"]})[:16]
class ReplayRequestMismatch(ValueError):
pass
class BoundReplay:
"""Return a saved reply only for its exact saved application-level request."""
mode = "exact-request offline replay"
def __init__(self, saved):
self.bindings = {}
questions, requests, replies = (saved[key] for key in ("questions", "requests", "model_replies"))
if len(questions) != len(requests) or len(requests) != len(replies):
raise ValueError("Every saved request needs one question and reply.")
for question, messages, reply in zip(questions, requests, replies, strict=True):
wire = json.loads(json.dumps(messages))
if not wire or wire[-1][0] != "human" or json.loads(wire[-1][1]).get("question") != question:
raise ValueError("The saved question does not match its request.")
key = digest(wire)
if key in self.bindings:
raise ValueError("Duplicate saved request binding.")
self.bindings[key] = (question, wire, copy.deepcopy(reply))
def generate(self, question, messages):
wire = json.loads(json.dumps(messages))
key = digest(wire)
found = self.bindings.get(key)
if found is None or found[0] != question or found[1] != wire:
raise ReplayRequestMismatch("No reply is recorded for this exact question and prompt.")
return copy.deepcopy(found[2])
def load_runtime(*, empty_index=False):
record_file = ROOT / "runs/2026-10-22-concept/recorded-run.json"
corpus_file = ROOT / "data/docs.json"
raw = record_file.read_bytes()
saved = json.loads(raw)
corpus_sha = hashlib.sha256(corpus_file.read_bytes()).hexdigest()
if corpus_sha != saved["corpus_sha256"]:
raise ValueError("The corpus differs from the saved embeddings run.")
prepared = [] if empty_index else chunks()
descriptor = {"corpus_sha256": corpus_sha, "embedding_model": saved["models"]["qwen3-embedding:0.6b"],
"saved_vectors_sha256": digest({doc.page_content: saved["embeddings"]["documents"][doc.page_content] for doc in prepared}),
"splitter": {"name": "RecursiveCharacterTextSplitter", "chunk_size": 280, "chunk_overlap": 40},
"chunk_ids": [chunk_id(doc) for doc in prepared]}
identity = {"saved_run_sha256": hashlib.sha256(raw).hexdigest(),
"corpus_sha256": corpus_sha, "logical_index_id": digest(descriptor),
"index_descriptor": descriptor, "model_digests": saved["models"]}
index = InMemoryVectorStore(SavedEmbeddings(saved["embeddings"])) if empty_index else build_index(saved["embeddings"])
return saved, index, BoundReplay(saved), identity
def trace_request(question, index, adapter, identity, *, events=None):
events = [] if events is None else events
if events:
raise ValueError("Start each request with an empty event list.")
def emit(kind, **data):
events.append({"sequence": len(events) + 1, "kind": kind, **data})
if not isinstance(question, str) or not question.strip():
emit("request_rejected", reason="blank_question")
raise ValueError("Ask a nonblank question.")
request_key = digest({"question": question, "index": identity["logical_index_id"]})
emit("request_accepted", question=question, request_key=request_key, identity=copy.deepcopy(identity))
# This is the real maintained vector-store call, using saved real embeddings.
try:
ranked = index.similarity_search_with_score(question, k=2)
except KeyError as error:
emit("retrieval_refused", reason="unrecorded_query_embedding", error=str(error))
raise ValueError("No query embedding is saved for this exact question.") from error
candidates = [{"rank": rank, "chunk_id": chunk_id(doc), "source_id": doc.metadata["id"],
"score": float(score), "text": doc.page_content, "url": doc.metadata["url"]}
for rank, (doc, score) in enumerate(ranked, start=1)]
emit("retrieval_completed", candidates=candidates, score_kind="InMemoryVectorStore cosine similarity")
selected = [doc for doc, _ in ranked]
emit("context_selected", chunk_ids=[chunk_id(doc) for doc in selected],
source_ids=[doc.metadata["id"] for doc in selected])
if not selected:
emit("model_skipped", reason="empty_context")
reply = Reply(status="not_in_docs", answer="No documentation was found.", source_ids=[])
else:
messages = prompt(question, selected)
emit("model_request_ready", request_sha256=digest(messages), messages=messages,
schema=Reply.model_json_schema(), adapter_mode=adapter.mode)
try:
raw_reply = adapter.generate(question, messages)
except ReplayRequestMismatch as error:
emit("replay_refused", error=str(error))
raise
emit("model_reply_returned", reply=copy.deepcopy(raw_reply), mode=adapter.mode)
try:
reply = Reply.model_validate(raw_reply)
except ValueError as error:
emit("reply_rejected", reason="invalid_shape", error=str(error))
raise
try:
check_reply(reply, selected)
except ValueError as error:
emit("reply_rejected", reason="source_contract", error=str(error))
raise
emit("reply_checked", status=reply.status, source_ids=reply.source_ids,
checks=["reply_shape", "source_membership", "status_source_consistency"])
lookup = {doc.metadata["id"]: doc.metadata["url"] for doc in selected}
result = {"question": question, "context_ids": [doc.metadata["id"] for doc in selected],
**reply.model_dump(), "source_links": [lookup[source] for source in reply.source_ids]}
emit("source_links_resolved", source_links=result["source_links"])
emit("request_completed", status=reply.status)
return {"request_key": request_key, "result": result, "events": events}The logical index identifier hashes its corpus, saved document vectors, splitter configuration and ordered chunk IDs. It is a reproducible description of this local replay index, not a remote deployment version. The trace also stores the complete saved-run hash. A changed corpus refuses loading against the old embeddings record instead of silently mixing versions.
request_key combines the question and logical index identifier. Replaying the same question against the same index repeats that key. Sequence numbers restart for each request. Neither field claims to be a globally unique invocation ID.
Check source IDs before resolving links
The reused Pydantic Reply schema accepts answered or not_in_docs, answer text and a list of source IDs. check_reply requires sources for an answered result, rejects IDs outside the selected context and rejects source links attached to a not_in_docs reply. Only after those checks does the application map IDs to selected source URLs.
Source outside selected context. A controlled reply mutation substitutes json-text. That source exists in the corpus but was not selected for this request. The application emits reply_rejected before resolving any links. Compare the reply IDs with this request’s context, rather than allowing every known corpus ID. The mutation is deliberately supplied by the driver; it is not attributed to the saved model run.
A linked answer can still be false. One test supplies an incorrect statement about appending to a file with a valid selected source ID. The source-membership check accepts the link. Checking whether a source supports the wording needs a different check and suitable questions; membership alone cannot do it.
Execute the pipeline and inspect a changed request
Download the complete lab. It includes the pilot’s unchanged corpus, saved vectors and original model requests and replies. The new driver executes retrieval and trace logic against those records. Its saved query vectors and replies support the two recorded questions. An unrecorded question is refused; this is not an arbitrary-question generation service. It needs no model server or weight download.
python -m venv .venv
.venv/bin/pip install -r requirements-lock.txt
export PYTHONPATH=src
.venv/bin/python trace_run.py
.venv/bin/python -m unittest discover -s tests -vThe local run returned the saved UTF-8 answer, returned not_in_docs for the backup question and skipped the adapter for empty context. Retrieval and checks execute locally; generation is exact-request replay.
actual local retrieval and request trace; saved model replies replayed, no fresh inference
Question: How do I save UTF-8 text with pathlib?
Status: answered
Events: request_accepted -> retrieval_completed -> context_selected -> model_request_ready -> model_reply_returned -> reply_checked -> source_links_resolved -> request_completed
Sources: https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.write_text
Question: How many days does this service keep backups?
Status: not_in_docs
Events: request_accepted -> retrieval_completed -> context_selected -> model_request_ready -> model_reply_returned -> reply_checked -> source_links_resolved -> request_completed
Sources: (none)
Empty context: model_skipped
Changed replay prompt: ReplayRequestMismatch
Source outside selected context: ValueErrorTry changing one passage in the assembled request before replay. Expect a mismatch, not a newly generated answer. Then inspect the empty-index trace: it should contain model_skipped and omit model_request_ready. These are different branches even though both can end with not_in_docs.
The executed suite passed 23 checks. It verifies exact replay binding, actual ranking and selected text, branch order, source handling and identity. 5 isolated source changes were rejected by named tests, including accepting a changed prompt and calling the model adapter on empty context. Tests do not turn this small example into a quality benchmark.
test_bad_reply_shape_stops_before_link_resolution (test_request_trace.RequestTraceTests.test_bad_reply_shape_stops_before_link_resolution) ... ok
test_blank_question_stops_before_search_or_model (test_request_trace.RequestTraceTests.test_blank_question_stops_before_search_or_model) ... ok
test_changed_context_cannot_receive_a_saved_reply (test_request_trace.RequestTraceTests.test_changed_context_cannot_receive_a_saved_reply) ... ok
test_changed_context_order_is_a_different_replay_request (test_request_trace.RequestTraceTests.test_changed_context_order_is_a_different_replay_request) ... ok
test_changed_policy_cannot_receive_a_saved_reply (test_request_trace.RequestTraceTests.test_changed_policy_cannot_receive_a_saved_reply) ... ok
test_changed_question_cannot_receive_a_saved_reply (test_request_trace.RequestTraceTests.test_changed_question_cannot_receive_a_saved_reply) ... ok
test_corpus_or_record_change_changes_the_identity_or_refuses_loading (test_request_trace.RequestTraceTests.test_corpus_or_record_change_changes_the_identity_or_refuses_loading) ... ok
test_duplicate_saved_request_is_refused (test_request_trace.RequestTraceTests.test_duplicate_saved_request_is_refused) ... ok
test_each_replayed_result_matches_the_preserved_original_run (test_request_trace.RequestTraceTests.test_each_replayed_result_matches_the_preserved_original_run) ... ok
test_emitted_order_and_request_match_the_operations_that_ran (test_request_trace.RequestTraceTests.test_emitted_order_and_request_match_the_operations_that_ran) ... ok
test_link_membership_does_not_check_answer_truth (test_request_trace.RequestTraceTests.test_link_membership_does_not_check_answer_truth) ... ok
test_mismatched_replay_emits_refusal_without_returned_reply (test_request_trace.RequestTraceTests.test_mismatched_replay_emits_refusal_without_returned_reply) ... ok
test_missing_reply_binding_is_refused (test_request_trace.RequestTraceTests.test_missing_reply_binding_is_refused) ... ok
test_not_in_docs_cannot_attach_a_retrieved_but_unrelated_source (test_request_trace.RequestTraceTests.test_not_in_docs_cannot_attach_a_retrieved_but_unrelated_source) ... ok
test_real_empty_index_skips_the_model_and_has_a_distinct_identity (test_request_trace.RequestTraceTests.test_real_empty_index_skips_the_model_and_has_a_distinct_identity) ... ok
test_real_ranked_candidates_match_the_actual_index_scores (test_request_trace.RequestTraceTests.test_real_ranked_candidates_match_the_actual_index_scores) ... ok
test_replay_is_bound_to_question_not_iteration_order (test_request_trace.RequestTraceTests.test_replay_is_bound_to_question_not_iteration_order) ... ok
test_request_key_varies_by_question_and_logical_index (test_request_trace.RequestTraceTests.test_request_key_varies_by_question_and_logical_index) ... ok
test_saved_binding_is_not_changed_by_the_callers_objects (test_request_trace.RequestTraceTests.test_saved_binding_is_not_changed_by_the_callers_objects) ... ok
test_saved_question_must_match_saved_prompt (test_request_trace.RequestTraceTests.test_saved_question_must_match_saved_prompt) ... ok
test_source_outside_selected_context_is_rejected_even_when_in_corpus (test_request_trace.RequestTraceTests.test_source_outside_selected_context_is_rejected_even_when_in_corpus) ... ok
test_unrecorded_question_is_refused_without_fabricating_an_embedding (test_request_trace.RequestTraceTests.test_unrecorded_question_is_refused_without_fabricating_an_embedding) ... ok
test_unsupported_question_still_retrieves_but_returns_no_links (test_request_trace.RequestTraceTests.test_unsupported_question_still_retrieves_but_returns_no_links) ... ok
----------------------------------------------------------------------
Ran 23 tests in 0.009s
OKRead the explanation
This adapter checks the complete saved message list and question. Matching does not claim identical provider transport settings or a fresh generation.
What’s next
Continue with Build a RAG Eval Dataset.
References
The implementation was checked against installed library signatures and the official LangChain in-memory vector store, retrieval guidance and LangChain Core package metadata on 22 October 2026. The preserved local run used Qwen3-Embedding-0.6B and Gemma 4 E2B, both with Apache-2.0 model-source licenses. Original application code is MIT; no model weights are redistributed.