Document Intelligence with RAG

The brief advanced

You have already lived the failure this project fixes. Someone asks "did we ever agree to a right-to-audit clause with this vendor?" and the honest answer is a shrug and forty minutes of Ctrl-F across a shared drive. The information exists — it is just scattered across ten thousand PDFs that no one can query. Multiply that by every question every week and you are burning expensive human hours re-reading documents nobody remembers signing.

The instinct is "just make the file names searchable" or "throw it all into keyword search." That is exactly where naive attempts die: keyword search matches words, not meaning, so a query for "termination" misses the clause that says "either party may end this agreement." The brief here is therefore sharper than "make it searchable" — it is let people ask questions in plain English and get an answer they can trust and verify. That last word carries the weight: an answer with no source is worse than no answer at all in a legal or financial setting, because it looks authoritative while being unaccountable. Read the whole project through that lens — every design choice below exists to make answers both found and provable.

"We have twenty years of documents and no way to ask them anything." The knowledge is trapped in PDFs on a share drive. People re-read the same contracts, miss renewal dates, and can't answer "what did we agree with this vendor?" without an afternoon of digging. A document-intelligence-with-RAG agent turns that pile into something you can query in plain English — with citations you can trust.

1 · Discovery — what questions do people actually ask? advanced

The tempting mistake is to build the impressive thing before you know what people need. A team hears "RAG over our contracts" and immediately reaches for "the agent reads the contract and tells you whether the indemnity clause is favourable." That is the one request you should refuse — legal judgment is exactly what a machine should not own, and shipping it invites a confidently wrong opinion that a human then rubber-stamps.

Discovery is the cheap step that saves the expensive rebuild: before designing anything, you sort the real requests by who should be accountable for the answer. Think of it like triage in an ER — not everything that walks in gets the same treatment. Some questions are pure retrieval ("find the clause about X across all contracts"), where the agent is astonishingly useful and low-risk because it only surfaces text a human already trusts. Others are metadata questions ("which contracts auto-renew in Q3?") that are really filters, not reading. And a thin, dangerous slice asks for interpretation or risk judgment. The table below is that triage: the leverage column is really a risk-and-fit column, and the single most valuable line in it is the low-leverage one — because deciding what not to automate is what keeps the whole system trustworthy.

2 · Architecture advanced

The architecture question people get wrong is "why are there two pipelines instead of one?" The obvious design — read the documents when the user asks — feels simpler, and it works right up until the moment a real user waits thirty seconds while you parse, chunk, and embed four thousand PDFs to answer one question. You cannot afford to do the expensive work per query.

So the shape of every serious RAG system is a split: an offline ingest pipeline that does the slow, one-time prep, and an online query pipeline that does the fast, per-question work — connected by a shared vector-plus-metadata store. The kitchen analogy is exact: ingest is the prep the kitchen does before service (chop, marinate, portion), and query is plating a dish to order. You would never make a diner wait while you butcher the meat. The store in the middle is the pantry both sides share — fill it once, serve from it many times. The deeper lesson hidden in this split is that retrieval quality is decided at ingest time, not at query time: how you chunk and what metadata you attach is fixed before a single question is ever asked, which is why the ingest side gets its own careful section (§4) below. Read the diagram as two rows — top runs once, bottom runs every time — and note that the language model appears in exactly one box, because a RAG system is mostly plumbing with a small, well-fenced use of the LLM at the end.

Two pipelines. Ingest (offline): parse each document, chunk it, extract structured metadata with the Project 4 pattern (vendor, dates, type), embed the chunks, and store both vectors and metadata. Query (online): retrieve relevant chunks — optionally filtered by metadata — and have the LLM answer grounded in them, citing the source document and page for every claim.

3 · Risk & safety model advanced

Here is the failure that gets a document-RAG system pulled from production. A user asks whether a contract permits early termination; the agent answers "Yes, with 30 days' notice" — fluent, confident, and completely invented, because the retrieval missed and the model filled the gap from its training data. Nobody checks, because the answer sounded like it came from the documents. That is not a hypothetical; it is the default behaviour of a language model asked a question it can't ground, and over legal or financial text it is the difference between a helpful tool and a liability.

So the safety model here is not a list of nice-to-haves — it is the load-bearing wall. Think of each control below as answering a specific way the system can betray a user's trust: a claim with no basis (fix: cite every claim or refuse), a claim citing the wrong document (fix: carry doc ID + page on every chunk), a stale answer from an out-of-date index (fix: re-index and show an "as-of" date), and — the one that ends careers — a user seeing a document they were never allowed to see (fix: filter by permission before the model sees a single chunk). The through-line is that trust must be built into the retrieval and generation path, not bolted on as a disclaimer. The one rule that subsumes the rest: no citation, no claim — and "not found in the documents" is always a valid, and often the correct, answer.

4 · The ingestion pipeline — where quality is won advanced

Almost everyone who ships a disappointing RAG system tries to fix it at the wrong end. The answers come back vague or wrong, so they swap the model, tune the prompt, crank up top-k — and nothing improves, because the damage was done long before the query ran. The chunk they needed was never a clean, retrievable unit in the first place. The most common culprit is blind fixed-size chunking: cut the document every 500 characters and you routinely slice a clause in half, or glue the end of one section to the start of the next, so the vector that should mean "termination clause" now means "half a termination clause plus a payment schedule header."

That is why this section carries the "where quality is won" title. Retrieval can only ever return chunks you created at ingest time; if the boundaries are wrong, no amount of clever querying recovers the lost meaning — garbage in, garbage retrieved. The fix is to chunk on the document's own structure (sections, clauses, pages) so each chunk is a self-contained thought, and to attach rich metadata — vendor, type, dates, and crucially the source doc ID and page — to every chunk as it goes in. The metadata is what later makes both permission filtering and citations possible: a chunk that doesn't know where it came from can never be cited or access-controlled. The sketch below shows the four jobs every document runs through — parse, extract metadata, chunk, index — and is a shape, not runnable code, so read it for the flow rather than the API.

class _Any:
    '''stands in for any undefined demo value; supports call/attr/index/
    iteration and basic arithmetic (as 0.7) so demo snippets run.'''
    def __call__(self, *a, **k): return _Any()
    def __getattr__(self, k): return _Any()
    def __getitem__(self, k): return _Any()
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __contains__(self, o): return True
    def __enter__(self, *a): return _Any()
    def __exit__(self, *a): return False
    def __float__(self): return 0.7
    def __int__(self): return 1
    def __lt__(self, o): return True
    def __gt__(self, o): return False
    def __le__(self, o): return True
    def __ge__(self, o): return False
    def __add__(self, o): return o
    def __radd__(self, o): return o
    def __bool__(self): return True
    def __repr__(self): return 'demo'
    def __str__(self): return 'demo'
def chunk_by_section(*a, **k):  # demo stub
    return _Any()
corpus = [
    {"id": "x1", "text": "demo one", "name": "api-7f9c", "status": "Running"},
    {"id": "x2", "text": "demo two", "name": "worker-2d", "status": "CrashLoopBackOff"},
]
def embed(*a, **k):  # demo stub
    return _Any()
def extract_metadata(*a, **k):  # demo stub
    return _Any()
class _index_t:
    add = 'demo'
    def add(self, *a, **k): return 'demo'
    def __getattr__(self, k): return 'demo'
index = _index_t()
def parse(*a, **k):  # demo stub
    return _Any()

ingest.py (shape)for doc in corpus:
    text = parse(doc)                       # native text / OCR / vision (P4)
    meta = extract_metadata(text)           # vendor, type, dates — Project 4 schema
    for chunk in chunk_by_section(text):    # respect headings, not blind 500-char cuts
        index.add(
            embedding=embed(chunk.text),
            text=chunk.text,
            metadata={**meta, "doc_id": doc.id, "page": chunk.page},
        )

5 · Tool surface advanced

The instinct is to give the agent one big "answer the question" tool and let it figure everything out. That feels powerful and is actually a trap — a single opaque capability is impossible to permission, impossible to test, and impossible to reason about when it does something wrong. The discipline is the opposite: break the agent's abilities into a small set of named, single-purpose tools, each with a clear job and a clear risk label.

Think of the tool surface as the agent's job description — it lists exactly what the agent is allowed to do and nothing more. Most of what a document-RAG agent needs is read-only: search the corpus, fetch a document, extract fields on demand, scope a query by metadata. Those are green because reading can't corrupt or leak (as long as permission filtering happens first — see §3). The one that earns a warning is anything that writes — export_report to a file or a ticket — because a write action can act on the world, so it is gated rather than granted freely. The reason this earns its own section: the tool surface is your security boundary made concrete. Every capability you add widens what the agent can do and therefore what can go wrong, so the smallest surface that solves the discovered needs (§1) is the right one. Read the table as a contract: capability, what it does, and how much you have to trust it.

6 · Evaluation advanced

When a RAG answer is wrong, the useless response is "the AI messed up." That tells you nothing you can act on, because there are two completely different failures hiding behind one bad answer: either retrieval never surfaced the right chunk, or retrieval found it and the model ignored or contradicted it. Fix the wrong one and you waste a week — you can't tune generation your way out of a retrieval miss, and you can't chunk your way out of a model that hallucinates over good sources.

So evaluation here is deliberately split into two dials you measure separately, like a doctor checking pulse and blood pressure as distinct readings rather than one vague "health." First, retrieval recall@k: on a labelled set of questions, does the chunk that actually answers each one appear in the top-k? That number is the ceiling on everything downstream — if the right chunk isn't retrieved, the best model on earth cannot answer faithfully. Only once retrieval clears the bar do you measure the second dial: answer faithfulness (does every claim trace to a cited chunk?), citation correctness (do the cited doc and page really contain the claim?), and abstention (does it say "not in the documents" instead of inventing?). The reason to test them in that order is causal: retrieval is upstream of generation, so you diagnose upstream first. The table below is these dials laid out as a checklist — read it as "measure retrieval, then judge the answer."

7 · Phased rollout expert

The failure mode of ambitious teams is shipping Phase 3 on day one. They build the full "ask any question, get a synthesized answer, run cross-document analytics, export reports" system, launch it, and then spend months firefighting because the retrieval wasn't trustworthy enough to carry synthesis, and every wrong answer erodes the credibility they needed to keep the project funded. You don't earn trust by promising it; you earn it one verifiable step at a time.

So the rollout is a ladder, and each rung is independently useful and independently trustworthy — you can stop at any rung and still have shipped something valuable. Phase 1 just returns relevant passages with citations and lets a human read them; it is pure retrieval, the lowest-risk thing you can do, and it is instantly useful because "find me the passage" already beats an afternoon of folder-hunting. Phase 2 adds synthesis — the model writes the answer, but every claim is cited and it abstains when unsupported — which you only unlock once Phase 1's retrieval is measurably good (that's what §6's recall@k proved). Phase 3 adds cross-document analytics and report export, the highest-leverage and highest-risk tier. The point of laddering this way is that each rung de-risks the next: you never automate a decision until the layer beneath it has earned the right. Note the final rung is a hard "Never" — some things (uncited claims presented as fact, retrieval past a permission boundary) are not later phases, they are permanent red lines.

Skills & course map expert

How to use this page expert

Do the steps in order, top to bottom. A terminal block means type those commands and press Enter. A file block means create that exact file and paste in the whole contents. Expected output follows each command so you can check yourself. Skip nothing — each step builds on the last.

Step 1 · Create the project folder expert

terminalmkdir -p doc-rag/samples doc-rag/tests
cd doc-rag

# no output — you are now inside doc-rag/

Step 2 · Virtual environment expert

A virtual environment isolates this project's packages. After activating, your prompt shows (.venv).

terminalpython3 -m venv .venv
source .venv/bin/activate

terminalpy -m venv .venv
.venv\Scripts\Activate.ps1

(.venv) /Users/you/doc-rag $

Step 3 · Install dependencies expert

We need pytest for tests and the Anthropic SDK for the optional live mode (Step 9). The core pipeline uses only the Python standard library.

terminalpip install pytest "anthropic>=0.40"
pip freeze > requirements.txt

Successfully installed anthropic-0.69.0 pytest-8.3.4 ...

Step 4 · The store — chunks with metadata + a stub embedder expert

Create store.py. It defines a Chunk, a pure-Python embedder (so similarity works offline), cosine similarity, and a Store that filters by metadata before ranking. Paste the whole file.

store.py"""In-memory vector store with a deterministic, offline embedder."""
from dataclasses import dataclass, field
import math, re


def embed(text: str, dims: int = 64) -> list[float]:
    """A tiny bag-of-words hashing embedder. Not smart, but deterministic
    and offline: same words -> similar vectors. Good enough to LEARN RAG.
    In Step 9 you swap this for a real embedding model."""
    vec = [0.0] * dims
    for word in re.findall(r"[a-z0-9]+", text.lower()):
        vec[hash(word) % dims] += 1.0
    return vec


def cosine(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    return 0.0 if na == 0 or nb == 0 else dot / (na * nb)


@dataclass
class Chunk:
    text: str
    doc_id: str
    page: int
    meta: dict = field(default_factory=dict)
    embedding: list[float] = field(default_factory=list)


class Store:
    def __init__(self):
        self.chunks: list[Chunk] = []

    def add(self, chunk: Chunk) -> None:
        if not chunk.embedding:
            chunk.embedding = embed(chunk.text)
        self.chunks.append(chunk)

    def search(self, query: str, k: int = 4, where: dict = None) -> list[Chunk]:
        """Filter by metadata FIRST (access control + scoping), THEN rank."""
        pool = [c for c in self.chunks
                if not where or all(c.meta.get(k2) == v for k2, v in where.items())]
        q = embed(query)
        ranked = sorted(pool, key=lambda c: cosine(q, c.embedding), reverse=True)
        # drop zero-similarity hits so out-of-corpus queries return nothing
        return [c for c in ranked[:k] if cosine(q, c.embedding) > 0]

Step 5 · Ingest — structure-aware chunking expert

Create ingest.py. It splits documents on blank-line sections (not blind character counts) and attaches metadata to every chunk.

ingest.py"""Turn raw document text into chunks with metadata + embeddings."""
from store import Store, Chunk


def chunk_sections(text: str):
    """Yield (section_text, page_number). Splits on blank lines so each
    chunk is a self-contained idea, not a mid-sentence cut."""
    page = 1
    for block in text.split("\n\n"):
        block = block.strip()
        if block:
            yield block, page
            page += 1


def ingest(store: Store, doc_id: str, text: str, meta: dict) -> int:
    """Add every section of one document to the store. Returns chunk count."""
    n = 0
    for section, page in chunk_sections(text):
        store.add(Chunk(text=section, doc_id=doc_id, page=page, meta=meta))
        n += 1
    return n

Step 6 · Sample documents expert

Create two tiny sample files so you have a corpus to query. Create samples/contract_acme.txt:

samples/contract_acme.txtPayment terms: ACME pays Net 30 from the invoice date.

Renewal: this contract auto-renews annually unless cancelled 60 days prior.

samples/policy_refunds.txtRefund policy: customers may request a refund within 30 days of purchase.

Exceptions: digital goods are non-refundable once downloaded.

Step 7 · The answer engine — grounded, cited, abstaining expert

Create answer.py. It retrieves relevant chunks, builds a numbered context, and (in mock mode) produces a grounded answer with citations — or abstains when nothing is retrieved.

answer.py"""Grounded question-answering over the store. Mock by default."""
import os
from store import Store

MODEL = "claude-opus-4-8"
ABSTAIN = "I don't have that in the documents."


def _context(hits) -> str:
    return "\n".join(
        f"[{i}] ({h.doc_id} p{h.page}) {h.text}" for i, h in enumerate(hits))


def _mock_llm(question: str, hits) -> str:
    """Offline stand-in: echoes the top chunk with its citation so you can
    see the grounding + citation flow without an API key."""
    top = hits[0]
    return f"{top.text} [0]"


def _real_llm(question: str, hits) -> str:
    import anthropic
    client = anthropic.Anthropic()
    system = ("Answer ONLY from the numbered sources. Cite the source id in "
              "[brackets] after each claim. If the sources do not contain the "
              f"answer, reply exactly: {ABSTAIN}")
    msg = client.messages.create(
        model=MODEL, max_tokens=400, system=system,
        messages=[{"role": "user",
                   "content": f"Sources:\n{_context(hits)}\n\nQuestion: {question}"}])
    return msg.content[0].text


def answer(question: str, store: Store, where: dict = None) -> dict:
    hits = store.search(question, k=4, where=where)
    if not hits:
        return {"text": ABSTAIN, "sources": []}      # the abstention path
    use_real = os.environ.get("USE_REAL_API") == "1"
    text = _real_llm(question, hits) if use_real else _mock_llm(question, hits)
    return {"text": text, "sources": [(h.doc_id, h.page) for h in hits]}


def find_docs(store: Store, where: dict) -> list[str]:
    """Cross-document analytics: a metadata filter, not a search."""
    return sorted({c.doc_id for c in store.chunks
                    if all(c.meta.get(k) == v for k, v in where.items())})

if __name__ == "__main__":
    from ingest import ingest
    s = Store()
    ingest(s, "contract_acme.txt", open("samples/contract_acme.txt").read(),
           {"type": "contract", "vendor": "ACME"})
    ingest(s, "policy_refunds.txt", open("samples/policy_refunds.txt").read(),
           {"type": "policy"})
    print("Q: What are ACME's payment terms?")
    print("A:", answer("payment terms ACME Net", s)["text"])
    print("Q: What is the capital of France? (out of corpus)")
    print("A:", answer("capital of France", s)["text"])
    print("contracts:", find_docs(s, {"type": "contract"}))

answer.py"""Grounded question-answering over the store. Mock by default."""
import os
from store import Store

MODEL = "gpt-5.5"
ABSTAIN = "I don't have that in the documents."


def _context(hits) -> str:
    return "\n".join(
        f"[{i}] ({h.doc_id} p{h.page}) {h.text}" for i, h in enumerate(hits))


def _mock_llm(question: str, hits) -> str:
    """Offline stand-in: echoes the top chunk with its citation so you can
    see the grounding + citation flow without an API key."""
    top = hits[0]
    return f"{top.text} [0]"


def _real_llm(question: str, hits) -> str:
    import openai
    client = openai.OpenAI()
    system = ("Answer ONLY from the numbered sources. Cite the source id in "
              "[brackets] after each claim. If the sources do not contain the "
              f"answer, reply exactly: {ABSTAIN}")
    resp = client.responses.create(
        model=MODEL, max_output_tokens=400, instructions=system,
        input=f"Sources:\n{_context(hits)}\n\nQuestion: {question}")
    return resp.output_text


def answer(question: str, store: Store, where: dict = None) -> dict:
    hits = store.search(question, k=4, where=where)
    if not hits:
        return {"text": ABSTAIN, "sources": []}      # the abstention path
    use_real = os.environ.get("USE_REAL_API") == "1"
    text = _real_llm(question, hits) if use_real else _mock_llm(question, hits)
    return {"text": text, "sources": [(h.doc_id, h.page) for h in hits]}


def find_docs(store: Store, where: dict) -> list[str]:
    """Cross-document analytics: a metadata filter, not a search."""
    return sorted({c.doc_id for c in store.chunks
                    if all(c.meta.get(k) == v for k, v in where.items())})

if __name__ == "__main__":
    from ingest import ingest
    s = Store()
    ingest(s, "contract_acme.txt", open("samples/contract_acme.txt").read(),
           {"type": "contract", "vendor": "ACME"})
    ingest(s, "policy_refunds.txt", open("samples/policy_refunds.txt").read(),
           {"type": "policy"})
    print("Q: What are ACME's payment terms?")
    print("A:", answer("payment terms ACME Net", s)["text"])
    print("Q: What is the capital of France? (out of corpus)")
    print("A:", answer("capital of France", s)["text"])
    print("contracts:", find_docs(s, {"type": "contract"}))

Run it — works immediately, no key:

terminalpython answer.py

Q: What are ACME's payment terms?
A: Payment terms: ACME pays Net 30 from the invoice date. [0]
Q: What is the capital of France? (out of corpus)
A: I don't have that in the documents.
contracts: ['contract_acme.txt']

Step 8 · Tests (no key) expert

Create tests/test_docrag.py and paste the whole file.

tests/test_docrag.py"""Offline tests — deterministic embedder, no API key."""
from store import Store, Chunk
from ingest import ingest
from answer import answer, find_docs, ABSTAIN


def _store():
    s = Store()
    ingest(s, "c.txt", "ACME pays Net 30 from invoice date.",
           {"type": "contract", "owner": "alice"})
    ingest(s, "p.txt", "Refunds allowed within 30 days of purchase.",
           {"type": "policy", "owner": "bob"})
    return s


def test_relevant_chunk_is_retrieved():
    s = _store()
    assert answer("payment terms Net invoice", s)["sources"][0][0] == "c.txt"


def test_metadata_filter_scopes_results():
    s = _store()
    res = answer("refund", s, where={"type": "policy"})
    assert all(src[0] == "p.txt" for src in res["sources"])


def test_permission_filter_hides_other_owners():
    s = _store()
    res = answer("anything", s, where={"owner": "alice"})
    assert all(src[0] == "c.txt" for src in res["sources"])


def test_out_of_corpus_abstains():
    s = _store()
    assert answer("xyzzy quux nothing", s)["text"] == ABSTAIN


def test_every_hit_has_doc_and_page():
    s = _store()
    for src in answer("payment", s)["sources"]:
        assert isinstance(src[0], str) and isinstance(src[1], int)


def test_find_docs_by_metadata():
    s = _store()
    assert find_docs(s, {"type": "contract"}) == ["c.txt"]


def test_answer_is_grounded_in_a_source():
    s = _store()
    res = answer("payment terms Net invoice", s)
    assert "[0]" in res["text"]        # cites a retrieved chunk

terminalpython -m pytest tests/ -v

tests/test_docrag.py::test_relevant_chunk_is_retrieved PASSED
tests/test_docrag.py::test_metadata_filter_scopes_results PASSED
tests/test_docrag.py::test_permission_filter_hides_other_owners PASSED
tests/test_docrag.py::test_out_of_corpus_abstains PASSED
tests/test_docrag.py::test_every_hit_has_doc_and_page PASSED
tests/test_docrag.py::test_find_docs_by_metadata PASSED
tests/test_docrag.py::test_answer_is_grounded_in_a_source PASSED

7 passed in 0.06s

Step 9 · Go live: real embeddings + Claude (optional) expert

The stub embedder is enough to learn RAG, but real embeddings understand meaning (synonyms, paraphrase). To go live you change two things — no structural code changes:

terminalexport ANTHROPIC_API_KEY="sk-ant-your-key-here"
export USE_REAL_API=1
python answer.py

Q: What are ACME's payment terms?
A: ACME pays Net 30 from the invoice date [0].

Troubleshooting — every error you might hit expert

import math, hashlib
from dataclasses import dataclass, field

def embed(text, dims=64):
    v = [0.0] * dims
    for tok in text.lower().split():
        h = int(hashlib.md5(tok.encode()).hexdigest(), 16)
        v[h % dims] += 1.0
    return v

def cosine(a, b):
    dot = sum(x*y for x, y in zip(a, b))
    na = math.sqrt(sum(x*x for x in a)); nb = math.sqrt(sum(y*y for y in b))
    return dot / (na*nb) if na and nb else 0.0

@dataclass
class Chunk:
    text: str; doc_id: str; page: int = 0
    meta: dict = field(default_factory=dict)
    embedding: list = None

class Store:
    def __init__(self): self.chunks = []
    def add(self, ch):
        ch.embedding = ch.embedding or embed(ch.text)
        self.chunks.append(ch)
    def search(self, query, k=4):
        q = embed(query)
        ranked = sorted(self.chunks, key=lambda c: cosine(q, c.embedding), reverse=True)
        return ranked[:k]

s = Store()
s.add(Chunk("Refunds are processed within 30 days.", "policy", 1))
s.add(Chunk("The office is closed on public holidays.", "hr", 2))
print(s.search("how long do refunds take?", k=1)[0].text)
# Refunds are processed within 30 days.

def chunk_sections(text):
    page = 1
    for block in text.split("\n\n"):
        block = block.strip()
        if not block:
            continue
        if block.startswith("[page]"):        # explicit page markers advance page
            page += 1; continue
        yield block, page

def ingest(store, doc_id, text, meta=None):
    n = 0
    for section, page in chunk_sections(text):
        store.add(Chunk(section, doc_id, page, dict(meta or {})))
        n += 1
    return n

doc = "Refund policy.\nRefunds within 30 days.\n\n[page]\nShipping.\nGround is 5 days."
print(ingest(s, "policy", doc, {"tenant": "acme"}))   # 2 sections indexed

def matches(meta, where):
    return all(meta.get(k) == v for k, v in where.items())

def search(store, query, k=4, where=None):
    cands = [c for c in store.chunks if not where or matches(c.meta, where)]
    q = embed(query)
    cands.sort(key=lambda c: cosine(q, c.embedding), reverse=True)
    return cands[:k]

def find_docs(store, where):
    return sorted({c.doc_id for c in store.chunks if matches(c.meta, where)})

st = Store()
st.add(Chunk("Acme secret roadmap Q3", "road", 1, {"tenant": "acme"}))
st.add(Chunk("Globex secret roadmap Q3", "road", 1, {"tenant": "globex"}))
hits = search(st, "roadmap", where={"tenant": "globex"})
print([h.text for h in hits])       # only Globex — Acme never considered
assert all(h.meta["tenant"] == "globex" for h in hits)

MIN_SIM = 0.05   # below this, retrieval is noise -> abstain

def answer(question, store, where=None):
    q = embed(question)
    ranked = [(cosine(q, c.embedding), c) for c in store.chunks
              if not where or matches(c.meta, where)]
    ranked = [(sim, c) for sim, c in ranked if sim >= MIN_SIM]
    ranked.sort(reverse=True, key=lambda t: t[0])
    top = [c for _, c in ranked[:3]]
    if not top:
        return {"text": "I don't have information on that.", "sources": []}
    ctx = "\n".join(f"[{c.doc_id} p{c.page}] {c.text}" for c in top)
    # --- needs API key: real grounded generation (Claude or OpenAI) ---
    # Claude:  msg = client.messages.create(model="claude-opus-4-8", max_tokens=400,
    #   system="Answer ONLY from context. Cite [doc pN]. If absent, say you don't know.",
    #   messages=[{"role":"user","content":f"{ctx}\n\nQ: {question}"}])
    #          text = msg.content[0].text
    # OpenAI:  resp = client.responses.create(model="gpt-5.5", max_output_tokens=400,
    #   instructions="Answer ONLY from context. Cite [doc pN]. If absent, say you don't know.",
    #   input=f"{ctx}\n\nQ: {question}")
    #          text = resp.output_text
    text = f"Based on {top[0].doc_id}: {top[0].text}"     # offline stand-in
    return {"text": text, "sources": [f"{c.doc_id} p{c.page}" for c in top]}

print(answer("refund window?", st, where={"tenant": "acme"}))   # abstains: no match

GOLD = [("refund window", "policy"), ("shipping time", "policy")]

def recall_at_k(store, gold, k=4):
    hits = 0
    for q, want in gold:
        got = {c.doc_id for c in search(store, q, k=k)}
        hits += (want in got)
    return hits / len(gold)

def unfaithful_sentences(answer_text, sources):
    # every claim sentence should map to at least one cited source id
    bad = []
    for sent in answer_text.split("."):
        sent = sent.strip()
        if sent and not any(src.split()[0] in sent or sources for src in sources):
            bad.append(sent)
    return bad

r = recall_at_k(st, GOLD, k=4)
print(f"recall@4 = {r:.2f}")
BAR = 0.80
raise SystemExit(0 if r >= BAR else 1)   # CI gate: block regressions

def term_overlap(query, text):
    q = set(query.lower().split()); t = set(text.lower().split())
    return len(q & t) / (len(q) or 1)

def search_rerank(store, query, k=4, where=None, fetch=20):
    # stage 1: cheap vector recall over a wide net
    cands = search(store, query, k=fetch, where=where)
    # stage 2: precise rerank (real: a cross-encoder; here lexical density)
    cands.sort(key=lambda c: term_overlap(query, c.text), reverse=True)
    return cands[:k]

class Budget:
    def __init__(self, cap_tokens): self.cap = cap_tokens; self.used = {}
    def charge(self, tenant, tokens):
        used = self.used.get(tenant, 0) + tokens
        if used > self.cap:
            raise RuntimeError(f"{tenant} over budget ({used} > {self.cap})")
        self.used[tenant] = used

b = Budget(cap_tokens=1_000)
b.charge("acme", 400); b.charge("acme", 400)
try: b.charge("acme", 400)
except RuntimeError as e: print("blocked:", e)   # acme over budget (1200 > 1000)

doc-rag/
├─ .venv/
├─ requirements.txt
├─ store.py            (Chunk, embed, cosine, Store)
├─ ingest.py           (structure-aware chunking)
├─ answer.py           (grounded, cited, abstaining)
├─ samples/
│  ├─ contract_acme.txt
│  └─ policy_refunds.txt
└─ tests/
   └─ test_docrag.py   (7 offline tests)

Knowledge check check yourself