How to Build a RAG Pipeline From Scratch: A Python Tutorial

A RAG pipeline is six stages: load, chunk, embed, store, retrieve, generate. The first four run once at index-time to build a searchable store; the last two run at query-time to answer each question against it. This page builds all six in plain Python you can paste and run — but the point is not the code. Every stage is a choice, and every ranking tutorial hides that by defaulting it silently. Here each choice is stated, and the pipeline ships with an evaluation harness so you measure your own build instead of trusting mine. This assumes you have already decided RAG is the right tool; if you are still weighing it against fine-tuning, start at RAG vs fine-tuning.

How do you set up the environment and pin versions?

Install four packages and set one key. The versions are pinned on purpose — a tutorial whose code silently breaks in six months taught you nothing. If a newer release is out when you read this, upgrade deliberately, not by accident.

pip install langchain==0.3.14 langchain-openai==0.2.14 chromadb==0.5.23 ragas==0.2.9

export OPENAI_API_KEY="sk-..."   # or set it in a .env you never commit

How do you load and chunk your documents?

Load the raw text, then split it into overlapping passages. Chunk size and overlap are the two knobs that decide retrieval quality: too large and the answer is buried in noise the model has to wade through; too small and a fact gets severed from the context that makes it meaningful. The 500-token / 50-overlap defaults below are a starting point to measure, not a law — comparing chunkers properly is its own topic, covered under chunking, and a mis-split chunk is the most common retrieval failure (the wrong chunk).

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

docs = TextLoader("handbook.txt").load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # the choice: passage length in characters
    chunk_overlap=50,    # the choice: carry-over so facts aren't severed at the seam
)
chunks = splitter.split_documents(docs)
print(f"{len(docs)} document(s) -> {len(chunks)} chunks")

Which embedding model should you use?

Any model that fits your language and budget — the real decision is dimensions vs cost vs recall, not brand. A hosted model is the least work; a local one is free and keeps your text on your own hardware. Poor model choice is a top-cited pipeline failure, so make it deliberately. The two common options, priced from their public rates:

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# swap for: from langchain_huggingface import HuggingFaceEmbeddings
#           HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")  # local, $0

What an embedding is, and how to compare models properly, lives at embeddings.

How do you store and query the vectors?

Put the vectors in a store that does nearest-neighbour search. For a first build, an embedded store like Chroma runs in-process and needs no server — the whole index is a folder on disk. State the choice honestly: embedded is zero-ops for a laptop or a small corpus; a server database is what you graduate to for scale, multi-tenancy, or built-in hybrid search.

from langchain_chroma import Chroma

store = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./rag_index",   # the index is just this folder
)

How do you retrieve the right chunks?

At query-time, embed the question with the same model and pull the top-k nearest chunks. Top-k is a recall-vs-cost dial: too low and you miss the evidence; too high and you flood the prompt with irrelevant text and pay for the extra tokens. Start at k=4 and move it based on your eval. When dense retrieval keeps missing exact terms — IDs, error codes, rare tokens — the fix is not a bigger k, it is adding a lexical retriever.

retriever = store.as_retriever(search_kwargs={"k": 4})   # the choice: k

hits = retriever.invoke("How do I reset my password?")
for h in hits:
    print(h.page_content[:80])

How do you assemble the prompt and generate the answer?

Stuff the retrieved chunks into a prompt that tells the model to answer only from the supplied context, and to say it does not know otherwise. That single instruction is what separates RAG from a chatbot: it grounds the answer in your documents instead of the model’s memory, which is how you keep hallucination out. This block completes the runnable pipeline end to end.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template(
    "Answer the question using ONLY the context below. "
    "If the context does not contain the answer, say you don't know.nn"
    "Context:n{context}nnQuestion: {question}"
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def answer(question: str) -> str:
    hits = retriever.invoke(question)
    context = "nn".join(h.page_content for h in hits)
    msg = prompt.format_messages(context=context, question=question)
    return llm.invoke(msg).content

print(answer("How do I reset my password?"))

How do you know the pipeline actually works?

You measure it. A pipeline you cannot score is a demo, not a system — and “it looked right on three questions I tried” is not a score. Build a small eval set of question/ground-truth pairs and run a harness that reports faithfulness (did the answer stick to the retrieved context) and context precision / recall (did retrieval fetch the right chunks). These are the numbers that tell you which stage to fix; what each metric means is defined at evaluation, and the tools that compute them are compared at evaluation tools.

from ragas import evaluate
from ragas.metrics import faithfulness, context_precision, context_recall
from datasets import Dataset

# one row per eval question — expand this to your real set
rows = [{
    "question": "How do I reset my password?",
    "answer": answer("How do I reset my password?"),
    "contexts": [h.page_content for h in retriever.invoke("How do I reset my password?")],
    "ground_truth": "Use the 'Forgot password' link on the sign-in page.",
}]

report = evaluate(Dataset.from_list(rows),
                  metrics=[faithfulness, context_precision, context_recall])
print(report)   # the scores you watch — run before and after every change

When should you add hybrid search or reranking?

Not by default. Add an enhancement only when the eval above shows a specific miss — bolting all three on unconditionally just buys you latency and cost:

• Hybrid search (add BM25) — when retrieval misses exact terms, IDs, or error codes that embeddings smear together. How to fuse the two: hybrid retrieval.
• Reranking — when recall is fine (the right chunk is in the top 20) but it is not ranked first. A cross-encoder reorders the shortlist: reranking.
• Multi-query — when questions are phrased far from the source wording, so one query vector misses. You generate several phrasings and union the hits.

If your eval already passes, add none of them.

Should you build from scratch or use a framework?

Build from scratch to understand the six stages — which is what this page is for. Reach for a framework when you want that plumbing maintained for you. It is the same six stages either way; the trade is control for speed:

Note this tutorial already uses LangChain’s components (loaders, splitters) while wiring the flow by hand — a common middle path: borrow the utilities, own the control flow.

What does it cost to run in production?

Three recurring costs, each tied to a stage. Embeddings are an index-time cost you pay once per document (and again when a document changes). Generation is a per-query LLM cost and is usually the largest line. The vector store host is a standing cost once you leave an embedded database. Latency is dominated by the generation call, not retrieval. The levers, each pointing back at a stage you already met: cache repeated answers, batch embeddings at index-time, drop to a smaller generation model, and lower top-k so you send fewer tokens. Production scaling and monitoring is a topic of its own — this is the shape of the bill, not the ops manual.