Documents (PDFs, text, markdown)
        ↓
  Chunker (RecursiveTextSplitter)
        ↓
  Embedder (Voyage AI via Claude SDK)
        ↓
  Vector Store (ChromaDB — local)
        ↓
  Query Engine
        ↓
  Claude API (claude-sonnet-4-6) → Answer

bashpip install anthropic chromadb voyageai pypdf

bashexport ANTHROPIC_API_KEY="sk-ant-..."
export VOYAGE_API_KEY="pa-..."   # free tier available at voyageai.com

pythonimport os
from pathlib import Path
from pypdf import PdfReader

def load_pdf(file_path: str) -> str:
    """Extract all text from a PDF."""
    reader = PdfReader(file_path)
    text = ""
    for page in reader.pages:
        text += page.extract_text() + "\n"
    return text

def chunk_text(text: str, chunk_size: int = 512, overlap: int = 64) -> list[str]:
    """Split text into overlapping chunks by word count."""
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunk = " ".join(words[start:end])
        chunks.append(chunk)
        start += chunk_size - overlap
    return chunks

# Example: load a PDF and chunk it
raw_text = load_pdf("docs/anthropic_usage_policy.pdf")
chunks = chunk_text(raw_text)
print(f"Loaded {len(chunks)} chunks")

pythonimport voyageai
import chromadb

# Initialize clients
voyage_client = voyageai.Client()
chroma_client = chromadb.PersistentClient(path="./chroma_db")

# Create (or load) a collection
collection = chroma_client.get_or_create_collection(
    name="my_documents",
    metadata={"hnsw:space": "cosine"}  # cosine similarity for text
)

def embed_and_store(chunks: list[str], source: str = "unknown"):
    """Embed chunks and upsert into ChromaDB."""
    # Voyage AI supports batch embedding (up to 128 at once)
    batch_size = 64
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i : i + batch_size]
        result = voyage_client.embed(batch, model="voyage-3", input_type="document")
        embeddings = result.embeddings

        collection.upsert(
            ids=[f"{source}-chunk-{i + j}" for j in range(len(batch))],
            embeddings=embeddings,
            documents=batch,
            metadatas=[{"source": source, "chunk_index": i + j} for j in range(len(batch))]
        )
    print(f"Stored {len(chunks)} chunks from '{source}'")

# Index your documents
embed_and_store(chunks, source="anthropic_usage_policy")

pythondef retrieve(query: str, top_k: int = 5) -> list[dict]:
    """Embed a query and return the top-k most relevant chunks."""
    query_embedding = voyage_client.embed(
        [query], model="voyage-3", input_type="query"
    ).embeddings[0]

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        include=["documents", "metadatas", "distances"]
    )

    chunks = []
    for doc, meta, dist in zip(
        results["documents"][0],
        results["metadatas"][0],
        results["distances"][0]
    ):
        chunks.append({
            "text": doc,
            "source": meta["source"],
            "score": 1 - dist  # convert cosine distance to similarity
        })
    return chunks

pythonimport anthropic

claude_client = anthropic.Anthropic()

RAG_SYSTEM_PROMPT = """You are a helpful assistant that answers questions based ONLY on the provided context.

Rules:
1. Answer using only the information in the context sections below.
2. If the context does not contain enough information to answer, say "I don't have enough information in my knowledge base to answer that."
3. Cite the source at the end of your answer using [Source: <source name>].
4. Be concise but complete — include all relevant details from the context."""

def ask(query: str) -> str:
    """Full RAG pipeline: retrieve context, then generate an answer."""
    # 1. Retrieve relevant chunks
    retrieved = retrieve(query, top_k=5)

    # 2. Format context block
    context_blocks = []
    for i, chunk in enumerate(retrieved):
        context_blocks.append(
            f"[Context {i+1} | Source: {chunk['source']} | Score: {chunk['score']:.2f}]\n{chunk['text']}"
        )
    context_str = "\n\n---\n\n".join(context_blocks)

    # 3. Build the user message
    user_message = f"""Context:
{context_str}

---

Question: {query}"""

    # 4. Call Claude
    response = claude_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=RAG_SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_message}]
    )

    return response.content[0].text

# Try it
answer = ask("What types of content does Anthropic's usage policy prohibit?")
print(answer)

python# rag_pipeline.py
import os
import anthropic
import chromadb
import voyageai
from pypdf import PdfReader

# --- Config ---
EMBED_MODEL = "voyage-3"
CLAUDE_MODEL = "claude-sonnet-4-6"
CHUNK_SIZE = 512
CHUNK_OVERLAP = 64
TOP_K = 5

# --- Clients ---
voyage = voyageai.Client()
chroma = chromadb.PersistentClient(path="./chroma_db")
claude = anthropic.Anthropic()
collection = chroma.get_or_create_collection("docs", metadata={"hnsw:space": "cosine"})

# --- Core functions (from above) ---
# load_pdf(), chunk_text(), embed_and_store(), retrieve(), ask()
# ... (paste functions here)

if __name__ == "__main__":
    import sys

    if sys.argv[1] == "index":
        path = sys.argv[2]
        text = load_pdf(path) if path.endswith(".pdf") else open(path).read()
        chunks = chunk_text(text)
        embed_and_store(chunks, source=os.path.basename(path))

    elif sys.argv[1] == "ask":
        query = " ".join(sys.argv[2:])
        print(ask(query))

bash# Index a document
python rag_pipeline.py index docs/my_policy.pdf

# Ask a question
python rag_pipeline.py ask "What are the key safety requirements?"