Ship a RAG assistant over your own docs

A repository with pinned dependencies, the API key loaded from the environment, and a docs folder you can ask real questions about.

1. 1.1
   Create the repository and a virtual environment. uv is used here because it resolves and installs in seconds; plain python -m venv and pip work identically.

   bashCopy

   mkdir docs-assistant && cd docs-assistant && git init -b main
   curl -LsSf https://astral.sh/uv/install.sh | sh    # skip if you have uv
   uv venv --python 3.12 && source .venv/bin/activate
   mkdir -p assistant evals tests docs traces .github/workflows
   touch assistant/__init__.py

2. 1.2
   Declare the dependencies in pyproject.toml with upper bounds, and install. The embedding model downloads on first use (about 130 MB).

   tomlCopy

   [project]
   name = "docs-assistant"
   version = "0.1.0"
   requires-python = ">=3.12"
   dependencies = [
     "anthropic>=1.0,<2",
     "chromadb>=1.0,<2",
     "sentence-transformers>=3.3,<4",
     "fastapi>=0.115,<1",
     "uvicorn[standard]>=0.34,<1",
     "pydantic>=2.9,<3",
     "rank-bm25>=0.2,<1",
     "prometheus-fastapi-instrumentator>=7,<8",
     "python-dotenv>=1.0,<2",
   ]
   
   [project.optional-dependencies]
   dev = ["pytest>=8,<9", "httpx>=0.27,<1", "ruff>=0.8,<1"]
   
   [tool.ruff]
   line-length = 100
   
   [tool.setuptools]
   packages = ["assistant"]

3. 1.3
   Install, then put the API key in a git-ignored .env. Nothing else in the project is secret.

   bashCopy

   uv pip install -e '.[dev]'
   printf 'ANTHROPIC_API_KEY=sk-ant-...\n' > .env
   printf '.venv/\n.env\ntraces/\n.chroma/\n__pycache__/\n.pytest_cache/\n.ruff_cache/\nevals/report*.json\n' > .gitignore
   python -c "import anthropic; print(anthropic.Anthropic().models.list().data[0].id)"

   The last line proves the key works by listing models. If you get an authentication error, the key is wrong or the .env was not loaded — export $(cat .env) for a one-off check.
4. 1.4
   Bring in the corpus. Copy Markdown from projects you did (or any docs you know well) into docs/; a few dozen files is plenty. The questions you ask later must be answerable from these files, so pick documents with real content, not stubs.

   bashCopy

   cp ../zero-to-prod/slo/*.md ../zero-to-prod/README.md docs/ 2>/dev/null || true
   cp -r ../zero-to-prod/incidents docs/ 2>/dev/null || true
   cp ../aws-platform/README.md docs/aws-platform.md 2>/dev/null || true
   find docs -name '*.md' | wc -l
   wc -w $(find docs -name '*.md') | tail -1

   No docs of your own yet? git clone --depth 1 https://github.com/prometheus/docs prom-docs && cp -r prom-docs/content/docs docs/prometheus gives a real, well-structured corpus.
5. 1.5
   Write assistant/config.py so every tunable lives in one place and can be overridden by the environment — the evals will sweep some of these.

   pythonCopy

   # assistant/config.py
   import os
   from dataclasses import dataclass, field
   from pathlib import Path
   
   from dotenv import load_dotenv
   
   load_dotenv()
   
   
   @dataclass(frozen=True)
   class Settings:
       docs_dir: Path = field(default_factory=lambda: Path(os.getenv("DOCS_DIR", "docs")))
       chroma_dir: Path = field(default_factory=lambda: Path(os.getenv("CHROMA_DIR", ".chroma")))
       traces_dir: Path = field(default_factory=lambda: Path(os.getenv("TRACES_DIR", "traces")))
       collection: str = os.getenv("COLLECTION", "docs")
       embedding_model: str = os.getenv("EMBEDDING_MODEL", "BAAI/bge-small-en-v1.5")
       llm_model: str = os.getenv("LLM_MODEL", "claude-opus-5")
       judge_model: str = os.getenv("JUDGE_MODEL", "claude-opus-5")
       chunk_words: int = int(os.getenv("CHUNK_WORDS", "300"))
       chunk_overlap: int = int(os.getenv("CHUNK_OVERLAP", "40"))
       top_k: int = int(os.getenv("TOP_K", "5"))
       retrieval: str = os.getenv("RETRIEVAL", "vector")  # vector | bm25 | hybrid
       max_tokens: int = int(os.getenv("MAX_TOKENS", "1024"))
   
   
   settings = Settings()

   claude-opus-5 gives the best answers and the most trustworthy judge. If you want cheaper eval loops while iterating, set LLM_MODEL=claude-sonnet-5; keep the judge on Opus so the grader is stronger than the graded.

Every Markdown file becomes heading-aware chunks with metadata, embedded locally and stored in a persistent Chroma collection that can be rebuilt or refreshed with one command.

1. 2.1
   Write the chunker. Splitting on headings keeps each chunk about one topic, and carrying the heading path into the chunk text gives the embedding (and the model) the context a bare paragraph lacks. Long sections are further split by word count with overlap.

   pythonCopy

   # assistant/chunking.py
   import hashlib
   import re
   from dataclasses import dataclass
   from pathlib import Path
   
   HEADING = re.compile(r"^(#{1,6})\s+(.*)$", re.MULTILINE)
   
   
   @dataclass
   class Chunk:
       id: str
       text: str
       path: str
       heading: str
       position: int
   
   
   def _sections(markdown: str):
       """Yield (heading_path, body) pairs, one per heading, keeping the hierarchy."""
       stack: list[str] = []
       matches = list(HEADING.finditer(markdown))
       if not matches or matches[0].start() > 0:
           yield "", markdown[: matches[0].start() if matches else len(markdown)]
       for i, m in enumerate(matches):
           level = len(m.group(1))
           stack = stack[: level - 1] + [m.group(2).strip()]
           end = matches[i + 1].start() if i + 1 < len(matches) else len(markdown)
           body = markdown[m.end() : end].strip()
           if body:
               yield " > ".join(stack), body
   
   
   def _windows(words: list[str], size: int, overlap: int):
       step = max(1, size - overlap)
       for start in range(0, max(1, len(words) - overlap), step):
           yield words[start : start + size]
           if start + size >= len(words):
               break
   
   
   def chunk_file(path: Path, root: Path, size: int = 300, overlap: int = 40) -> list[Chunk]:
       rel = str(path.relative_to(root))
       title = path.stem.replace("-", " ").replace("_", " ")
       chunks: list[Chunk] = []
       for heading, body in _sections(path.read_text(encoding="utf-8", errors="replace")):
           words = body.split()
           for window in _windows(words, size, overlap):
               text = f"{title} — {heading}\n\n{' '.join(window)}" if heading else f"{title}\n\n{' '.join(window)}"
               cid = hashlib.sha1(f"{rel}|{heading}|{text}".encode()).hexdigest()[:16]
               chunks.append(Chunk(cid, text, rel, heading, len(chunks)))
       return chunks
   
   
   def chunk_dir(root: Path, size: int = 300, overlap: int = 40) -> list[Chunk]:
       out: list[Chunk] = []
       for p in sorted(root.rglob("*.md")):
           out.extend(chunk_file(p, root, size, overlap))
       return out

2. 2.2
   Test the chunker before anything downstream depends on it. Fixtures are tiny Markdown strings, so the tests run in milliseconds and never touch the network.

   pythonCopy

   # tests/test_chunking.py
   from pathlib import Path
   
   from assistant.chunking import chunk_dir, chunk_file
   
   DOC = """# Runbook\n\nIntro line.\n\n## Symptoms\n\nLatency is high. """ + "word " * 700 + """\n\n## Fix\n\nRestart it.\n"""
   
   
   def test_chunks_carry_heading_path(tmp_path: Path):
       (tmp_path / "runbook.md").write_text(DOC)
       chunks = chunk_file(tmp_path / "runbook.md", tmp_path, size=300, overlap=40)
       headings = {c.heading for c in chunks}
       assert "Runbook > Symptoms" in headings and "Runbook > Fix" in headings
       assert all(c.text.startswith("runbook — ") for c in chunks if c.heading)
   
   
   def test_long_sections_are_windowed_with_overlap(tmp_path: Path):
       (tmp_path / "runbook.md").write_text(DOC)
       symptoms = [c for c in chunk_file(tmp_path / "runbook.md", tmp_path, 300, 40) if c.heading.endswith("Symptoms")]
       assert len(symptoms) >= 3
       assert len(set(c.id for c in symptoms)) == len(symptoms)
   
   
   def test_chunk_dir_is_deterministic(tmp_path: Path):
       (tmp_path / "a.md").write_text("# A\n\nalpha\n")
       (tmp_path / "b.md").write_text("# B\n\nbeta\n")
       assert [c.id for c in chunk_dir(tmp_path)] == [c.id for c in chunk_dir(tmp_path)]

3. 2.3
   Write the store: one embedding model loaded once, one Chroma collection with cosine distance, and an upsert keyed by chunk id so re-indexing unchanged files is a no-op. bge models expect a query prefix for retrieval, so the store owns that detail.

   pythonCopy

   # assistant/store.py
   from functools import lru_cache
   
   import chromadb
   from sentence_transformers import SentenceTransformer
   
   from assistant.chunking import Chunk
   from assistant.config import settings
   
   QUERY_PREFIX = "Represent this sentence for searching relevant passages: "
   
   
   @lru_cache(maxsize=1)
   def embedder() -> SentenceTransformer:
       return SentenceTransformer(settings.embedding_model)
   
   
   def embed_texts(texts: list[str]) -> list[list[float]]:
       return embedder().encode(texts, normalize_embeddings=True, batch_size=32).tolist()
   
   
   def embed_query(q: str) -> list[float]:
       return embedder().encode([QUERY_PREFIX + q], normalize_embeddings=True)[0].tolist()
   
   
   @lru_cache(maxsize=1)
   def collection():
       client = chromadb.PersistentClient(path=str(settings.chroma_dir))
       return client.get_or_create_collection(settings.collection, metadata={"hnsw:space": "cosine"})
   
   
   def upsert(chunks: list[Chunk]) -> int:
       col = collection()
       existing = set(col.get(ids=[c.id for c in chunks], include=[])["ids"]) if chunks else set()
       new = [c for c in chunks if c.id not in existing]
       for i in range(0, len(new), 64):
           batch = new[i : i + 64]
           col.upsert(
               ids=[c.id for c in batch],
               documents=[c.text for c in batch],
               embeddings=embed_texts([c.text for c in batch]),
               metadatas=[{"path": c.path, "heading": c.heading, "position": c.position} for c in batch],
           )
       return len(new)
   
   
   def reset() -> None:
       chromadb.PersistentClient(path=str(settings.chroma_dir)).delete_collection(settings.collection)
       collection.cache_clear()
   
   
   def prune(keep_ids: set[str]) -> int:
       col = collection()
       all_ids = col.get(include=[])["ids"]
       stale = [i for i in all_ids if i not in keep_ids]
       if stale:
           col.delete(ids=stale)
       return len(stale)
   
   
   def all_documents() -> tuple[list[str], list[str], list[dict]]:
       got = collection().get(include=["documents", "metadatas"])
       return got["ids"], got["documents"], got["metadatas"]

   Chunk ids are content hashes, so an edited paragraph gets a new id, the old one is pruned, and untouched chunks are skipped. That is incremental indexing in fifteen lines.
4. 2.4
   Write the CLI and build the index. Timing and counts are printed because you will change the chunk size later and want to see the effect.

   pythonCopy

   # index.py
   import argparse
   import time
   from pathlib import Path
   
   from assistant.chunking import chunk_dir
   from assistant.config import settings
   from assistant.store import collection, prune, reset, upsert
   
   
   def main() -> None:
       ap = argparse.ArgumentParser(description="Build or refresh the docs index")
       ap.add_argument("--docs", default=str(settings.docs_dir))
       ap.add_argument("--rebuild", action="store_true", help="delete the collection first")
       args = ap.parse_args()
   
       t0 = time.perf_counter()
       if args.rebuild:
           reset()
       chunks = chunk_dir(Path(args.docs), settings.chunk_words, settings.chunk_overlap)
       added = upsert(chunks)
       removed = prune({c.id for c in chunks})
       files = len({c.path for c in chunks})
       print(f"{files} files, {len(chunks)} chunks, {added} added, {removed} removed, "
             f"{collection().count()} in collection, {time.perf_counter() - t0:.1f}s")
   
   
   if __name__ == "__main__":
       main()

5. 2.5
   Run it twice: the second run should add nothing. Then poke at the collection to see what a chunk looks like — this is what the model will be given.

   bashCopy

   python index.py
   python index.py           # ... 0 added, 0 removed
   python - <<'EOF'
   from assistant.store import collection, embed_query
   col = collection()
   r = col.query(query_embeddings=[embed_query("what do I do when the error budget burns too fast")], n_results=3, include=["documents", "metadatas", "distances"])
   for doc, meta, dist in zip(r["documents"][0], r["metadatas"][0], r["distances"][0]):
       print(f"{dist:.3f}  {meta['path']}  |  {meta['heading']}\n    {doc[:160]}...\n")
   EOF

   Ask three or four questions you know the answer to and check the right chunk is in the top three. If it is not, the problem is upstream of any prompt and no prompt will fix it.

A question goes in; a grounded answer with numbered citations to real chunks comes out, and the model is told to say so when the docs do not contain the answer.

1. 3.1
   Write the retriever with a single interface the API and the evals will share. Vector search now; the other strategies fill in during the improvement phase.

   pythonCopy

   # assistant/retrieve.py
   from dataclasses import dataclass
   
   from assistant.config import settings
   from assistant.store import collection, embed_query
   
   
   @dataclass
   class Hit:
       id: str
       text: str
       path: str
       heading: str
       score: float
   
   
   def vector_search(question: str, k: int) -> list[Hit]:
       r = collection().query(
           query_embeddings=[embed_query(question)], n_results=k,
           include=["documents", "metadatas", "distances"],
       )
       return [
           Hit(i, d, m["path"], m["heading"], 1.0 - dist)
           for i, d, m, dist in zip(r["ids"][0], r["documents"][0], r["metadatas"][0], r["distances"][0])
       ]
   
   
   def retrieve(question: str, k: int | None = None, strategy: str | None = None) -> list[Hit]:
       k = k or settings.top_k
       strategy = strategy or settings.retrieval
       if strategy == "vector":
           return vector_search(question, k)
       raise ValueError(f"unknown retrieval strategy: {strategy}")

2. 3.2
   Write the answerer. The system prompt is the contract: answer only from the numbered sources, cite them inline as [n], and refuse cleanly when they do not cover the question. The sources go in the user turn so the stable system prompt can be cached across requests.

   pythonCopy

   # assistant/answer.py
   import re
   from dataclasses import dataclass, field
   
   import anthropic
   
   from assistant.config import settings
   from assistant.retrieve import Hit, retrieve
   
   SYSTEM = """You answer questions about a set of internal documents.
   
   Rules:
   - Use ONLY the numbered sources provided. Do not use outside knowledge.
   - Cite every claim with the source number in square brackets, like [2]. Cite at most three sources per sentence.
   - If the sources do not contain the answer, reply exactly: "I couldn't find this in the docs." and, if useful, say what the closest source covers.
   - Be concise: a short paragraph or a list of steps. Quote commands verbatim from the sources."""
   
   CITATION = re.compile(r"\[(\d+)\]")
   
   
   @dataclass
   class Answer:
       text: str
       hits: list[Hit]
       cited: list[int] = field(default_factory=list)
       input_tokens: int = 0
       output_tokens: int = 0
       model: str = ""
   
   
   def format_sources(hits: list[Hit]) -> str:
       return "\n\n".join(f"[{i}] {h.path} — {h.heading}\n{h.text}" for i, h in enumerate(hits, start=1))
   
   
   def build_user_message(question: str, hits: list[Hit]) -> str:
       return f"Sources:\n\n{format_sources(hits)}\n\nQuestion: {question}"
   
   
   def parse_citations(text: str, n: int) -> list[int]:
       return sorted({int(m) for m in CITATION.findall(text) if 1 <= int(m) <= n})
   
   
   _client: anthropic.Anthropic | None = None
   
   
   def client() -> anthropic.Anthropic:
       global _client
       if _client is None:
           _client = anthropic.Anthropic()
       return _client
   
   
   def answer(question: str, k: int | None = None, strategy: str | None = None) -> Answer:
       hits = retrieve(question, k, strategy)
       if not hits:
           return Answer("I couldn't find this in the docs.", [], model=settings.llm_model)
       resp = client().messages.create(
           model=settings.llm_model,
           max_tokens=settings.max_tokens,
           system=[{"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}}],
           messages=[{"role": "user", "content": build_user_message(question, hits)}],
       )
       text = "".join(b.text for b in resp.content if b.type == "text").strip()
       return Answer(
           text=text, hits=hits, cited=parse_citations(text, len(hits)),
           input_tokens=resp.usage.input_tokens, output_tokens=resp.usage.output_tokens,
           model=resp.model,
       )
   
   
   def answer_stream(question: str, k: int | None = None, strategy: str | None = None):
       """Yield (event, payload): ('sources', hits) first, then ('delta', text)..., then ('done', Answer)."""
       hits = retrieve(question, k, strategy)
       yield "sources", hits
       if not hits:
           yield "done", Answer("I couldn't find this in the docs.", [], model=settings.llm_model)
           return
       with client().messages.stream(
           model=settings.llm_model,
           max_tokens=settings.max_tokens,
           system=[{"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}}],
           messages=[{"role": "user", "content": build_user_message(question, hits)}],
       ) as stream:
           for delta in stream.text_stream:
               yield "delta", delta
           final = stream.get_final_message()
       text = "".join(b.text for b in final.content if b.type == "text").strip()
       yield "done", Answer(text, hits, parse_citations(text, len(hits)),
                            final.usage.input_tokens, final.usage.output_tokens, final.model)

   cache_control on the system block means repeated requests pay a fraction of the input price for that prefix. The sources change per request, so they sit after it — order matters for caching.
3. 3.3
   Test the prompt assembly and citation parsing without the network. A fake client stands in for Anthropic so the test suite is free and deterministic.

   pythonCopy

   # tests/test_answer.py
   from types import SimpleNamespace
   
   import pytest
   
   from assistant import answer as mod
   from assistant.retrieve import Hit
   
   HITS = [Hit("a", "Restart with `systemctl restart app`.", "runbook.md", "Runbook > Fix", 0.9),
           Hit("b", "Latency alert fires above 400 ms.", "slo.md", "SLO > Latency", 0.8)]
   
   
   class FakeMessages:
       def __init__(self, text):
           self.text = text
           self.calls = []
   
       def create(self, **kw):
           self.calls.append(kw)
           return SimpleNamespace(content=[SimpleNamespace(type="text", text=self.text)],
                                  usage=SimpleNamespace(input_tokens=100, output_tokens=20), model="fake")
   
   
   @pytest.fixture
   def fake(monkeypatch):
       fm = FakeMessages("Restart it [1]. The alert threshold is 400 ms [2][7].")
       monkeypatch.setattr(mod, "_client", SimpleNamespace(messages=fm))
       monkeypatch.setattr(mod, "retrieve", lambda q, k=None, s=None: HITS)
       return fm
   
   
   def test_sources_are_numbered_and_question_last():
       msg = mod.build_user_message("how do I fix it?", HITS)
       assert msg.index("[1] runbook.md") < msg.index("[2] slo.md") < msg.index("Question: how do I fix it?")
   
   
   def test_answer_parses_only_valid_citations(fake):
       a = mod.answer("how do I fix it?")
       assert a.cited == [1, 2]            # [7] is out of range and dropped
       assert a.input_tokens == 100 and a.model == "fake"
       assert fake.calls[0]["system"][0]["cache_control"] == {"type": "ephemeral"}
   
   
   def test_no_hits_short_circuits(monkeypatch):
       monkeypatch.setattr(mod, "retrieve", lambda q, k=None, s=None: [])
       a = mod.answer("anything")
       assert a.text.startswith("I couldn't find") and a.hits == []

4. 3.4
   Ask it something for real from the command line, and something the docs cannot answer. Read the whole output: the citations must point at chunks that actually support the sentence.

   bashCopy

   python - <<'EOF'
   from assistant.answer import answer
   for q in ["What burn rate pages the on-call, and over which windows?", "What is the capital of France?"]:
       a = answer(q)
       print("Q:", q)
       print(a.text)
       for i in a.cited:
           h = a.hits[i - 1]
           print(f"  [{i}] {h.path} — {h.heading}")
       print(f"  tokens in/out: {a.input_tokens}/{a.output_tokens}\n")
   EOF

   The second question must come back as "I couldn't find this in the docs." If the model answers it anyway, the refusal rule is being ignored — tighten the wording before building anything else on top.
5. 3.5
   Commit the core and run the suite.

   bashCopy

   ruff check . && pytest -q
   git add . && git commit -m "feat: chunking, local embeddings, Chroma index, grounded answers with citations"
   gh repo create docs-assistant --public --source=. --remote=origin --push

POST /ask returns a structured answer, POST /ask/stream streams tokens as server-sent events, every request writes a trace you can replay, and /metrics exposes request counts and latency for Prometheus.

1. 4.1
   Write the tracer first: one JSON line per request with the question, retrieval hits and scores, the citations, tokens, an estimated cost and timings. This is the record you will open when someone says "it gave a wrong answer yesterday".

   pythonCopy

   # assistant/tracing.py
   import json
   import time
   import uuid
   from datetime import datetime, timezone
   
   from assistant.config import settings
   
   # USD per million tokens; keep current with the pricing page
   PRICES = {"claude-opus-5": (5.0, 25.0), "claude-sonnet-5": (3.0, 15.0)}
   
   
   def estimate_cost(model: str, tokens_in: int, tokens_out: int) -> float:
       base = next((v for k, v in PRICES.items() if model.startswith(k)), (0.0, 0.0))
       return round((tokens_in * base[0] + tokens_out * base[1]) / 1_000_000, 6)
   
   
   class Trace:
       def __init__(self, question: str, strategy: str, k: int):
           self.id = uuid.uuid4().hex[:12]
           self.t0 = time.perf_counter()
           self.data: dict = {
               "trace_id": self.id, "ts": datetime.now(timezone.utc).isoformat(),
               "question": question, "strategy": strategy, "k": k, "timings_ms": {},
           }
   
       def mark(self, name: str) -> None:
           self.data["timings_ms"][name] = round((time.perf_counter() - self.t0) * 1000, 1)
   
       def finish(self, answer) -> dict:
           self.mark("total")
           self.data.update({
               "hits": [{"id": h.id, "path": h.path, "heading": h.heading, "score": round(h.score, 4)} for h in answer.hits],
               "answer": answer.text, "cited": answer.cited, "model": answer.model,
               "tokens": {"in": answer.input_tokens, "out": answer.output_tokens},
               "cost_usd": estimate_cost(answer.model, answer.input_tokens, answer.output_tokens),
           })
           settings.traces_dir.mkdir(parents=True, exist_ok=True)
           with open(settings.traces_dir / f"{datetime.now(timezone.utc):%Y-%m-%d}.jsonl", "a") as f:
               f.write(json.dumps(self.data) + "\n")
           return self.data

2. 4.2
   Write the API. Pydantic models define the contract; the streaming endpoint sends the sources first (so a UI can render them immediately), then text deltas, then a final event with the trace id.

   pythonCopy

   # assistant/api.py
   import json
   
   from fastapi import FastAPI
   from fastapi.responses import StreamingResponse
   from prometheus_fastapi_instrumentator import Instrumentator
   from pydantic import BaseModel, Field
   
   from assistant import answer as core
   from assistant.config import settings
   from assistant.tracing import Trace
   
   app = FastAPI(title="docs-assistant")
   Instrumentator().instrument(app).expose(app, endpoint="/metrics")
   
   
   class AskRequest(BaseModel):
       question: str = Field(min_length=3, max_length=2000)
       k: int | None = Field(default=None, ge=1, le=20)
       strategy: str | None = None
   
   
   class Source(BaseModel):
       n: int
       path: str
       heading: str
       score: float
   
   
   class AskResponse(BaseModel):
       trace_id: str
       answer: str
       sources: list[Source]
       cited: list[int]
       model: str
       tokens_in: int
       tokens_out: int
   
   
   @app.get("/health")
   def health():
       return {"status": "ok", "model": settings.llm_model, "retrieval": settings.retrieval}
   
   
   @app.post("/ask", response_model=AskResponse)
   def ask(req: AskRequest):
       trace = Trace(req.question, req.strategy or settings.retrieval, req.k or settings.top_k)
       a = core.answer(req.question, req.k, req.strategy)
       trace.finish(a)
       return AskResponse(
           trace_id=trace.id, answer=a.text, cited=a.cited, model=a.model,
           tokens_in=a.input_tokens, tokens_out=a.output_tokens,
           sources=[Source(n=i, path=h.path, heading=h.heading, score=round(h.score, 4)) for i, h in enumerate(a.hits, 1)],
       )
   
   
   @app.post("/ask/stream")
   def ask_stream(req: AskRequest):
       trace = Trace(req.question, req.strategy or settings.retrieval, req.k or settings.top_k)
   
       def events():
           for kind, payload in core.answer_stream(req.question, req.k, req.strategy):
               if kind == "sources":
                   trace.mark("retrieval")
                   data = [{"n": i, "path": h.path, "heading": h.heading} for i, h in enumerate(payload, 1)]
                   yield f"event: sources\ndata: {json.dumps(data)}\n\n"
               elif kind == "delta":
                   yield f"event: delta\ndata: {json.dumps(payload)}\n\n"
               else:
                   trace.finish(payload)
                   yield f"event: done\ndata: {json.dumps({'trace_id': trace.id, 'cited': payload.cited})}\n\n"
   
       return StreamingResponse(events(), media_type="text/event-stream",
                                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

3. 4.3
   Test the API with the same fake model, then run it for real and stream an answer with curl.

   pythonCopy

   # tests/test_api.py
   from types import SimpleNamespace
   
   from fastapi.testclient import TestClient
   
   from assistant import answer as core
   from assistant.api import app
   from assistant.retrieve import Hit
   
   
   def test_ask_returns_sources_and_trace(monkeypatch, tmp_path):
       from assistant import tracing
       monkeypatch.setattr(tracing, "settings", SimpleNamespace(traces_dir=tmp_path))
       hits = [Hit("a", "text", "runbook.md", "Runbook > Fix", 0.9)]
       monkeypatch.setattr(core, "answer", lambda q, k=None, s=None: core.Answer("Do X [1].", hits, [1], 10, 5, "fake"))
       r = TestClient(app).post("/ask", json={"question": "how do I fix it?"})
       assert r.status_code == 200
       body = r.json()
       assert body["cited"] == [1] and body["sources"][0]["path"] == "runbook.md" and len(body["trace_id"]) == 12
       assert list(tmp_path.glob("*.jsonl"))
   
   
   def test_ask_rejects_short_question():
       assert TestClient(app).post("/ask", json={"question": "hi"}).status_code == 422

   The tracer's settings is swapped for a stand-in that only has traces_dir, so tests never write into traces/. Settings is frozen, which is exactly why the module reference is patched rather than the attribute.
4. 4.4
   Run the server and try both endpoints. Then open the trace the request wrote.

   bashCopy

   uvicorn assistant.api:app --port 8000 &
   sleep 3
   curl -s localhost:8000/ask -H 'content-type: application/json' \
     -d '{"question":"How do I acknowledge a page and where is the runbook?"}' | python3 -m json.tool
   curl -sN localhost:8000/ask/stream -H 'content-type: application/json' \
     -d '{"question":"What are the severity levels?"}'
   tail -1 traces/$(date -u +%F).jsonl | python3 -m json.tool | head -40
   curl -s localhost:8000/metrics | grep http_request_duration_seconds_count
   kill %1

A golden set of questions with known sources and answers, retrieval metrics (hit@k, MRR) that cost nothing, answer metrics (faithfulness, correctness) graded by a stronger model, and a report you can compare between commits.

1. 5.1
   Write the golden set: at least twenty questions, each with the file that answers it and a reference answer written by you from the docs. Include three or four questions the docs do *not* answer, with "expected_answer": null, so refusals are measured too.

   textCopy

   # evals/golden.jsonl — one JSON object per line, e.g.
   {"id": "q01", "question": "Which burn rates page and over which windows?", "expected_paths": ["SLO.md"], "expected_answer": "14.4x over 1h/5m and 6x over 6h/30m page; 3x over 1d/2h and 1x over 3d/6h open tickets."}
   {"id": "q02", "question": "What is the first thing the incident commander does after declaring?", "expected_paths": ["incidents/PROCESS.md", "incidents/roles/commander.md"], "expected_answer": "Assigns roles (investigator, comms, scribe) and posts the first status update within 10 minutes."}
   {"id": "q18", "question": "How do I reset the Grafana admin password?", "expected_paths": [], "expected_answer": null}

   Writing the reference answers is the slowest part of the whole project and the most valuable: it forces you to read the docs the way the assistant will. Keep questions phrased the way a colleague would ask, not the way the document is worded.
2. 5.2
   Write the judge. Two rubrics, each returning a single integer, on the answer and the sources it was shown — faithfulness asks whether every claim is supported by the sources; correctness compares against your reference answer.

   pythonCopy

   # evals/judge.py
   import re
   
   import anthropic
   
   from assistant.config import settings
   
   FAITHFULNESS = """You are grading whether an answer is supported by the sources it cites.
   Score 1 if every factual claim in the answer is supported by the sources, 0 otherwise.
   An answer of "I couldn't find this in the docs." is always supported (score 1).
   
   Sources:
   {sources}
   
   Answer:
   {answer}
   
   Reply with only the digit 0 or 1."""
   
   CORRECTNESS = """You are grading an answer against a reference written by a domain expert.
   Score 2 if the answer conveys the same facts as the reference (wording may differ),
   1 if it is partially correct or incomplete, 0 if it is wrong or answers a different question.
   If the reference is "NONE", the correct behaviour is to say the docs do not contain the answer: score 2 for that, 0 for anything else.
   
   Question: {question}
   Reference: {reference}
   Answer: {answer}
   
   Reply with only the digit 0, 1 or 2."""
   
   _client = anthropic.Anthropic()
   
   
   def _score(prompt: str, valid: str) -> int:
       r = _client.messages.create(model=settings.judge_model, max_tokens=5,
                                   messages=[{"role": "user", "content": prompt}])
       text = "".join(b.text for b in r.content if b.type == "text")
       m = re.search(f"[{valid}]", text)
       return int(m.group(0)) if m else 0
   
   
   def faithfulness(answer: str, sources: str) -> int:
       return _score(FAITHFULNESS.format(sources=sources, answer=answer), "01")
   
   
   def correctness(question: str, reference: str | None, answer: str) -> int:
       return _score(CORRECTNESS.format(question=question, reference=reference or "NONE", answer=answer), "012")

3. 5.3
   Write the runner. Retrieval metrics run over every question for free; the answer metrics call the model, so --retrieval-only exists for quick loops. The report is JSON with a summary and every per-question row, so a regression can be traced to one question.

   pythonCopy

   # evals/run.py
   import argparse
   import json
   import statistics
   import sys
   from pathlib import Path
   
   from assistant.answer import answer, format_sources
   from assistant.config import settings
   from assistant.retrieve import retrieve
   from evals.judge import correctness, faithfulness
   
   
   def retrieval_metrics(expected: list[str], hits) -> dict:
       ranks = [i for i, h in enumerate(hits, 1) if any(h.path.endswith(p) for p in expected)]
       return {"hit@k": int(bool(ranks)), "rr": 1.0 / ranks[0] if ranks else 0.0}
   
   
   def main() -> int:
       ap = argparse.ArgumentParser()
       ap.add_argument("--golden", default="evals/golden.jsonl")
       ap.add_argument("--retrieval-only", action="store_true")
       ap.add_argument("--strategy", default=settings.retrieval)
       ap.add_argument("--k", type=int, default=settings.top_k)
       ap.add_argument("--out", default="evals/report.json")
       ap.add_argument("--min-hit", type=float, default=0.0)
       ap.add_argument("--min-correct", type=float, default=0.0)
       args = ap.parse_args()
   
       rows = []
       for line in Path(args.golden).read_text().splitlines():
           if not line.strip():
               continue
           q = json.loads(line)
           row = {"id": q["id"], "question": q["question"]}
           if q["expected_paths"]:
               row.update(retrieval_metrics(q["expected_paths"], retrieve(q["question"], args.k, args.strategy)))
           if not args.retrieval_only:
               a = answer(q["question"], args.k, args.strategy)
               row["answer"] = a.text
               row["faithful"] = faithfulness(a.text, format_sources(a.hits))
               row["correct"] = correctness(q["question"], q["expected_answer"], a.text) / 2
               row["tokens_in"] = a.input_tokens
           rows.append(row)
           print(f"{q['id']}: " + " ".join(f"{k}={v}" for k, v in row.items() if k in ('hit@k', 'rr', 'faithful', 'correct')))
   
       def mean(key):
           vals = [r[key] for r in rows if key in r]
           return round(statistics.mean(vals), 3) if vals else None
   
       summary = {"n": len(rows), "strategy": args.strategy, "k": args.k, "model": settings.llm_model,
                  "hit@k": mean("hit@k"), "mrr": mean("rr"), "faithfulness": mean("faithful"), "correctness": mean("correct")}
       Path(args.out).write_text(json.dumps({"summary": summary, "rows": rows}, indent=2))
       print(json.dumps(summary))
   
       failed = (summary["hit@k"] is not None and summary["hit@k"] < args.min_hit) or \
                (summary["correctness"] is not None and summary["correctness"] < args.min_correct)
       return 1 if failed else 0
   
   
   if __name__ == "__main__":
       sys.exit(main())

4. 5.4
   Run the retrieval-only evals first; they take seconds and cost nothing. Then the full run. Write the numbers down — they are your baseline.

   bashCopy

   touch evals/__init__.py
   python -m evals.run --retrieval-only
   python -m evals.run --out evals/report-baseline.json
   python3 -c "import json; print(json.load(open('evals/report-baseline.json'))['summary'])"

   Typical first numbers on a small corpus: hit@5 around 0.8, MRR 0.6, faithfulness above 0.9, correctness around 0.7. Whatever yours are, open the rows with correct = 0 and read the answer next to the reference — most failures are retrieval misses, not model mistakes.
5. 5.5
   Look at one failure properly, using the trace rather than guessing: was the right chunk retrieved but not cited, retrieved and mis-read, or never retrieved? Each has a different fix.

   bashCopy

   python3 - <<'EOF'
   import json
   rep = json.load(open('evals/report-baseline.json'))
   bad = [r for r in rep['rows'] if r.get('correct', 1) < 1]
   for r in bad[:5]:
       print(r['id'], '| hit@k', r.get('hit@k'), '| faithful', r.get('faithful'))
       print('  Q:', r['question'])
       print('  A:', r.get('answer', '')[:300].replace('\n', ' '))
   EOF

Add keyword search fused with vector search, sweep chunk size and k, and keep whichever configuration the golden set says is best.

1. 6.1
   Add BM25 and reciprocal-rank fusion to the retriever. Vectors find paraphrases; BM25 finds exact identifiers, flags and error strings that embeddings blur. Fusing by rank rather than score sidesteps the fact that the two scores are not comparable.

   pythonCopy

   # assistant/retrieve.py — add below vector_search, and extend retrieve()
   from functools import lru_cache
   
   from rank_bm25 import BM25Okapi
   
   from assistant.store import all_documents
   
   
   @lru_cache(maxsize=1)
   def _bm25():
       ids, docs, metas = all_documents()
       return ids, docs, metas, BM25Okapi([d.lower().split() for d in docs])
   
   
   def bm25_search(question: str, k: int) -> list[Hit]:
       ids, docs, metas, index = _bm25()
       scores = index.get_scores(question.lower().split())
       top = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:k]
       return [Hit(ids[i], docs[i], metas[i]["path"], metas[i]["heading"], float(scores[i])) for i in top if scores[i] > 0]
   
   
   def rrf(*rankings: list[Hit], k: int, c: int = 60) -> list[Hit]:
       fused: dict[str, float] = {}
       by_id: dict[str, Hit] = {}
       for ranking in rankings:
           for rank, hit in enumerate(ranking, 1):
               fused[hit.id] = fused.get(hit.id, 0.0) + 1.0 / (c + rank)
               by_id.setdefault(hit.id, hit)
       return [Hit(i, by_id[i].text, by_id[i].path, by_id[i].heading, s)
               for i, s in sorted(fused.items(), key=lambda kv: kv[1], reverse=True)[:k]]
   
   
   def retrieve(question: str, k: int | None = None, strategy: str | None = None) -> list[Hit]:
       k = k or settings.top_k
       strategy = strategy or settings.retrieval
       if strategy == "vector":
           return vector_search(question, k)
       if strategy == "bm25":
           return bm25_search(question, k)
       if strategy == "hybrid":
           return rrf(vector_search(question, k * 2), bm25_search(question, k * 2), k=k)
       raise ValueError(f"unknown retrieval strategy: {strategy}")

   Replace the earlier retrieve() with this one. The BM25 index is built from the collection on first use and cached; after re-indexing, restart the process (or call _bm25.cache_clear()).
2. 6.2
   Compare the three strategies on retrieval metrics — free and fast — then run the full evals only for the winner.

   bashCopy

   for s in vector bm25 hybrid; do
     python -m evals.run --retrieval-only --strategy "$s" --out "evals/report-$s.json" | tail -1
   done
   for k in 3 5 8; do
     python -m evals.run --retrieval-only --strategy hybrid --k "$k" --out "evals/report-hybrid-k$k.json" | tail -1
   done

3. 6.3
   Try one chunk-size change as well, since it is the other lever that moves retrieval most. Re-index with a different CHUNK_WORDS, re-run, and restore whichever was better. Then set the winning defaults in config.py (or the environment) and run the full evals once more to get the new baseline.

   bashCopy

   CHUNK_WORDS=180 CHUNK_OVERLAP=30 python index.py --rebuild
   python -m evals.run --retrieval-only --strategy hybrid --out evals/report-hybrid-180.json | tail -1
   python index.py --rebuild     # back to the default size if 180 was not better
   RETRIEVAL=hybrid python -m evals.run --out evals/report-hybrid-full.json | tail -1
   git add assistant/retrieve.py evals && git commit -m "feat(retrieval): BM25 + RRF hybrid; evals show hit@5 x -> y"

   Put the actual numbers in the commit message. A retrieval change without a measured delta is a guess; with one it is engineering.

Pull requests run the unit tests and the retrieval evals; merges to main also run the graded evals against a threshold; the image on GHCR runs anywhere with one environment variable.

1. 7.1
   Write the workflow. Unit tests and retrieval evals need no key, so they run on every PR from anyone. The graded evals use the repository secret and run on main and on PRs from the repo itself, with thresholds a little under your baseline so noise does not block merges.

   yamlCopy

   name: ci
   
   on:
     pull_request:
     push:
       branches: [main]
   
   jobs:
     test:
       runs-on: ubuntu-latest
       steps:
         - uses: actions/checkout@v4
         - uses: astral-sh/setup-uv@v5
         - run: uv venv --python 3.12 && uv pip install -e '.[dev]'
         - run: .venv/bin/ruff check .
         - run: .venv/bin/pytest -q
         - name: Retrieval evals (no API key needed)
           run: |
             .venv/bin/python index.py
             .venv/bin/python -m evals.run --retrieval-only --strategy hybrid --min-hit 0.75
   
     graded-evals:
       needs: test
       if: github.event_name == 'push' || github.event.pull_request.head.repo.full_name == github.repository
       runs-on: ubuntu-latest
       env:
         ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
         RETRIEVAL: hybrid
       steps:
         - uses: actions/checkout@v4
         - uses: astral-sh/setup-uv@v5
         - run: uv venv --python 3.12 && uv pip install -e '.[dev]'
         - run: .venv/bin/python index.py
         - run: .venv/bin/python -m evals.run --min-correct 0.65 --out evals/report.json
         - uses: actions/upload-artifact@v4
           if: always()
           with:
             name: eval-report
             path: evals/report.json
   
     image:
       needs: test
       if: github.event_name == 'push'
       runs-on: ubuntu-latest
       permissions:
         contents: read
         packages: write
       steps:
         - uses: actions/checkout@v4
         - uses: docker/login-action@v3
           with:
             registry: ghcr.io
             username: ${{ github.actor }}
             password: ${{ secrets.GITHUB_TOKEN }}
         - uses: docker/build-push-action@v6
           with:
             context: .
             push: true
             tags: |
               ghcr.io/${{ github.repository }}:latest
               ghcr.io/${{ github.repository }}:${{ github.sha }}

   gh secret set ANTHROPIC_API_KEY before the first push. The graded job costs roughly a few cents per run with twenty questions; the retrieval job is free, which is why it runs everywhere.
2. 7.2
   Write the Dockerfile. The embedding model is downloaded at build time so the container starts in seconds and works without internet access to Hugging Face. The docs are copied in and indexed at build as well, so the image is self-contained; mount a volume over /app/docs and re-run index.py to change the corpus.

   dockerfileCopy

   # syntax=docker/dockerfile:1
   FROM python:3.12-slim AS build
   COPY --from=ghcr.io/astral-sh/uv:0.5 /uv /bin/uv
   WORKDIR /app
   COPY pyproject.toml .
   COPY assistant ./assistant
   COPY index.py .
   RUN uv venv /venv && VIRTUAL_ENV=/venv uv pip install --no-cache .
   ENV PATH="/venv/bin:$PATH" HF_HOME=/models
   RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('BAAI/bge-small-en-v1.5')"
   COPY docs ./docs
   RUN python index.py
   
   FROM python:3.12-slim
   RUN useradd --create-home --uid 10001 app
   WORKDIR /app
   COPY --from=build /venv /venv
   COPY --from=build /models /models
   COPY --from=build /app /app
   RUN chown -R app:app /app
   ENV PATH="/venv/bin:$PATH" HF_HOME=/models HF_HUB_OFFLINE=1 PYTHONUNBUFFERED=1 TRACES_DIR=/app/traces
   USER app
   EXPOSE 8000
   HEALTHCHECK --interval=30s --timeout=3s CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health')"
   CMD ["uvicorn", "assistant.api:app", "--host", "0.0.0.0", "--port", "8000"]

   The image is about 1.5 GB because of PyTorch; that is the cost of local embeddings. A hosted embedding API makes the image small and the per-request cost non-zero — a trade-off worth knowing, not a mistake.
3. 7.3
   Build and run it locally with Compose, mounting a traces volume so the JSONL survives restarts.

   yamlCopy

   # compose.yml
   services:
     assistant:
       build: .
       image: docs-assistant:dev
       environment:
         ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY}
         RETRIEVAL: hybrid
         LLM_MODEL: ${LLM_MODEL:-claude-opus-5}
       ports: ["8000:8000"]
       volumes:
         - traces:/app/traces
       restart: unless-stopped
   
   volumes:
     traces:

4. 7.4
   Run it, ask through the container, and confirm it starts with the network to Hugging Face blocked (the HF_HUB_OFFLINE=1 guarantee).

   bashCopy

   docker compose up -d --build
   docker compose logs -f assistant &
   sleep 5; kill %1
   curl -s localhost:8000/health
   curl -s localhost:8000/ask -H 'content-type: application/json' -d '{"question":"What are the severity levels?"}' | python3 -m json.tool | head -20
   docker compose down

5. 7.5
   Push, let CI build the image, and deploy it somewhere real. Two free-tier options: a Hugging Face Docker Space (the same route as the SkillMatch project — add ANTHROPIC_API_KEY as a Space secret and set the port to 8000 in the README front matter), or the EC2 host from the zero-to-prod project, where it is one more service in deploy/compose.yml using the GHCR image. Either way, write the URL and the eval baseline into the README.

   bashCopy

   gh secret set ANTHROPIC_API_KEY < <(grep ANTHROPIC_API_KEY .env | cut -d= -f2)
   git add . && git commit -m "ci: tests, eval gate, image; docker: self-contained image with model and index"
   git push
   gh run watch
   # On the zero-to-prod host (deploy/compose.yml), add:
   #   assistant:
   #     image: ghcr.io/YOUR_USER/docs-assistant:latest
   #     environment: {ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY}, RETRIEVAL: hybrid}
   #     ports: ["8080:8000"]
   # then: docker compose pull assistant && docker compose up -d assistant

   Point Prometheus at /metrics on the new service and you have request rate, latency and error ratio for free — the same SLO machinery from the SRE track applies to an LLM service without changes.

python index.py is very slow or the first request takes a minute

The embedding model downloads on first use (about 130 MB) and PyTorch initialises; both are one-time. If every run is slow, HF_HOME may point at a non-persistent directory. On Apple Silicon, PYTORCH_ENABLE_MPS_FALLBACK=1 avoids occasional errors; CPU is fine for this model size.

Answers cite [1] for everything or make claims the sources do not contain

Look at the trace: if the retrieved chunks do not contain the fact, the model is filling gaps — that is a retrieval miss, so improve chunking or switch to hybrid. If they do contain it and the citations are still wrong, shorten the chunks (long chunks dilute) and restate the citation rule as the last line of the system prompt, where instructions are followed most reliably.

evals.run faithfulness is high but correctness is low

The assistant is honest but under-informed: it says less than the reference or refuses. Check hit@k for those rows — almost always the right chunk was not in the top k. Raise k, use hybrid retrieval, or split the question's target document into smaller sections with headings.

RateLimitError or overloaded_error during the graded evals

The SDK retries twice by default; for a 20-question run that is usually enough. Add max_retries=5 on the client in judge.py, or run the evals with fewer parallel jobs (they are sequential here). Reduce JUDGE_MODEL load by caching judgments per (question, answer) hash if you re-run often.

The container answers in English even when the docs are in another language, or embeddings score everything similarly

bge-small-en-v1.5 is English-only. For other languages switch EMBEDDING_MODEL to a multilingual model such as BAAI/bge-m3 (larger) and rebuild the index; the rest of the code does not change.

chromadb fails to import or the collection is empty after a rebuild inside Docker

Chroma stores the collection in CHROMA_DIR (.chroma by default). Inside the image that path is /app/.chroma, built in the build stage; if you mounted a volume over /app, it hid the index. Mount only /app/docs or /app/traces, never the whole app directory.

• Add a reranker (BAAI/bge-reranker-base through sentence-transformers' CrossEncoder) over the top 20 hybrid hits and measure whether MRR improves enough to justify the latency.
• Capture user feedback: a POST /feedback with the trace id and a thumbs up/down, then a weekly script that turns thumbs-down traces into new golden-set candidates — the loop from the LLM observability module.
• Give it a chat memory: pass the last three turns and rewrite the question with them before retrieval (query rewriting), and add multi-turn cases to the golden set.
• Swap the JSONL tracer for OpenTelemetry spans exported to the Grafana stack from the SRE track, so retrieval and generation latency show up next to your other services.
• Turn the answerer into an MCP server (see the MCP module) so any MCP-capable client can query your docs with the same grounding rules.