Logging retrieved chunks RAG pipelines is essential for debugging retrieval failures, yet most teams either log nothing or dump entire documents into append-only files. The fix is to treat chunk metadata as first-class structured data and push bulky text to a separate store keyed by content hash. This how-to walks through a concrete setup you can ship in an afternoon.
Step 1: Define a structured schema for chunk metadata
When logging retrieved chunks RAG metadata, resist the urge to include the embedding or the full text. Capture enough to reason about relevance: chunk ID, parent document, similarity score, rank, length, a short snippet, and a content hash. The hash is the bridge to the full text stored elsewhere.
from dataclasses import dataclass
import hashlib
@dataclass
class RetrievedChunk:
chunk_id: str
doc_id: str
score: float
rank: int
char_len: int
snippet: str
content_hash: str
def make_chunk(record: dict, rank: int) -> RetrievedChunk:
text = record["text"]
h = hashlib.sha256(text.encode()).hexdigest()[:16]
return RetrievedChunk(
chunk_id=record["id"],
doc_id=record["doc_id"],
score=record["score"],
rank=rank,
char_len=len(text),
snippet=text[:80],
content_hash=h,
)The snippet should be a deterministic prefix, not a random window. Determinism lets you spot the same chunk across traces without pulling the full body.
Step 2: Configure a dedicated logger with sampling
A raw retrieval logger at 500 QPS will generate gigabytes daily if each entry carries even a 200-character snippet. Use a dedicated JSON logger and a sampling filter that keeps all errors but samples routine hits.
import logging
import json
import random
from logging import LogRecord
class SamplingFilter:
def __init__(self, rate: float):
self.rate = rate
def filter(self, record: LogRecord) -> bool:
if record.levelno >= logging.ERROR:
return True
return random.random() < self.rate
logger = logging.getLogger("rag.retrieval")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(message)s'))
logger.addFilter(SamplingFilter(0.1)) # keep ~10% of info logs
logger.addHandler(handler)
def log_chunk(chunk: RetrievedChunk, trace_id: str):
payload = {
"trace_id": trace_id,
"event": "chunk_retrieved",
"chunk": chunk.__dict__,
}
logger.info(json.dumps(payload))Sampling is not optional for production RAG. You can always reconstruct the missing 90% on demand from the sidecar store using the hash. Debugging is an interactive process; passive log hoarding is not.
Step 3: Integrate retrieval logging into your RAG function
Centralize the pattern for logging retrieved chunks RAG calls inside the retrieval wrapper. Never scatter print statements across your vector store client and reranker.
def retrieve(query_vec, top_k=5, trace_id="") -> list[RetrievedChunk]:
results = vector_db.search(query_vec, top_k)
chunks = []
for i, r in enumerate(results):
c = make_chunk(r, i)
chunks.append(c)
log_chunk(c, trace_id)
store_full_text(r["text"]) # see Step 4
return chunksIf you use a reranker, log the post-rerank rank separately. Add a rerank_score field to the dataclass rather than overwriting the vector score. The two numbers explain why a low-similarity chunk surfaced to the prompt.
Substep: propagate trace context
Use contextvars so the trace ID flows from the HTTP request into the retrieval call without threading it through every signature.
import contextvars
trace_id_var = contextvars.ContextVar("trace_id")
def retrieve_with_ctx(query_vec, top_k=5):
tid = trace_id_var.get("no-trace")
return retrieve(query_vec, top_k, tid)Step 4: Offload full chunk text to a sidecar store
The sidecar is a key-value store indexed by content hash. SQLite is enough for most single-node services; for distributed systems use an object store bucket with the hash as key. Write-only-if-absent avoids duplication.
import sqlite3, hashlib
conn = sqlite3.connect("chunk_store.db")
conn.execute("CREATE TABLE IF NOT EXISTS chunks (hash TEXT PRIMARY KEY, text TEXT)")
def store_full_text(text: str) -> str:
h = hashlib.sha256(text.encode()).hexdigest()[:16]
conn.execute("INSERT OR IGNORE INTO chunks VALUES (?,?)", (h, text))
conn.commit()
return hCall store_full_text inside make_chunk or right after retrieval. The log never contains text, only content_hash. This single decision cuts log volume by 95% in typical document RAG where chunks average 500–1500 characters.
Step 5: Correlate logs with request and model calls
A retrieval log孤立 from the generation step is half-useful. Attach the same trace_id to your LLM inference call. If you route through a gateway, pass the trace ID in a header. The retrieval chunks and the final answer now share a key for post-hoc analysis.
# example with OpenAI-compatible client
import openai
def generate(prompt, trace_id):
return openai.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
extra_headers={"x-trace-id": trace_id},
)When logging retrieved chunks RAG pipelines, the trace ID is the only join key you need. Do not log the prompt body in the retrieval logger; keep concerns separated.
Step 6: Build a debug CLI to reconstruct logged chunks
A debug tool completes the logging retrieved chunks RAG loop. Write a small script that tails a trace ID and pulls full text from the sidecar.
import json, sqlite3, sys
conn = sqlite3.connect("chunk_store.db")
def debug(trace_id, log_file="rag.log"):
with open(log_file) as f:
for line in f:
rec = json.loads(line)
if rec.get("trace_id") != trace_id:
continue
c = rec["chunk"]
row = conn.execute("SELECT text FROM chunks WHERE hash=?", (c["content_hash"],)).fetchone()
print(f"Rank {c['rank']} | score {c['score']:.3f} | {c['char_len']} chars")
print(row[0][:300])
print("-" * 40)
if __name__ == "__main__":
debug(sys.argv[1])Run it with python debug_rag.py trace-abc123. You get the exact text the model saw, without ever having stored it in the log pipeline.
Step 7: Verify success and measure bloat
Verification is concrete. After deploying the above:
- Issue 1,000 synthetic queries against a staging index.
- Measure log growth:
wc -c rag.log. With 10% sampling and ~120-byte JSON per entry, expect ~120 KB, not MB. - Confirm zero full-text leakage:
grep -c '"text"' rag.logshould be0. - Pick a trace ID, run the debug CLI, and confirm the printed snippet matches the live chunk from your vector store.
If those hold, your logging retrieved chunks RAG setup is correct. You can rotate the sidecar weekly; the logs alone are sufficient for triage of score distribution and rank shifts.
Operational notes
- Set a retention policy on the sidecar separate from logs. Chunks go stale when you re-embed; expire by hash prefix or timestamp.
- Alert on
event="chunk_retrieved"withscore < 0.2at ERROR level (override sampling) to catch index drift. - Never log PII-containing snippets. If a chunk might contain email addresses, redact the snippet in
make_chunkbefore it enters the logger.
The discipline here is separation: metadata in logs, content in a fetch-on-demand store. That is how you keep RAG observable without drowning in your own context windows.