Mini-Project: RAG System with Intelligent Chunking

Project overview

You'll build a complete RAG system that implements intelligent chunking (recursive), embeddings (OpenAI), retrieval with vector search (in-memory numpy), and evaluation with metrics (nDCG, Recall@K). This system is the foundation for production-ready RAG.

By completing this project, you'll have a functional RAG with all the optimizations you've learned.

---

Objectives

1. ✅ Recursive chunking with metadata
2. ✅ Vector embeddings (OpenAI)
3. ✅ Vector search (cosine similarity)
4. ✅ Evaluation with metrics
5. ✅ CLI for interactive testing

---

Structure

rag-system/
├── src/
│   ├── chunker.py      # Recursive chunking
│   ├── embedder.py     # OpenAI embeddings
│   ├── retriever.py    # Vector search
│   └── evaluator.py    # Metrics
├── data/
│   └── documents.txt   # Corpus
├── main.py             # CLI
└── requirements.txt

---

Implementation

chunker.py:

from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken

class SmartChunker:
    """Recursive chunker with metadata"""
    
    def __init__(self, chunk_size=500, overlap=50):
        def tiktoken_len(text):
            encoding = tiktoken.encoding_for_model("gpt-4")
            return len(encoding.encode(text))
        
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=overlap,
            length_function=tiktoken_len,
            separators=["\n\n", "\n", ". ", " ", ""]
        )
    
    def chunk(self, documents):
        """Chunking with metadata"""
        chunks = []
        
        for doc_id, doc in enumerate(documents):
            texts = self.splitter.split_text(doc['text'])
            
            for chunk_id, text in enumerate(texts):
                chunks.append({
                    'id': f"{doc_id}_{chunk_id}",
                    'text': text,
                    'doc_id': doc_id,
                    'source': doc.get('source', 'unknown')
                })
        
        return chunks

---

embedder.py:

from openai import OpenAI
import numpy as np
import os
from dotenv import load_dotenv

load_dotenv()

class Embedder:
    """OpenAI embeddings"""
    
    def __init__(self):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    def embed(self, texts):
        """Batch embedding"""
        response = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=texts
        )
        
        return np.array([item.embedding for item in response.data])

---

retriever.py:

import numpy as np

class VectorRetriever:
    """Cosine similarity search"""
    
    def __init__(self, embedder):
        self.embedder = embedder
        self.chunks = None
        self.embeddings = None
    
    def index(self, chunks):
        """Index chunks"""
        self.chunks = chunks
        texts = [c['text'] for c in chunks]
        self.embeddings = self.embedder.embed(texts)
        print(f"✅ Indexed {len(chunks)} chunks")
    
    def search(self, query, k=5):
        """Retrieve top-K"""
        query_emb = self.embedder.embed([query])[0]
        
        # Cosine similarity
        similarities = np.dot(self.embeddings, query_emb) / (
            np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(query_emb)
        )
        
        # Top-K indices
        top_k_idx = np.argsort(similarities)[::-1][:k]
        
        results = []
        for idx in top_k_idx:
            results.append({
                'chunk': self.chunks[idx],
                'score': float(similarities[idx])
            })
        
        return results

---

main.py:

from src.chunker import SmartChunker
from src.embedder import Embedder
from src.retriever import VectorRetriever

def main():
    print("=== RAG System with Smart Chunking ===\n")
    
    # Load documents
    documents = [
        {
            'text': "Python is a programming language. Install from python.org. Created in 1991.",
            'source': 'python_guide.md'
        },
        {
            'text': "JavaScript is used for web development. Node.js runs JS on servers.",
            'source': 'js_guide.md'
        }
    ]
    
    # Chunking
    print("1. Chunking documents...")
    chunker = SmartChunker(chunk_size=100, overlap=10)
    chunks = chunker.chunk(documents)
    print(f"   Generated {len(chunks)} chunks\n")
    
    # Embedding + Indexing
    print("2. Embedding and indexing...")
    embedder = Embedder()
    retriever = VectorRetriever(embedder)
    retriever.index(chunks)
    print()
    
    # Interactive search
    while True:
        query = input("Query (or 'quit'): ")
        if query.lower() == 'quit':
            break
        
        results = retriever.search(query, k=3)
        
        print(f"\nTop-3 results:")
        for i, result in enumerate(results):
            print(f"\n{i+1}. Score: {result['score']:.4f}")
            print(f"   Text: {result['chunk']['text'][:100]}...")
            print(f"   Source: {result['chunk']['source']}")

if __name__ == "__main__":
    main()

---

Execution

python main.py

Output:

=== RAG System with Smart Chunking ===

1. Chunking documents...
   Generated 2 chunks

2. Embedding and indexing...
✅ Indexed 2 chunks

Query (or 'quit'): How to install Python?

Top-3 results:

1. Score: 0.8523
   Text: Python is a programming language. Install from python.org. Created in 1991....
   Source: python_guide.md

2. Score: 0.4521
   Text: JavaScript is used for web development. Node.js runs JS on servers....
   Source: js_guide.md

---

Evaluation

Add an evaluation dataset:

eval_dataset = [
    {
        'query': 'How to install Python?',
        'relevant_chunks': ['0_0']  # Relevant chunk ID
    }
]

# Evaluate
from src.evaluator import RetrievalEvaluator

evaluator = RetrievalEvaluator(k=5)
retrieved_ids = [r['chunk']['id'] for r in retriever.search(query, k=5)]
relevant_ids = eval_dataset[0]['relevant_chunks']

metrics = evaluator.evaluate_all(retrieved_ids, relevant_ids)
print(f"Recall@5: {metrics['recall@k']:.3f}")
print(f"nDCG@5: {metrics['ndcg@k']:.3f}")

---

Optional extensions

1. Vector database (Pinecone, Weaviate)
2. Reranking with a cross-encoder
3. Hybrid search (BM25 + embeddings)
4. LLM generation (GPT-4 for answers)

---

Module 4 summary

What you implemented:

• ✅ Fixed-size chunking (token-based)
• ✅ Semantic chunking (recursive)
• ✅ Retrieval metrics (nDCG, Recall@K)
• ✅ Evaluation datasets
• ✅ A/B testing
• ✅ Complete RAG system

Lines of code: ~800 (production-ready)

---

Next module

Module 5: Distance Metrics Deep Dive

You'll learn:

• Cosine similarity (why it's the standard)
• Euclidean distance (when to use it)
• Dot product (normalized embeddings)
• Manhattan, Hamming distances
• Trade-offs and performance

From RAG to advanced distance metrics.

---

Module 4 completed ✅ Chunking + Evaluation: production-ready RAG