Build a RAG pipeline in pure Python
Chunk, embed, retrieve, and assemble a grounded prompt - the retrieval pipeline without frameworks.
Why it matters: RAG is the most common GenAI system. Building the steps once helps you debug bad retrieval and explain the design.
Basics
The five steps
- Chunk documents into passages
- Embed each chunk as a vector
- Index the vectors
- Retrieve top-k similar chunks for a query
- Assemble a prompt with only those chunks, then generate
Bad answers usually come from steps 1-4, not the model. Print retrieved chunks before blaming the LLM.
Tiny example
Print the RAG stages in order.
steps = ["ingest", "chunk", "embed", "retrieve", "generate"]
for i, s in enumerate(steps, 1):
print(i, s)Basics
Chunking
Too small loses meaning. Too large wastes tokens.
Start around 400-800 tokens with about 10-15% overlap. Prefer splitting on paragraphs or headings, then by size.
Keep metadata: source, page, section. Users ask where an answer came from.
Tiny example
Chunk text with size and overlap.
text = "abcdefghijklmnopqrstuvwxyz"
size, overlap = 8, 2
step = size - overlap
chunks = [text[i:i + size] for i in range(0, len(text), step)]
print(chunks)Basics
Retrieval quality
Pure vector search can miss exact terms like error codes. Hybrid search mixes keyword and vector scores.
A similarity floor helps: if nothing is close enough, say you do not know instead of stuffing weak context into the prompt.
Tiny example
Rank docs with a simple keyword score.
docs = ["python mlops", "rag with embeddings", "css flexbox"]
query = "rag"
scored = sorted(((d.count(query), d) for d in docs), reverse=True)
print(scored[0][1])Then AI / ML
AI / ML examples
Same idea, used in real AI work. Press Try in lab to run it.
AI / ML example
AI / ML: retrieve with a tiny embedder
A bag-of-words vector stands in for a real embedding model.
In AI / ML: Real systems swap this embed() for OpenAI or sentence-transformers.
example.py
Python
import numpy as np
VOCAB = ["mlops", "rag", "python", "retrieval", "model", "prompt"]
def embed(text):
vec = np.zeros(len(VOCAB), dtype=np.float32)
for word in text.lower().split():
if word in VOCAB:
vec[VOCAB.index(word)] += 1.0
n = np.linalg.norm(vec)
return vec / n if n > 0 else vec
corpus = [
{"id": "c1", "text": "MLOps handles model deployment and monitoring"},
{"id": "c2", "text": "RAG uses retrieval to ground a prompt"},
{"id": "c3", "text": "Python is used to train and serve models"},
]
matrix = np.vstack([embed(c["text"]) for c in corpus])
query = "how does rag build a prompt"
scores = matrix @ embed(query)
for i in np.argsort(-scores)[:2]:
print(corpus[i]["id"], round(float(scores[i]), 3), corpus[i]["text"])AI / ML example 2
AI / ML: assemble a grounded prompt
Numbered sources plus a refuse instruction.
In AI / ML: This is the prompt shape most RAG services send to the LLM.
example.py
Python
def build_rag_prompt(question, chunks):
if not chunks:
return None
sources = "\n\n".join(
f"[{i}] ({c['id']})\n{c['text']}"
for i, c in enumerate(chunks, start=1)
)
return (
"Answer ONLY from the sources. Cite like [1].\n"
"If missing, reply: I do not know.\n\n"
f"SOURCES:\n{sources}\n\nQUESTION: {question}\nANSWER:"
)
chunks = [
{"id": "c2", "text": "RAG uses retrieval to ground a prompt."},
]
print(build_rag_prompt("What is RAG?", chunks))Takeaways
- 1RAG is chunk, embed, index, retrieve, assemble.
- 2Debug retrieval before blaming the model.
- 3Use overlap, source metadata, and a similarity floor.