How to Build a RAG Pipeline in Python

The four-stage pipeline

Retrieval-Augmented Generation (RAG) grounds an LLM in a specific corpus so it answers from your data rather than from training memory. Every RAG system - from a basic prototype to a production knowledge tool - follows the same four stages:

1. Chunk - split documents into retrievable pieces (~200 chars each)
2. Index - embed all chunks into vectors (TF-IDF or dense embeddings)
3. Retrieve - vectorize the question, rank chunks by cosine similarity, return top-k
4. Generate - pass the retrieved chunks as a system message and call the LLM

Build it in Python

The simplest production-valid implementation uses TfidfVectorizer from scikit-learn. TF-IDF is offline, deterministic, and requires no external embedding API - a solid first choice for small-to-medium corpora.

import numpy as np
import requests
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

LLM_URL = "http://llm-service:8080/v1/chat/completions"

_vectorizer = None
_matrix = None
_chunks = []

def chunk(text):
    """Split a document into ~paragraph-sized chunks."""
    return [p.strip() for p in text.split("\n\n") if p.strip()] or [text.strip()]

def index(corpus):
    """Fit TF-IDF on all chunks and store the matrix."""
    global _vectorizer, _matrix, _chunks
    _chunks = [c for doc in corpus for c in chunk(doc)]
    _vectorizer = TfidfVectorizer().fit(_chunks)
    _matrix = _vectorizer.transform(_chunks)

def retrieve(question, k=2):
    """Return the top-k chunks most similar to the question."""
    qv = _vectorizer.transform([question])
    sims = cosine_similarity(qv, _matrix)[0]
    idxs = np.argsort(sims)[::-1][:k]
    return [_chunks[i] for i in idxs]

def answer(question):
    """Retrieve context and call the LLM."""
    context = "\n\n".join(retrieve(question, k=2))
    payload = {
        "model": "my-model",
        "messages": [
            {"role": "system", "content": f"Answer using ONLY the context below.\n\nContext:\n{context}"},
            {"role": "user", "content": question},
        ],
        "temperature": 0,
        "max_tokens": 80,
    }
    r = requests.post(LLM_URL, json=payload, timeout=60)
    r.raise_for_status()
    return r.json()["choices"][0]["message"]["content"]

Call index(corpus) once at startup, then answer(question) for every query.

Why each stage matters

• Chunking - chunks that are too large dilute the similarity signal; too small and you lose context. Paragraph splits (~200 chars) are a reliable starting point.
• Cosine similarity - np.argsort(sims)[::-1][:k] is the full retrieval path: sort descending, take top-k. It runs in milliseconds on a few thousand chunks.
• The system message - putting retrieved context in "role": "system" keeps it separate from the user's question and reduces hallucination. The instruction "answer using ONLY the context below" is load-bearing.

Upgrading the pipeline

Once this shape is working, each stage upgrades independently:

• Embed - swap TfidfVectorizer for sentence-transformers or openai.embeddings for dense vectors that understand semantics.
• Store - move from an in-memory numpy matrix to pgvector, Pinecone, or Qdrant for millions of chunks.
• Retrieve - add a reranker (Cohere rerank, cross-encoder) between retrieve and generate to boost precision.
• Eval - measure answer quality with RAGAS (faithfulness, context recall) before shipping to production.