Module 10 — Project: a complete document assistant

Nine modules of building blocks. This module wires them into a single service that a colleague can actually use: send a question, receive an answer with citations, respecting who they are and what they can see. The code below is intentionally short — production robustness is a matter of putting the pieces of modules 2 to 9 in the right order, not of inventing new ones.

What "done" looks like

The finished service exposes one endpoint. A user posts a question with their identity; the service returns an answer, its citations, and whether the answer came from cache. Behind the endpoint sit the ingestion pipeline (run separately, on a schedule), the vector index (persistent), the retriever, the reranker, the prompt builder, the language model client, and two caches.

# expected request
{
  "question": "How long do we keep accident reports?",
  "user_id": "u_1042"
}

# expected response
{
  "answer": "According to QUAL-047 section 3.2, accident reports are kept for seven years [S1].",
  "citations": [
    {"index": 1, "source_name": "QUAL-047.pdf", "section": "3.2", "page": 8}
  ],
  "from_cache": false,
  "audit": {"cited_indices": [1], "invalid_indices": []}
}

Permissions: enforced at retrieval, not at rendering

The one architectural mistake to avoid: computing an answer, and then filtering the citations for what the user is allowed to see. If the model was shown a passage from a confidential procedure, its answer already carries that content, whether or not you strip the citation link. Permissions must be applied before anything is retrieved.

from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: str
    language: str
    clearance: int       # 1 = public, 2 = internal, 3 = confidential
    departments: set[str]

def load_user_context(user_id: str) -> UserContext:
    row = users_db.get(user_id)
    return UserContext(
        user_id=user_id,
        language=row["language"],
        clearance=row["clearance"],
        departments=set(row["departments"]),
    )

def retrieval_filter(ctx: UserContext) -> dict:
    return {
        "language": ctx.language,
        "access_class": {"$lte": ctx.clearance},
        "department": {"$in": list(ctx.departments | {"all"})},
        "status": "in_force",
    }

The retrieval_filter is passed to coll.query (module 4). No confidential passage ever enters the prompt of a user without clearance; therefore no confidential content leaks into the answer, regardless of what the model tries to do.

The service in one file

from fastapi import FastAPI, HTTPException

app = FastAPI(title="Document Assistant")

@app.post("/ask")
def ask(payload: dict):
    question = payload.get("question", "").strip()
    user_id  = payload.get("user_id", "").strip()
    if not question or not user_id:
        raise HTTPException(400, "question and user_id are required")

    ctx = load_user_context(user_id)
    key = answer_cache_key(question, {"language": ctx.language,
                                       "clearance": ctx.clearance,
                                       "departments": tuple(sorted(ctx.departments))})

    hit = answer_cache.get(key, max_age=3600)
    if hit is not None:
        log_event({**hit, "from_cache": True, "user_id": user_id})
        return {**hit, "from_cache": True}

    passages = retrieve_and_rerank(
        question,
        user_context=vars(ctx),
        first_pass_k=40,
        rerank_keep=5,
    )
    if not passages:
        answer_text = "I do not have enough information in the provided documents to answer."
        result = {"answer": answer_text, "citations": [], "audit": {}}
    else:
        passages = order_by_bracketing(passages)
        prompt   = SYSTEM + "\n\n" + build_prompt(question, passages)
        reply    = llm.generate(prompt, temperature=0.1)
        result   = {
            "answer": reply,
            "citations": [
                {"index": i + 1, "source_name": p["source_name"],
                 "section": p.get("section"), "page": p.get("page")}
                for i, p in enumerate(passages)
            ],
            "audit": audit_citations(reply, passages),
        }

    answer_cache.put(key, result)
    log_event({**result, "from_cache": False, "user_id": user_id})
    return {**result, "from_cache": False}

Two hundred lines of setup make this work — the imports, the model loads, the SQLite connection, the pgvector schema. The core is here, and it is small on purpose. Add complexity only when a measurement (module 8) or a bill (module 9) says you must.

The ingestion pipeline as a scheduled job

Ingestion should never run inside the request handler. Run it on a schedule (once a night, or triggered by a webhook when a document is uploaded).

def ingest_all(docs_dir: str, previous_hashes: dict) -> dict:
    docs = list_documents(docs_dir)   # walks PDF, HTML, DOCX, XLSX, PPTX
    stats = {"ingested": 0, "extraction_errors": 0}

    current_docs = []
    for path in docs:
        try:
            blocks = extract(path)         # module 2
            meta   = infer_metadata(path)  # source_name, language, access_class, ...
            current_docs.append({"path": path, "blocks": blocks, "meta": meta})
            stats["ingested"] += 1
        except Exception as exc:
            log_extraction_error(path, exc)
            stats["extraction_errors"] += 1

    stats.update(incremental_reindex(current_docs, previous_hashes))
    return stats

Every stage from module 2 to module 4 is a function call. Log the return of ingest_all — added, updated, deleted, unchanged, errors — into the same log stream as the answers. A silent drop in "added" from 40 to 0 for a week is how you notice a source folder has moved.

A minimal interface

For internal use, a small HTML page is enough. Two textareas, a submit button, an answer panel with clickable citations that open the source PDF at the right page. Below the answer, the small print users almost never read, but that the compliance team will:

• "Answer generated on 2026-09-06 at 14:32 from N passages of X documents."
• "Cached response, computed on 2026-09-06 at 12:11."
• "The abstention is a design choice, not a technical error."

The interface is where the abstention answer stops looking like a bug. A user who asks something outside the corpus, and reads "I do not have enough information", must be shown that this is expected behaviour and offered a "contact a human" path. A polished refusal is worth a hundred confident hallucinations.

Failure analysis in practice

Every week, sample 30 questions from the log at random and read them by hand. Sort the failing ones into the four buckets from module 8:

• Missed retrieval → look at extraction or chunking (modules 2, 3)
• Ignored context → tighten the abstention rule or lower temperature (module 7)
• Contradictions → check metadata status and effective_date
• Question outside the corpus → is the answer really outside, or is a document missing?

Track the bucket counts over time. A healthy assistant sees the counts stabilise; a drifting corpus sees the "outside" bucket grow, which is a hint that a source folder has stopped being maintained.

Deployment checklist

Not a script — a list of things that must be true before a colleague uses the service:

• Every embedding is normalised and the search uses cosine (module 4).
• Every retrieval passes user_context and enforces the filter (this module).
• The system prompt contains the exact abstention string (module 7).
• temperature is at 0.0 to 0.2 for generation.
• Ingestion errors are logged, not swallowed (module 2).
• Cache TTL is short enough that the last edited document is served within one working day (module 9).
• The evaluation set has been re-run and the four metrics are non-regressing (module 8).
• Every response is logged with hashes, tokens and costs (module 9).

Ten lines. The first breach of any of them is the bug you spend a Saturday chasing three months later.

An "assistant" that never abstains is a fabrication generator

The first time you demo the assistant to a stakeholder, ask a question you know the corpus does not cover. If the answer is not the abstention string, close the demo, fix the prompt, and demo again. A single confident hallucination in the launch meeting can override every metric on your dashboard.

A weekly 30-minute review does more than a rewrite

Reading 30 real interactions per week catches problems no metric ever will: a user's expectation the corpus does not match, a source that has quietly become authoritative, an entire class of questions no one anticipated. Book the slot. Keep the slot.

In summary

• The service is one endpoint that composes retrieval, rerank, prompt and cache; permissions are enforced at retrieval time, never at rendering.
• Ingestion is a scheduled job, separate from the request handler, and its incremental_reindex output is logged alongside every question.
• A minimal interface with clickable citations, cache indicators and a compliance line makes the honest "I do not know" acceptable to users.
• Follow a short deployment checklist and a weekly 30-minute log review — most production RAG failures show up there long before they show up in metrics.

Next module: the course recap, a diagnostic tree for the answer that comes back wrong, and the 40-question exam that certifies you can build and operate a RAG system.