Create Your Own AI Agent – ElizaOS No GPU Required, 100% Free!
January 15, 2025Hide and unhide files in your system.
March 9, 2025Smart Way to Use Your LLM (Without Breaking the Bank)
Let's look at it this way, you have a mountain of documents—PDFs, web pages, tables,, or even images. Now, you need ChatGPT to make sense of all that data.
You have two choices:
1.It's either you fine-tune a Large Language Model (LLM) to learn your data inside out— which is complex, costly, and also time-consuming.
2.Or usse RAG (Retrieval-Augmented Generation)— a smarter, faster, and more efficient approach.
So, how does RAG work?
Well, it doesn’t just dump all your data into an LLM. Instead, it indexes your documents and retrieves only the relevant information - using methods like similarity search. That is, indexing your search query and chunked-result and comparing values.
Here’s how it works:
•Your documents are split into smaller pieces (chunks).
•When you ask a question, the system finds the most relevant chunk instead of searching through everything.
•That chunk, along with your query, is then fed into an LLM (like ChatGPT or DeepSeek), which generates a refined, context-aware answer based on your document.
Think of it this way - you have a detailed encyclopedia about the UAE, covering Dubai, Abu Dhabi, Sharjah, and other emirates. Chunking means breaking it down by emirate. Now, if you ask about Dubai, the system fetches only the Dubai section—not everything about the UAE. This same logic is applied if you ask questions about Abu Dhabi or Sharjah.
Now, to the most important question. Why does this matter? Because LLMs are expensive to run and they all have token -limits, meaning you only send a limited number of file/text/information! So instead of feeding an entire encyclopedia into an AI every time you ask a question, RAG ensures that you only send what’s necessary. Less data = lower costs and faster responses.
let's start by installing and importing the necessary dependencies.
!pip install faiss-cpu --upgrade pymupdf !pip install pytesseract pdf2image !apt-get install -y poppler-utils !apt-get install -y poppler-utils tesseract-ocr # Install dependencies !pip install qdrant-client sentence-transformers numpy
import faiss
import fitz # PyMuPDF for PDF reading
import nltk # For sentence tokenization
import faiss # For vector search
import numpy as np
import nltk
from nltk.tokenize import sent_tokenize
from transformers import AutoTokenizer # For token counting
from sentence_transformers import SentenceTransformer
import textwrap
import pytesseract
from pdf2image import convert_from_path
# Initialize a tokenizer (using GPT-2 as an approximation for token counts)
tokenizer = AutoTokenizer.from_pretrained("gpt2") #Using gpt2 to tokenize
nltk.download('punkt')
nltk.download('punkt_tab')
In this case, we will be importing a pdf file (including text and images).
def extract_text_from_pdf(pdf_path):
doc = fitz.open(pdf_path)
text = ""
for page in doc:
text += page.get_text("text") + "\n"
# Basic cleaning: remove excessive whitespace
clean_text = " ".join(text.split())
return clean_text
The next and most important part will be how to chunk our texts accurately?
So far, this has been the most difficult part for me. Choosing an ideal chunking method that can handle different use-cases.
When we rely solely on sentence-based chunking, we risk isolating context that spans multiple sentences. Conversely, chunking by paragraphs might scatter key information across neighboring sections, leading to incomplete query responses. Balancing these approaches is crucial to preserve the full context for accurate and comprehensive retrieval.
Choosing the right chunking approach depends on carefully analysing our texts/documents to find a better match.
#Please note: It is important to preprocess our texts before chunking. Each data requires a different preprocessing step, hence the reason why I will be skipping this part. But then again, preprocessing your data heavily influences your chunking result.
Here are some chunking methods:
Fixed-Length : Chunks based on a particular length, with overlaps added to preserve context.
#Fixed_length chunking
def fixed_length_chunking(text, chunk_size=500, overlap=50):
chunks = []
for i in range(0, len(text), chunk_size - overlap):
chunks.append(text[i:i + chunk_size])
return chunks
#use
text = "How can we effectively chunk large documents without missing context "
chunks = fixed_length_chunking(text, chunk_size=100, overlap=20)
print(chunks)
#sentence based chunking
#import nltk
#nltk.download('punkt')
#from nltk.tokenize import sent_tokenize
def sentence_chunking(text, max_sentences=5):
sentences = sent_tokenize(text)
chunks = [" ".join(sentences[i:i + max_sentences]) for i in range(0, len(sentences), max_sentences)]
return chunks
# Use
text = "This is sentence one. This is sentence two. This is sentence three..."
chunks = sentence_chunking(text, max_sentences=2)
print(chunks)
#token_based chunking
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
def token_based_chunking(text, max_tokens=100):
tokens = tokenizer.encode(text, add_special_tokens=False)
chunks = [tokenizer.decode(tokens[i:i + max_tokens]) for i in range(0, len(tokens), max_tokens)]
return chunks
# Use
text = "This is a long document that needs to be chunked..."
chunks = token_based_chunking(text, max_tokens=50)
print(chunks)
#semantic based chunking.
from langchain.text_splitter import RecursiveCharacterTextSplitter
def semantic_chunking(text):
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
return text_splitter.split_text(text)
# Use
text = "This is a long document that needs to be chunked..."
chunks = semantic_chunking(text)
print(chunks)
#Paragrapgh based chunking
import re
def paragraph_chunking(text):
return re.split(r'\n\n+', text.strip())
#Use
text = "Paragraph 1.\n\nParagraph 2.\n\nParagraph 3."
chunks = paragraph_chunking(text)
print(chunks)
#Heading based chunking
import re
def heading_based_chunking(text):
chunks = re.split(r'(?:\n|^)#\s+', text) # Splitting at Markdown-style headers
return [chunk.strip() for chunk in chunks if chunk.strip()]
# Use
text = "# Introduction\nThis is the intro.\n\n# Methods\nHere are the methods."
chunks = heading_based_chunking(text)
print(chunks)
#table-based chunking
import pandas as pd
def table_chunking(csv_file):
df = pd.read_csv(csv_file)
return df.to_dict(orient="records") # Convert each row into a dictionary
# use
# chunks = table_chunking("data.csv")
# print(chunks)
#image-based chunking using pyteserract.
import pytesseract
from PIL import Image
def image_chunking(image_path):
return pytesseract.image_to_string(Image.open(image_path))
# Use
# text = image_chunking("image.png")
# print(text)
Fixed-Length: Chunking based on a fixed length with an overlap to preserve context.
Sentence-Based. Chunking based on sentences using (.) to differentiate between sentences.
Token-Based: Splitting chunks based on a particular number of tokens.
More efficient methods include Colipali and VLM chunker.
import nltk
import numpy as np
from nltk.tokenize import sent_tokenize
from transformers import CLIPTokenizer, CLIPModel, AutoTokenizer
from sklearn.metrics.pairwise import cosine_similarity
import torch
# Download required NLTK data.
nltk.download('punkt')
# 1. VLMChunker: Using CLIP’s text encoder
class VLMChunker:
def __init__(self, similarity_threshold=0.75, device='cpu'):
self.similarity_threshold = similarity_threshold
self.device = device
# Load CLIP tokenizer and model from Hugging Face.
self.tokenizer = CLIPTokenizer.from_pretrained("openai/clip-vit-base-patch32")
self.model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32").to(self.device)
def _encode_sentences(self, sentences):
inputs = self.tokenizer(sentences, padding=True, truncation=True, return_tensors="pt").to(self.device)
with torch.no_grad():
outputs = self.model.get_text_features(**inputs)
# Normalize the embeddings for cosine similarity.
embeddings = outputs / outputs.norm(dim=-1, keepdim=True)
return embeddings.cpu().numpy()
def chunk(self, text):
sentences = sent_tokenize(text)
if not sentences:
return []
# Get embeddings for all sentences.
embeddings = self._encode_sentences(sentences)
chunks = []
current_chunk = [sentences[0]]
# Compare each sentence with its previous one.
for i in range(1, len(sentences)):
sim = cosine_similarity(embeddings[i-1].reshape(1, -1), embeddings[i].reshape(1, -1))[0][0]
# If similarity is above threshold, merge sentences.
if sim >= self.similarity_threshold:
current_chunk.append(sentences[i])
else:
chunks.append(" ".join(current_chunk))
current_chunk = [sentences[i]]
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
#Colipali Chunker
class ColipaliChunker:
def __init__(self, max_tokens=100, overlap_sentences=1, tokenizer_name="gpt2"):
self.max_tokens = max_tokens
self.overlap_sentences = overlap_sentences
self.tokenizer = AutoTokenizer.from_pretrained(tokenizer_name)
def _count_tokens(self, text):
tokens = self.tokenizer.encode(text)
return len(tokens)
def chunk(self, text):
sentences = sent_tokenize(text)
chunks = []
current_chunk = []
current_token_count = 0
for sentence in sentences:
sentence_token_count = self._count_tokens(sentence)
# If adding the sentence stays within the token budget, do so.
if current_token_count + sentence_token_count <= self.max_tokens:
current_chunk.append(sentence)
current_token_count += sentence_token_count
else:
# Append the current chunk as one unit.
if current_chunk:
chunks.append(" ".join(current_chunk))
# Overlap: take the last few sentences from the current chunk.
if self.overlap_sentences > 0:
current_chunk = current_chunk[-self.overlap_sentences:]
current_token_count = sum(self._count_tokens(s) for s in current_chunk)
else:
current_chunk = []
current_token_count = 0
# Add the current sentence after resetting (or overlapping).
current_chunk.append(sentence)
current_token_count += sentence_token_count
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
if __name__ == "__main__":
sample_text = (
"Artificial intelligence is transforming industries. "
"Deep learning has enabled breakthroughs in image and language processing. "
"Vision-language models are now widely used for tasks that require understanding both modalities. "
"These models combine visual and textual inputs to generate meaningful representations. "
"In parallel, advanced chunking techniques ensure that large texts are split without losing context. "
"Efficient chunking allows LLMs to process long documents by focusing on semantically coherent segments. "
"This method improves both the speed and quality of downstream tasks."
)
vlm_chunker = VLMChunker(similarity_threshold=0.75, device='cpu')
vlm_chunks = vlm_chunker.chunk(sample_text)
for i, chunk in enumerate(vlm_chunks, start=1):
print(f"\nChunk {i}:\n{chunk}\n{'-'*40}")
# Using ColipaliChunker
print("\nColipaliChunker Output ")
colipali_chunker = ColipaliChunker(max_tokens=50, overlap_sentences=1, tokenizer_name="gpt2")
colipali_chunks = colipali_chunker.chunk(sample_text)
for i, chunk in enumerate(colipali_chunks, start=1):
print(f"\nChunk {i}:\n{chunk}\n")
After chunking, the next approach is the similarity search. Which simply converts our chunks to tokens (numbers) with proper indexing. So when we input our query( questions), they are then indexed and compared with the indexed-chunks.
While there is so much hype on FAISS (due to it's low inference) and Weaviate, I have gotten better results with Qdrant.
#Qdrant
def build_qdrant_index(chunks):
embeddings = embedding_model.encode(chunks)
# Normalize embeddings for cosine similarity
embeddings /= np.linalg.norm(embeddings, axis=1, keepdims=True)
points = [
PointStruct(id=str(uuid.uuid4()), vector=emb.tolist(), payload={"text": chunk})
for chunk, emb in zip(chunks, embeddings)
]
client.upsert(collection_name=collection_name, points=points)
print(f"{len(points)} chunks indexed in Qdrant!")
def search_qdrant(query, top_k=3):
query_vector = embedding_model.encode([query])[0]
query_vector /= np.linalg.norm(query_vector) # Normalize
results = client.search(collection_name=collection_name, query_vector=query_vector, limit=top_k)
print("\n Search Results:")
for result in results:
print(f"{result.payload['text']} (Score: {result.score:.4f})")
And here is the FAISS model that I used.
def build_faiss_index(chunks, embedding_model):
embeddings = embedding_model.encode(chunks)
# Normalize embeddings for cosine similarity
embeddings /= np.linalg.norm(embeddings, axis=1, keepdims=True)
dimension = embeddings.shape[1]
index = faiss.IndexFlatIP(dimension) # Cosine similarity (when normalized)
index.add(embeddings) # No need to convert to np.array explicitly
return index
def search_faiss(index, chunks, query, top_k, embedding_model):
query_embedding = embedding_model.encode([query])
distances, indices = index.search(np.array(query_embedding), top_k)
print(f"FAISS Output: indices={indices}, distances={distances}")
# Ensure indices are correct
if np.isscalar(indices[0]):
indices = [[indices[0]]]
# Debug: Print retrieved chunks
retrieved_chunks = [chunks[i] for i in indices[0] if i < len(chunks)]
print(f"Retrieved Chunks: {retrieved_chunks}")
return retrieved_chunks
Now to put everything together.
Let's use pytesseract to process our pdf. This ensures that it handles both texts and images in our pdf.
client = QdrantClient(":memory:") # Use in-memory storage (for testing)
# client = QdrantClient("http://localhost:6333") # Use local Qdrant server
# client = QdrantClient("https://your-qdrant-server.com", api_key="your-api-key") # Cloud Qdrant
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
collection_name = "my_embeddings"
client.recreate_collection(
collection_name=collection_name,
vectors_config=VectorParams(size=embedding_model.get_sentence_embedding_dimension(), distance=Distance.COSINE)
)
# Convert PDF to images
images = convert_from_path(pdf_path) #This is one of the functions imported.
# Extract text from each image
text = " ".join([pytesseract.image_to_string(img) for img in images])
chunks = paragraph_chunking(text, max_tokens=1024, overlap_sentences=1, tokenizer=tokenizer)
build_qdrant_index(chunks)
search_qdrant("Location of the ministry of health and prevention, What is the address ")
