Retrieval-Augmented Generation (RAG)

import anthropic
import numpy as np
from typing import List

client = anthropic.Anthropic()

# ─── Indexing ───────────────────────────────────────────────

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
    """Simple sliding-window chunking."""
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        if chunk:
            chunks.append(chunk)
    return chunks

def embed(texts: List[str]) -> np.ndarray:
    """Embed texts using Voyage AI via Anthropic."""
    # In practice, use a dedicated embedding API
    # Here we simulate with a placeholder
    import random
    return np.array([[random.random() for _ in range(1024)] for _ in texts])

def cosine_similarity(a: np.ndarray, b: np.ndarray) -> np.ndarray:
    a_norm = a / np.linalg.norm(a, axis=1, keepdims=True)
    b_norm = b / np.linalg.norm(b, axis=1, keepdims=True)
    return a_norm @ b_norm.T

# Build the index
documents = [
    "Machine learning is a subset of artificial intelligence...",
    "Transformers use self-attention mechanisms to process sequences...",
    "Large language models are trained on massive text corpora...",
    # ... more documents
]

all_chunks = []
for doc in documents:
    all_chunks.extend(chunk_text(doc))

chunk_embeddings = embed(all_chunks)  # shape: (n_chunks, 1024)

# ─── Retrieval + Generation ──────────────────────────────────

def rag_query(user_query: str, top_k: int = 3) -> str:
    # 1. Embed the query
    query_embedding = embed([user_query])  # (1, 1024)

    # 2. Compute similarities
    similarities = cosine_similarity(query_embedding, chunk_embeddings)[0]

    # 3. Retrieve top-k chunks
    top_indices = np.argsort(similarities)[::-1][:top_k]
    retrieved_chunks = [all_chunks[i] for i in top_indices]
    top_scores = [similarities[i] for i in top_indices]

    # 4. Build prompt with retrieved context
    context = "\n\n".join([
        f"[Source {i+1} (relevance: {score:.2f})]\n{chunk}"
        for i, (chunk, score) in enumerate(zip(retrieved_chunks, top_scores))
    ])

    # 5. Generate with grounded context
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="""Answer the user's question based ONLY on the provided context.
If the context doesn't contain the answer, say "I don't have information about that in my knowledge base."
Always cite which source(s) you used.""",
        messages=[{
            "role": "user",
            "content": f"Context:\n{context}\n\nQuestion: {user_query}"
        }],
    )
    return response.content[0].text

answer = rag_query("How do transformers work?")
print(answer)

import chromadb
import anthropic

chroma = chromadb.Client()
collection = chroma.create_collection("knowledge_base")
client = anthropic.Anthropic()

# Index documents
docs = ["AI learns from data...", "Transformers use attention...", "RAG retrieves context..."]
ids = [f"doc_{i}" for i in range(len(docs))]
metadatas = [{"source": "textbook", "chapter": i+1} for i in range(len(docs))]

collection.add(documents=docs, ids=ids, metadatas=metadatas)

def rag(query: str, n_results: int = 3) -> str:
    # Retrieve with optional metadata filter
    results = collection.query(
        query_texts=[query],
        n_results=n_results,
        where={"source": "textbook"},  # optional filter
    )
    context = "\n\n".join(results["documents"][0])

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"Answer using only this context:\n{context}\n\nQuestion: {query}"
        }],
    )
    return response.content[0].text

print(rag("How does attention work in AI?"))