Self-Optimizing Production RAG Platform

This repository is the executable implementation of the architecture in 03-self-optimizing-production-rag-platform.md. It provides versioned ingestion, tenant-aware hybrid retrieval, citation-or-abstention behavior, and a small browser client.

By default the stores and models are deterministic in-process implementations that sit behind the same boundaries the production adapters use: IndexCatalog stands in for PostgreSQL plus the Qdrant, OpenSearch and Neo4j indexes, and PiiProcessor stands in for a Presidio analyzer. Swapping an adapter does not change the ingestion contract, the authorization filters or the index-version lifecycle.

Real adapters for those stores live in rag_platform.adapters and install through the stores extra — see Production stores.

Run locally

python -m venv .venv
python -m pip install -e ".[dev]"
python -m rag_platform

Open http://127.0.0.1:8000. The demo UI sends the tenant-acme tenant and employees access label. API callers must send X-Tenant-ID; X-Access-Labels is an optional comma-separated list that defaults to public. In production, a trusted identity gateway must derive and overwrite these headers.

The demo corpus is ingested through the real pipeline on start-up, so every run exercises source versioning, chunking, sensitive-data processing and staged index activation.

API

Endpoint	Purpose
POST /v1/query	Answer a question from authorized evidence, or abstain.
POST /v1/sources	Register and ingest a source. Requires the data-steward label.
GET /v1/index-versions/active	Report the tenant's active index version.
POST /v1/index-versions/rollback	Restore the previous index version without reingestion.
POST /v1/feedback	Record feedback on an answer, tied to its trace.
GET /v1/feedback	List a tenant's feedback. Requires the data-steward label.
POST /v1/feedback/{id}/review	Accept or dismiss feedback as a named reviewer.
GET /healthz	Report the active pipeline configuration version.
GET /readyz	Report readiness and the configured external dependencies.

Ingestion

IngestionPipeline.ingest follows specification section 7: hash the content, reuse the existing source version when the hash is unchanged, detect document type and language, pseudonymize sensitive values into a vault held apart from the indexes, chunk structurally with parent context, stage a new index version, validate it, and only then activate it. Validation rejects cross-tenant chunks, unlabelled chunks and indexes that fail a retrieval smoke check; a rejected index version is retired rather than activated.

Retrieval

A query runs dense and lexical search under mandatory tenant and access filters, fuses the two rankings with a versioned method (reciprocal-rank or normalized weighted fusion), expands the candidate set along entity relationships in the knowledge graph, reranks a bounded candidate set, and then constructs the context: authorization is rechecked, near-duplicates are dropped, evidence is kept diverse across sources, parent sections replace fragments only when they fit, and the character budget is enforced before citation identifiers are attached.

Fusion weights, graph depth, reranker revision, diversity and budget are all PipelineConfig fields rather than code, so each is a candidate configuration that evaluation can accept or reject. Every answer carries the retrieval strategy, the retrieved and context chunk identifiers, the graph-expanded identifiers and the policy decisions that dropped evidence.

Query workflow

QueryWorkflow follows specification section 9 as an explicit state machine rather than a linear chain: classify, clarify, rewrite, decompose, retrieve, build context, generate, verify, repair and fallback are separate states, each recorded on the answer's workflow_path. An ambiguous question is clarified instead of answered; a grounded-but-unverified answer is repaired by narrowing to its single best-supported sentence before it is re-verified, and only falls back after the configured repair budget is spent. This is the structure LangGraph owns in the deployed topology — the executor can be swapped without changing the states or their tests.

The states call a ProgramSuite (specification section 10): typed classifier, rewriter, decomposer, synthesizer, verifier and clarification programs, each with an independent revision. The default implementations are deterministic — the synthesizer is extractive, so every claim it produces is a sentence copied from a chunk it cites — and a DSPy-compiled module can replace an implementation without changing the signature the workflow depends on or the program_revisions an evaluation run records.

Evaluation and optimization

Evaluator (specification section 16) runs a versioned EvaluationCase set against a query service and checks each answer deterministically rather than by LLM judgment: retrieval recall and reciprocal rank against the required evidence, grounding and forbidden claims against the answer text, and the expected outcome against the actual status. A failed case is tagged with the taxonomy a reviewer would assign by hand (section 13) — an unauthorized chunk in context is always authorization; a wrong outcome or missing evidence is retrieval; an ungrounded or forbidden claim is citation; a correct, grounded answer missing an expected term is generation.

OptimizationRun runs the control loop in section 12 over the bounded candidate space in section 11: CandidateSpace perturbs one PipelineConfig field at a time within reviewer-approved bounds, Constraints rejects any candidate that lets an unauthorized chunk into context or regresses latency or quality against the frozen baseline, and the non-dominated survivors among what remains are marked Pareto-optimal. canary performs the loop's last two steps: one approved candidate is evaluated again and promoted only if it still clears every constraint, otherwise it is rolled back. PipelineConfigRegistry mirrors IndexCatalog's staged-activation pattern for PipelineConfig, so promotion and one-action rollback apply to the production configuration the same way they apply to an index version (section 17).

Optimization runs outside production (section 3, "Optimization service"): nothing in this module changes which configuration answers a live query, so it is exercised as a library against its own test corpus rather than through the HTTP API.

Production stores

Adapters for Qdrant, OpenSearch, Neo4j, PostgreSQL, MinIO and Redis implement the protocols the platform already depends on, so running on the real stores changes no retrieval, workflow or authorization code (specification sections 6, 7, 8 and 18):

python -m pip install -e ".[dev,stores]"
docker compose up -d --wait
set -a && . deploy/local.env && set +a
pytest tests/integration/test_stores.py tests/integration/test_services.py --no-cov

Adapter	Protocol or base it satisfies	Role
QdrantDenseIndex	DenseIndex	Vector search, one collection per IndexVersion
OpenSearchSparseIndex	SparseIndex	BM25, one index per IndexVersion
CompositeChunkRepository	ChunkRepository	Presents both halves to HybridRetriever
Neo4jGraphRetriever	GraphExpander	Entity and provenance graph, one subgraph per IndexVersion
QdrantMirror / OpenSearchMirror / Neo4jGraphMirror	IndexMirror	Ingestion's "build the indexes" step
PostgresIndexCatalog	IndexCatalog	Durable source, index-version and lineage metadata
MinioOriginalStore	OriginalStore	Immutable, content-addressed originals
RedisAnswerCache + CachedQueryService	AnswerCache	Answer cache keyed by everything that produced the answer
HashingEmbedder / SentenceTransformerEmbedder	TextEmbedder	Versioned embeddings

Two properties are worth calling out because they are what the tests are about.

The database is never the authorization boundary. Specification section 4 puts that out of scope deliberately, so every adapter sends the tenant filter to the store and re-applies the authoritative is_authorized check in Python before a chunk leaves the adapter. A store that is stale, misconfigured or simply wrong cannot widen access on its own. OpenSearch can express the label subset test exactly (terms_set against required_label_count); Qdrant cannot, so there the store filter only narrows and the Python check decides. The integration tests assert the outcome is identical either way, including that a chunk labelled finance and management is withheld from a caller holding only finance.

Index versions stay immutable. Each adapter addresses the store by the physical_dense_index / physical_sparse_index identifier its IndexVersion already records, and that name is derived from the version fingerprint. A new index version is therefore written into its own collection and only then activated, so activation and rollback switch between whole indexes instead of mutating the one serving traffic. If a store refuses the write, the index version is retired and the previous one keeps answering.

The graph means the same thing in either store. Neo4jGraphRetriever and the in-process GraphRetriever share one extractor, and an integration test asserts they return the same expansion for the same seeds at every depth — including the case a Cypher path cannot express on its own, walking out and back to the seed entity.

The catalog is durable, not different. PostgresIndexCatalog inherits the staging, validation, activation and rollback rules and adds write-through persistence, so the metadata survives a restart and rollback still needs no reingestion.

A cached answer is only reused when nothing that produced it changed. The cache key covers the tenant, the caller's exact label set, the question, the configuration version, the program revisions and the active index version, so activation, rollback or promotion miss the cache by construction, and two callers with different labels never share an entry. Redis sits behind a CircuitBreaker: if it is down, queries are answered directly.

Embeddings are versioned because the revision is part of an IndexVersion's fingerprint. HashingEmbedder is the default: a genuine vector embedding (feature hashing over token unigrams and bigrams, L2-normalized) that needs no model download, so the stores can be populated and queried offline. It captures lexical and co-occurrence similarity, not learned semantics — install the embeddings extra and use SentenceTransformerEmbedder for that, behind the same protocol.

Language programs and sensitive-data recognition

Two more stand-ins have real implementations behind their protocols.

DSPy programs (rag_platform.adapters.dspy_programs, the llm extra) implement the six candidate modules in specification section 10 — classification, rewriting, decomposition, synthesis, claim verification and clarification — with everything that section requires of each: a typed signature, curated examples, an explicit metric, a training and a held-out split, and a versioned compiled artifact.

import dspy
from pathlib import Path
from rag_platform.adapters import dspy_examples
from rag_platform.adapters.dspy_programs import (
    ClassifyQuery, build_program_suite, compile_program, intent_metric,
)

lm = dspy.LM("<any model identifier dspy.LM accepts>")
compiled = compile_program(
    "classifier", ClassifyQuery, dspy_examples.classifier_examples(), intent_metric, lm
)
compiled.save(Path("artifacts/programs"))        # records revision and held-out score
programs = build_program_suite(lm, artifacts=Path("artifacts/programs"))

A program's revision is a hash of its compiled state, so the program_revisions on an AnswerTrace identify exactly the prompt and demonstrations that produced an answer. The adapters keep the guarantees the deterministic programs gave for free: a citation the model invents is discarded, an answer with no surviving citation is not an answer, and the verifier only sees the evidence the answer cites. Classification, rewriting, decomposition and clarification fall back to the deterministic program if the model call fails; synthesis and verification raise instead, and the workflow's circuit breakers turn that into abstention. The suite is tested offline against DSPy's DummyLM, including the full query workflow.

Presidio (rag_platform.adapters.presidio, the privacy extra) is a PiiRecognizer, so it replaces the regex recognizers inside PiiProcessor without touching pseudonymization, the vault or rehydration. It runs in-process and finds what no pattern can — a person's name:

from rag_platform.adapters.presidio import PresidioRecognizerFactory, build_analyzer
from rag_platform.pii import PiiProcessor

processor = PiiProcessor(recognizer_factory=PresidioRecognizerFactory(analyzer=build_analyzer()))

A PiiPolicy opts into a kind by naming it (recognizers=("email", "phone", "person")); a kind that is not named is never analyzed.

Observability and experiment governance

Install with python -m pip install -e ".[observability,governance]".

Tracing. QueryWorkflow reports to a WorkflowObserver; the core has no telemetry dependency. rag_platform.adapters.tracing.OpenTelemetryObserver turns each query into one trace in OpenInference conventions: a root rag.query span and one child span per workflow state (RETRIEVER for retrieval, LLM for generation, GUARDRAIL for verification). Spans are exported over OTLP to the OpenTelemetry Collector (deploy/otel-collector.yaml), which forwards them to Phoenix, so the backend is a deployment decision.

Traces leave the trusted boundary, so the adapter is narrow about what it exports (specification section 15):

• chunk text is never exported; retrieval spans carry chunk ids, scores and versions;
• the question and answer are exported only if TracePolicy.capture_text allows it, after identifiers are redacted and the text is truncated;
• a degraded dependency is named, but its error message is not exported;
• sampling is decided at the root span (sample_ratio), so a query is traced whole or not at all, and an unsampled query has no trace_id.

AnswerTrace carries the trace_id, measured latency_ms, the model_route that generated the answer and whitespace-token estimates (prompt_tokens, completion_tokens).

Feedback. POST /v1/feedback records a rating against a trace. Free text is redacted on the way in. Feedback becomes evaluation material only after a steward reviews it; Feedback.to_case drafts an EvaluationCase from accepted feedback, with the reviewer supplying the expected evidence.

Versioned evaluation set. data/evaluation/cases.jsonl is the reviewed evaluation set for the demo corpus. Its revision is the MD5 of the file, which is the hash DVC records in dvc.lock, so the revision an evaluation or an MLflow run reports is the one DVC tracks.

python -m rag_platform evaluate     # run the set, write reports/, exit 1 if the gate fails
dvc repro                           # the same, as the DVC stage in dvc.yaml
dvc metrics show
python -m rag_platform optimize --mlflow-tracking-uri http://127.0.0.1:5000

evaluate applies ReleaseGate: every case passes, every answer is grounded and no unauthorized chunk reaches context. The reports DVC tracks contain no timing, so they are byte-identical for identical data and code. CI reruns the stage and fails if the gate fails or if dvc.lock no longer matches what the platform answers. Evaluation outputs are pushed to an S3-compatible DVC remote (MinIO in the local stack; credentials come from the standard AWS environment variables).

MLflow. MlflowRunLogger records every evaluation and every optimization candidate, including rejected ones, with its configuration, metrics, dataset revision, program revisions, constraint results and promotion disposition. MlflowConfigRegistry persists PipelineConfigRegistry semantics in the MLflow model registry: each configuration is a model version, the champion alias is the active one, and promotion and rollback move the alias, so a restarted process reads the active configuration back.

docker compose up -d phoenix otel-collector mlflow starts the three services; the Store adapters CI job runs tests/integration/test_observability.py against them and round-trips the DVC outputs through MinIO.

Workflow executor, ingestion composition and events

Install with python -m pip install -e ".[workflow,ingestion,events]".

LangGraph. The states and every allowed edge between them are declared once in rag_platform.workflow (TRANSITIONS). rag_platform.adapters.langgraph_workflow. LangGraphQueryWorkflow runs those same state functions as a LangGraph StateGraph: one node per state and a conditional edge per declared transition, so clarification, repair and fallback are branches of the graph. Pass workflow_class=LangGraphQueryWorkflow to QueryService to use it. The answer and trace are identical under either executor; the equivalence tests assert that for every path. A state that routes outside the declared transitions raises WorkflowTransitionError under both.

LlamaIndex. IngestionPipeline.piece_builder is the chunking seam. rag_platform.adapters.llamaindex_ingestion.LlamaIndexPieceBuilder composes LlamaIndex's MarkdownNodeParser and SentenceSplitter: sections follow the document structure and long sections are cut on sentence boundaries, within ChunkingConfig.max_characters. Its output is the same ChunkPiece contract, so identifiers, deduplication, parent links, validation and staged activation are unchanged.

Events. IngestionPipeline.events receives source.changed and ingestion.completed (rag_platform.events). Events carry identifiers and counts, never document text. Publishing is best effort: a broker failure is recorded in the ingestion report's validation_checks and never fails or undoes the ingest. rag_platform.adapters.kafka_events.KafkaEventPublisher delivers them to Kafka, one topic per event type, keyed by tenant. docker compose up -d kafka starts a single-node broker.

Orchestration and evaluation frameworks

Install with python -m pip install -e ".[orchestration,evaluation,governance]".

RAGAS and DeepEval. rag_platform.adapters.eval_frameworks scores the same responses the deterministic Evaluator judged, with two independent implementations. Both are used reference-based only: they compare the run against the reviewer-authored expectations in each EvaluationCase, and no metric calls a model, so no release decision rests on an LLM judge (specification section 16).

• ragas_scores: RAGAS id-based context recall and precision of the retrieved chunks against the required evidence.
• deepeval_regression: one DeepEval BaseMetric per release property (expected outcome, evidence recall, groundedness, authorization, required terms), usable with DeepEval's own assert_test.
• framework_gate: both results as release-blocking findings, in ReleaseGate's format.

Test cases carry chunk ids, not chunk text.

Prefect. rag_platform.adapters.prefect_flows wraps existing steps in flows; the flows hold no logic of their own.

Flow	Does
ingestion_flow(documents)	Ingests sources in order. Transient failures are retried; a validation failure is not.
evaluation_flow(dataset, mlflow_tracking_uri=None)	Deterministic evaluation, RAGAS and DeepEval on the same run, optional MLflow record, an evaluation.completed event, then the gate. A failed gate fails the flow run with GateFailedError.
optimization_flow(dataset, max_candidates, mlflow_tracking_uri, promote)	Section 12's loop: evaluate candidates, canary the best Pareto-optimal one, record every disposition, and with promote=True move the registry champion only if the canary passed.

Flows act on the platform returned by prefect_flows.PLATFORM_FACTORY. docker compose up -d prefect starts a Prefect server; point PREFECT_API_URL at http://127.0.0.1:4200/api to record runs there.

from rag_platform.adapters.prefect_flows import evaluation_flow
evaluation_flow("data/evaluation/cases.jsonl", mlflow_tracking_uri="http://127.0.0.1:5000")

Configuration and deployment

Runtime configuration. rag_platform.settings.Settings.from_env reads the environment once at start-up and rag_platform.runtime.build_runtime assembles the platform from it. With nothing set the platform runs in-process; each variable swaps one boundary to its production adapter. An unusable value is a start-up error that names the variable.

Variable	Selects
RAG_POSTGRES_DSN	Durable catalog and feedback log. Loaded on start; replicas follow each other's activations within RAG_CATALOG_SYNC_SECONDS.
RAG_QDRANT_URL, RAG_OPENSEARCH_URL, RAG_NEO4J_URL (+ RAG_NEO4J_USER, RAG_NEO4J_PASSWORD)	Dense, sparse and graph stores. Ingestion writes each index version to its own collection, index and subgraph; queries read whichever version is active. An unconfigured leg is served from the catalog.
RAG_MINIO_ENDPOINT (+ RAG_MINIO_ACCESS_KEY, RAG_MINIO_SECRET_KEY, RAG_MINIO_SECURE, RAG_MINIO_BUCKET)	Immutable originals.
RAG_REDIS_URL (+ RAG_CACHE_TTL_SECONDS)	Answer cache.
RAG_KAFKA_BOOTSTRAP_SERVERS (+ RAG_KAFKA_TOPIC_PREFIX)	Event publishing.
RAG_OTLP_ENDPOINT (+ RAG_TRACE_SAMPLE_RATIO, RAG_TRACE_CAPTURE_TEXT)	Tracing. Question and answer text is exported only if RAG_TRACE_CAPTURE_TEXT=true.
RAG_PII_RECOGNIZER=presidio	Presidio recognition.
RAG_LLM_MODEL (+ RAG_PROGRAM_ARTIFACTS)	DSPy language programs on that model.
RAG_WORKFLOW_EXECUTOR=langgraph, RAG_CHUNKER=llamaindex, RAG_EMBEDDER=sentence-transformers	LangGraph executor, LlamaIndex chunking, learned embeddings.
RAG_MLFLOW_TRACKING_URI	Reported by /readyz; pass it to evaluate/optimize or the flows to record runs.
RAG_SEED_DEMO_SOURCES	Ingest the demo corpus on start. Defaults to on only without a durable catalog.

GET /readyz is served once start-up has connected to every configured dependency and lists them. The API, the evaluate and optimize commands and the Prefect flows all assemble the platform this way. deploy/local.env holds the values for the local compose stack:

docker compose up -d --wait
set -a && . deploy/local.env && set +a
python -m pip install -e ".[stores,observability,events]"
python -m rag_platform

Container image. The default image runs in-process. Build with the extras a deployment's settings need: docker build --build-arg RAG_EXTRAS=stores,observability,events,workflow .

Kubernetes. deploy/helm/rag-platform is the Helm chart for the topology in specification section 18.

• API Deployment (non-root, read-only root filesystem, no service-account token), Service, HorizontalPodAutoscaler, PodDisruptionBudget, NetworkPolicy, optional Ingress.
• Settings in a ConfigMap, credentials from secrets.existingSecret (recommended) or a chart managed Secret; pods roll when either changes.
• Optional OpenTelemetry Collector, a scheduled evaluation CronJob that fails when the release gate fails, and a Prefect worker.
• endpoints.* points at services you already run. backingServices.enabled=true instead runs PostgreSQL, Qdrant, OpenSearch, Neo4j, MinIO, Redis, Kafka, Phoenix, MLflow and Prefect in-cluster as single-replica development instances and wires the application to them; it requires an explicit backingServices.devPassword and is not for production data.

helm upgrade --install rag deploy/helm/rag-platform --set image.tag=1.0.0
helm test rag
# whole topology in one namespace, for development:
helm upgrade --install rag deploy/helm/rag-platform -f deploy/helm/rag-platform/values-dev.yaml

The API trusts X-Tenant-ID and X-Access-Labels. Only an identity gateway that derives and overwrites them may reach it; list it in networkPolicy.ingressFrom.

CI lints the chart, validates every manifest of the full topology against a Kubernetes API server, installs the chart on a kind cluster and runs helm test against the running API.

Reliability

Graph expansion, reranking, answer synthesis and claim verification are all optional enrichments over an already-authorized candidate set, not the retrieval or authorization boundary itself (specification section 17). Each is called behind a CircuitBreaker: a failing or slow dependency degrades to a declared fallback — no graph expansion, the unranked candidates, an empty synthesis, or (for verification) "not grounded" — rather than crashing the query. A verifier that cannot run is never treated as having confirmed grounding, so a failure there still routes through the same repair-then-fallback path an ungrounded answer does. Every degradation is recorded on AnswerTrace.degraded_dependencies. After enough consecutive failures the breaker opens and fails fast for a cooldown period instead of retrying a dependency that is down on every query; one trial call after the cooldown decides whether it closes again.

Test

ruff check .
mypy src
pytest --cov --cov-report=term-missing
npm ci
npx playwright install chromium
npm run test:e2e

Unit tests cover policy and retrieval behavior, integration tests exercise the HTTP boundary, and Playwright tests verify the user journey in a real browser. GitHub Actions runs all three layers and verifies the production container build.

Delivery policy

All changes are made on a branch and delivered by pull request. The CI workflow is the required quality gate. Successful repository-owner and Dependabot pull requests are merged automatically after CI while preserving their individual commits. Repository branch protection should require the CI / Python quality and tests, CI / Playwright end-to-end, CI / Store adapters, CI / Container build and CI / Helm chart on Kubernetes checks.