Skip to main content

Project: Build a Local RAG Pipeline

This capstone answers one policy question from five repository-owned chunks. It runs offline and keeps the unstable generative component behind a fixed-response port. The result proves the orchestration contract—not model quality.

Prerequisites​

Complete the preceding generative-AI systems lessons. The project uses only the Python standard library and local JSONL fixtures.

Learning goals​

  • retrieve local evidence and assemble context under provenance and access rules.
  • validate response shape, citations, and literal grounding.
  • keep untrusted content away from tools and secrets.

Worked example​

The retrieval score is a transparent token overlap with a stable tie-break. kb-attack deliberately ranks first, proving that retrieval and trust are separate decisions. The model port receives admitted evidence only and has zero tools or secrets.

import json
import os
import re
from pathlib import Path

fixture_root = Path(os.environ.get("RAG_FIXTURE_ROOT", "tests/fixtures/modern-ai-rag"))
documents = [json.loads(line) for line in (fixture_root / "documents.jsonl").read_text().splitlines()]
query = "How long are expense receipts retained?"

def tokens(text):
    return set(re.findall(r"[a-z0-9]+", text.lower()))

query_tokens = tokens(query)
ranked = sorted(
    documents,
    key=lambda row: (-len(query_tokens & tokens(row["text"])), row["id"]),
)
retrieved = ranked[:3]
admitted = [
    row for row in retrieved
    if row["trust"] == "approved"
    and "employee" in row["audience"]
    and row["license"] == "CC-BY-4.0"
]
context = [{"id": row["id"], "text": row["text"]} for row in admitted]

def fixed_model_port(request):
    assert set(request) == {"question", "evidence", "response_contract"}
    return {"answer": "Expense receipts are retained for seven years.", "citations": ["kb-retention"]}

response = fixed_model_port({
    "question": query,
    "evidence": context,
    "response_contract": {"answer": "string", "citations": "non-empty string list"},
})
schema_valid = (
    set(response) == {"answer", "citations"}
    and isinstance(response["answer"], str)
    and isinstance(response["citations"], list)
    and bool(response["citations"])
    and all(isinstance(item, str) for item in response["citations"])
)
admitted_ids = {row["id"] for row in admitted}
citation_valid = set(response["citations"]) <= admitted_ids
cited_text = " ".join(row["text"] for row in admitted if row["id"] in response["citations"])
grounded = citation_valid and "seven years" in response["answer"].lower() and "seven years" in cited_text.lower()

print(f"query={query}")
print(f"retrieved={[row['id'] for row in retrieved]} admitted={[row['id'] for row in admitted]}")
print(f"answer={response['answer']} citations={response['citations']} schema_valid={schema_valid} grounded={grounded}")
print("model_port=fixed-response tools=0 secrets=0")

Output:

query=How long are expense receipts retained?
retrieved=['kb-attack', 'kb-retention', 'kb-travel'] admitted=['kb-retention', 'kb-travel']
answer=Expense receipts are retained for seven years. citations=['kb-retention'] schema_valid=True grounded=True
model_port=fixed-response tools=0 secrets=0

The fixed response is mocked and is not evidence about a real model. Before replacing the port, define an evaluation set, pin the model and decoding configuration, repeat stochastic cases, measure latency and cost, and require the same schema, citation, authorization, refusal, and human-review boundaries. No hosted API, credential, or download is required.

Common mistakes​

  • Treating top-ranked content as approved content.
  • Giving the model access to secrets or side-effecting tools for a read-only task.
  • Claiming this fixture demonstrates semantic retrieval or real-model faithfulness.
  • Failing open when the response is malformed, uncited, unsupported, or conflicting.
  • Forgetting source licensing, document versions, deletion, and per-tenant access control.

Practice​

Run all four cases in evaluation.jsonl. Add retrieval recall, citation validity, supported-claim, refusal, latency, and cost records without collapsing them into one score. Route the unknown parental-leave case to a clear fallback.

Next steps​

The 40-lesson path is complete. Return to the Deep Learning & Modern AI overview and use the capstones as templates for versioned, testable system boundaries.

Primary sources​

Checked 2026-10-16: Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks, JSON Schema Draft 2020-12, OWASP RAG Security, and NIST AI RMF Generative AI Profile.