Building a RAG System From Scratch

Architecture Overview

A RAG system has two pipelines: an offline indexing pipeline that runs once (or periodically) to build the knowledge base, and an online query pipeline that runs for every user request.

INDEXING PIPELINE
Documents → Parse → Chunk → Embed → Store in vector DB

QUERY PIPELINE
User query → Embed → ANN search → Rerank → Assemble context → LLM → Response

Step 1: Document Ingestion and Parsing

pythonCopy
from pathlib import Path
import pypdf

def extract_text_from_pdf(pdf_path: str) -> list[dict]:
    """Extract text page by page, preserving metadata."""
    reader = pypdf.PdfReader(pdf_path)
    pages = []
    for i, page in enumerate(reader.pages):
        text = page.extract_text()
        if text.strip():
            pages.append({
                "text": text,
                "source": Path(pdf_path).name,
                "page": i + 1,
            })
    return pages

For web content:

pythonCopy
import httpx
from bs4 import BeautifulSoup

def scrape_webpage(url: str) -> dict:
    response = httpx.get(url, timeout=10)
    soup = BeautifulSoup(response.text, "html.parser")
    # Remove nav, footer, ads
    for tag in soup(["nav", "footer", "script", "style", "aside"]):
        tag.decompose()
    text = " ".join(soup.get_text().split())
    return {"text": text, "source": url}

Step 2: Chunking

pythonCopy
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=["\n\n", "\n", ". ", " "],
)

def chunk_documents(documents: list[dict]) -> list[dict]:
    chunks = []
    for doc in documents:
        doc_chunks = splitter.split_text(doc["text"])
        for i, chunk_text in enumerate(doc_chunks):
            chunks.append({
                "text": chunk_text,
                "source": doc["source"],
                "chunk_index": i,
                "page": doc.get("page"),
            })
    return chunks

Step 3: Embedding and Indexing

pythonCopy
from sentence_transformers import SentenceTransformer
import numpy as np
import json

embed_model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")

def build_index(chunks: list[dict]) -> tuple[np.ndarray, list[dict]]:
    texts = [c["text"] for c in chunks]
    embeddings = embed_model.encode(texts, batch_size=64, show_progress_bar=True)
    return embeddings, chunks

embeddings, chunk_metadata = build_index(all_chunks)
np.save("embeddings.npy", embeddings)
with open("chunk_metadata.json", "w") as f:
    json.dump(chunk_metadata, f)

For production: store embeddings in a vector database. pgvector (PostgreSQL extension) is a solid choice for teams already on Postgres:

pythonCopy
import psycopg2
from pgvector.psycopg2 import register_vector

conn = psycopg2.connect("postgresql://user:pass@localhost/rag_db")
register_vector(conn)

with conn.cursor() as cur:
    cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
    cur.execute("""
        CREATE TABLE IF NOT EXISTS chunks (
            id BIGSERIAL PRIMARY KEY,
            text TEXT NOT NULL,
            source VARCHAR(512),
            page INTEGER,
            embedding vector(384)
        )
    """)
    for chunk, emb in zip(chunk_metadata, embeddings):
        cur.execute(
            "INSERT INTO chunks (text, source, page, embedding) VALUES (%s, %s, %s, %s)",
            (chunk["text"], chunk["source"], chunk.get("page"), emb.tolist())
        )
conn.commit()

Step 4: Query Pipeline - Retrieval

pythonCopy
def retrieve(query: str, top_k: int = 20) -> list[dict]:
    query_embedding = embed_model.encode([query])[0]

    with conn.cursor() as cur:
        cur.execute("""
            SELECT id, text, source, page,
                   1 - (embedding <=> %s::vector) AS cosine_similarity
            FROM chunks
            ORDER BY embedding <=> %s::vector
            LIMIT %s
        """, (query_embedding.tolist(), query_embedding.tolist(), top_k))
        rows = cur.fetchall()

    return [
        {"id": r[0], "text": r[1], "source": r[2], "page": r[3], "score": r[4]}
        for r in rows
    ]

Step 5: Reranking

pythonCopy
from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
    pairs = [(query, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)
    for candidate, score in zip(candidates, scores):
        candidate["rerank_score"] = float(score)
    return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)[:top_k]

Step 6: Generation With Citation

pythonCopy
import anthropic

client = anthropic.Anthropic()

def generate_response(query: str, context_chunks: list[dict]) -> dict:
    context = "\n\n".join([
        f"[Source {i+1}: {c['source']}, page {c.get('page', 'N/A')}]\n{c['text']}"
        for i, c in enumerate(context_chunks)
    ])

    prompt = f"""Answer the question using ONLY the provided context.
Cite sources using [Source N] notation for every factual claim.
If the answer is not in the context, say "I don't have information about this."

Context:
{context}

Question: {query}"""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    answer = response.content[0].text

    return {
        "answer": answer,
        "sources": [{"source": c["source"], "page": c.get("page")} for c in context_chunks],
    }

# Full pipeline
def rag_query(query: str) -> dict:
    candidates = retrieve(query, top_k=20)
    top_chunks = rerank(query, candidates, top_k=5)
    return generate_response(query, top_chunks)

Evaluating the System

pythonCopy
eval_set = [
    {"query": "What is the return policy?", "relevant_source": "policies.pdf"},
    {"query": "How do I reset my password?", "relevant_source": "faq.pdf"},
]

for case in eval_set:
    candidates = retrieve(case["query"], top_k=20)
    top_chunks = rerank(case["query"], candidates, top_k=5)
    sources_retrieved = [c["source"] for c in top_chunks]
    hit = case["relevant_source"] in sources_retrieved
    print(f"Query: {case['query'][:50]} | Hit@5: {hit}")

Track Recall@5 (is the relevant source in the top 5?), then run generation evaluation using LLM-as-judge for groundedness and correctness.

Common Mistakes

Using one chunk size for all document types. A 512-token chunk is reasonable for dense prose but will split code examples or table rows mid-structure, making them unretrieval-friendly. Dense technical PDFs, conversational transcripts, and markdown docs each have different natural segmentation boundaries. Always profile your corpus structure before fixing chunk size.

Skipping retrieval recall measurement and only measuring answer quality. A RAG system can produce fluent, plausible answers even when the retrieved chunks are wrong - the LLM hallucinates. Measuring only answer quality hides retrieval failures. Establish a ground-truth Q&A set and measure recall@k before optimizing generation.

Not L2-normalizing embeddings before cosine similarity. Raw cosine similarity between un-normalized vectors is equivalent to dot product, which is dominated by vector magnitude rather than angle. Most embedding models expect you to normalize; skipping this step produces subtly wrong similarity rankings that are hard to diagnose without a deliberate retrieval audit.

What to Practice Next

• Build a 50-question evaluation set for a RAG system you control: 25 questions with clear retrievable answers and 25 that require reasoning across chunks. Measure recall@5 against your ground truth.
• Compare at least two chunking strategies (fixed token, recursive character, sentence-boundary) on the same document set and measure how recall@5 changes.
• Add L2 normalization to your embedding pipeline and verify that nearest-neighbor rankings change - then confirm the new rankings look semantically correct.