Building RAG from Scratch: No Frameworks Needed
Building RAG from Scratch: No Frameworks Needed Understand Retrieval-Augmented Generation by building it yourself — pure Python, zero dependencies. Introduction LangChain, LlamaIndex, Haystack — great tools
Building RAG from Scratch: No Frameworks Needed
Understand Retrieval-Augmented Generation by building it yourself — pure Python, zero dependencies.
Introduction
LangChain, LlamaIndex, Haystack — great tools, but they hide how RAG actually works. This tutorial builds a production-ready RAG system from scratch using only requests, numpy, and a vector database client.
RAG Architecture
User Query → Embed → Vector Search → Retrieve Context → LLM Generate → Answer
Four components:
- Embedder — Text → Vector
- Vector Store — Store & Search vectors
- Retriever — Top-k similar chunks
- Generator — LLM + Context → Answer
Step 1: Embedding Function
# embedder.py
import os
import requests
import numpy as np
from typing import List
class Embedder:
def __init__(self, provider: str = "openrouter"):
self.provider = provider
self.configs = {
"openrouter": {
"url": "https://openrouter.ai/api/v1/embeddings",
"key": os.getenv("OPENROUTER_API_KEY"),
"model": "text-embedding-3-small" # 1536 dims
},
"local": {
"url": "http://localhost:11434/api/embeddings",
"model": "nomic-embed-text" # 768 dims via Ollama
}
}
def embed(self, texts: List[str]) -> np.ndarray:
config = self.configs[self.provider]
if self.provider == "openrouter":
headers = {
"Authorization": f"Bearer {config['key']}",
"Content-Type": "application/json"
}
response = requests.post(
config["url"],
headers=headers,
json={"model": config["model"], "input": texts},
timeout=30
)
response.raise_for_status()
data = response.json()["data"]
embeddings = [d["embedding"] for d in sorted(data, key=lambda x: x["index"])]
elif self.provider == "local":
embeddings = []
for text in texts:
r = requests.post(config["url"], json={"model": config["model"], "prompt": text})
r.raise_for_status()
embeddings.append(r.json()["embedding"])
return np.array(embeddings, dtype=np.float32)
def embed_one(self, text: str) -> np.ndarray:
return self.embed([text])[0]
Step 2: Vector Store Interface
# vector_store.py
import numpy as np
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
class VectorStore(ABC):
@abstractmethod
def add(self, vectors: np.ndarray, metadatas: List[Dict], ids: List[str]): ...
@abstractmethod
def search(self, query: np.ndarray, top_k: int = 5,
filter: Optional[Dict] = None) -> List[Dict]: ...
@abstractmethod
def delete(self, ids: List[str]): ...
# Qdrant Implementation
class QdrantStore(VectorStore):
def __init__(self, url: str, api_key: str, collection: str = "rag_docs"):
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams
self.client = QdrantClient(url=url, api_key=api_key)
self.collection = collection
# Create collection if not exists
collections = self.client.get_collections().collections
if collection not in [c.name for c in collections]:
self.client.create_collection(
collection_name=collection,
vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)
def add(self, vectors: np.ndarray, metadatas: List[Dict], ids: List[str]):
from qdrant_client.models import PointStruct
points = [
PointStruct(id=id_, vector=vec.tolist(), payload=meta)
for id_, vec, meta in zip(ids, vectors, metadatas)
]
self.client.upsert(collection_name=self.collection, points=points)
def search(self, query: np.ndarray, top_k: int = 5,
filter: Optional[Dict] = None) -> List[Dict]:
from qdrant_client.models import Filter, FieldCondition, MatchValue
q_filter = None
if filter:
conditions = [
FieldCondition(key=k, match=MatchValue(value=v))
for k, v in filter.items()
]
q_filter = Filter(must=conditions)
results = self.client.query_points(
collection_name=self.collection,
query=query.tolist(),
query_filter=q_filter,
limit=top_k,
with_payload=True
)
return [
{"id": str(r.id), "score": r.score, "metadata": r.payload}
for r in results.points
]
def delete(self, ids: List[str]):
self.client.delete(collection_name=self.collection, points_selector=ids)
# Simple NumPy Store (for testing, no external DB)
class NumpyStore(VectorStore):
def __init__(self, dim: int = 1536):
self.vectors = np.empty((0, dim), dtype=np.float32)
self.metadatas = []
self.ids = []
def add(self, vectors: np.ndarray, metadatas: List[Dict], ids: List[str]):
self.vectors = np.vstack([self.vectors, vectors])
self.metadatas.extend(metadatas)
self.ids.extend(ids)
def search(self, query: np.ndarray, top_k: int = 5,
filter: Optional[Dict] = None) -> List[Dict]:
# Cosine similarity
norms = np.linalg.norm(self.vectors, axis=1)
query_norm = np.linalg.norm(query)
if query_norm == 0 or np.any(norms == 0):
return []
similarities = (self.vectors @ query) / (norms * query_norm)
# Apply filter
if filter:
mask = np.ones(len(self.metadatas), dtype=bool)
for k, v in filter.items():
mask &= np.array([m.get(k) == v for m in self.metadatas])
similarities[~mask] = -1
top_indices = np.argsort(similarities)[-top_k:][::-1]
return [
{"id": self.ids[i], "score": float(similarities[i]), "metadata": self.metadatas[i]}
for i in top_indices if similarities[i] > 0
]
def delete(self, ids: List[str]):
indices = [self.ids.index(id_) for id_ in ids if id_ in self.ids]
self.vectors = np.delete(self.vectors, indices, axis=0)
for i in sorted(indices, reverse=True):
del self.metadatas[i]
del self.ids[i]
Step 3: Document Chunker
# chunker.py
from typing import List
import re
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
"""Split text into overlapping chunks by sentences."""
sentences = re.split(r'(?<=[.!?])\s+', text)
chunks = []
current_chunk = []
current_length = 0
for sentence in sentences:
sent_len = len(sentence)
if current_length + sent_len > chunk_size and current_chunk:
chunks.append(" ".join(current_chunk))
# Keep overlap
overlap_text = " ".join(current_chunk)[-overlap:]
current_chunk = [overlap_text, sentence]
current_length = len(overlap_text) + sent_len
else:
current_chunk.append(sentence)
current_length += sent_len
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
def chunk_markdown(text: str, chunk_size: int = 500) -> List[str]:
"""Chunk preserving markdown structure."""
# Split by headers first
sections = re.split(r'(^#{1,3}\s+.+$)', text, flags=re.MULTILINE)
chunks = []
current = ""
for section in sections:
if not section.strip():
continue
if len(current) + len(section) > chunk_size and current:
chunks.append(current.strip())
current = section
else:
current += "\n" + section
if current.strip():
chunks.append(current.strip())
return chunks
Step 4: Retriever
# retriever.py
import numpy as np
from typing import List, Dict, Optional
from embedder import Embedder
from vector_store import VectorStore
class Retriever:
def __init__(self, embedder: Embedder, store: VectorStore):
self.embedder = embedder
self.store = store
def add_documents(self, texts: List[str], metadatas: List[Dict], ids: List[str]):
"""Embed and store documents."""
vectors = self.embedder.embed(texts)
self.store.add(vectors, metadatas, ids)
def add_file(self, filepath: str, metadata: Dict = None, chunk_size: int = 500):
"""Read, chunk, embed, and store a file."""
from chunker import chunk_text
with open(filepath, 'r') as f:
content = f.read()
chunks = chunk_text(content, chunk_size=chunk_size)
base_id = filepath.replace("/", "_").replace(".", "_")
ids = [f"{base_id}_{i}" for i in range(len(chunks))]
metadatas = [
{**(metadata or {}), "source": filepath, "chunk": i, "text": chunk}
for i, chunk in enumerate(chunks)
]
self.add_documents(chunks, metadatas, ids)
return len(chunks)
def retrieve(self, query: str, top_k: int = 5,
filter: Optional[Dict] = None) -> List[Dict]:
"""Retrieve relevant chunks for a query."""
query_vector = self.embedder.embed_one(query)
return self.store.search(query_vector, top_k=top_k, filter=filter)
Step 5: Generator (LLM)
# generator.py
import os
import requests
from typing import List, Dict
class Generator:
def __init__(self, provider: str = "openrouter"):
self.provider = provider
self.configs = {
"openrouter": {
"url": "https://openrouter.ai/api/v1/chat/completions",
"key": os.getenv("OPENROUTER_API_KEY"),
"model": "google/gemini-2.0-flash-exp:free"
},
"groq": {
"url": "https://api.groq.com/openai/v1/chat/completions",
"key": os.getenv("GROQ_API_KEY"),
"model": "llama-3.1-8b-instant"
}
}
def generate(self, query: str, context_chunks: List[Dict],
system_prompt: str = None) -> str:
config = self.configs[self.provider]
# Build context
context = "\n\n".join([
f"[Source: {c['metadata'].get('source', 'unknown')}]\n{c['metadata'].get('text', '')}"
for c in context_chunks
])
# Default system prompt
if system_prompt is None:
system_prompt = """You are a helpful assistant. Answer the question using ONLY the provided context.
If the context doesn't contain the answer, say "I don't have enough information to answer."
Cite sources using [Source: filename] format."""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"}
]
headers = {
"Authorization": f"Bearer {config['key']}",
"Content-Type": "application/json"
}
response = requests.post(
config["url"],
headers=headers,
json={
"model": config["model"],
"messages": messages,
"temperature": 0.1,
"max_tokens": 1000
},
timeout=60
)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
def generate_stream(self, query: str, context_chunks: List[Dict]):
"""Streaming version (for UIs)."""
config = self.configs[self.provider]
context = "\n\n".join([
f"[Source: {c['metadata'].get('source', 'unknown')}]\n{c['metadata'].get('text', '')}"
for c in context_chunks
])
messages = [
{"role": "system", "content": "Answer using only provided context. Cite sources."},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"}
]
headers = {
"Authorization": f"Bearer {config['key']}",
"Content-Type": "application/json"
}
response = requests.post(
config["url"],
headers=headers,
json={
"model": config["model"],
"messages": messages,
"temperature": 0.1,
"max_tokens": 1000,
"stream": True
},
stream=True,
timeout=60
)
response.raise_for_status()
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
break
try:
import json
chunk = json.loads(data)
delta = chunk["choices"][0]["delta"]
if "content" in delta:
yield delta["content"]
except:
pass
Step 6: Complete RAG Pipeline
# rag.py
from typing import List, Dict, Optional
from embedder import Embedder
from vector_store import QdrantStore, NumpyStore
from retriever import Retriever
from generator import Generator
class RAGPipeline:
def __init__(self,
embedder_provider: str = "openrouter",
store_type: str = "qdrant",
generator_provider: str = "openrouter",
**store_kwargs):
self.embedder = Embedder(embedder_provider)
if store_type == "qdrant":
self.store = QdrantStore(**store_kwargs)
elif store_type == "numpy":
self.store = NumpyStore()
else:
raise ValueError(f"Unknown store: {store_type}")
self.retriever = Retriever(self.embedder, self.store)
self.generator = Generator(generator_provider)
def ingest(self, filepath: str, metadata: Dict = None, chunk_size: int = 500) -> int:
"""Add a document to the knowledge base."""
return self.retriever.add_file(filepath, metadata, chunk_size)
def ingest_text(self, text: str, metadata: Dict = None, chunk_size: int = 500) -> int:
"""Add raw text to the knowledge base."""
from chunker import chunk_text
chunks = chunk_text(text, chunk_size)
ids = [f"text_{i}" for i in range(len(chunks))]
metadatas = [
{**(metadata or {}), "chunk": i, "text": chunk}
for i, chunk in enumerate(chunks)
]
self.retriever.add_documents(chunks, metadatas, ids)
return len(chunks)
def query(self, question: str, top_k: int = 5,
filter: Optional[Dict] = None) -> Dict:
"""Full RAG query: retrieve + generate."""
# Retrieve
chunks = self.retriever.retrieve(question, top_k=top_k, filter=filter)
if not chunks:
return {
"answer": "No relevant documents found.",
"sources": [],
"query": question
}
# Generate
answer = self.generator.generate(question, chunks)
return {
"answer": answer,
"sources": [
{
"source": c["metadata"].get("source", "unknown"),
"chunk": c["metadata"].get("chunk", 0),
"score": c["score"]
}
for c in chunks
],
"query": question
}
def query_stream(self, question: str, top_k: int = 5):
"""Streaming query for real-time UIs."""
chunks = self.retriever.retrieve(question, top_k=top_k)
if not chunks:
yield "No relevant documents found."
return
for token in self.generator.generate_stream(question, chunks):
yield token
Usage Example
# main.py
import os
from rag import RAGPipeline
# Initialize
rag = RAGPipeline(
embedder_provider="openrouter",
store_type="qdrant",
generator_provider="openrouter",
url=os.getenv("QDRANT_URL"),
api_key=os.getenv("QDRANT_API_KEY"),
collection="my_docs"
)
# Ingest documents
rag.ingest("docs/api_docs.md", {"category": "api"})
rag.ingest("docs/tutorial.md", {"category": "tutorial"})
rag.ingest_text("""
# Company Policy
Remote work is allowed 3 days/week.
Vacation: 20 days/year.
""", {"category": "policy"})
# Query
result = rag.query("What's the remote work policy?")
print(result["answer"])
print("\nSources:")
for s in result["sources"]:
print(f" - {s['source']} (chunk {s['chunk']}, score: {s['score']:.3f})")
# Streaming
print("\nStreaming answer:")
for token in rag.query_stream("How many vacation days?"):
print(token, end="", flush=True)
Run with Simple NumPy Store (No DB Needed)
# simple_rag.py
from rag import RAGPipeline
rag = RAGPipeline(
embedder_provider="local", # Uses Ollama
store_type="numpy",
generator_provider="openrouter"
)
rag.ingest_text("Your document text here...")
result = rag.query("Your question")
print(result["answer"])
Requirements
pip install requests numpy qdrant-client
# For local embeddings:
# ollama pull nomic-embed-text
Cost: $0 (Free Tier)
| Component | Free Option |
|---|---|
| Embeddings | OpenRouter (text-embedding-3-small) or Ollama |
| Vector DB | Qdrant Cloud (1M vectors) or self-hosted |
| LLM | OpenRouter/Groq/Together free models |
| Hosting | Local / Railway / Render free tier |
Conclusion
You now understand every piece of RAG. No black boxes — just embeddings, vector search, and prompt engineering. This foundation lets you:
- Customize chunking for your domain
- Add hybrid search (keyword + vector)
- Implement re-ranking
- Build domain-specific retrievers
Next: Add cross-encoder re-ranking, hybrid BM25+vector search, or agentic RAG with tool use.
Part of the RAG from Scratch series on Dev.to
📰 Read the original article on Dev.to AI
Originally published by Dev.to AI. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.