Building Your First RAG System: From Zero to QA Hero
RAG grounds an LLM in real documents, which cuts hallucination and lets you serve fresh information. It matters to me because I want to build MoE-driven systems that retrieve and synthesize the way a person does, not just repeat patterns. This post builds a RAG system from scratch, one stage at a time: loading, chunking, embedding, vector storage and retrieval, and the prompt that turns retrieved context into an answer. Raw API calls, minimal dependencies, and enough of the reasoning to see why each piece is there.
Building Your First RAG System: From Zero to QA Hero
As a CSE student buried in algorithms and data structures, I keep coming back to the brain. How does it recall a fact, reason with it, and adapt? Large language models are impressive, but they're static snapshots of their training data, prone to hallucinating the moment you push past it. We don't just memorize everything; we retrieve what's relevant from a large, changing store of knowledge and then reason over it.
That's what Retrieval-Augmented Generation (RAG) gives you: a practical way to make an LLM more dynamic and grounded. For me it isn't only a patch for the model's limits. It points toward architectures where specialized expert modules, as in a Mixture of Experts (MoE) model, each draw on their own tuned memory.
Let's build one.
Why RAG?
Ask an LLM about your company's latest internal policy. Unless it happened to train on that exact document, it will guess or hallucinate. And fine-tuning the model for every new document is slow and expensive, like rewriting a textbook each time a footnote changes.
RAG handles this differently:
- It gives the model verifiable external context at inference time.
- You update the knowledge base without retraining anything.
- With facts in front of it, the model invents fewer answers.
- You avoid the cost of continuous fine-tuning.
For a competitive programmer, this is mostly about efficiency. Why rebuild the model when a good retrieval strategy gets you there?
The RAG pipeline
A RAG system has a few distinct but connected stages. We'll build each from scratch with raw API calls and minimal dependencies. I'll use Python here, but the same steps port to TypeScript or JavaScript with node-fetch and similar libraries.
1. Document loading
First, get your data into a usable format. For this guide it's a single text file. In practice you'd pull from databases, APIs, or formats like PDF and Markdown.
Let's assume we have a document named my_document.txt:
The quick brown fox jumps over the lazy dog.
This is a sample document to demonstrate RAG.
RAG systems combine retrieval and generation for better answers.
It helps LLMs by providing relevant context.
Mixture of Experts (MoE) models can enhance this by routing queries to specialized sub-models.
Each expert could have its own RAG system, mimicking specialized cognitive domains.
Performance is key in AI systems, demanding efficient data pipelines.# document_loader.py
def load_document(file_path: str) -> str:
"""Loads a text document from the specified path."""
try:
with open(file_path, 'r', encoding='utf-8') as f:
return f.read()
except FileNotFoundError:
print(f"Error: Document not found at {file_path}")
return ""
if __name__ == "__main__":
document_content = load_document("my_document.txt")
print(f"Loaded document (first 100 chars):\n{document_content[:100]}...")2. Chunking
LLMs and embedding models have token limits, so you can't feed a whole book as one chunk. Break the document into smaller, meaningful pieces. The simple starting strategy is fixed-size chunks with a small overlap so context carries across boundaries.
# chunker.py
from typing import List
def chunk_text(text: str, chunk_size: int = 200, chunk_overlap: int = 50) -> List[str]:
"""
Splits text into fixed-size chunks with overlap.
A simple, character-based chunker. For production, consider sentence-aware splitting.
"""
if not text:
return []
chunks = []
start = 0
while start < len(text):
end = min(start + chunk_size, len(text))
chunk = text[start:end]
chunks.append(chunk)
if end == len(text):
break
# Move start position back by overlap for the next chunk
start += chunk_size - chunk_overlap
# Ensure start doesn't go negative if chunk_size < chunk_overlap
if start < 0:
start = 0
return chunks
if __name__ == "__main__":
sample_text = "This is a long sentence that needs to be chunked into smaller pieces. We want to ensure that context is maintained across chunks. Overlap helps with this."
chunks = chunk_text(sample_text, chunk_size=50, chunk_overlap=10)
print("Generated Chunks:")
for i, chunk in enumerate(chunks):
print(f"Chunk {i+1}: '{chunk}'")Competitive programmer's note: this character-based chunking is O(N) in text length. Simple and fast. For production, a recursive character splitter or semantic chunking with spaCy/NLTK gives better results at the cost of complexity. Keep it simple for 0-to-hero.
3. Embedding
Now convert each chunk into a numerical vector. This is what lets us find similar chunks: text that means similar things ends up close together in the vector space.
We'll use a sentence-transformers model, which runs locally. You could use OpenAI's embedding API instead for cloud embeddings. Either way, skip high-level wrappers like LangChain's OpenAIEmbeddings and call the library or API directly.
# embedder.py
from typing import List
import numpy as np
# Prefer a fast, local model for quick iteration.
# If you don't have it, run: pip install sentence-transformers
from sentence_transformers import SentenceTransformer
class Embedder:
def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
"""
Initializes the embedding model.
'all-MiniLM-L6-v2' is a good balance of speed and performance for many tasks.
"""
print(f"Loading embedding model: {model_name}...")
self.model = SentenceTransformer(model_name)
print("Model loaded.")
def embed_chunks(self, chunks: List[str]) -> List[np.ndarray]:
"""
Embeds a list of text chunks into vectors.
"""
if not chunks:
return []
print(f"Embedding {len(chunks)} chunks...")
embeddings = self.model.encode(chunks, convert_to_numpy=True)
print("Embedding complete.")
return embeddings.tolist() # Convert to list of lists/ndarrays for easier storage
if __name__ == "__main__":
embedder = Embedder()
sample_chunks = [
"The quick brown fox jumps over the lazy dog.",
"A fast mammal with reddish-brown fur leaps over a sleepy canine."
]
embeddings = embedder.embed_chunks(sample_chunks)
for i, emb in enumerate(embeddings):
print(f"Embedding {i+1} shape: {len(emb)}")
print(f"Embedding {i+1} (first 5 values): {emb[:5]}")Performance note: all-MiniLM-L6-v2 is lightweight, so embedding is fast. On larger datasets, batch your calls; SentenceTransformer does this internally when you pass a list.
4. Vector storage and retrieval
With chunks and embeddings in hand, we need to store them and retrieve the most relevant ones for a query. For a from-zero guide, in-memory is fine. A plain list with brute-force search works, but on any non-trivial dataset a vector index like FAISS (Facebook AI Similarity Search) is far faster.
# vector_store.py
from typing import List, Tuple
import numpy as np
# pip install faiss-cpu
import faiss
class VectorStore:
def __init__(self, dimension: int):
"""
Initializes an in-memory FAISS index.
For larger datasets, consider `faiss.IndexFlatL2` for L2 distance, or more advanced indices.
"""
self.index = faiss.IndexFlatIP(dimension) # IP for Inner Product, suitable for normalized cosine similarity
self.texts: List[str] = []
def add_vectors(self, embeddings: List[np.ndarray], texts: List[str]):
"""
Adds vectors and their corresponding texts to the store.
"""
if not embeddings or not texts:
return
if len(embeddings) != len(texts):
raise ValueError("Number of embeddings must match number of texts.")
embeddings_np = np.array(embeddings).astype('float32')
# Normalize embeddings for cosine similarity with Inner Product index
faiss.normalize_L2(embeddings_np)
self.index.add(embeddings_np)
self.texts.extend(texts)
print(f"Added {len(embeddings)} vectors to the store.")
def search(self, query_embedding: np.ndarray, k: int = 3) -> List[Tuple[str, float]]:
"""
Searches for the k most similar texts to the query embedding.
Returns a list of (text, similarity_score) tuples.
"""
if self.index.ntotal == 0:
return []
query_embedding_np = np.array([query_embedding]).astype('float32')
faiss.normalize_L2(query_embedding_np) # Normalize query embedding too
distances, indices = self.index.search(query_embedding_np, k)
results = []
for i, dist in zip(indices[0], distances[0]):
if i != -1: # -1 indicates no result found (shouldn't happen if k <= ntotal)
results.append((self.texts[i], dist))
print(f"Retrieved {len(results)} relevant chunks.")
return results
if __name__ == "__main__":
# Example usage (requires an Embedder instance)
from embedder import Embedder
embedder = Embedder()
sample_chunks = [
"The quick brown fox jumps over the lazy dog.",
"A fast mammal with reddish-brown fur leaps over a sleepy canine.",
"RAG systems combine retrieval and generation for better answers.",
"Performance is key in AI systems."
]
embeddings = embedder.embed_chunks(sample_chunks)
# Initialize VectorStore with the dimension of our embeddings
vector_store = VectorStore(dimension=len(embeddings[0]))
vector_store.add_vectors(embeddings, sample_chunks)
query = "What is RAG?"
query_embedding = embedder.embed_chunks([query])[0]
search_results = vector_store.search(query_embedding, k=2)
print("\nSearch Results for 'What is RAG?':")
for text, score in search_results:
print(f"Score: {score:.4f}, Text: '{text}'")Competitive programmer's note: FAISS runs optimized C++ for nearest-neighbor search and beats Python loops badly on large datasets. IndexFlatIP (inner product), with L2-normalized embeddings, gives you cosine similarity, a common and effective metric.
5. Prompt engineering
Here the retrieved context meets the LLM. The prompt sets the model's role, hands it the context, and then asks the user's question. Clear, tight prompts matter for good answers.
We'll use OpenAI's API directly. If you prefer a local LLM, ollama or transformers can be used similarly.
# llm_client.py
import os
from typing import List, Dict
import openai # pip install openai
class LLMClient:
def __init__(self, api_key: str, model_name: str = "gpt-3.5-turbo"):
"""
Initializes the OpenAI LLM client.
Ensure OPENAI_API_KEY is set in your environment or passed directly.
"""
self.client = openai.OpenAI(api_key=api_key)
self.model_name = model_name
def generate_response(self, prompt_messages: List[Dict[str, str]]) -> str:
"""
Sends a list of messages to the LLM and returns the generated response.
"""
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=prompt_messages,
temperature=0.0 # For factual QA, lower temperature is usually better
)
return response.choices[0].message.content
except openai.AuthenticationError:
print("Error: OpenAI API key is invalid or not provided.")
return "Error: Could not authenticate with OpenAI. Please check your API key."
except Exception as e:
print(f"Error calling LLM: {e}")
return "Error: Could not generate response."
if __name__ == "__main__":
# For testing, ensure OPENAI_API_KEY is set in your environment
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
print("Please set the OPENAI_API_KEY environment variable.")
else:
llm_client = LLMClient(api_key=api_key)
test_messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is the capital of France?"}
]
response = llm_client.generate_response(test_messages)
print(f"LLM Response: {response}")Opinion: calling the API directly gives you control over every parameter. Framework layers often hide the ones that matter and add overhead. For performance and precision, stay close to the raw interface.
6. Tying it together
Now assemble the components into a working RAG system.
# rag_system.py
import os
from document_loader import load_document
from chunker import chunk_text
from embedder import Embedder
from vector_store import VectorStore
from llm_client import LLMClient
from typing import List, Dict
class RAGSystem:
def __init__(self, document_path: str, openai_api_key: str):
self.document_path = document_path
self.embedder = Embedder()
self.llm_client = LLMClient(api_key=openai_api_key)
self.vector_store: VectorStore = None
self._initialize_knowledge_base()
def _initialize_knowledge_base(self):
"""Loads, chunks, and embeds the document to set up the vector store."""
print("Initializing RAG knowledge base...")
document_content = load_document(self.document_path)
if not document_content:
raise ValueError("Failed to load document.")
chunks = chunk_text(document_content)
if not chunks:
raise ValueError("No chunks generated from document.")
embeddings = self.embedder.embed_chunks(chunks)
if not embeddings:
raise ValueError("No embeddings generated.")
self.vector_store = VectorStore(dimension=len(embeddings[0]))
self.vector_store.add_vectors(embeddings, chunks)
print("RAG knowledge base initialized.")
def ask(self, query: str, k_retrievals: int = 3) -> str:
"""
Performs a RAG query:
1. Embeds the user query.
2. Retrieves relevant chunks from the vector store.
3. Constructs a prompt with the retrieved context.
4. Generates a response using the LLM.
"""
if not self.vector_store:
return "RAG system not initialized. Please check document loading."
print(f"\nProcessing query: '{query}'")
# 1. Embed the user query
query_embedding = self.embedder.embed_chunks([query])[0]
# 2. Retrieve relevant chunks
retrieved_chunks_info = self.vector_store.search(query_embedding, k=k_retrievals)
retrieved_texts = [text for text, score in retrieved_chunks_info]
# Combine retrieved texts into a single context string
context = "\n---\n".join(retrieved_texts)
# 3. Construct the prompt
system_message = {
"role": "system",
"content": (
"You are an intelligent QA assistant. "
"Use the provided context to answer the user's question. "
"If the answer is not in the context, state that you don't have enough information."
"Be concise and direct."
)
}
user_message = {
"role": "user",
"content": (
f"Context:\n{context}\n\n"
f"Question: {query}"
)
}
prompt_messages: List[Dict[str, str]] = [system_message, user_message]
print("Sending prompt to LLM...")
# 4. Generate response
response = self.llm_client.generate_response(prompt_messages)
print("LLM response received.")
return response
if __name__ == "__main__":
# Create a dummy document for demonstration
with open("my_document.txt", "w") as f:
f.write("""The quick brown fox jumps over the lazy dog.
This is a sample document to demonstrate RAG.
RAG systems combine retrieval and generation for better answers.
It helps LLMs by providing relevant context.
Mixture of Experts (MoE) models can enhance this by routing queries to specialized sub-models.
Each expert could have its own RAG system, mimicking specialized cognitive domains.
Performance is key in AI systems, demanding efficient data pipelines.
My favorite fictional character is Sherlock Holmes, a brilliant detective.
He uses deductive reasoning to solve complex cases.
His methods are a great example of structured problem-solving.
""")
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
print("Please set the OPENAI_API_KEY environment variable to run the RAG system.")
else:
try:
rag_system = RAGSystem(document_path="my_document.txt", openai_api_key=api_key)
print("\n--- RAG QA Session ---")
queries = [
"What is RAG and how does it help LLMs?",
"What is the significance of MoE models in this context?",
"Who is Sherlock Holmes?",
"What is the capital of Mars?" # This should fail gracefully
]
for q in queries:
answer = rag_system.ask(q)
print(f"\nQuestion: {q}")
print(f"Answer: {answer}")
except ValueError as e:
print(f"Initialization Error: {e}")
finally:
# Clean up the dummy document
if os.path.exists("my_document.txt"):
os.remove("my_document.txt")
What I learned
Building this from scratch drove home a few things:
- Keep it modular. Each component (loader, chunker, embedder, vector store, LLM client) is a separate, testable unit, which is what makes the system easy to scale and maintain.
- Directness buys performance. With no framework in the way, you can profile each step and optimize where it counts, which matters for anything real-time.
- RAG mirrors how we actually think. We retrieve relevant information before we answer, rather than reasoning in a vacuum.
- It scales to MoE. This pipeline is a small version of what I want from MoE architectures: a router sends a query to a specialized expert (finance, medical, fiction), and each expert runs its own RAG system tuned to its domain. You get specificity and recency together, with the router deciding which module to engage.
This first RAG is a small thing, but the fundamentals under it are the ones that carry all the way up. The path from a simple QA bot to something genuinely adaptive is long, and getting these building blocks right, with a real eye on efficiency, is how you walk it.
Next steps
- Try semantic chunking or recursive text splitters to preserve context better.
- Test other embedding models (open-source ones via
HuggingFaceEmbeddingsfortransformers, or fine-tune your own). - Swap the in-memory store for a persistent one like
Qdrant,Weaviate,Pinecone, orpgvector. - Add evaluation metrics (ROUGE, faithfulness, answer relevancy) to measure performance.
- Port it to TypeScript with
node-fetchand a server-side embedding service.
Grounded, fast AI is built on RAG systems that are actually well engineered. Let's build.