Chapter 05 · Section II · 28 min read
Building a RAG pipeline step by step
The complete code, walked one function at a time. Load, chunk, embed, retrieve, prompt, cite. About 200 lines that generalise to any RAG project.
This section is the code. We take the mental model from Section 1 and turn it into a working RAG pipeline for a corpus of Nepali documents. Every step is walked, every design decision named. By the end you’ll have around 200 lines of Python that can answer questions from any folder of documents you point it at, with source citations, in Nepali or English, honestly saying “I don’t know” when the documents don’t cover the question.
The full architecture
┌──────────────────────────────────────────────────────────────┐
│ BUILD PHASE (once per corpus update) │
├──────────────────────────────────────────────────────────────┤
│ 1. Load documents from folder │
│ 2. Chunk each document into passages │
│ 3. Embed all passages (cached) │
│ 4. Save vectors + metadata to disk │
└──────────────────────────────────────────────────────────────┘
┌──────────────────────────────────────────────────────────────┐
│ QUERY PHASE (every user question) │
├──────────────────────────────────────────────────────────────┤
│ 1. Embed the user's query │
│ 2. Retrieve top-k similar passages │
│ 3. Format them into the RAG prompt │
│ 4. Call the LLM │
│ 5. Return the answer + cited sources │
└──────────────────────────────────────────────────────────────┘Build once, query many. Everything below is code.
Step 1 — data structures
Small, honest types for what we’re passing around:
from dataclasses import dataclass
from pathlib import Path
@dataclass
class Passage:
text: str
source: str # filename
chunk_index: int # position within the source
@dataclass
class RetrievedResult:
passage: Passage
similarity: float
@dataclass
class RAGAnswer:
answer: str
sources: list[str] # deduplicated source filenames used
passages_used: list[RetrievedResult]
prompt_tokens: int
completion_tokens: int
cost_usd: floatSimple and typed. Now every function has a clear input and output.
Step 2 — load and chunk
Reuse the chunker from Chapter 4 Section 3 without changes:
import re
def split_into_paragraphs(text: str) -> list[str]:
parts = re.split(r"\n\s*\n", text.strip())
return [p.strip() for p in parts if p.strip()]
def approximate_token_count(text: str) -> int:
return max(1, len(text) // 3) # rough heuristic
def chunk_document(text: str, target_size: int = 400, max_size: int = 700) -> list[str]:
paragraphs = split_into_paragraphs(text)
chunks: list[str] = []
current: list[str] = []
current_size = 0
for p in paragraphs:
p_size = approximate_token_count(p)
if current and (current_size + p_size > max_size):
chunks.append("\n\n".join(current))
current = [p]
current_size = p_size
continue
current.append(p)
current_size += p_size
if current_size >= target_size:
chunks.append("\n\n".join(current))
current = []
current_size = 0
if current:
chunks.append("\n\n".join(current))
return chunks
def load_corpus(folder: str) -> list[Passage]:
passages: list[Passage] = []
for path in sorted(Path(folder).glob("*.txt")):
text = path.read_text(encoding="utf-8")
for i, chunk in enumerate(chunk_document(text)):
passages.append(Passage(text=chunk, source=path.name, chunk_index=i))
return passagesStep 3 — embedding + persistence
Same pattern as Chapter 4 Section 3, packaged as one class:
import hashlib
import json
import numpy as np
from openai import OpenAI
_openai = OpenAI()
class EmbeddingIndex:
def __init__(self, cache_path: str = "data/embed_cache.json"):
self.cache_path = Path(cache_path)
self.cache: dict[str, list[float]] = {}
if self.cache_path.exists():
with open(self.cache_path, encoding="utf-8") as f:
self.cache = json.load(f)
self.passages: list[Passage] = []
self.vectors: np.ndarray = np.empty((0, 1536))
@staticmethod
def _key(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def _embed(self, texts: list[str], batch_size: int = 100) -> list[list[float]]:
results = [None] * len(texts)
needed_i = []
needed_texts = []
for i, t in enumerate(texts):
k = self._key(t)
if k in self.cache:
results[i] = self.cache[k]
else:
needed_i.append(i)
needed_texts.append(t)
for start in range(0, len(needed_texts), batch_size):
batch = needed_texts[start:start + batch_size]
response = _openai.embeddings.create(
model="text-embedding-3-small",
input=batch,
)
for offset, item in enumerate(response.data):
idx_in_needed = start + offset
overall_i = needed_i[idx_in_needed]
results[overall_i] = item.embedding
self.cache[self._key(needed_texts[idx_in_needed])] = item.embedding
return results
def build(self, passages: list[Passage]):
self.passages = passages
vectors = self._embed([p.text for p in passages])
self.vectors = np.array(vectors)
self._save_cache()
def _save_cache(self):
with open(self.cache_path, "w", encoding="utf-8") as f:
json.dump(self.cache, f, ensure_ascii=False)
def search(self, query: str, k: int = 5) -> list[RetrievedResult]:
query_vec = np.array(self._embed([query])[0])
norms = np.linalg.norm(self.vectors, axis=1) * np.linalg.norm(query_vec)
sims = self.vectors @ query_vec / norms
top_idx = np.argsort(sims)[-k:][::-1]
return [RetrievedResult(self.passages[i], float(sims[i])) for i in top_idx]Step 4 — the RAG prompt
The system prompt tells the LLM to answer only from the passages, cite sources, and say “I don’t know” when needed:
RAG_SYSTEM_PROMPT = """You are a helpful assistant that answers questions
using ONLY the retrieved passages provided.
Rules:
1. Answer only using information from the passages. Do not use outside
knowledge, even if you know the topic.
2. If the passages don't contain the answer, respond with: "I don't have
information about that in the provided documents." Do not guess.
3. Cite the sources you used by their [Source: filename] tag. Every
factual claim should be traceable to a source.
4. Respond in the same language the user asked in (English or Nepali).
5. Keep the answer concise: 2-5 sentences unless the question requires
more.
If the question and passages are in different languages, translate
your understanding but respond in the user's language.
"""
def format_passages_for_prompt(results: list[RetrievedResult]) -> str:
lines = []
for i, r in enumerate(results, 1):
lines.append(f"[Source: {r.passage.source}]")
lines.append(r.passage.text)
lines.append("") # blank line
return "\n".join(lines)
def build_rag_prompt(question: str, results: list[RetrievedResult]) -> str:
passages = format_passages_for_prompt(results)
return f"""Retrieved passages:
{passages}
---
Question: {question}
Answer:"""Step 5 — the answer function
Tie it all together with a clean API:
import anthropic
_anthropic = anthropic.Anthropic()
HAIKU_INPUT_PER_M = 0.25
HAIKU_OUTPUT_PER_M = 1.25
def answer_question(
question: str,
index: EmbeddingIndex,
k: int = 5,
model: str = "claude-haiku-4-5-20251001",
) -> RAGAnswer:
results = index.search(question, k=k)
prompt = build_rag_prompt(question, results)
response = _anthropic.messages.create(
model=model,
max_tokens=800,
temperature=0.0, # deterministic for factual Q&A
system=RAG_SYSTEM_PROMPT,
messages=[{"role": "user", "content": prompt}],
)
answer_text = response.content[0].text
sources_used = sorted({r.passage.source for r in results})
cost = (
response.usage.input_tokens / 1_000_000 * HAIKU_INPUT_PER_M +
response.usage.output_tokens / 1_000_000 * HAIKU_OUTPUT_PER_M
)
return RAGAnswer(
answer=answer_text,
sources=sources_used,
passages_used=results,
prompt_tokens=response.usage.input_tokens,
completion_tokens=response.usage.output_tokens,
cost_usd=cost,
)Step 6 — a working CLI
def main():
print("Building RAG index over data/corpus/ ...")
passages = load_corpus("data/corpus/")
print(f"Loaded {len(passages)} passages.")
index = EmbeddingIndex(cache_path="data/embed_cache.json")
index.build(passages)
print("Index ready.\n")
while True:
try:
q = input("Question (or 'quit'): ").strip()
except (KeyboardInterrupt, EOFError):
print()
break
if q.lower() in {"quit", "exit"}:
break
if not q:
continue
result = answer_question(q, index)
print(f"\n{result.answer}\n")
print(f"[Sources: {', '.join(result.sources)}]")
print(f"[Cost: ${result.cost_usd:.5f} — "
f"{result.prompt_tokens} in / {result.completion_tokens} out]\n")
if __name__ == "__main__":
main()Running it
Put some .txt documents in data/corpus/. Nepali government-style circulars, company policies, whatever you have.
python rag.pyBuilding RAG index over data/corpus/ ...
Loaded 143 passages.
Index ready.
Question: What are the disclosure timelines for listed companies?The bot fires off a semantic search, retrieves the top 5 passages, hands them to Claude with the “answer only from these” instruction, and prints:
Listed companies must disclose material information within 24 hours of
occurrence, and file quarterly financial reports within 30 days of
quarter-end. Certain events (mergers, capital changes) require immediate
disclosure regardless of the 24-hour window.
[Sources: nepse_listing_bylaw_2024.txt, disclosure_amendments_2025.txt]
[Cost: $0.00082 — 1,850 in / 210 out]
Question: What's the meaning of life?
I don't have information about that in the provided documents.
[Sources: nepse_listing_bylaw_2024.txt, disclosure_amendments_2025.txt]
[Cost: $0.00048 — 1,320 in / 60 out]Two things to note:
- The answer is grounded, cited, and specific.
- When asked something the corpus doesn’t cover, the bot says so honestly. This is the behaviour the RAG system prompt engineered.
Common failure modes and their fixes
Six things that go wrong with real RAG systems, and their fixes:
1. Retrieval misses the right passage. → Improve chunking, embed with a better model, or add hybrid keyword search.
2. The LLM ignores the “answer only from passages” rule and hallucinates. → Strengthen the system prompt with capitalised MUSTs, and add a validator that flags responses without any citation.
3. The LLM refuses even when the answer is in the passages. → Usually a prompt phrasing issue; test with the retrieved passages inserted directly and see whether the model can answer at all.
4. Answer is technically correct but too long. → Add explicit length constraint in the prompt.
5. Sources are listed but not actually cited in the text. → Adjust the prompt to require inline [Source] markers.
6. Costs balloon. → Reduce k (fewer retrieved passages), use a smaller model, or cache answers to common questions.
Every one of these is a prompt or configuration change, not a fundamental redesign. Iteration is fast.
Check your understanding
Quick check
—A teammate's RAG pipeline sometimes says 'I don't have information about that' even when the answer is clearly in one of the retrieved passages. Investigation shows the passage does contain the answer verbatim. What's the most likely cause?
Quick check
—A RAG pipeline is answering with `[Cost: $0.05]` per question — much higher than expected. Investigation shows the pipeline is retrieving k=20 passages, most of which are long. What are the two-part fix?
What comes next
You have a working RAG pipeline. The last section of Chapter 5 wraps it in a specific project: a question-answering bot for local Nepali content — the kind of thing a small business, NGO, or agency could actually deploy. We add a small web UI, per-conversation memory, and honest failure handling.