How to Build a RAG Pipeline from Scratch in Python

What RAG Actually Is (And Isn't)

You've heard the buzz. Every AI company claims they do RAG. But when you ask what's actually happening under the hood, you get blank stares and a link to LangChain's docs.

This guide is different. We're building a complete RAG pipeline from scratch — no LangChain, no LlamaIndex, no abstractions hiding the magic. Just Python, numpy, and one API call to OpenAI (or Ollama for local).

By the end, you'll have a production-ready system that:

• Ingests PDFs, Markdown, and web pages
• Chunks documents intelligently (not just splitting by character count)
• Embeds with OpenAI text-embedding-3-small or local models
• Stores vectors in a searchable HNSW index
• Retrieves with hybrid search (semantic + BM25 keyword)
• Reranks results for maximum relevance
• Generates answers with proper citations

Architecture Overview

Our RAG pipeline has 8 components, split into two phases:

Indexing phase (run once per document update):

1. Ingestion — Parse PDFs, Markdown, text into Document objects
2. Chunking — Split into overlapping semantic chunks
3. Embedding — Convert chunks to vectors
4. Storage — Index vectors with HNSW for fast search

Query phase (run per user question):

5. Hybrid Search — Combine semantic + BM25 keyword search
6. Reranking — Re-score top candidates with cross-encoder
7. Generation — Generate answer with citations from LLM

Step 1: Document Ingestion

We support three input formats: PDF, Markdown, and raw text. Each gets normalized into a standard Document object with metadata.

# rag/ingest.py
import os, re, hashlib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

@dataclass
class Document:
    content: str
    source: str
    doc_type: str
    doc_id: str = ""
    metadata: dict = field(default_factory=dict)
    
    def __post_init__(self):
        if not self.doc_id:
            self.doc_id = hashlib.md5(
                f"{self.source}:{self.content[:200]}".encode()
            ).hexdigest()

class DocumentIngester:
    """Ingest documents from multiple formats."""
    
    def ingest_file(self, path: str) -> list[Document]:
        path = Path(path)
        suffix = path.suffix.lower()
        if suffix == ".pdf":
            return self._ingest_pdf(path)
        elif suffix in (".md", ".markdown"):
            return self._ingest_markdown(path)
        elif suffix in (".txt", ".text"):
            return self._ingest_text(path)
        raise ValueError(f"Unsupported: {suffix}")
    
    def ingest_directory(self, dir_path: str, recursive=True) -> list[Document]:
        documents = []
        for file_path in Path(dir_path).glob("**/*" if recursive else "*"):
            if file_path.suffix.lower() in (".pdf", ".md", ".markdown", ".txt"):
                try:
                    documents.extend(self.ingest_file(str(file_path)))
                except Exception as e:
                    print(f"  ✗ {file_path.name}: {e}")
        return documents
    
    def _ingest_pdf(self, path):
        import fitz  # pymupdf
        doc = fitz.open(str(path))
        pages = []
        for i, page in enumerate(doc):
            text = page.get_text("text").strip()
            if text:
                pages.append(Document(
                    content=text, source=f"{path}#page={i+1}",
                    doc_type="pdf", metadata={"page": i+1}
                ))
        doc.close()
        return pages
    
    def _ingest_markdown(self, path):
        content = path.read_text(encoding="utf-8")
        sections = re.split(r'\n(?=#{1,3}\s)', content)
        return [Document(
            content=s.strip(), source=f"{path}#{i}",
            doc_type="markdown", metadata={"section_index": i}
        ) for i, s in enumerate(sections) if s.strip()]
    
    def _ingest_text(self, path):
        return [Document(
            content=path.read_text(encoding="utf-8"),
            source=str(path), doc_type="text"
        )]

Step 2: Intelligent Chunking

This is where most RAG systems fail. Naive chunking (split every 500 characters) destroys semantic meaning. We respect document structure with sentence-boundary awareness and 20% overlap.

# rag/chunk.py
class IntelligentChunker:
    def __init__(self, chunk_size=512, chunk_overlap=100, min_chunk_size=100):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.min_chunk_size = min_chunk_size
    
    def chunk_document(self, document) -> list[Chunk]:
        text = document.content
        chunks, start, idx = [], 0, 0
        
        while start < len(text):
            end = start + self.chunk_size
            if end >= len(text):
                remaining = text[start:].strip()
                if len(remaining) >= self.min_chunk_size:
                    chunks.append(self._create_chunk(remaining, document, idx, start, len(text)))
                break
            
            split = self._find_split_point(text, start, end)
            chunk_text = text[start:split].strip()
            if len(chunk_text) >= self.min_chunk_size:
                chunks.append(self._create_chunk(chunk_text, document, idx, start, split))
                idx += 1
            start = max(split - self.chunk_overlap, 0)
        
        return chunks
    
    def _find_split_point(self, text, start, target_end):
        window = text[start:min(target_end + 100, len(text))]
        # Priority: paragraph > sentence > word boundary
        for pattern in [r'\n\n', r'[.!?][\s\n]', r' ']:
            matches = [m.end() for m in re.finditer(pattern, window[:target_end - start + 50])]
            if matches:
                best = min(matches, key=lambda x: abs(x - (target_end - start)))
                if best > self.min_chunk_size:
                    return start + best
        return target_end

Step 3: Embedding Generation

We support both OpenAI and local embeddings with disk-based caching and batching to minimize API costs.

# rag/embed.py
class EmbeddingEngine:
    def __init__(self, model="text-embedding-3-small", use_local=False, batch_size=100):
        self.model = model
        self.use_local = use_local
        self.batch_size = batch_size
        self.cache = EmbeddingCache()
        self._dimension = 1536 if "3-small" in model else 3072
    
    def embed_texts(self, texts: list[str]) -> np.ndarray:
        embeddings = np.zeros((len(texts), self._dimension), dtype=np.float32)
        uncached, uncached_idx = [], []
        
        for i, text in enumerate(texts):
            cached = self.cache.get(text, self.model)
            if cached is not None:
                embeddings[i] = cached
            else:
                uncached.append(text)
                uncached_idx.append(i)
        
        # Batch process uncached
        for batch_start in range(0, len(uncached), self.batch_size):
            batch = uncached[batch_start:batch_start + self.batch_size]
            embs = self._embed_batch(batch)
            for idx, text, emb in zip(uncached_idx[batch_start:batch_start+len(batch)], batch, embs):
                embeddings[idx] = emb
                self.cache.put(text, self.model, emb)
        
        return embeddings

Step 4: Vector Storage with HNSW

HNSW (Hierarchical Navigable Small World) provides sub-linear approximate nearest neighbor search — fast enough for production, accurate enough for RAG.

# rag/store.py
class VectorStore:
    def __init__(self, dimension=1536, space="cosine", max_elements=100000):
        self.dimension = dimension
        self.space = space
        self.max_elements = max_elements
        self._index = None
        self._metadata = []
        self._next_id = 0
    
    def _init_index(self):
        import hnswlib
        self._index = hnswlib.Index(space=self.space, dim=self.dimension)
        self._index.init_index(max_elements=self.max_elements, ef_construction=200, M=16)
        self._index.set_ef(50)
    
    def add(self, embeddings: np.ndarray, metadata: list[dict]):
        if self._index is None:
            self._init_index()
        ids = np.arange(self._next_id, self._next_id + embeddings.shape[0])
        self._index.add_items(embeddings, ids)
        self._metadata.extend(metadata)
        self._next_id += embeddings.shape[0]
    
    def search(self, query_embedding: np.ndarray, k=10) -> list[dict]:
        if self._index is None:
            return []
        labels, distances = self._index.knn_query(query_embedding, k=k)
        return [
            {"id": int(idx), "distance": float(dist), "metadata": self._metadata[idx]}
            for idx, dist in zip(labels[0], distances[0])
            if idx < len(self._metadata)
        ]

Step 5: Hybrid Search (Semantic + BM25)

This is the secret sauce. Pure semantic search misses exact keyword matches. Pure BM25 misses conceptual similarity. We combine both with Reciprocal Rank Fusion.

# rag/search.py
class HybridSearcher:
    def __init__(self, vector_store, bm25_index, rrf_k=60):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
        self.rrf_k = rrf_k
    
    def search(self, query_embedding, query_text, k=10):
        # Get results from both retrievers
        vector_results = self.vector_store.search(query_embedding, k=k * 2)
        bm25_results = self.bm25_index.search(query_text, k=k * 2)
        
        # Reciprocal Rank Fusion
        scores = {}
        for rank, result in enumerate(vector_results):
            scores[result["id"]] = scores.get(result["id"], 0) + 0.6 / (self.rrf_k + rank + 1)
        for rank, result in enumerate(bm25_results):
            scores[result["id"]] = scores.get(result["id"], 0) + 0.4 / (self.rrf_k + rank + 1)
        
        sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)[:k]
        return [
            {"id": i, "score": scores[i], "metadata": self.vector_store._metadata[i]}
            for i in sorted_ids if i < len(self.vector_store._metadata)
        ]

Step 6: Reranking

After retrieval, we rerank the top candidates using a cross-encoder. More expensive than bi-encoder retrieval, but significantly more accurate.

# rag/rerank.py
class CrossEncoderReranker:
    def __init__(self, model_name="cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.model_name = model_name
        self._model = None
    
    def rerank(self, query, chunks, top_k=5):
        from sentence_transformers import CrossEncoder
        if self._model is None:
            self._model = CrossEncoder(self.model_name)
        
        pairs = [(query, chunk) for chunk in chunks]
        scores = self._model.predict(pairs)
        
        ranked = sorted(zip(chunks, scores), key=lambda x: x[1], reverse=True)
        return [{"chunk": c, "score": float(s)} for c, s in ranked[:top_k]]

Step 7: Answer Generation with Citations

The final step generates answers grounded in retrieved chunks, with proper citations.

# rag/generate.py
class AnswerGenerator:
    SYSTEM_PROMPT = """Answer using ONLY the provided context.
    1. If context doesn't contain the answer, say so.
    2. Always cite sources using [1], [2], etc.
    3. Be concise but complete.
    4. Format code blocks properly."""
    
    def generate(self, query, chunks):
        context = "\n\n".join(
            f"[{i+1}] (src: {c.get('source', '?')})\n{c['content']}"
            for i, c in enumerate(chunks[:5])
        )
        
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": self.SYSTEM_PROMPT},
                {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"}
            ],
            temperature=0.1
        )
        
        return {
            "answer": response.choices[0].message.content,
            "citations": [{"id": i+1, "source": c.get("source")} for i, c in enumerate(chunks[:5])],
            "tokens_used": response.usage.total_tokens
        }

Step 8: Complete Pipeline

Here's the full pipeline tying everything together:

# rag/pipeline.py
class RAGPipeline:
    def __init__(self, config=None):
        config = config or {}
        self.ingester = DocumentIngester()
        self.chunker = IntelligentChunker(chunk_size=512, chunk_overlap=100)
        self.embedder = EmbeddingEngine(model="text-embedding-3-small")
        self.vector_store = VectorStore(dimension=1536)
        self.bm25 = BM25Index()
        self.reranker = SimpleReranker(self.embedder)
        self.generator = AnswerGenerator(model="gpt-4o-mini")
    
    def index(self, source):
        docs = self.ingester.ingest_directory(source)
        chunks = []
        for doc in docs:
            chunks.extend(self.chunker.chunk_document(doc))
        
        embeddings = self.embedder.embed_texts([c.content for c in chunks])
        self.vector_store.add(embeddings, [{"content": c.content, "source": c.source} for c in chunks])
        self.bm25.add_documents([c.content for c in chunks])
        self._searcher = HybridSearcher(self.vector_store, self.bm25)
        return len(chunks)
    
    def query(self, question, top_k=5):
        query_emb = self.embedder.embed_query(question)
        results = self._searcher.search(query_emb, question, k=top_k * 3)
        reranked = self.reranker.rerank(question, [r["metadata"]["content"] for r in results], top_k)
        top_chunks = [r["metadata"] for r in results[:top_k]]
        return self.generator.generate(question, top_chunks)

# Usage
pipeline = RAGPipeline()
pipeline.index("./knowledge_base/")
result = pipeline.query("How does authentication work?")
print(result["answer"])

Production Considerations

When we deploy RAG systems for ZOO clients, these are the issues that actually matter:

1. Chunk Size Matters More Than You Think

Size	Effect
< 200 chars	Loses context, poor retrieval
400-600 tokens	Sweet spot for most content
> 1000 chars	Dilutes relevance, wastes tokens

2. Embedding Model Choice

Model	Cost	Quality
text-embedding-3-small	$0.02/1M tokens	Good
text-embedding-3-large	$0.13/1M tokens	Excellent
nomic-embed-text (local)	Free	Decent

3. The "Lost in the Middle" Problem

LLMs pay more attention to the beginning and end of the context window. Put the most relevant chunks first.

4. Evaluation is Non-Negotiable

Always measure retrieval accuracy and answer correctness before optimizing. We maintain test suites of 50-100 questions per client project.

Performance Benchmarks

Tested on a real client project: 2,400 pages of technical documentation, 18,600 chunks.

Metric	Value
Indexing time	47 min (OpenAI embeddings)
Query latency (p50)	340ms
Query latency (p95)	890ms
Retrieval accuracy (top-5)	87%
Answer correctness	82%
Cost per query	~$0.003

Next Steps

You now have a complete RAG pipeline. Here's what to do next:

1. Start small: Index one document, ask 5 questions, evaluate manually
2. Measure before optimizing: Set up evaluation before tweaking parameters
3. Iterate on chunking: This has the highest impact on retrieval quality
4. Add metadata filtering: Filter by date, source, document type
5. Consider multi-query: Generate 3 query variations and merge results