Claude Code, part 5: RAG, giving Claude your own knowledge

Claude knows a lot, but not your runbooks, your wiki or last quarter’s incident notes, and some of that is too large to paste into a conversation. Retrieval-augmented generation (RAG) fixes that: find the few passages that matter, then hand only those to the model. This part explains the moving pieces and builds a working pipeline that Claude Code can query through MCP (part 4).

The idea in one picture

 INDEXING (once, and when docs change)
 docs ──▶ split into chunks ──▶ embed each chunk ──▶ store vectors + text

 ANSWERING (every question)
 question ──▶ embed ──▶ find the most similar chunks ──▶ put them in the prompt ──▶ Claude answers, citing them

• Retrieval finds relevant text.
• Augmentation adds it to the prompt.
• Generation is Claude answering from it.

The model never “learns” your documents; it reads the right page at the right moment, which is why RAG stays current: re-index and the answers change.

The concepts

Embeddings

An embedding model turns text into a vector (a list of numbers, here 384 of them) so that texts with similar meaning land close together. “How do I renew TLS certificates?” and “Caddy obtains and renews Let’s Encrypt certs” share almost no words, but their vectors are close.

Closeness is usually measured with cosine similarity: 1.0 means the same direction, around 0 means unrelated. If you normalise every vector to length 1, cosine similarity is a dot product, and a matrix multiply scores thousands of chunks at once.

Anthropic doesn’t offer its own embedding model. Its docs recommend Voyage AI (voyage-4, voyage-4-lite, and voyage-code-4 for code). For a free, private, local option, small open models like bge-small run on a laptop CPU.

Chunking

You embed chunks, not whole documents. A good chunk is about one idea:

• Too big and its vector blurs several topics, so it matches nothing well and wastes context.
• Too small and it loses the surrounding meaning (“it” refers to what?).
• Overlap a little between chunks so a sentence isn’t cut in half across two.
• Split on structure (headings, paragraphs) before splitting on length.

A few hundred to about a thousand characters with a small overlap is a sensible default for prose. Measure on your own questions.

Retrieval quality boosters

Anthropic’s contextual retrieval research reported 35% fewer retrieval failures from contextual embeddings alone, 49% when combined with contextual BM25, and 67% with reranking added.

When you don’t need RAG

• The data fits in context. A few files? Just let Claude read them. Claude Code already explores code with its own search tools (glob, grep, read), which works better for code than a vector index.
• The data lives in a system with a good API. An MCP server that queries the source directly (part 4) is fresher and simpler.

RAG earns its keep with large, mostly-prose knowledge, such as docs, wikis, tickets and notes, where keyword search misses things and full documents won’t fit.

Build it: a local RAG pipeline in one file

rag.py indexes a folder of Markdown or text files into SQLite, searches it, and serves that search to Claude over MCP. It uses NumPy for the maths and runs with no external services:

• Embeddings: the local bge-small-en-v1.5 model through fastembed (about 70 MB, downloaded on first use), or Voyage AI if VOYAGE_API_KEY is set.
• Storage: a single SQLite file; vectors are stored as bytes.
• Search: brute-force cosine similarity, which is fine up to tens of thousands of chunks.

python -m venv .venv && source .venv/bin/activate
pip install numpy fastembed "mcp[cli]"     # add voyageai to use Voyage instead

rag.py:

"""Minimal local RAG: index Markdown/text files, search them, and serve the search to Claude over MCP.

  python rag.py index ./docs        # chunk + embed + store in rag.db
  python rag.py search "question"   # try a query from the terminal
  python rag.py serve               # run as an MCP server (stdio) for Claude Code

Embeddings: a free local model (fastembed, bge-small) by default, or Voyage AI if VOYAGE_API_KEY is set.
"""
import json
import os
import re
import sqlite3
import sys
from pathlib import Path

import numpy as np

DB = Path(os.environ.get("RAG_DB", Path(__file__).with_name("rag.db")))
CHUNK_CHARS, OVERLAP = 1200, 200

# ---------- 1. Embeddings ---------------------------------------------------
_local = None

def embed(texts: list[str], kind: str) -> np.ndarray:
    """Return one L2-normalised vector per text. kind is "document" or "query"."""
    if os.environ.get("VOYAGE_API_KEY"):
        import voyageai  # pip install voyageai
        vecs = voyageai.Client().embed(texts, model="voyage-4", input_type=kind).embeddings
    else:
        global _local
        if _local is None:
            from fastembed import TextEmbedding  # pip install fastembed
            _local = TextEmbedding("BAAI/bge-small-en-v1.5")  # ~70 MB, downloaded once
        gen = _local.query_embed(texts) if kind == "query" else _local.passage_embed(texts)
        vecs = list(gen)
    m = np.asarray(vecs, dtype=np.float32)
    return m / np.linalg.norm(m, axis=1, keepdims=True)

# ---------- 2. Chunking -----------------------------------------------------
def chunk(text: str) -> list[str]:
    """Split on Markdown headings, then pack paragraphs into ~CHUNK_CHARS chunks with overlap."""
    sections = re.split(r"(?m)^(?=#{1,3} )", text)
    chunks: list[str] = []
    for sec in sections:
        buf = ""
        for para in re.split(r"\n\s*\n", sec):
            if buf and len(buf) + len(para) > CHUNK_CHARS:
                chunks.append(buf.strip())
                buf = buf[-OVERLAP:]  # carry a little context into the next chunk
            buf += para + "\n\n"
        if buf.strip():
            chunks.append(buf.strip())
    return [c for c in chunks if len(c) > 40]

# ---------- 3. Store --------------------------------------------------------
def connect() -> sqlite3.Connection:
    db = sqlite3.connect(DB)
    db.execute("CREATE TABLE IF NOT EXISTS chunks (id INTEGER PRIMARY KEY, source TEXT, text TEXT, vec BLOB)")
    return db

def index(folder: str) -> None:
    files = [p for p in Path(folder).rglob("*") if p.suffix in {".md", ".txt"}]
    db = connect()
    db.execute("DELETE FROM chunks")
    for f in files:
        parts = chunk(f.read_text(encoding="utf-8", errors="ignore"))
        if not parts:
            continue
        vecs = embed(parts, "document")
        db.executemany("INSERT INTO chunks (source, text, vec) VALUES (?, ?, ?)",
                       [(str(f), t, v.tobytes()) for t, v in zip(parts, vecs)])
        print(f"indexed {f} ({len(parts)} chunks)", file=sys.stderr)
    db.commit()

# ---------- 4. Retrieve -----------------------------------------------------
def search(query: str, k: int = 5) -> list[dict]:
    rows = connect().execute("SELECT source, text, vec FROM chunks").fetchall()
    if not rows:
        return []
    mat = np.vstack([np.frombuffer(r[2], dtype=np.float32) for r in rows])
    scores = mat @ embed([query], "query")[0]          # cosine similarity (vectors are normalised)
    best = np.argsort(-scores)[:k]
    return [{"source": rows[i][0], "score": round(float(scores[i]), 3), "text": rows[i][1]} for i in best]

# ---------- 5. Serve to Claude over MCP --------------------------------------
def serve() -> None:
    from mcp.server import MCPServer
    mcp = MCPServer("docs-rag", instructions="Search the indexed documentation before answering questions about it. Cite the source path.")

    @mcp.tool()
    def search_docs(query: str, k: int = 5) -> list[dict]:
        """Semantic search over the indexed documents. Returns the k most relevant chunks with their source file.

        Args:
            query: A natural-language question or keywords.
            k: Number of chunks to return (1-10).
        """
        return search(query, max(1, min(k, 10)))

    mcp.run(transport="stdio")

if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else ""
    if cmd == "index" and len(sys.argv) == 3:
        index(sys.argv[2])
    elif cmd == "search" and len(sys.argv) >= 3:
        print(json.dumps(search(" ".join(sys.argv[2:])), indent=2))
    elif cmd == "serve":
        serve()
    else:
        sys.exit(__doc__)

Run it

# 1. Index a folder (re-run whenever the docs change)
python rag.py index ~/notes

# 2. Sanity-check retrieval from the terminal
python rag.py search "how do I run a job every two minutes"

Example output:

[
  {
    "source": "/home/me/notes/systemd.md",
    "score": 0.81,
    "text": "# systemd timers\n\nA timer unit triggers a service unit on a schedule. OnUnitActiveSec=2min ..."
  }
]

Always look at what comes back before blaming the model. Most bad RAG answers are bad retrieval.

3. Give it to Claude Code

claude mcp add docs-rag --env RAG_DB=$HOME/rag/rag.db \
  -- /absolute/path/.venv/bin/python /absolute/path/rag.py serve

Now Claude can look things up on its own:

> according to my notes, what's the rollback procedure for the deploy timer?
  ● docs-rag - search_docs (query: "rollback deploy timer")
  Based on systemd.md and deploy.md: stop the timer with …

The server’s instructions ask Claude to cite source paths, so you can open the original and check.

Make retrieval smarter: contextual retrieval

A chunk like “Revenue grew 3% over the previous quarter” is ambiguous on its own: which company, which quarter? Contextual retrieval has a model write that missing context into the chunk before it is embedded. With prompt caching, the document is processed once and every following chunk costs very little:

import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY

def contextualize(document: str, chunk: str) -> str:
    """Prepend a short, model-written context line to a chunk before embedding it."""
    msg = client.messages.create(
        model="claude-haiku-5-5",
        max_tokens=150,
        # The whole document is cached, so every chunk after the first is cheap.
        system=[{"type": "text", "text": f"<document>\n{document}\n</document>",
                 "cache_control": {"type": "ephemeral"}}],
        messages=[{"role": "user", "content":
            f"Here is a chunk from the document above:\n<chunk>\n{chunk}\n</chunk>\n"
            "Write one or two sentences that situate this chunk within the whole document, "
            "to improve search retrieval. Reply with only those sentences."}],
    )
    return f"{msg.content[0].text}\n\n{chunk}"

In rag.py’s index(), call contextualize(full_text, part) on each chunk before embed(). It’s an API call per chunk, so use it when retrieval quality matters more than indexing cost.

Where to go next

• Scale up storage: swap SQLite for sqlite-vec, pgvector (Postgres) or a vector database once you pass roughly 100k chunks.
• Add keyword search: SQLite’s built-in FTS5 gives you BM25; merge its results with the vector results.
• Rerank: fetch the top 30 by vector, rerank, and return the best 5.
• Evaluate: write 20 real questions with known answers and track how often the right chunk lands in the top 5. Change one thing at a time.

The whole series

You now have the full path: install and first session (part 1), the daily workflow and CLAUDE.md (part 2), skills, subagents, hooks and plugins (part 3), MCP servers you can build yourself (part 4), and RAG over your own knowledge (part 5). The best next step is to pick one repetitive task from your week and automate it with whichever piece fits.