rag-timetravel

Time-travel debugging for the RAG retrieval layer, built on LanceDB's native dataset versioning.

Every retrieval and generation step is recorded as an immutable event, and the vector index is versioned automatically. You can take any past query, reconstruct the exact index version it ran against, re-run retrieval, and diff what was retrieved then versus now. The defensible, deterministic core is the retrieval diff; answer re-generation is a best-effort layer on top (see Determinism).

---

The problem

When a RAG answer regresses ("why did this get worse than last month?"), the hard question is what changed: the retrieved chunks, the index state at that time, or the generator. Tracing/eval tools (Langfuse, Arize Phoenix, LangSmith, TruLens) record traces and let you compare runs, but they treat the vector index as opaque: they cannot reconstruct the index as it existed at an arbitrary past moment and re-run retrieval against it.

rag-timetravel fills that specific gap. Because it owns ingestion, it can pin retrieval to a historical LanceDB version and show you exactly which chunks a query would have surfaced at any point in its history, with no full-index copies.

What is and isn't deterministic

• Retrieval replay is faithful. Pinning to a snapshot's LanceDB version reproduces the exact chunk set and scores that existed at that time. This is the core guarantee, and it is tested against a real index.
• Generation replay is best-effort. Remote models drift and sample non-deterministically, so a replayed answer reflects today's generator over the historical retrieval, not a byte-for-byte reproduction of the old answer. Treat the replayed answer as informational; trust the retrieval diff.

---

How it works

┌─────────────────────────────────────────────────────────────────┐
│                        RAGPipeline                              │
│                                                                 │
│  ingest(doc) ──► chunk ──► embed ──► LanceDB ──► emit event    │
│                                                                 │
│  query(text) ──► embed ──► search ──► LLM ──► emit 5 events    │
│                                         │                       │
│                                         ▼                       │
│                                   EventStore (SQLite)           │
└─────────────────────────────────────────────────────────────────┘
          │
          │  (at any later time)
          ▼
┌─────────────────────────────────────────────────────────────────┐
│                      ReplayEngine                               │
│                                                                 │
│  1. Load QueryReceived event → get original timestamp T         │
│  2. Find IndexSnapshot with max(ts) where ts <= T               │
│  3. table.checkout(snapshot.lancedb_version) on a fresh handle  │
│     (read-only view, O(1), no data copy)                        │
│  4. Re-run retrieval + generation                               │
│  5. Return ReplayResult                                         │
└─────────────────────────────────────────────────────────────────┘
          │
          ▼
┌─────────────────────────────────────────────────────────────────┐
│                       Comparator                                │
│                                                                 │
│  compare(left_id, right_id) → DiffReport                        │
│    - chunks added / removed / score-delta                       │
│    - answer similarity (embedding cosine + token Jaccard)       │
│    - latency delta                                              │
└─────────────────────────────────────────────────────────────────┘

Why LanceDB?

LanceDB's versioned storage creates a new dataset version on every write automatically. A "snapshot" in rag-timetravel is just a metadata record pointing to a specific LanceDB version integer. Checking out a past version is O(1): no data is copied. This is the key primitive that makes cheap time-travel possible.

Why SQLite for the event store?

Zero dependencies, file-portable, and fast enough for local workloads (< 1 ms per append). The EventStore interface is narrow enough to swap in a Postgres backend later with no changes to the pipeline.

---

Installation

pip install rag-timetravel

The base install is lightweight. The default PipelineConfig uses a local sentence-transformers embedder, which (because it pulls in torch) ships as an optional extra:

pip install 'rag-timetravel[local]'    # local embeddings + cross-encoder rerankers

If you use an OpenAI-compatible embedder instead (embedder_model="openai/..." or "ollama/..."), the base install is all you need.

Other optional extras: postgres (PostgreSQL event store), mcp (MCP server), otel (live OpenTelemetry span export; the OTLP/JSON export path needs no extra). Combine them, e.g. pip install 'rag-timetravel[local,otel]'.

Requirements: Python 3.11+

For local generation (default config): install Ollama and pull a model:

ollama pull gemma3

---

Quickstart

Python API

import asyncio
from rag_timetravel import RAGPipeline, PipelineConfig, ReplayEngine, Comparator

async def main():
    # Create a pipeline
    config = PipelineConfig(model="ollama/gemma3", top_k=4)
    pipeline = await RAGPipeline.create("./my_project", config)

    # Ingest documents
    await pipeline.ingest("policy_v1.txt", open("policy_v1.txt").read())
    snap1 = await pipeline.take_snapshot(label="v1-docs")

    # Query
    result = await pipeline.query("What is the return window?")
    print(result.answer)
    print(result.query_id)  # save this

    # Ingest updated documents
    await pipeline.ingest("policy_v2.txt", open("policy_v2.txt").read())
    await pipeline.take_snapshot(label="v2-docs")

    # Replay the original query against the v1 index
    engine = ReplayEngine(pipeline)
    replay = await engine.replay(result.query_id)
    print(replay.answer)  # answer using only v1 docs

    # Diff v1 vs v2 answers
    result_v2 = await pipeline.query("What is the return window?")
    cmp = Comparator(pipeline)
    report = await cmp.compare(result.query_id, result_v2.query_id)
    print(report.to_text())

asyncio.run(main())

CLI

# Ingest a directory of .txt and .md files
rag-timetravel ingest ./docs --project ./my_project

# Take a snapshot
rag-timetravel snapshot --label "after-v1" --project ./my_project

# Query
rag-timetravel query "What is the refund policy?" --project ./my_project

# Replay a past query
rag-timetravel replay --query-id <uuid> --project ./my_project

# Replay any text against the index at a past timestamp
rag-timetravel replay \
  --text "What is the refund policy?" \
  --as-of "2024-12-01T10:00:00" \
  --project ./my_project

# Diff two queries (add --html report.html to also write an HTML report)
rag-timetravel diff --left <uuid> --right <uuid> --project ./my_project

# Show the event trace for a query
rag-timetravel trace --query-id <uuid> --project ./my_project

# List all snapshots
rag-timetravel snapshots --project ./my_project

# Capture a golden-query suite, then gate CI on retrieval drift
rag-timetravel test capture "What is the refund policy?" --project ./my_project
rag-timetravel test run --project ./my_project --threshold 0.8   # exits non-zero on drift

# Explain why retrieval changed between two query runs (lineage back to ingest)
rag-timetravel lineage --left <uuid> --right <uuid> --project ./my_project

# Diff the chunk sets between two snapshots
rag-timetravel snapshot-diff --from <snap_id> --to <snap_id> --project ./my_project

# Detect embedding drift between two snapshots (catches silent model swaps)
rag-timetravel drift --from <snap_id> --to <snap_id> --threshold 0.9 --project ./my_project

# Compare the same queries across two pipeline configs over one corpus
rag-timetravel compare-configs --corpus ./docs --config-a a.json --config-b b.json -q "What is X?"

# Apply event-log retention (keep N snapshots, drop events older than a cutoff)
rag-timetravel prune --max-snapshots 50 --max-age 30d --project ./my_project

# Start the REST API
rag-timetravel serve --project ./my_project --port 8000

REST API

rag-timetravel serve --project ./my_project

Method	Path	Description
POST	/ingest	Ingest a document
POST	/query	Run a RAG query
GET	/query/{query_id}	Get event trace for a query
POST	/replay	Replay a past query against its original index
POST	/replay/as-of	Replay query text at a historical timestamp
POST	/diff	Diff two query executions (JSON)
GET	/diff/{left}/{right}/html	Rendered HTML diff report
GET	/snapshots	List all snapshots
POST	/snapshot	Take a manual snapshot
GET	/snapshots/{a}/{b}/diff	Chunk-set diff between two snapshots
GET	/lineage/{left}/{right}	Explain why retrieval changed, traced back to ingest
GET	/drift/{a}/{b}	Embedding drift between two snapshots
POST	/regression/run	Run a golden-query suite against the live index
POST	/compare-configs	Compare two configs over a corpus
GET	/events	Query the raw event log
GET	/health	Health check
GET	/ui	Web dashboard

Interactive docs: http://localhost:8000/docs

---

Pipeline features

Beyond replay and diff, rag-timetravel ships six features that turn it from a debugging tool into something you wire into a pipeline. All of them reuse the same primitives: the event log, LanceDB version pinning, and the retrieval diff.

For a complete, step-by-step deployment and usage guide (CI gating, MCP, the REST API, deployment patterns, backup, and troubleshooting), see docs/production.md.

Regression testing (golden-query CI gate)

Capture a set of golden queries with their expected top-k results, then run them against the live index as a CI gate. test capture records the current results as a baseline (matched on sources, which survive re-chunking, plus chunk ids); test run re-runs them and computes recall@k per query. It exits non-zero when any query drifts past --threshold, so you can fail a build on retrieval regressions.

from rag_timetravel.regression import RegressionRunner

runner = RegressionRunner(pipeline)
suite = await runner.capture(["What is the refund policy?"])
suite.save("golden.json")

report = await runner.run(suite, threshold=0.8)
print(report.to_text())
print(report.exit_code)   # 0 pass, 1 drift, 2 usage/IO error

Lineage tracing (retrieval back to ingest)

Replay tells you what changed between two query runs; lineage explains why. LineageTracer.explain diffs the two LanceDB snapshot versions behind the runs, then attributes each chunk that entered or dropped out of the top-k to the document.ingested event for its source, producing statements like "chunk Y dropped from top-k because source S was re-ingested at snapshot Z".

from rag_timetravel.lineage import LineageTracer

tracer = LineageTracer(pipeline)
report = await tracer.explain(left_query_id, right_query_id)
print(report.to_text())

Embedding drift detection

Track the same source document's embeddings across snapshots and surface cosine drift, catching silent regressions when someone swaps the embedder model or re-runs ingestion. Drift is computed on each source's centroid vector (robust to re-chunking); a dimension change between configs is reported as incomparable, and a changed embedder_model is flagged automatically.

from rag_timetravel.drift import EmbeddingDriftDetector

detector = EmbeddingDriftDetector(pipeline)
report = await detector.scan(version_a, version_b, threshold=0.9)
for s in report.sources:
    print(s.source, s.centroid_cosine)

Web dashboard

rag-timetravel serve exposes a server-rendered dashboard at /ui for browsing query event traces, taking snapshots, running the comparator, and bisecting, plus visual snapshot diffs and a drift view backed by the endpoints above. No build step or frontend toolchain is required.

Event-log retention and compaction

The event log is append-only and grows unbounded. A RetentionPolicy lets you keep the N most recent snapshots and drop events past an age cutoff (replay events are compacted first). RetentionScheduler runs the policy in the background on an interval, mirroring the snapshot scheduler.

from rag_timetravel.core.retention import RetentionPolicy, apply_retention

policy = RetentionPolicy(max_snapshots=50, max_event_age_seconds=30 * 86400)
result = await apply_retention(pipeline.store, policy)
print(result.to_dict())

Pruning snapshot metadata removes the time-travel pointer; it does not delete the underlying LanceDB dataset versions, so retained snapshots stay replayable.

Cross-config comparison

Answer "should I switch embedding model or chunk size?" with evidence. compare_configs ingests the same corpus under two PipelineConfigs, runs your queries against each, and diffs the results with the existing comparator.

from rag_timetravel.compare import compare_configs
from rag_timetravel.core.config import PipelineConfig

report = await compare_configs(
    corpus_path="./docs",
    config_a=PipelineConfig(chunk_size=256),
    config_b=PipelineConfig(chunk_size=512),
    queries=["What is the refund policy?"],
)
print(report.to_text())

Hybrid retrieval and versioned rerankers

Retrieval strategy is part of the pipeline's versioned config, so a switch from dense vector search to hybrid (or a reranker upgrade) is something you can replay, diff, and bisect across, not just a silent behaviour change.

• retrieval_mode: "vector" (dense ANN, default), "bm25" (a built-in, dependency-free lexical scorer), or "hybrid" (reciprocal-rank fusion of both, tuned by hybrid_alpha).
• reranker: a named, versioned reranker applied to the candidate pool before the final top-k cut. Built-ins: none, lexical-overlap-v1, and cross-encoder/<model> (real neural reranking via the [local] extra).

from rag_timetravel import PipelineConfig

config = PipelineConfig(
    retrieval_mode="hybrid",        # vector | bm25 | hybrid
    hybrid_alpha=0.5,               # 1.0 = pure vector, 0.0 = pure BM25
    reranker="lexical-overlap-v1",  # or "cross-encoder/ms-marco-MiniLM-L-6-v2"
    candidate_pool=20,              # candidates before fusion/rerank (0 = auto)
)

# Ad-hoc from the CLI, no config file needed:
rag-timetravel query "refund policy" --mode hybrid --alpha 0.4 --reranker lexical-overlap-v1

Because the reranker name and mode are persisted with every snapshot, a replay reconstructs the exact strategy that produced a past answer.

Cloud storage backends (S3 / GCS / Azure)

Point the LanceDB index at an object store while keeping the event log local. Time-travel works unchanged: LanceDB versions the dataset manifests in the bucket, so checkout / replay resolve a historical version straight from the cloud.

pipeline = await RAGPipeline.create(
    "./project",                         # events.db stays here
    index_uri="s3://my-bucket/rag-index",  # index lives in the bucket
)

Credentials and region come from storage_options (passed explicitly) merged with the standard environment variables (AWS_*, GOOGLE_*, AZURE_*). From the CLI, set RAG_TIMETRAVEL_INDEX_URI and every command targets the cloud index transparently.

Observability: export traces to OpenTelemetry

Every query already records a precise, timestamped trace in the event log. Ship it to any OTLP/HTTP collector (Phoenix, Langfuse, Jaeger, Tempo, ...) as OpenTelemetry spans.

from rag_timetravel.observability import TraceExporter

exporter = TraceExporter(pipeline)
otlp = await exporter.to_otlp(query_id)   # dependency-free OTLP/JSON dict
# POST otlp to your collector's /v1/traces, or:
await exporter.export_otel(query_id)      # live SDK export ([otel] extra)

rag-timetravel export-traces --query-id <id> --out trace.json

The OTLP payload contains a root rag.query span with rag.retrieval and rag.generation children, carrying retrieval mode, reranker, token counts, and per-phase latency. Replays produce the same span shape, so originals and replays are directly comparable in a tracing UI.

---

Configuration

from rag_timetravel import PipelineConfig

config = PipelineConfig(
    embedder_model="all-MiniLM-L6-v2",   # local sentence-transformers model
    model="ollama/gemma3",                # litellm model string
    top_k=4,                              # chunks to retrieve per query
    chunk_size=512,                       # words per chunk
    chunk_overlap=64,                     # word overlap between chunks
    auto_snapshot_every=10,               # snapshot after every N docs (0=off)
    table_name="documents",               # LanceDB table name
    retrieval_mode="vector",              # vector | bm25 | hybrid
    hybrid_alpha=0.5,                     # vector/BM25 fusion weight (hybrid)
    reranker="none",                      # versioned reranker name
    candidate_pool=0,                     # candidates before rerank (0=auto)
)

Using OpenAI

import os
os.environ["OPENAI_API_KEY"] = "sk-..."

config = PipelineConfig(
    embedder_model="openai/text-embedding-3-small",
    model="gpt-4o",
)

Using a custom embedder

from rag_timetravel.index.embedder import Embedder

class MyEmbedder:
    @property
    def dim(self) -> int:
        return 768

    async def embed(self, texts: list[str]) -> list[list[float]]:
        # your implementation
        ...

pipeline = await RAGPipeline.create("./project", config, embedder=MyEmbedder())

---

Project structure

my_project/
├── events.db          # SQLite event store (append-only log)
└── lancedb/           # LanceDB vector index (versioned automatically)
    └── documents.lance/

The event store and index are self-contained in a single directory. Copy or rsync the directory to back up the full history. With index_uri (or RAG_TIMETRAVEL_INDEX_URI), the lancedb/ index instead lives on an object store while events.db stays local.

---

Event types

Event	When	Key payload fields
query.received	Query enters the pipeline	text, pipeline_config_id
retrieval.executed	Vector search completes	chunk_ids, scores, latency_ms
context.assembled	Chunks merged into prompt	chunk_ids, token_count
generation.executed	LLM produces answer	answer, model, token counts
query.completed	Full round-trip done	total_latency_ms
document.ingested	Document written to index	doc_id, source, chunk_count
index.snapshot_taken	Snapshot recorded	snapshot_id, lancedb_version
replay.*	Replay events (same structure, separate prefix)

---

Development

git clone https://github.com/Ar-maan05/rag-timetravel
cd rag-timetravel
pip install -e ".[dev]"

# Run tests: they exercise a real local LanceDB but mock the LLM and use a
# stub embedder, so no GPU, model download, or API key is needed.
pytest tests/

# Lint and type-check
ruff check src/rag_timetravel/
mypy src/rag_timetravel/

---

Roadmap

• SQLite event store
• LanceDB index versioning via checkout
• Query, replay, diff (JSON + HTML report)
• CLI + FastAPI server
• PostgreSQL event store backend
• Snapshot scheduling (cron-style)
• bisect command: binary search across snapshots to find when an answer changed
• Web UI for browsing event traces
• MCP server wrapper
• Regression testing: golden-query suites with a drift-gated test run exit code
• Lineage tracing: attribute retrieval changes back to the ingest that caused them
• Embedding drift detection across re-indexing
• Event-log retention and compaction (RetentionPolicy + scheduler)
• Cross-config comparison over a shared corpus
• Advanced retrieval replay: hybrid search (vector + BM25) and versioned rerankers
• Cloud storage backends: open the LanceDB index directly from S3/GCS/Azure
• Observability: export time-travel traces as OpenTelemetry/OTLP spans

Future Explorations

• Streaming ingestion: incremental snapshots on a live document feed
• Multi-index federation: replay a query across several indexes at once

---

License

MIT