CC12: RAG for Claude Code

def chunk_fixed(text: str, size: int = 500, overlap: int = 100) -> list[str]:
    words = text.split()
    chunks = []
    i = 0
    while i < len(words):
        chunk = " ".join(words[i:i + size])
        chunks.append(chunk)
        i += size - overlap
    return chunks

import re
def chunk_by_heading(md: str) -> list[str]:
    # Split on H1/H2/H3
    parts = re.split(r"(?=^#{1,3} )", md, flags=re.MULTILINE)
    return [p.strip() for p in parts if p.strip()]

import ast
def chunk_python(src: str) -> list[dict]:
    tree = ast.parse(src)
    chunks = []
    lines = src.splitlines()
    for node in tree.body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            start = node.lineno - 1
            end   = node.end_lineno
            chunks.append({
                "name": node.name,
                "kind": type(node).__name__,
                "code": "\n".join(lines[start:end]),
            })
    return chunks

import numpy as np
import voyageai

vo = voyageai.Client()

def embed(texts: list[str]) -> np.ndarray:
    r = vo.embed(texts, model="voyage-3")
    return np.array(r.embeddings)

# Ingest
chunks = chunk_fixed(open("docs.md").read())
chunk_vecs = embed(chunks)   # shape: (N, 1024)

# Query
def search(query: str, k: int = 5) -> list[tuple[float, str]]:
    q = embed([query])[0]
    sims = chunk_vecs @ q / (np.linalg.norm(chunk_vecs, axis=1) * np.linalg.norm(q))
    top = np.argsort(-sims)[:k]
    return [(float(sims[i]), chunks[i]) for i in top]

print(search("how is money represented?"))

from rank_bm25 import BM25Okapi
import re

def tokenize(s: str) -> list[str]:
    return re.findall(r"\w+", s.lower())

# Ingest
chunks = chunk_fixed(open("docs.md").read())
bm25 = BM25Okapi([tokenize(c) for c in chunks])

# Query
def bm25_search(query: str, k: int = 5) -> list[tuple[float, str]]:
    scores = bm25.get_scores(tokenize(query))
    top = sorted(range(len(chunks)), key=lambda i: -scores[i])[:k]
    return [(scores[i], chunks[i]) for i in top]

print(bm25_search("PO-100023 status"))

def rrf(rankings: list[list[int]], k: int = 60) -> list[int]:
    """Merge multiple ranked lists of chunk indices using RRF."""
    scores = {}
    for ranking in rankings:
        for rank, idx in enumerate(ranking):
            scores[idx] = scores.get(idx, 0) + 1.0 / (k + rank + 1)
    return sorted(scores, key=lambda i: -scores[i])

q = "how do I refund an order?"
bm25_top = sorted(range(len(chunks)), key=lambda i: -bm25.get_scores(tokenize(q))[i])[:50]
vec_top  = np.argsort(-(chunk_vecs @ embed([q])[0]))[:50].tolist()
merged   = rrf([bm25_top, vec_top])[:10]
final_chunks = [chunks[i] for i in merged]

---
name: docs-qa
description: Answers questions about internal docs. Use whenever a user asks about company policy, runbooks, or onboarding.
model: claude-sonnet-4-6
tools: ["mcp__internal_docs__search_docs"]   # MCP tool from Pattern B
---

You answer questions strictly from internal documentation.

Process:
1. Call `search_docs(query)` to retrieve top 8 passages.
2. If no passage is relevant, say "I don't have that documented" — do not guess.
3. Cite each claim with the doc title. Do not invent doc titles.

# mcp_rag_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("internal_docs")

@mcp.tool()
def search_docs(query: str, k: int = 8) -> list[dict]:
    """Search internal docs and return top K relevant passages with citations.

    Use whenever a user mentions runbooks, policies, or onboarding.
    Returns: list of {title, source, snippet, score}.
    """
    bm25_top = bm25_rank(query, top=50)
    vec_top  = vec_rank(query, top=50)
    merged   = rrf([bm25_top, vec_top])[:k]
    return [{"title": meta[i].title, "source": meta[i].path,
             "snippet": chunks[i], "score": float(scores[i])} for i in merged]

if __name__ == "__main__":
    mcp.run()

$ mkdir docs-rag && cd docs-rag
$ python -m venv .venv && source .venv/bin/activate
$ pip install "mcp[cli]" rank-bm25 voyageai numpy
$ export VOYAGE_API_KEY="..."     # any embedding provider

# server.py
import os, re, glob
import numpy as np
from rank_bm25 import BM25Okapi
from mcp.server.fastmcp import FastMCP
import voyageai

DOCS_DIR = os.environ.get("DOCS_DIR", "./docs")

# --- Ingest at startup ---
def chunk_md(text, size=500, overlap=100):
    parts = re.split(r"(?=^#{1,3} )", text, flags=re.MULTILINE) or [text]
    out = []
    for p in parts:
        words = p.split()
        i = 0
        while i < len(words):
            out.append(" ".join(words[i:i+size]))
            i += size - overlap
    return out

CHUNKS, META = [], []
for path in glob.glob(f"{DOCS_DIR}/**/*.md", recursive=True):
    text = open(path).read()
    for ch in chunk_md(text):
        CHUNKS.append(ch)
        META.append({"source": path, "title": os.path.basename(path)})

bm25 = BM25Okapi([re.findall(r"\w+", c.lower()) for c in CHUNKS])
vo = voyageai.Client()
VECS = np.array(vo.embed(CHUNKS, model="voyage-3").embeddings)

def rrf(rankings, k=60):
    s = {}
    for r in rankings:
        for rank, idx in enumerate(r):
            s[idx] = s.get(idx, 0) + 1 / (k + rank + 1)
    return sorted(s, key=lambda i: -s[i])

# --- MCP tool ---
mcp = FastMCP("internal_docs")

@mcp.tool()
def search_docs(query: str, k: int = 6) -> list[dict]:
    """Search internal markdown docs by hybrid BM25 + vector retrieval. Returns top-K passages with source.

    Use when answering questions about internal documentation, runbooks, policies, onboarding, FAQs.
    """
    bm = sorted(range(len(CHUNKS)),
                key=lambda i: -bm25.get_scores(re.findall(r"\w+", query.lower()))[i])[:30]
    qv = np.array(vo.embed([query], model="voyage-3").embeddings[0])
    vc = np.argsort(-(VECS @ qv) / (np.linalg.norm(VECS, axis=1) * np.linalg.norm(qv)))[:30].tolist()
    top = rrf([bm, vc])[:k]
    return [{"title": META[i]["title"], "source": META[i]["source"],
             "snippet": CHUNKS[i][:600]} for i in top]

if __name__ == "__main__":
    mcp.run()

$ claude mcp add docs-rag --scope project \
    --env VOYAGE_API_KEY="$VOYAGE_API_KEY" \
    --env DOCS_DIR=/path/to/your/docs \
    -- python "$PWD/server.py"

$ claude
> What does our refund policy say about refunds older than 30 days?

[Claude calls search_docs("refund policy 30 days")]
[6 chunks returned, top from policies/refunds-v3.md]

According to policies/refunds-v3.md, refunds are not honored after 30 days
unless... [grounded answer with cited source]