INTRODUCTION
Retrieval-Augmented Generation, commonly abbreviated as RAG, represents a paradigm shift in how we build intelligent question-answering systems. Instead of relying solely on the knowledge encoded in a language model's parameters during training, RAG systems dynamically retrieve relevant information from external knowledge bases and use that context to generate accurate, grounded responses. This approach addresses one of the fundamental limitations of large language models: their knowledge cutoff date and inability to access proprietary or domain-specific information.
Imagine you have thousands of internal company documents, technical manuals, research papers, or customer support tickets. A traditional language model cannot answer questions about this specific content unless you fine-tune it, which is expensive and time-consuming. A RAG system solves this elegantly by breaking the problem into two stages. First, it searches through your document collection to find relevant passages. Second, it feeds these passages to the language model as context, enabling it to generate informed answers based on your actual data.
LangChain is a framework specifically designed to simplify the development of applications powered by language models. It provides abstractions and utilities for common tasks like document loading, text splitting, embeddings generation, vector storage, and orchestrating chains of operations. Rather than writing hundreds of lines of boilerplate code to handle different document formats or integrate with various vector databases, LangChain offers a consistent interface that dramatically accelerates development.
This article will guide you through building a production-ready RAG system that can ingest documents in multiple formats including PDF, Word documents, HTML pages, and Markdown files. We will explore each component in depth, understand the rationale behind design decisions, and examine how to support different hardware configurations including NVIDIA CUDA, AMD ROCm, Intel GPUs, and Apple Metal Performance Shaders. By the end, you will have both a thorough conceptual understanding and a complete working implementation.
UNDERSTANDING THE RAG ARCHITECTURE
Before diving into implementation details, we need to understand the fundamental architecture of a RAG system. The system consists of two primary phases: the indexing phase and the query phase.
During the indexing phase, we process our document collection. We load documents from various sources and formats, split them into manageable chunks, convert these chunks into numerical vector representations called embeddings, and store these embeddings in a vector database. This preprocessing happens once for each document or whenever documents are updated.
The query phase occurs when a user asks a question. We convert the user's question into an embedding using the same embedding model used during indexing. We then search the vector database for document chunks whose embeddings are most similar to the question embedding. These relevant chunks are retrieved and combined with the original question to form a prompt for the language model. The language model generates a response based on both the question and the retrieved context.
The power of this architecture lies in its ability to ground the language model's responses in actual source material. Instead of hallucinating or relying on potentially outdated training data, the model works with fresh, relevant information retrieved specifically for each query.
SETTING UP THE DEVELOPMENT ENVIRONMENT
A production-ready RAG system requires careful attention to dependencies and hardware compatibility. We need to support multiple GPU architectures because different organizations use different hardware. Some run on cloud providers with NVIDIA GPUs, others use AMD hardware, some deploy on Apple Silicon devices, and increasingly, Intel's discrete GPUs are entering the market.
The core dependencies include LangChain itself, which provides the orchestration framework. We need document loaders for different file formats, an embedding model to convert text to vectors, a vector database to store and search embeddings efficiently, and a language model for generation. Additionally, we require format-specific libraries like PyPDF for PDF files, python-docx for Word documents, and BeautifulSoup for HTML parsing.
For GPU support, the approach differs by manufacturer. NVIDIA GPUs work with CUDA and the standard PyTorch builds. AMD GPUs require PyTorch built with ROCm support. Apple Silicon devices use Metal Performance Shaders through PyTorch's MPS backend. Intel GPUs can leverage Intel Extension for PyTorch. Our system must detect the available hardware and configure itself accordingly.
Here is how we structure the initial setup and hardware detection:
import torch
import platform
import subprocess
import sys
from typing import Optional, Literal
class HardwareDetector:
"""
Detects available GPU hardware and configures the appropriate
compute backend for optimal performance across different architectures.
"""
def __init__(self):
self.device_type: Optional[Literal['cuda', 'rocm', 'mps', 'xpu', 'cpu']] = None
self.device = None
self.device_name = None
self._detect_hardware()
def _detect_hardware(self):
"""
Performs comprehensive hardware detection across all supported
GPU architectures and falls back to CPU if no GPU is available.
"""
# Check for NVIDIA CUDA support
if torch.cuda.is_available():
self.device_type = 'cuda'
self.device = torch.device('cuda')
self.device_name = torch.cuda.get_device_name(0)
print(f"Detected NVIDIA GPU: {self.device_name}")
return
# Check for Apple Metal Performance Shaders
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
self.device_type = 'mps'
self.device = torch.device('mps')
self.device_name = "Apple Metal Performance Shaders"
print(f"Detected Apple Silicon with MPS support")
return
# Check for Intel XPU support
try:
import intel_extension_for_pytorch as ipex
if torch.xpu.is_available():
self.device_type = 'xpu'
self.device = torch.device('xpu')
self.device_name = torch.xpu.get_device_name(0)
print(f"Detected Intel GPU: {self.device_name}")
return
except ImportError:
pass
# Check for AMD ROCm support
# ROCm uses the same torch.cuda namespace but requires special build
if torch.version.hip is not None:
self.device_type = 'rocm'
self.device = torch.device('cuda') # ROCm uses cuda namespace
self.device_name = "AMD GPU with ROCm"
print(f"Detected AMD GPU with ROCm support")
return
# Fallback to CPU
self.device_type = 'cpu'
self.device = torch.device('cpu')
self.device_name = "CPU"
print("No GPU detected, using CPU")
def get_device(self) -> torch.device:
"""Returns the configured PyTorch device."""
return self.device
def get_device_type(self) -> str:
"""Returns a string identifier for the device type."""
return self.device_type
def optimize_model(self, model):
"""
Applies device-specific optimizations to a model.
Different hardware benefits from different optimization strategies.
"""
if self.device_type == 'xpu':
import intel_extension_for_pytorch as ipex
model = ipex.optimize(model)
return model.to(self.device)
The hardware detection logic systematically checks for each GPU type in a specific order. We first check for NVIDIA CUDA because it is the most common in production environments. The check uses PyTorch's built-in cuda.is_available function which queries the CUDA runtime. If CUDA is available, we retrieve the GPU name for logging purposes.
Next, we check for Apple's Metal Performance Shaders. This is relevant for developers working on MacBooks with M1, M2, or M3 chips. The MPS backend provides GPU acceleration on Apple Silicon. We verify both that the backend exists in the PyTorch build and that it reports availability.
Intel GPU support requires the Intel Extension for PyTorch, which is an optional dependency. We wrap the import in a try-except block to gracefully handle its absence. If the extension is available and detects an Intel GPU, we configure the XPU device.
AMD ROCm support is more nuanced. PyTorch built for ROCm reuses the CUDA namespace for compatibility, but we can detect ROCm by checking if torch.version.hip is not None. The HIP version string indicates a ROCm build.
Finally, if no GPU is detected, we fall back to CPU execution. While slower, CPU execution ensures the system works everywhere, which is crucial for development and testing.
DOCUMENT LOADING AND FORMAT HANDLING
A robust RAG system must handle diverse document formats because organizational knowledge exists in many forms. Technical documentation might be in Markdown, legal documents in PDF, reports in Word format, and web content in HTML. Each format requires specialized parsing logic to extract clean text while preserving semantic structure.
The document loading phase is critical because poor quality input leads to poor quality retrieval. If we extract text with formatting artifacts, broken sentences, or missing content, the embeddings will be less meaningful and retrieval accuracy will suffer.
LangChain provides document loaders for many formats, but we need to understand what happens under the hood. A document loader's job is to read a file and produce Document objects. Each Document contains the extracted text content and metadata like source filename, page numbers, or section headers. This metadata proves valuable later for citation and provenance tracking.
For PDF files, we use PyPDF2 or pdfplumber. PDFs are complex because they encode visual layout rather than semantic structure. Text might be scattered across the page in rendering order rather than reading order. Tables and multi-column layouts pose additional challenges. We need robust extraction that handles these cases.
Word documents use the python-docx library. The DOCX format is actually a ZIP archive containing XML files. The library parses the document.xml file to extract paragraphs, tables, headers, and footers. This structured approach generally yields cleaner text than PDF extraction.
HTML parsing uses BeautifulSoup. Web pages contain navigation menus, advertisements, and other content we typically want to exclude. We need to extract the main content while filtering noise. This often involves identifying the primary content container or stripping specific tags.
Markdown is the simplest case. Since Markdown is plain text with lightweight formatting, we can read it directly. However, we might want to parse the structure to preserve heading hierarchy in metadata.
Here is the document loader implementation:
from langchain.docstore.document import Document
from typing import List
import os
from pathlib import Path
class MultiFormatDocumentLoader:
"""
Loads documents from multiple file formats and converts them
into a unified Document representation suitable for RAG processing.
"""
def __init__(self):
self.supported_extensions = {
'.pdf': self._load_pdf,
'.docx': self._load_docx,
'.doc': self._load_doc,
'.html': self._load_html,
'.htm': self._load_html,
'.md': self._load_markdown,
'.txt': self._load_text
}
def load_documents(self, file_path: str) -> List[Document]:
"""
Loads a document from the specified file path, automatically
detecting the format and applying the appropriate parser.
"""
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"File not found: {file_path}")
extension = path.suffix.lower()
if extension not in self.supported_extensions:
raise ValueError(
f"Unsupported file format: {extension}. "
f"Supported formats: {list(self.supported_extensions.keys())}"
)
loader_func = self.supported_extensions[extension]
return loader_func(file_path)
def _load_pdf(self, file_path: str) -> List[Document]:
"""
Extracts text from PDF files page by page.
Each page becomes a separate document to preserve granularity.
"""
try:
import pdfplumber
except ImportError:
raise ImportError(
"pdfplumber is required for PDF support. "
"Install it with: pip install pdfplumber"
)
documents = []
with pdfplumber.open(file_path) as pdf:
for page_num, page in enumerate(pdf.pages, start=1):
text = page.extract_text()
if text and text.strip():
# Clean up common PDF extraction artifacts
text = self._clean_pdf_text(text)
doc = Document(
page_content=text,
metadata={
'source': file_path,
'page': page_num,
'total_pages': len(pdf.pages),
'format': 'pdf'
}
)
documents.append(doc)
return documents
def _clean_pdf_text(self, text: str) -> str:
"""
Removes common artifacts from PDF text extraction such as
excessive whitespace and broken hyphenation.
"""
# Remove excessive whitespace while preserving paragraph breaks
lines = text.split('\n')
cleaned_lines = []
for line in lines:
line = line.strip()
if line:
# Fix broken hyphenation at line ends
if cleaned_lines and cleaned_lines[-1].endswith('-'):
cleaned_lines[-1] = cleaned_lines[-1][:-1] + line
else:
cleaned_lines.append(line)
return '\n'.join(cleaned_lines)
def _load_docx(self, file_path: str) -> List[Document]:
"""
Extracts text from Word DOCX files, preserving paragraph structure.
"""
try:
from docx import Document as DocxDocument
except ImportError:
raise ImportError(
"python-docx is required for DOCX support. "
"Install it with: pip install python-docx"
)
docx = DocxDocument(file_path)
# Extract all paragraphs
paragraphs = [para.text for para in docx.paragraphs if para.text.strip()]
# Combine into a single document
# For very large documents, you might want to split by sections
full_text = '\n\n'.join(paragraphs)
doc = Document(
page_content=full_text,
metadata={
'source': file_path,
'format': 'docx',
'paragraphs': len(paragraphs)
}
)
return [doc]
def _load_doc(self, file_path: str) -> List[Document]:
"""
Handles legacy DOC format by converting to DOCX using antiword
or by raising an informative error.
"""
raise NotImplementedError(
"Legacy DOC format requires conversion. "
"Please convert to DOCX format or use a conversion tool like antiword."
)
def _load_html(self, file_path: str) -> List[Document]:
"""
Extracts main content from HTML files, filtering out navigation,
scripts, and other non-content elements.
"""
try:
from bs4 import BeautifulSoup
except ImportError:
raise ImportError(
"beautifulsoup4 is required for HTML support. "
"Install it with: pip install beautifulsoup4"
)
with open(file_path, 'r', encoding='utf-8') as f:
html_content = f.read()
soup = BeautifulSoup(html_content, 'html.parser')
# Remove script and style elements
for element in soup(['script', 'style', 'nav', 'footer', 'header']):
element.decompose()
# Extract text from main content areas
# Try to find main content container
main_content = soup.find('main') or soup.find('article') or soup.find('body')
if main_content:
text = main_content.get_text(separator='\n', strip=True)
else:
text = soup.get_text(separator='\n', strip=True)
# Clean up excessive whitespace
lines = [line.strip() for line in text.split('\n') if line.strip()]
text = '\n'.join(lines)
doc = Document(
page_content=text,
metadata={
'source': file_path,
'format': 'html',
'title': soup.title.string if soup.title else None
}
)
return [doc]
def _load_markdown(self, file_path: str) -> List[Document]:
"""
Loads Markdown files and optionally parses structure for metadata.
"""
with open(file_path, 'r', encoding='utf-8') as f:
text = f.read()
doc = Document(
page_content=text,
metadata={
'source': file_path,
'format': 'markdown'
}
)
return [doc]
def _load_text(self, file_path: str) -> List[Document]:
"""
Loads plain text files.
"""
with open(file_path, 'r', encoding='utf-8') as f:
text = f.read()
doc = Document(
page_content=text,
metadata={
'source': file_path,
'format': 'text'
}
)
return [doc]
def load_directory(self, directory_path: str, recursive: bool = True) -> List[Document]:
"""
Loads all supported documents from a directory.
Optionally recurses into subdirectories.
"""
documents = []
path = Path(directory_path)
if not path.is_dir():
raise ValueError(f"Not a directory: {directory_path}")
# Determine file iteration method based on recursive flag
file_iterator = path.rglob('*') if recursive else path.glob('*')
for file_path in file_iterator:
if file_path.is_file() and file_path.suffix.lower() in self.supported_extensions:
try:
docs = self.load_documents(str(file_path))
documents.extend(docs)
print(f"Loaded {len(docs)} document(s) from {file_path.name}")
except Exception as e:
print(f"Error loading {file_path.name}: {str(e)}")
return documents
The MultiFormatDocumentLoader class provides a unified interface for loading different file types. The design uses a dictionary mapping file extensions to loader methods, making it easy to add support for new formats. Each loader method returns a list of Document objects because some formats like PDFs naturally split into multiple documents per page.
The PDF loader uses pdfplumber because it handles complex layouts better than alternatives. We extract text page by page, creating separate documents for each page. This granularity helps during retrieval because we can pinpoint specific pages. The cleaning function addresses common PDF artifacts like broken hyphenation where words split across lines with hyphens.
The DOCX loader extracts all paragraphs and combines them. For very large documents, you might want to split by sections or chapters, but for most use cases, treating the entire document as one unit works well. The paragraph structure is preserved through double newlines.
The HTML loader demonstrates content extraction challenges. We use BeautifulSoup to parse the HTML tree and remove non-content elements like scripts, styles, and navigation. We attempt to find the main content container using common HTML5 semantic tags. This heuristic approach works for most well-structured HTML but might need customization for specific websites.
The directory loading method provides batch processing capability. It walks through a directory tree, identifies supported files by extension, and loads each one. Error handling ensures that one problematic file does not stop the entire process.
TEXT CHUNKING STRATEGIES
After loading documents, we face a critical decision: how to split the text into chunks. This step profoundly impacts retrieval quality and system performance. Chunks that are too small lack context and may not contain enough information to answer questions. Chunks that are too large dilute relevance signals and may exceed the language model's context window.
The chunking strategy must balance several competing concerns. We want chunks large enough to be semantically meaningful but small enough to be specific. We want to avoid splitting in the middle of sentences or paragraphs. We want some overlap between chunks so that information near boundaries is not lost.
A common approach uses recursive character splitting. We define a target chunk size and a set of separators in priority order. The splitter first tries to split on double newlines to preserve paragraph boundaries. If the resulting chunks are still too large, it tries single newlines, then sentences, and finally individual characters as a last resort.
The overlap parameter creates redundancy at chunk boundaries. If we use a chunk size of one thousand characters with a two hundred character overlap, each chunk shares its last two hundred characters with the next chunk's first two hundred characters. This overlap ensures that concepts spanning chunk boundaries appear in multiple chunks, improving retrieval recall.
Different document types may benefit from different chunking strategies. Code should split on function boundaries. Legal documents might split on section numbers. Academic papers could split on subsections. Our implementation provides a flexible framework that can accommodate these variations.
Here is the text chunking implementation:
from typing import List, Optional
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.docstore.document import Document
class DocumentChunker:
"""
Splits documents into optimally-sized chunks for embedding and retrieval.
Uses recursive splitting to preserve semantic boundaries.
"""
def __init__(
self,
chunk_size: int = 1000,
chunk_overlap: int = 200,
separators: Optional[List[str]] = None
):
"""
Initializes the chunker with specified parameters.
The chunk_size determines the target size in characters.
The chunk_overlap creates redundancy at boundaries to prevent
information loss. Separators define the split priority hierarchy.
"""
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
# Default separators in priority order
# Try to split on larger semantic units first
if separators is None:
self.separators = [
"\n\n", # Paragraph breaks
"\n", # Line breaks
". ", # Sentences
"! ", # Exclamations
"? ", # Questions
"; ", # Semicolons
", ", # Commas
" ", # Words
"" # Characters (last resort)
]
else:
self.separators = separators
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
separators=self.separators,
length_function=len
)
def chunk_documents(self, documents: List[Document]) -> List[Document]:
"""
Splits a list of documents into chunks while preserving metadata.
Each chunk inherits the metadata from its source document with
additional chunk-specific information.
"""
chunked_documents = []
for doc in documents:
# Split the document into chunks
chunks = self.splitter.split_text(doc.page_content)
# Create Document objects for each chunk
for i, chunk_text in enumerate(chunks):
# Copy original metadata and add chunk information
chunk_metadata = doc.metadata.copy()
chunk_metadata.update({
'chunk_index': i,
'total_chunks': len(chunks),
'chunk_size': len(chunk_text)
})
chunk_doc = Document(
page_content=chunk_text,
metadata=chunk_metadata
)
chunked_documents.append(chunk_doc)
return chunked_documents
def chunk_text(self, text: str, metadata: Optional[dict] = None) -> List[Document]:
"""
Chunks a raw text string into documents.
Useful for processing text that doesn't come from a file.
"""
chunks = self.splitter.split_text(text)
base_metadata = metadata or {}
documents = []
for i, chunk_text in enumerate(chunks):
chunk_metadata = base_metadata.copy()
chunk_metadata.update({
'chunk_index': i,
'total_chunks': len(chunks),
'chunk_size': len(chunk_text)
})
doc = Document(
page_content=chunk_text,
metadata=chunk_metadata
)
documents.append(doc)
return documents
def optimize_chunk_size_for_content(self, sample_documents: List[Document]) -> int:
"""
Analyzes sample documents to suggest an optimal chunk size.
This is useful when dealing with documents of varying lengths
and structures.
"""
if not sample_documents:
return self.chunk_size
# Calculate average paragraph length
paragraph_lengths = []
for doc in sample_documents:
paragraphs = doc.page_content.split('\n\n')
for para in paragraphs:
if para.strip():
paragraph_lengths.append(len(para))
if not paragraph_lengths:
return self.chunk_size
# Use median paragraph length as a guide
paragraph_lengths.sort()
median_length = paragraph_lengths[len(paragraph_lengths) // 2]
# Aim for chunks that contain 2-3 paragraphs
suggested_size = median_length * 2.5
# Clamp to reasonable bounds
suggested_size = max(500, min(2000, suggested_size))
return int(suggested_size)
The DocumentChunker class wraps LangChain's RecursiveCharacterTextSplitter with additional functionality. The recursive splitting algorithm works by trying each separator in order. It splits the text using the first separator, then recursively processes any resulting chunks that exceed the target size using the next separator in the list.
The separator list prioritizes semantic boundaries. We first try to split on paragraph breaks (double newlines), which typically represent complete thoughts or topic shifts. If paragraphs are too large, we split on single newlines, then sentences, and so on. This hierarchy ensures we preserve as much semantic coherence as possible.
The chunk_documents method processes a list of Document objects. For each document, it splits the text and creates new Document objects for each chunk. Critically, it preserves the original metadata and adds chunk-specific information. This metadata proves essential later for tracking which chunks came from which sources and for citation purposes.
The optimize_chunk_size_for_content method demonstrates adaptive chunking. By analyzing sample documents, we can estimate appropriate chunk sizes for a particular corpus. Documents with long paragraphs might benefit from larger chunks, while documents with short, dense paragraphs might work better with smaller chunks. This method calculates the median paragraph length and suggests a chunk size that captures roughly two to three paragraphs.
EMBEDDINGS AND VECTOR REPRESENTATIONS
Embeddings are the mathematical foundation of semantic search. An embedding is a dense vector representation of text that captures semantic meaning. Words or phrases with similar meanings have similar embeddings in vector space. This property enables us to find relevant documents by comparing vector similarity rather than exact keyword matching.
The embedding model is a neural network trained to map text to vectors. Modern embedding models use transformer architectures and are trained on massive text corpora using objectives like contrastive learning. The training process teaches the model to place semantically similar texts close together in vector space.
Choosing an embedding model involves several trade-offs. Larger models generally produce better quality embeddings but require more computational resources. The embedding dimension affects both quality and storage requirements. Common dimensions range from 384 to 1536. Higher dimensions can capture more nuance but increase vector database size and search time.
For a production system, we need to support both local and remote embedding models. Local models run on your infrastructure, giving you control and privacy but requiring GPU resources. Remote models use APIs from providers like OpenAI or Cohere, offering convenience and scalability but introducing latency and cost.
We will use Sentence Transformers for local embeddings because they offer excellent quality and support multiple languages. For remote embeddings, we will support OpenAI's embedding API as an example, but the pattern extends to other providers.
Here is the embedding implementation:
from typing import List, Optional, Union
import numpy as np
from langchain.embeddings.base import Embeddings
class MultiBackendEmbeddings(Embeddings):
"""
Provides a unified interface for generating embeddings using either
local models or remote API services. Automatically handles hardware
acceleration across different GPU architectures.
"""
def __init__(
self,
model_name: str = "all-MiniLM-L6-v2",
use_local: bool = True,
api_key: Optional[str] = None,
hardware_detector: Optional[HardwareDetector] = None
):
"""
Initializes the embedding backend.
For local models, model_name should be a Sentence Transformers model.
For remote models, model_name should be the API model identifier.
"""
self.model_name = model_name
self.use_local = use_local
self.api_key = api_key
self.hardware_detector = hardware_detector or HardwareDetector()
if self.use_local:
self._initialize_local_model()
else:
self._initialize_remote_client()
def _initialize_local_model(self):
"""
Loads a local Sentence Transformers model and configures it
for the available hardware.
"""
try:
from sentence_transformers import SentenceTransformer
except ImportError:
raise ImportError(
"sentence-transformers is required for local embeddings. "
"Install it with: pip install sentence-transformers"
)
# Load the model
self.model = SentenceTransformer(self.model_name)
# Move model to appropriate device
device = self.hardware_detector.get_device()
self.model = self.model.to(device)
# Apply hardware-specific optimizations
self.model = self.hardware_detector.optimize_model(self.model)
print(f"Loaded local embedding model '{self.model_name}' on {device}")
def _initialize_remote_client(self):
"""
Initializes the client for remote embedding API.
"""
if not self.api_key:
raise ValueError("API key required for remote embeddings")
try:
import openai
self.client = openai.OpenAI(api_key=self.api_key)
print(f"Initialized remote embedding client for model '{self.model_name}'")
except ImportError:
raise ImportError(
"openai is required for remote embeddings. "
"Install it with: pip install openai"
)
def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""
Generates embeddings for a list of documents.
Uses batch processing for efficiency.
"""
if self.use_local:
return self._embed_local(texts)
else:
return self._embed_remote(texts)
def embed_query(self, text: str) -> List[float]:
"""
Generates an embedding for a single query.
Some models use different processing for queries vs documents.
"""
if self.use_local:
# Sentence Transformers treats queries and documents the same
return self._embed_local([text])[0]
else:
return self._embed_remote([text])[0]
def _embed_local(self, texts: List[str]) -> List[List[float]]:
"""
Generates embeddings using the local model with batch processing.
"""
# Sentence Transformers handles batching internally
embeddings = self.model.encode(
texts,
convert_to_numpy=True,
show_progress_bar=len(texts) > 10,
batch_size=32
)
# Convert to list of lists for consistency
return embeddings.tolist()
def _embed_remote(self, texts: List[str]) -> List[List[float]]:
"""
Generates embeddings using a remote API with batching.
"""
embeddings = []
# Process in batches to respect API limits
batch_size = 100
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
response = self.client.embeddings.create(
model=self.model_name,
input=batch
)
batch_embeddings = [item.embedding for item in response.data]
embeddings.extend(batch_embeddings)
return embeddings
def get_embedding_dimension(self) -> int:
"""
Returns the dimensionality of the embedding vectors.
"""
if self.use_local:
# Get dimension from model
return self.model.get_sentence_embedding_dimension()
else:
# For OpenAI, dimension depends on model
dimension_map = {
'text-embedding-ada-002': 1536,
'text-embedding-3-small': 1536,
'text-embedding-3-large': 3072
}
return dimension_map.get(self.model_name, 1536)
The MultiBackendEmbeddings class provides a unified interface regardless of whether we use local or remote models. This abstraction allows the rest of the system to remain agnostic about the embedding source.
For local embeddings, we use Sentence Transformers, a library that provides pre-trained models optimized for semantic similarity tasks. The all-MiniLM-L6-v2 model is a good default choice because it balances quality and speed. It produces 384-dimensional embeddings and runs efficiently even on modest hardware.
The initialization process loads the model and moves it to the appropriate device using our hardware detector. This ensures the model runs on the GPU if available. The encode method handles batching internally, which is important for efficiency when embedding large document collections.
For remote embeddings, we use OpenAI's API as an example. The implementation batches requests to respect API rate limits. Most embedding APIs charge per token, so batching reduces the number of API calls and improves throughput.
The embed_documents and embed_query methods provide the interface expected by LangChain. Some embedding models differentiate between document embeddings and query embeddings, using different processing or prompts for each. Sentence Transformers treats them identically, but the interface supports models that do not.
VECTOR STORAGE AND SIMILARITY SEARCH
Once we have embeddings, we need a way to store them and perform efficient similarity searches. A vector database is a specialized database optimized for storing high-dimensional vectors and finding nearest neighbors. Traditional databases struggle with this task because vector similarity search requires computing distances between the query vector and potentially millions of stored vectors.
Vector databases use specialized indexing structures like HNSW (Hierarchical Navigable Small World) graphs or IVF (Inverted File) indexes. These structures enable approximate nearest neighbor search that is orders of magnitude faster than brute force comparison while maintaining high accuracy.
For production systems, popular vector databases include Pinecone, Weaviate, Qdrant, and Milvus. For development and smaller deployments, FAISS (Facebook AI Similarity Search) provides an excellent in-memory option. ChromaDB offers a lightweight persistent option that works well for moderate scale.
We will implement support for both FAISS and ChromaDB. FAISS excels at raw search speed and supports various index types. ChromaDB provides persistence and a simple API, making it ideal for applications that need to save and reload indexes.
The vector store needs to support adding documents, performing similarity searches, and retrieving metadata. The similarity search returns the k most similar documents to a query, where similarity is typically measured using cosine similarity or Euclidean distance.
Here is the vector store implementation:
from typing import List, Optional, Tuple, Union
from langchain.vectorstores import FAISS, Chroma
from langchain.docstore.document import Document
import os
class VectorStoreManager:
"""
Manages vector storage and retrieval using multiple backend options.
Provides a unified interface for different vector database implementations.
"""
def __init__(
self,
embeddings: MultiBackendEmbeddings,
backend: str = "faiss",
persist_directory: Optional[str] = None
):
"""
Initializes the vector store with the specified backend.
Backend options:
- 'faiss': Fast in-memory search, optional persistence
- 'chroma': Persistent storage with built-in metadata filtering
"""
self.embeddings = embeddings
self.backend = backend
self.persist_directory = persist_directory
self.vector_store = None
def create_from_documents(self, documents: List[Document]) -> None:
"""
Creates a new vector store from a list of documents.
This involves embedding all documents and building the index.
"""
if not documents:
raise ValueError("Cannot create vector store from empty document list")
print(f"Creating vector store with {len(documents)} documents...")
if self.backend == "faiss":
self._create_faiss_store(documents)
elif self.backend == "chroma":
self._create_chroma_store(documents)
else:
raise ValueError(f"Unsupported backend: {self.backend}")
print("Vector store created successfully")
def _create_faiss_store(self, documents: List[Document]) -> None:
"""
Creates a FAISS vector store.
FAISS provides extremely fast similarity search using optimized
indexing structures.
"""
self.vector_store = FAISS.from_documents(
documents=documents,
embedding=self.embeddings
)
def _create_chroma_store(self, documents: List[Document]) -> None:
"""
Creates a ChromaDB vector store with persistence.
Chroma automatically handles persistence if a directory is specified.
"""
if self.persist_directory:
os.makedirs(self.persist_directory, exist_ok=True)
self.vector_store = Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory=self.persist_directory
)
def add_documents(self, documents: List[Document]) -> None:
"""
Adds new documents to an existing vector store.
"""
if self.vector_store is None:
raise ValueError("Vector store not initialized. Call create_from_documents first.")
print(f"Adding {len(documents)} documents to vector store...")
self.vector_store.add_documents(documents)
# Persist if using Chroma
if self.backend == "chroma" and self.persist_directory:
self.vector_store.persist()
print("Documents added successfully")
def similarity_search(
self,
query: str,
k: int = 4,
filter_metadata: Optional[dict] = None
) -> List[Document]:
"""
Performs similarity search to find the k most relevant documents.
The filter_metadata parameter allows filtering results based on
document metadata, useful for restricting search to specific sources
or document types.
"""
if self.vector_store is None:
raise ValueError("Vector store not initialized")
if filter_metadata and self.backend == "chroma":
# Chroma supports native metadata filtering
results = self.vector_store.similarity_search(
query=query,
k=k,
filter=filter_metadata
)
else:
# FAISS doesn't support filtering, so we retrieve more results
# and filter in memory
retrieve_k = k * 3 if filter_metadata else k
results = self.vector_store.similarity_search(
query=query,
k=retrieve_k
)
if filter_metadata:
results = self._filter_documents(results, filter_metadata)
results = results[:k]
return results
def similarity_search_with_score(
self,
query: str,
k: int = 4
) -> List[Tuple[Document, float]]:
"""
Performs similarity search and returns documents with their
similarity scores. Lower scores indicate higher similarity.
"""
if self.vector_store is None:
raise ValueError("Vector store not initialized")
results = self.vector_store.similarity_search_with_score(
query=query,
k=k
)
return results
def _filter_documents(
self,
documents: List[Document],
filter_metadata: dict
) -> List[Document]:
"""
Filters documents based on metadata criteria.
This is a fallback for backends that don't support native filtering.
"""
filtered = []
for doc in documents:
match = True
for key, value in filter_metadata.items():
if key not in doc.metadata or doc.metadata[key] != value:
match = False
break
if match:
filtered.append(doc)
return filtered
def save(self, path: str) -> None:
"""
Saves the vector store to disk.
"""
if self.vector_store is None:
raise ValueError("No vector store to save")
if self.backend == "faiss":
self.vector_store.save_local(path)
print(f"FAISS index saved to {path}")
elif self.backend == "chroma":
# Chroma persists automatically if persist_directory was set
if self.persist_directory:
self.vector_store.persist()
print(f"Chroma database persisted to {self.persist_directory}")
else:
print("Warning: Chroma was initialized without persistence")
def load(self, path: str) -> None:
"""
Loads a vector store from disk.
"""
if self.backend == "faiss":
self.vector_store = FAISS.load_local(
path,
self.embeddings,
allow_dangerous_deserialization=True
)
print(f"FAISS index loaded from {path}")
elif self.backend == "chroma":
self.vector_store = Chroma(
persist_directory=path,
embedding_function=self.embeddings
)
print(f"Chroma database loaded from {path}")
def get_statistics(self) -> dict:
"""
Returns statistics about the vector store.
"""
if self.vector_store is None:
return {"status": "not_initialized"}
stats = {
"backend": self.backend,
"embedding_dimension": self.embeddings.get_embedding_dimension()
}
if self.backend == "faiss":
stats["num_vectors"] = self.vector_store.index.ntotal
elif self.backend == "chroma":
# Chroma doesn't expose count directly, but we can get it
try:
collection = self.vector_store._collection
stats["num_vectors"] = collection.count()
except:
stats["num_vectors"] = "unknown"
return stats
The VectorStoreManager class abstracts the differences between vector database backends. This design allows switching between FAISS and ChromaDB without changing the rest of the system.
The create_from_documents method is the primary entry point for building a new index. It takes a list of Document objects, generates embeddings for each, and builds the vector index. For FAISS, this creates an in-memory index using the IndexFlatL2 structure by default, which performs exact nearest neighbor search. For larger datasets, you might switch to an approximate index like IndexIVFFlat.
ChromaDB handles persistence automatically when you provide a persist_directory. It stores both the vectors and metadata in a local SQLite database. This makes it easy to stop and restart your application without rebuilding the index.
The similarity_search method is the core retrieval function. It takes a query string, generates an embedding for it, and searches the index for the k nearest neighbors. The results are Document objects containing both the text content and metadata.
Metadata filtering allows restricting search to specific subsets of documents. For example, you might want to search only within PDF documents or only within documents from a specific time period. ChromaDB supports this natively through its filter parameter. FAISS does not, so we implement a fallback that retrieves extra results and filters them in memory.
The similarity_search_with_score variant returns both documents and their distance scores. This is useful for implementing relevance thresholds or debugging retrieval quality. Lower scores indicate higher similarity.
LANGUAGE MODEL INTEGRATION
The language model is the component that generates final answers based on retrieved context. We need to support both local models that run on your infrastructure and remote models accessed through APIs. Local models provide privacy and control but require significant computational resources. Remote models offer convenience and access to the latest capabilities but introduce latency and ongoing costs.
For local models, we will use the Hugging Face Transformers library, which provides access to thousands of open-source models. Popular choices include Llama 2, Mistral, and Phi models. These models require GPU acceleration for reasonable performance, which is where our hardware detection becomes critical.
For remote models, we will support OpenAI's API as the primary example, but the pattern extends to Anthropic, Cohere, and other providers. Remote models typically offer better quality than local models of similar size because providers can deploy much larger models than most organizations can run locally.
The language model integration must handle prompt construction, response generation, and error handling. We need to format the retrieved context and user question into a prompt that elicits accurate, grounded responses.
Here is the language model implementation:
from typing import Optional, List, Dict, Any
from langchain.llms.base import LLM
from langchain.callbacks.manager import CallbackManagerForLLMRun
class MultiBackendLLM:
"""
Provides a unified interface for language models running locally
or accessed through remote APIs. Handles hardware acceleration
and prompt formatting.
"""
def __init__(
self,
model_name: str = "gpt-3.5-turbo",
use_local: bool = False,
api_key: Optional[str] = None,
hardware_detector: Optional[HardwareDetector] = None,
temperature: float = 0.7,
max_tokens: int = 512
):
"""
Initializes the language model backend.
For local models, model_name should be a Hugging Face model ID.
For remote models, model_name should be the API model identifier.
"""
self.model_name = model_name
self.use_local = use_local
self.api_key = api_key
self.hardware_detector = hardware_detector or HardwareDetector()
self.temperature = temperature
self.max_tokens = max_tokens
if self.use_local:
self._initialize_local_model()
else:
self._initialize_remote_client()
def _initialize_local_model(self):
"""
Loads a local language model using Hugging Face Transformers.
Configures the model for the available hardware.
"""
try:
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
except ImportError:
raise ImportError(
"transformers is required for local models. "
"Install it with: pip install transformers torch"
)
print(f"Loading local model '{self.model_name}'...")
# Load tokenizer
self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
# Ensure tokenizer has a pad token
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
# Load model with appropriate dtype based on hardware
device = self.hardware_detector.get_device()
# Use float16 for GPU, float32 for CPU
dtype = torch.float16 if device.type != 'cpu' else torch.float32
self.model = AutoModelForCausalLM.from_pretrained(
self.model_name,
torch_dtype=dtype,
device_map="auto" if device.type != 'cpu' else None,
low_cpu_mem_usage=True
)
if device.type == 'cpu':
self.model = self.model.to(device)
# Apply hardware-specific optimizations
self.model = self.hardware_detector.optimize_model(self.model)
self.model.eval()
print(f"Model loaded on {device}")
def _initialize_remote_client(self):
"""
Initializes the client for remote language model API.
"""
if not self.api_key:
raise ValueError("API key required for remote models")
try:
import openai
self.client = openai.OpenAI(api_key=self.api_key)
print(f"Initialized remote model client for '{self.model_name}'")
except ImportError:
raise ImportError(
"openai is required for remote models. "
"Install it with: pip install openai"
)
def generate(
self,
prompt: str,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None
) -> str:
"""
Generates a response to the given prompt.
"""
temp = temperature if temperature is not None else self.temperature
max_tok = max_tokens if max_tokens is not None else self.max_tokens
if self.use_local:
return self._generate_local(prompt, temp, max_tok)
else:
return self._generate_remote(prompt, temp, max_tok)
def _generate_local(
self,
prompt: str,
temperature: float,
max_tokens: int
) -> str:
"""
Generates a response using the local model.
"""
import torch
# Tokenize input
inputs = self.tokenizer(
prompt,
return_tensors="pt",
padding=True,
truncation=True,
max_length=2048
)
# Move inputs to device
device = self.hardware_detector.get_device()
inputs = {k: v.to(device) for k, v in inputs.items()}
# Generate response
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
do_sample=temperature > 0,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id
)
# Decode response
response = self.tokenizer.decode(
outputs[0][inputs['input_ids'].shape[1]:],
skip_special_tokens=True
)
return response.strip()
def _generate_remote(
self,
prompt: str,
temperature: float,
max_tokens: int
) -> str:
"""
Generates a response using a remote API.
"""
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "user", "content": prompt}
],
temperature=temperature,
max_tokens=max_tokens
)
return response.choices[0].message.content.strip()
def generate_with_context(
self,
question: str,
context_documents: List[Document],
system_prompt: Optional[str] = None
) -> str:
"""
Generates a response using retrieved context documents.
This is the primary method for RAG applications.
"""
# Build context from documents
context_parts = []
for i, doc in enumerate(context_documents, 1):
source = doc.metadata.get('source', 'Unknown')
page = doc.metadata.get('page', '')
page_info = f" (Page {page})" if page else ""
context_parts.append(
f"Document {i} [{source}{page_info}]:\n{doc.page_content}"
)
context = "\n\n".join(context_parts)
# Build the prompt
if system_prompt is None:
system_prompt = (
"You are a helpful assistant that answers questions based on "
"the provided context. If the answer cannot be found in the "
"context, say so clearly. Always cite which document you are "
"using for your answer."
)
prompt = f"""{system_prompt}
Context:
{context}
Question: {question}
Answer:"""
return self.generate(prompt)
The MultiBackendLLM class provides a consistent interface for both local and remote models. The initialization process differs significantly between the two.
For local models, we use Hugging Face Transformers to load the model and tokenizer. The device_map="auto" parameter automatically distributes the model across available GPUs if it is too large to fit on one. We use float16 precision on GPUs to reduce memory usage and increase speed. The model is set to evaluation mode to disable dropout and other training-specific behaviors.
For remote models, we initialize the OpenAI client with the API key. The implementation is straightforward because the API handles all the complexity of model serving.
The generate method is the basic interface for text generation. It takes a prompt and returns the model's response. The local implementation tokenizes the input, moves it to the appropriate device, runs generation, and decodes the output. We slice the output to exclude the input tokens, returning only the newly generated text.
The generate_with_context method is specifically designed for RAG. It takes a question and a list of retrieved documents, constructs a prompt that includes the context, and generates an answer. The prompt engineering is crucial here. We format each document with its source information, provide a system prompt that instructs the model to use the context and cite sources, and structure the prompt to clearly separate context from question.
BUILDING THE COMPLETE RAG PIPELINE
Now we integrate all the components into a cohesive RAG system. The pipeline orchestrates document loading, chunking, embedding, indexing, retrieval, and generation. It provides a high-level interface that hides the complexity of the individual components.
The RAG system needs to support two primary workflows. The indexing workflow processes documents and builds the vector store. This typically runs offline or as a batch job. The query workflow handles user questions in real-time, retrieving relevant context and generating answers.
We also need to consider error handling, logging, and monitoring. Production systems must gracefully handle failures at any stage, provide visibility into what is happening, and collect metrics for performance optimization.
Here is the complete RAG pipeline implementation:
from typing import List, Optional, Dict, Any
import time
from pathlib import Path
class RAGSystem:
"""
Complete Retrieval-Augmented Generation system that integrates
document loading, embedding, vector storage, and language model
generation into a unified pipeline.
"""
def __init__(
self,
embedding_model: str = "all-MiniLM-L6-v2",
llm_model: str = "gpt-3.5-turbo",
use_local_embeddings: bool = True,
use_local_llm: bool = False,
vector_store_backend: str = "faiss",
persist_directory: Optional[str] = None,
api_key: Optional[str] = None,
chunk_size: int = 1000,
chunk_overlap: int = 200
):
"""
Initializes the RAG system with specified configuration.
"""
print("Initializing RAG system...")
# Initialize hardware detector
self.hardware_detector = HardwareDetector()
# Initialize document loader
self.document_loader = MultiFormatDocumentLoader()
# Initialize chunker
self.chunker = DocumentChunker(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap
)
# Initialize embeddings
self.embeddings = MultiBackendEmbeddings(
model_name=embedding_model,
use_local=use_local_embeddings,
api_key=api_key,
hardware_detector=self.hardware_detector
)
# Initialize vector store manager
self.vector_store_manager = VectorStoreManager(
embeddings=self.embeddings,
backend=vector_store_backend,
persist_directory=persist_directory
)
# Initialize language model
self.llm = MultiBackendLLM(
model_name=llm_model,
use_local=use_local_llm,
api_key=api_key,
hardware_detector=self.hardware_detector
)
self.indexed = False
print("RAG system initialized successfully")
def index_documents(
self,
file_paths: Optional[List[str]] = None,
directory_path: Optional[str] = None,
recursive: bool = True
) -> Dict[str, Any]:
"""
Indexes documents from files or a directory.
This is the offline processing phase that builds the vector store.
"""
start_time = time.time()
# Load documents
print("Loading documents...")
documents = []
if file_paths:
for file_path in file_paths:
try:
docs = self.document_loader.load_documents(file_path)
documents.extend(docs)
except Exception as e:
print(f"Error loading {file_path}: {str(e)}")
if directory_path:
try:
docs = self.document_loader.load_directory(
directory_path,
recursive=recursive
)
documents.extend(docs)
except Exception as e:
print(f"Error loading directory {directory_path}: {str(e)}")
if not documents:
raise ValueError("No documents loaded")
print(f"Loaded {len(documents)} document(s)")
# Chunk documents
print("Chunking documents...")
chunked_documents = self.chunker.chunk_documents(documents)
print(f"Created {len(chunked_documents)} chunk(s)")
# Create vector store
self.vector_store_manager.create_from_documents(chunked_documents)
self.indexed = True
elapsed_time = time.time() - start_time
stats = {
"num_documents": len(documents),
"num_chunks": len(chunked_documents),
"indexing_time_seconds": elapsed_time,
"vector_store_stats": self.vector_store_manager.get_statistics()
}
print(f"Indexing completed in {elapsed_time:.2f} seconds")
return stats
def add_documents(
self,
file_paths: Optional[List[str]] = None,
directory_path: Optional[str] = None,
recursive: bool = True
) -> Dict[str, Any]:
"""
Adds new documents to an existing index.
"""
if not self.indexed:
raise ValueError(
"No existing index. Use index_documents for initial indexing."
)
start_time = time.time()
# Load documents
documents = []
if file_paths:
for file_path in file_paths:
try:
docs = self.document_loader.load_documents(file_path)
documents.extend(docs)
except Exception as e:
print(f"Error loading {file_path}: {str(e)}")
if directory_path:
try:
docs = self.document_loader.load_directory(
directory_path,
recursive=recursive
)
documents.extend(docs)
except Exception as e:
print(f"Error loading directory {directory_path}: {str(e)}")
if not documents:
print("No documents to add")
return {}
# Chunk documents
chunked_documents = self.chunker.chunk_documents(documents)
# Add to vector store
self.vector_store_manager.add_documents(chunked_documents)
elapsed_time = time.time() - start_time
stats = {
"num_documents_added": len(documents),
"num_chunks_added": len(chunked_documents),
"time_seconds": elapsed_time
}
return stats
def query(
self,
question: str,
num_results: int = 4,
return_sources: bool = True,
filter_metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Answers a question using the RAG pipeline.
This is the real-time query phase.
"""
if not self.indexed:
raise ValueError(
"No index available. Run index_documents first."
)
start_time = time.time()
# Retrieve relevant documents
print(f"Retrieving relevant documents for: {question}")
retrieved_docs = self.vector_store_manager.similarity_search(
query=question,
k=num_results,
filter_metadata=filter_metadata
)
if not retrieved_docs:
return {
"question": question,
"answer": "I could not find any relevant information to answer this question.",
"sources": [],
"retrieval_time_seconds": time.time() - start_time
}
retrieval_time = time.time() - start_time
# Generate answer
print("Generating answer...")
generation_start = time.time()
answer = self.llm.generate_with_context(
question=question,
context_documents=retrieved_docs
)
generation_time = time.time() - generation_start
# Prepare response
response = {
"question": question,
"answer": answer,
"retrieval_time_seconds": retrieval_time,
"generation_time_seconds": generation_time,
"total_time_seconds": time.time() - start_time
}
if return_sources:
sources = []
for doc in retrieved_docs:
source_info = {
"source": doc.metadata.get('source', 'Unknown'),
"page": doc.metadata.get('page'),
"chunk_index": doc.metadata.get('chunk_index'),
"content_preview": doc.page_content[:200] + "..."
}
sources.append(source_info)
response["sources"] = sources
return response
def query_with_scores(
self,
question: str,
num_results: int = 4
) -> Dict[str, Any]:
"""
Queries the system and returns results with similarity scores.
Useful for debugging and understanding retrieval quality.
"""
if not self.indexed:
raise ValueError("No index available. Run index_documents first.")
# Retrieve with scores
results = self.vector_store_manager.similarity_search_with_score(
query=question,
k=num_results
)
# Format results
formatted_results = []
for doc, score in results:
formatted_results.append({
"content": doc.page_content,
"score": float(score),
"metadata": doc.metadata
})
return {
"question": question,
"results": formatted_results
}
def save_index(self, path: str) -> None:
"""
Saves the vector store index to disk.
"""
if not self.indexed:
raise ValueError("No index to save")
self.vector_store_manager.save(path)
def load_index(self, path: str) -> None:
"""
Loads a previously saved vector store index.
"""
self.vector_store_manager.load(path)
self.indexed = True
def get_system_info(self) -> Dict[str, Any]:
"""
Returns information about the system configuration and status.
"""
info = {
"hardware": {
"device_type": self.hardware_detector.get_device_type(),
"device_name": self.hardware_detector.device_name
},
"embeddings": {
"model": self.embeddings.model_name,
"local": self.embeddings.use_local,
"dimension": self.embeddings.get_embedding_dimension()
},
"llm": {
"model": self.llm.model_name,
"local": self.llm.use_local
},
"vector_store": self.vector_store_manager.get_statistics(),
"indexed": self.indexed
}
return info
The RAGSystem class ties everything together into a cohesive interface. The initialization process creates instances of all the component classes, passing configuration parameters and sharing the hardware detector across components.
The index_documents method implements the offline indexing workflow. It loads documents from specified files or directories, chunks them, and builds the vector store. The method returns statistics about the indexing process, including the number of documents and chunks processed and the time taken. This information is valuable for monitoring and optimization.
The add_documents method allows incremental updates to the index. This is important for production systems where new documents arrive regularly. Instead of rebuilding the entire index, we can add new documents to the existing store.
The query method implements the real-time question-answering workflow. It retrieves relevant documents using similarity search, generates an answer using the language model, and returns a structured response. The response includes the answer, timing information, and optionally the source documents used. This transparency helps users understand where information came from and allows them to verify answers.
The query_with_scores variant returns similarity scores along with documents. This is useful for debugging retrieval quality. If scores are consistently high (indicating low similarity), it suggests the retrieval is not finding relevant content, which might indicate problems with chunking, embedding, or the query itself.
The save_index and load_index methods provide persistence. For production deployments, you typically build the index once and load it when the application starts, rather than rebuilding it every time.
The get_system_info method returns comprehensive information about the system configuration and status. This is useful for logging, monitoring, and debugging.
ADVANCED FEATURES AND OPTIMIZATIONS
A production-ready RAG system benefits from several advanced features beyond the basic pipeline. These include query rewriting, result reranking, hybrid search, and caching.
Query rewriting addresses the vocabulary mismatch problem. Users often phrase questions differently than how information appears in documents. A query rewriter can expand the query with synonyms, rephrase it for better retrieval, or break complex questions into sub-questions.
Reranking improves retrieval quality by applying a more sophisticated model to the initial retrieval results. The vector search provides fast approximate retrieval, then a cross-encoder model reranks the top results by computing relevance scores based on the interaction between query and document.
Hybrid search combines vector similarity with traditional keyword search. Some queries benefit from exact keyword matching, especially for technical terms, product names, or specific phrases. Combining both approaches often yields better results than either alone.
Caching reduces latency and costs for repeated queries. If multiple users ask the same or similar questions, we can serve cached answers instead of running the full pipeline.
Here is an implementation of these advanced features:
from typing import List, Dict, Any, Optional, Tuple
import hashlib
import json
from collections import OrderedDict
class AdvancedRAGSystem(RAGSystem):
"""
Extended RAG system with advanced features including query rewriting,
reranking, hybrid search, and caching.
"""
def __init__(
self,
enable_reranking: bool = False,
enable_caching: bool = True,
cache_size: int = 100,
**kwargs
):
"""
Initializes the advanced RAG system with additional features.
"""
super().__init__(**kwargs)
self.enable_reranking = enable_reranking
self.enable_caching = enable_caching
# Initialize cache as an ordered dict for LRU behavior
self.cache = OrderedDict()
self.cache_size = cache_size
self.cache_hits = 0
self.cache_misses = 0
if self.enable_reranking:
self._initialize_reranker()
def _initialize_reranker(self):
"""
Initializes a cross-encoder model for reranking.
Cross-encoders compute relevance by processing query and document
together, providing more accurate scores than vector similarity.
"""
try:
from sentence_transformers import CrossEncoder
except ImportError:
raise ImportError(
"sentence-transformers is required for reranking. "
"Install it with: pip install sentence-transformers"
)
print("Loading reranking model...")
self.reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
# Move to appropriate device
device = self.hardware_detector.get_device()
self.reranker.model = self.reranker.model.to(device)
print(f"Reranker loaded on {device}")
def _rerank_documents(
self,
query: str,
documents: List[Document],
top_k: int
) -> List[Document]:
"""
Reranks retrieved documents using a cross-encoder model.
"""
if not documents:
return documents
# Prepare query-document pairs
pairs = [[query, doc.page_content] for doc in documents]
# Compute relevance scores
scores = self.reranker.predict(pairs)
# Sort documents by score (higher is better for cross-encoders)
doc_score_pairs = list(zip(documents, scores))
doc_score_pairs.sort(key=lambda x: x[1], reverse=True)
# Return top k documents
reranked_docs = [doc for doc, score in doc_score_pairs[:top_k]]
return reranked_docs
def _get_cache_key(self, question: str, num_results: int) -> str:
"""
Generates a cache key for a query.
"""
key_data = f"{question}|{num_results}"
return hashlib.md5(key_data.encode()).hexdigest()
def _get_from_cache(self, cache_key: str) -> Optional[Dict[str, Any]]:
"""
Retrieves a cached response if available.
"""
if cache_key in self.cache:
self.cache_hits += 1
# Move to end for LRU
self.cache.move_to_end(cache_key)
return self.cache[cache_key]
self.cache_misses += 1
return None
def _add_to_cache(self, cache_key: str, response: Dict[str, Any]) -> None:
"""
Adds a response to the cache with LRU eviction.
"""
self.cache[cache_key] = response
self.cache.move_to_end(cache_key)
# Evict oldest entry if cache is full
if len(self.cache) > self.cache_size:
self.cache.popitem(last=False)
def query(
self,
question: str,
num_results: int = 4,
return_sources: bool = True,
filter_metadata: Optional[Dict[str, Any]] = None,
use_reranking: Optional[bool] = None,
rerank_top_k: int = 4
) -> Dict[str, Any]:
"""
Enhanced query method with reranking and caching support.
"""
# Check cache
if self.enable_caching:
cache_key = self._get_cache_key(question, num_results)
cached_response = self._get_from_cache(cache_key)
if cached_response is not None:
cached_response["from_cache"] = True
return cached_response
# Determine if we should use reranking
use_rerank = use_reranking if use_reranking is not None else self.enable_reranking
# Retrieve more documents if reranking
retrieve_k = num_results * 3 if use_rerank else num_results
if not self.indexed:
raise ValueError("No index available. Run index_documents first.")
start_time = time.time()
# Retrieve relevant documents
retrieved_docs = self.vector_store_manager.similarity_search(
query=question,
k=retrieve_k,
filter_metadata=filter_metadata
)
if not retrieved_docs:
response = {
"question": question,
"answer": "I could not find any relevant information to answer this question.",
"sources": [],
"retrieval_time_seconds": time.time() - start_time,
"from_cache": False
}
return response
retrieval_time = time.time() - start_time
# Rerank if enabled
if use_rerank and self.enable_reranking:
rerank_start = time.time()
retrieved_docs = self._rerank_documents(
question,
retrieved_docs,
rerank_top_k
)
rerank_time = time.time() - rerank_start
else:
rerank_time = 0
# Generate answer
generation_start = time.time()
answer = self.llm.generate_with_context(
question=question,
context_documents=retrieved_docs
)
generation_time = time.time() - generation_start
# Prepare response
response = {
"question": question,
"answer": answer,
"retrieval_time_seconds": retrieval_time,
"rerank_time_seconds": rerank_time,
"generation_time_seconds": generation_time,
"total_time_seconds": time.time() - start_time,
"from_cache": False
}
if return_sources:
sources = []
for doc in retrieved_docs:
source_info = {
"source": doc.metadata.get('source', 'Unknown'),
"page": doc.metadata.get('page'),
"chunk_index": doc.metadata.get('chunk_index'),
"content_preview": doc.page_content[:200] + "..."
}
sources.append(source_info)
response["sources"] = sources
# Add to cache
if self.enable_caching:
self._add_to_cache(cache_key, response)
return response
def get_cache_statistics(self) -> Dict[str, Any]:
"""
Returns statistics about cache performance.
"""
total_requests = self.cache_hits + self.cache_misses
hit_rate = self.cache_hits / total_requests if total_requests > 0 else 0
return {
"cache_enabled": self.enable_caching,
"cache_size": len(self.cache),
"max_cache_size": self.cache_size,
"cache_hits": self.cache_hits,
"cache_misses": self.cache_misses,
"hit_rate": hit_rate
}
def clear_cache(self) -> None:
"""
Clears the query cache.
"""
self.cache.clear()
self.cache_hits = 0
self.cache_misses = 0
The AdvancedRAGSystem extends the base RAGSystem with additional capabilities. The reranking feature uses a cross-encoder model, which is more accurate than bi-encoder similarity but slower. The strategy is to use fast vector search to get candidate documents, then apply the slower but more accurate cross-encoder to rerank the top results.
The caching implementation uses an OrderedDict to provide LRU (Least Recently Used) eviction. When the cache reaches its size limit, we remove the oldest entry. The cache key is a hash of the question and number of results, ensuring that identical queries hit the cache. Cache statistics help monitor effectiveness and tune the cache size.
EVALUATION AND MONITORING
A production RAG system requires ongoing evaluation and monitoring to ensure quality and identify issues. We need metrics for both retrieval quality and generation quality. We also need logging and observability to debug problems when they occur.
Retrieval metrics include precision, recall, and Mean Reciprocal Rank. Precision measures what fraction of retrieved documents are relevant. Recall measures what fraction of relevant documents are retrieved. MRR measures how quickly we find the first relevant document.
Generation metrics are more challenging because evaluating natural language quality is subjective. We can use automated metrics like BLEU or ROUGE if we have reference answers, but these correlate imperfectly with human judgment. In practice, human evaluation and user feedback are essential.
Monitoring should track query latency, error rates, cache hit rates, and resource utilization. Logging should capture queries, retrieved documents, and generated answers for debugging and analysis.
Here is an evaluation and monitoring framework:
import logging
from typing import List, Dict, Any, Optional, Callable
from datetime import datetime
import statistics
class RAGEvaluator:
"""
Provides evaluation metrics and monitoring capabilities for RAG systems.
"""
def __init__(self, rag_system: RAGSystem):
"""
Initializes the evaluator with a RAG system instance.
"""
self.rag_system = rag_system
self.query_log = []
# Set up logging
self.logger = logging.getLogger('RAGSystem')
self.logger.setLevel(logging.INFO)
# Create handler if not already present
if not self.logger.handlers:
handler = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def evaluate_retrieval(
self,
queries: List[str],
relevance_judgments: Dict[str, List[str]],
k: int = 4
) -> Dict[str, float]:
"""
Evaluates retrieval quality using precision and recall metrics.
relevance_judgments maps each query to a list of relevant document IDs.
"""
precisions = []
recalls = []
mrrs = []
for query in queries:
if query not in relevance_judgments:
continue
relevant_docs = set(relevance_judgments[query])
# Retrieve documents
retrieved = self.rag_system.vector_store_manager.similarity_search(
query=query,
k=k
)
retrieved_ids = [
doc.metadata.get('source', '') for doc in retrieved
]
# Calculate precision
relevant_retrieved = sum(
1 for doc_id in retrieved_ids if doc_id in relevant_docs
)
precision = relevant_retrieved / len(retrieved_ids) if retrieved_ids else 0
precisions.append(precision)
# Calculate recall
recall = relevant_retrieved / len(relevant_docs) if relevant_docs else 0
recalls.append(recall)
# Calculate MRR
for i, doc_id in enumerate(retrieved_ids, 1):
if doc_id in relevant_docs:
mrrs.append(1.0 / i)
break
else:
mrrs.append(0.0)
return {
"mean_precision": statistics.mean(precisions) if precisions else 0,
"mean_recall": statistics.mean(recalls) if recalls else 0,
"mean_reciprocal_rank": statistics.mean(mrrs) if mrrs else 0,
"num_queries": len(queries)
}
def log_query(
self,
question: str,
answer: str,
sources: List[Dict[str, Any]],
latency: float,
user_id: Optional[str] = None
) -> None:
"""
Logs a query for monitoring and analysis.
"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"question": question,
"answer": answer,
"num_sources": len(sources),
"latency_seconds": latency,
"user_id": user_id
}
self.query_log.append(log_entry)
self.logger.info(
f"Query processed - Latency: {latency:.2f}s - "
f"Sources: {len(sources)} - Question: {question[:50]}..."
)
def get_latency_statistics(self) -> Dict[str, float]:
"""
Computes latency statistics from the query log.
"""
if not self.query_log:
return {}
latencies = [entry["latency_seconds"] for entry in self.query_log]
return {
"mean_latency": statistics.mean(latencies),
"median_latency": statistics.median(latencies),
"min_latency": min(latencies),
"max_latency": max(latencies),
"p95_latency": statistics.quantiles(latencies, n=20)[18] if len(latencies) > 20 else max(latencies),
"total_queries": len(latencies)
}
def analyze_query_patterns(self) -> Dict[str, Any]:
"""
Analyzes patterns in the query log.
"""
if not self.query_log:
return {}
# Analyze query lengths
query_lengths = [len(entry["question"]) for entry in self.query_log]
# Analyze source counts
source_counts = [entry["num_sources"] for entry in self.query_log]
return {
"mean_query_length": statistics.mean(query_lengths),
"mean_sources_per_query": statistics.mean(source_counts),
"queries_with_no_sources": sum(1 for count in source_counts if count == 0),
"total_queries_analyzed": len(self.query_log)
}
def export_query_log(self, filepath: str) -> None:
"""
Exports the query log to a JSON file for further analysis.
"""
import json
with open(filepath, 'w') as f:
json.dump(self.query_log, f, indent=2)
print(f"Query log exported to {filepath}")
The RAGEvaluator class provides tools for measuring and monitoring system performance. The evaluate_retrieval method computes standard information retrieval metrics using relevance judgments. In practice, creating these judgments requires manual annotation or user feedback.
The logging functionality captures every query along with metadata like latency and sources used. This data is invaluable for debugging, understanding usage patterns, and identifying areas for improvement. The latency statistics help track performance over time and identify degradation.
The query pattern analysis reveals insights about how users interact with the system. If many queries return no sources, it suggests problems with retrieval or gaps in the document collection. If query lengths vary widely, you might need to optimize for different query types.
FULL RUNNING EXAMPLE
#!/usr/bin/env python3
"""
Production-Ready RAG System with Multi-Format Document Support
and Cross-Platform GPU Acceleration
This complete implementation demonstrates a fully functional RAG system
that supports PDF, Word, HTML, and Markdown documents, with automatic
hardware detection for NVIDIA CUDA, AMD ROCm, Intel XPU, and Apple MPS.
"""
import torch
import platform
import os
import sys
from typing import List, Optional, Dict, Any, Tuple, Literal, Union
from pathlib import Path
import time
import hashlib
import json
import logging
from collections import OrderedDict
import statistics
from datetime import datetime
# Document handling imports
try:
import pdfplumber
except ImportError:
pdfplumber = None
try:
from docx import Document as DocxDocument
except ImportError:
DocxDocument = None
try:
from bs4 import BeautifulSoup
except ImportError:
BeautifulSoup = None
# LangChain imports
from langchain.docstore.document import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings.base import Embeddings
from langchain.vectorstores import FAISS, Chroma
# Sentence transformers for embeddings
try:
from sentence_transformers import SentenceTransformer, CrossEncoder
except ImportError:
SentenceTransformer = None
CrossEncoder = None
# Transformers for local LLM
try:
from transformers import AutoTokenizer, AutoModelForCausalLM
except ImportError:
AutoTokenizer = None
AutoModelForCausalLM = None
# OpenAI for remote models
try:
import openai
except ImportError:
openai = None
class HardwareDetector:
"""
Detects available GPU hardware and configures the appropriate
compute backend for optimal performance across different architectures.
"""
def __init__(self):
self.device_type: Optional[Literal['cuda', 'rocm', 'mps', 'xpu', 'cpu']] = None
self.device = None
self.device_name = None
self._detect_hardware()
def _detect_hardware(self):
"""
Performs comprehensive hardware detection across all supported
GPU architectures and falls back to CPU if no GPU is available.
"""
if torch.cuda.is_available():
self.device_type = 'cuda'
self.device = torch.device('cuda')
self.device_name = torch.cuda.get_device_name(0)
print(f"Detected NVIDIA GPU: {self.device_name}")
return
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
self.device_type = 'mps'
self.device = torch.device('mps')
self.device_name = "Apple Metal Performance Shaders"
print(f"Detected Apple Silicon with MPS support")
return
try:
import intel_extension_for_pytorch as ipex
if torch.xpu.is_available():
self.device_type = 'xpu'
self.device = torch.device('xpu')
self.device_name = torch.xpu.get_device_name(0)
print(f"Detected Intel GPU: {self.device_name}")
return
except (ImportError, AttributeError):
pass
if torch.version.hip is not None:
self.device_type = 'rocm'
self.device = torch.device('cuda')
self.device_name = "AMD GPU with ROCm"
print(f"Detected AMD GPU with ROCm support")
return
self.device_type = 'cpu'
self.device = torch.device('cpu')
self.device_name = "CPU"
print("No GPU detected, using CPU")
def get_device(self) -> torch.device:
"""Returns the configured PyTorch device."""
return self.device
def get_device_type(self) -> str:
"""Returns a string identifier for the device type."""
return self.device_type
def optimize_model(self, model):
"""
Applies device-specific optimizations to a model.
"""
if self.device_type == 'xpu':
try:
import intel_extension_for_pytorch as ipex
model = ipex.optimize(model)
except ImportError:
pass
return model.to(self.device)
class MultiFormatDocumentLoader:
"""
Loads documents from multiple file formats and converts them
into a unified Document representation suitable for RAG processing.
"""
def __init__(self):
self.supported_extensions = {
'.pdf': self._load_pdf,
'.docx': self._load_docx,
'.doc': self._load_doc,
'.html': self._load_html,
'.htm': self._load_html,
'.md': self._load_markdown,
'.txt': self._load_text
}
def load_documents(self, file_path: str) -> List[Document]:
"""
Loads a document from the specified file path.
"""
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"File not found: {file_path}")
extension = path.suffix.lower()
if extension not in self.supported_extensions:
raise ValueError(
f"Unsupported file format: {extension}. "
f"Supported formats: {list(self.supported_extensions.keys())}"
)
loader_func = self.supported_extensions[extension]
return loader_func(file_path)
def _load_pdf(self, file_path: str) -> List[Document]:
"""Extracts text from PDF files page by page."""
if pdfplumber is None:
raise ImportError(
"pdfplumber is required for PDF support. "
"Install it with: pip install pdfplumber"
)
documents = []
with pdfplumber.open(file_path) as pdf:
for page_num, page in enumerate(pdf.pages, start=1):
text = page.extract_text()
if text and text.strip():
text = self._clean_pdf_text(text)
doc = Document(
page_content=text,
metadata={
'source': file_path,
'page': page_num,
'total_pages': len(pdf.pages),
'format': 'pdf'
}
)
documents.append(doc)
return documents
def _clean_pdf_text(self, text: str) -> str:
"""Removes common artifacts from PDF text extraction."""
lines = text.split('\n')
cleaned_lines = []
for line in lines:
line = line.strip()
if line:
if cleaned_lines and cleaned_lines[-1].endswith('-'):
cleaned_lines[-1] = cleaned_lines[-1][:-1] + line
else:
cleaned_lines.append(line)
return '\n'.join(cleaned_lines)
def _load_docx(self, file_path: str) -> List[Document]:
"""Extracts text from Word DOCX files."""
if DocxDocument is None:
raise ImportError(
"python-docx is required for DOCX support. "
"Install it with: pip install python-docx"
)
docx = DocxDocument(file_path)
paragraphs = [para.text for para in docx.paragraphs if para.text.strip()]
full_text = '\n\n'.join(paragraphs)
doc = Document(
page_content=full_text,
metadata={
'source': file_path,
'format': 'docx',
'paragraphs': len(paragraphs)
}
)
return [doc]
def _load_doc(self, file_path: str) -> List[Document]:
"""Handles legacy DOC format."""
raise NotImplementedError(
"Legacy DOC format requires conversion. "
"Please convert to DOCX format."
)
def _load_html(self, file_path: str) -> List[Document]:
"""Extracts main content from HTML files."""
if BeautifulSoup is None:
raise ImportError(
"beautifulsoup4 is required for HTML support. "
"Install it with: pip install beautifulsoup4"
)
with open(file_path, 'r', encoding='utf-8') as f:
html_content = f.read()
soup = BeautifulSoup(html_content, 'html.parser')
for element in soup(['script', 'style', 'nav', 'footer', 'header']):
element.decompose()
main_content = soup.find('main') or soup.find('article') or soup.find('body')
if main_content:
text = main_content.get_text(separator='\n', strip=True)
else:
text = soup.get_text(separator='\n', strip=True)
lines = [line.strip() for line in text.split('\n') if line.strip()]
text = '\n'.join(lines)
doc = Document(
page_content=text,
metadata={
'source': file_path,
'format': 'html',
'title': soup.title.string if soup.title else None
}
)
return [doc]
def _load_markdown(self, file_path: str) -> List[Document]:
"""Loads Markdown files."""
with open(file_path, 'r', encoding='utf-8') as f:
text = f.read()
doc = Document(
page_content=text,
metadata={
'source': file_path,
'format': 'markdown'
}
)
return [doc]
def _load_text(self, file_path: str) -> List[Document]:
"""Loads plain text files."""
with open(file_path, 'r', encoding='utf-8') as f:
text = f.read()
doc = Document(
page_content=text,
metadata={
'source': file_path,
'format': 'text'
}
)
return [doc]
def load_directory(self, directory_path: str, recursive: bool = True) -> List[Document]:
"""Loads all supported documents from a directory."""
documents = []
path = Path(directory_path)
if not path.is_dir():
raise ValueError(f"Not a directory: {directory_path}")
file_iterator = path.rglob('*') if recursive else path.glob('*')
for file_path in file_iterator:
if file_path.is_file() and file_path.suffix.lower() in self.supported_extensions:
try:
docs = self.load_documents(str(file_path))
documents.extend(docs)
print(f"Loaded {len(docs)} document(s) from {file_path.name}")
except Exception as e:
print(f"Error loading {file_path.name}: {str(e)}")
return documents
class DocumentChunker:
"""
Splits documents into optimally-sized chunks for embedding and retrieval.
"""
def __init__(
self,
chunk_size: int = 1000,
chunk_overlap: int = 200,
separators: Optional[List[str]] = None
):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
if separators is None:
self.separators = ["\n\n", "\n", ". ", "! ", "? ", "; ", ", ", " ", ""]
else:
self.separators = separators
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
separators=self.separators,
length_function=len
)
def chunk_documents(self, documents: List[Document]) -> List[Document]:
"""Splits documents into chunks while preserving metadata."""
chunked_documents = []
for doc in documents:
chunks = self.splitter.split_text(doc.page_content)
for i, chunk_text in enumerate(chunks):
chunk_metadata = doc.metadata.copy()
chunk_metadata.update({
'chunk_index': i,
'total_chunks': len(chunks),
'chunk_size': len(chunk_text)
})
chunk_doc = Document(
page_content=chunk_text,
metadata=chunk_metadata
)
chunked_documents.append(chunk_doc)
return chunked_documents
class MultiBackendEmbeddings(Embeddings):
"""
Unified interface for generating embeddings using local or remote models.
"""
def __init__(
self,
model_name: str = "all-MiniLM-L6-v2",
use_local: bool = True,
api_key: Optional[str] = None,
hardware_detector: Optional[HardwareDetector] = None
):
self.model_name = model_name
self.use_local = use_local
self.api_key = api_key
self.hardware_detector = hardware_detector or HardwareDetector()
if self.use_local:
self._initialize_local_model()
else:
self._initialize_remote_client()
def _initialize_local_model(self):
"""Loads a local Sentence Transformers model."""
if SentenceTransformer is None:
raise ImportError(
"sentence-transformers is required for local embeddings. "
"Install it with: pip install sentence-transformers"
)
self.model = SentenceTransformer(self.model_name)
device = self.hardware_detector.get_device()
self.model = self.model.to(device)
self.model = self.hardware_detector.optimize_model(self.model)
print(f"Loaded local embedding model '{self.model_name}' on {device}")
def _initialize_remote_client(self):
"""Initializes the client for remote embedding API."""
if not self.api_key:
raise ValueError("API key required for remote embeddings")
if openai is None:
raise ImportError(
"openai is required for remote embeddings. "
"Install it with: pip install openai"
)
self.client = openai.OpenAI(api_key=self.api_key)
print(f"Initialized remote embedding client for model '{self.model_name}'")
def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""Generates embeddings for a list of documents."""
if self.use_local:
return self._embed_local(texts)
else:
return self._embed_remote(texts)
def embed_query(self, text: str) -> List[float]:
"""Generates an embedding for a single query."""
if self.use_local:
return self._embed_local([text])[0]
else:
return self._embed_remote([text])[0]
def _embed_local(self, texts: List[str]) -> List[List[float]]:
"""Generates embeddings using the local model."""
embeddings = self.model.encode(
texts,
convert_to_numpy=True,
show_progress_bar=len(texts) > 10,
batch_size=32
)
return embeddings.tolist()
def _embed_remote(self, texts: List[str]) -> List[List[float]]:
"""Generates embeddings using a remote API."""
embeddings = []
batch_size = 100
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
response = self.client.embeddings.create(
model=self.model_name,
input=batch
)
batch_embeddings = [item.embedding for item in response.data]
embeddings.extend(batch_embeddings)
return embeddings
def get_embedding_dimension(self) -> int:
"""Returns the dimensionality of the embedding vectors."""
if self.use_local:
return self.model.get_sentence_embedding_dimension()
else:
dimension_map = {
'text-embedding-ada-002': 1536,
'text-embedding-3-small': 1536,
'text-embedding-3-large': 3072
}
return dimension_map.get(self.model_name, 1536)
class VectorStoreManager:
"""
Manages vector storage and retrieval using multiple backend options.
"""
def __init__(
self,
embeddings: MultiBackendEmbeddings,
backend: str = "faiss",
persist_directory: Optional[str] = None
):
self.embeddings = embeddings
self.backend = backend
self.persist_directory = persist_directory
self.vector_store = None
def create_from_documents(self, documents: List[Document]) -> None:
"""Creates a new vector store from documents."""
if not documents:
raise ValueError("Cannot create vector store from empty document list")
print(f"Creating vector store with {len(documents)} documents...")
if self.backend == "faiss":
self._create_faiss_store(documents)
elif self.backend == "chroma":
self._create_chroma_store(documents)
else:
raise ValueError(f"Unsupported backend: {self.backend}")
print("Vector store created successfully")
def _create_faiss_store(self, documents: List[Document]) -> None:
"""Creates a FAISS vector store."""
self.vector_store = FAISS.from_documents(
documents=documents,
embedding=self.embeddings
)
def _create_chroma_store(self, documents: List[Document]) -> None:
"""Creates a ChromaDB vector store with persistence."""
if self.persist_directory:
os.makedirs(self.persist_directory, exist_ok=True)
self.vector_store = Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory=self.persist_directory
)
def add_documents(self, documents: List[Document]) -> None:
"""Adds new documents to an existing vector store."""
if self.vector_store is None:
raise ValueError("Vector store not initialized")
print(f"Adding {len(documents)} documents to vector store...")
self.vector_store.add_documents(documents)
if self.backend == "chroma" and self.persist_directory:
self.vector_store.persist()
print("Documents added successfully")
def similarity_search(
self,
query: str,
k: int = 4,
filter_metadata: Optional[dict] = None
) -> List[Document]:
"""Performs similarity search to find relevant documents."""
if self.vector_store is None:
raise ValueError("Vector store not initialized")
if filter_metadata and self.backend == "chroma":
results = self.vector_store.similarity_search(
query=query,
k=k,
filter=filter_metadata
)
else:
retrieve_k = k * 3 if filter_metadata else k
results = self.vector_store.similarity_search(query=query, k=retrieve_k)
if filter_metadata:
results = self._filter_documents(results, filter_metadata)
results = results[:k]
return results
def similarity_search_with_score(
self,
query: str,
k: int = 4
) -> List[Tuple[Document, float]]:
"""Performs similarity search and returns documents with scores."""
if self.vector_store is None:
raise ValueError("Vector store not initialized")
return self.vector_store.similarity_search_with_score(query=query, k=k)
def _filter_documents(
self,
documents: List[Document],
filter_metadata: dict
) -> List[Document]:
"""Filters documents based on metadata criteria."""
filtered = []
for doc in documents:
match = True
for key, value in filter_metadata.items():
if key not in doc.metadata or doc.metadata[key] != value:
match = False
break
if match:
filtered.append(doc)
return filtered
def save(self, path: str) -> None:
"""Saves the vector store to disk."""
if self.vector_store is None:
raise ValueError("No vector store to save")
if self.backend == "faiss":
self.vector_store.save_local(path)
print(f"FAISS index saved to {path}")
elif self.backend == "chroma":
if self.persist_directory:
self.vector_store.persist()
print(f"Chroma database persisted to {self.persist_directory}")
def load(self, path: str) -> None:
"""Loads a vector store from disk."""
if self.backend == "faiss":
self.vector_store = FAISS.load_local(
path,
self.embeddings,
allow_dangerous_deserialization=True
)
print(f"FAISS index loaded from {path}")
elif self.backend == "chroma":
self.vector_store = Chroma(
persist_directory=path,
embedding_function=self.embeddings
)
print(f"Chroma database loaded from {path}")
def get_statistics(self) -> dict:
"""Returns statistics about the vector store."""
if self.vector_store is None:
return {"status": "not_initialized"}
stats = {
"backend": self.backend,
"embedding_dimension": self.embeddings.get_embedding_dimension()
}
if self.backend == "faiss":
stats["num_vectors"] = self.vector_store.index.ntotal
elif self.backend == "chroma":
try:
collection = self.vector_store._collection
stats["num_vectors"] = collection.count()
except:
stats["num_vectors"] = "unknown"
return stats
class MultiBackendLLM:
"""
Unified interface for language models running locally or via API.
"""
def __init__(
self,
model_name: str = "gpt-3.5-turbo",
use_local: bool = False,
api_key: Optional[str] = None,
hardware_detector: Optional[HardwareDetector] = None,
temperature: float = 0.7,
max_tokens: int = 512
):
self.model_name = model_name
self.use_local = use_local
self.api_key = api_key
self.hardware_detector = hardware_detector or HardwareDetector()
self.temperature = temperature
self.max_tokens = max_tokens
if self.use_local:
self._initialize_local_model()
else:
self._initialize_remote_client()
def _initialize_local_model(self):
"""Loads a local language model."""
if AutoTokenizer is None or AutoModelForCausalLM is None:
raise ImportError(
"transformers is required for local models. "
"Install it with: pip install transformers torch"
)
print(f"Loading local model '{self.model_name}'...")
self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
device = self.hardware_detector.get_device()
dtype = torch.float16 if device.type != 'cpu' else torch.float32
self.model = AutoModelForCausalLM.from_pretrained(
self.model_name,
torch_dtype=dtype,
device_map="auto" if device.type != 'cpu' else None,
low_cpu_mem_usage=True
)
if device.type == 'cpu':
self.model = self.model.to(device)
self.model = self.hardware_detector.optimize_model(self.model)
self.model.eval()
print(f"Model loaded on {device}")
def _initialize_remote_client(self):
"""Initializes the client for remote language model API."""
if not self.api_key:
raise ValueError("API key required for remote models")
if openai is None:
raise ImportError(
"openai is required for remote models. "
"Install it with: pip install openai"
)
self.client = openai.OpenAI(api_key=self.api_key)
print(f"Initialized remote model client for '{self.model_name}'")
def generate(
self,
prompt: str,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None
) -> str:
"""Generates a response to the given prompt."""
temp = temperature if temperature is not None else self.temperature
max_tok = max_tokens if max_tokens is not None else self.max_tokens
if self.use_local:
return self._generate_local(prompt, temp, max_tok)
else:
return self._generate_remote(prompt, temp, max_tok)
def _generate_local(
self,
prompt: str,
temperature: float,
max_tokens: int
) -> str:
"""Generates a response using the local model."""
inputs = self.tokenizer(
prompt,
return_tensors="pt",
padding=True,
truncation=True,
max_length=2048
)
device = self.hardware_detector.get_device()
inputs = {k: v.to(device) for k, v in inputs.items()}
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
do_sample=temperature > 0,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id
)
response = self.tokenizer.decode(
outputs[0][inputs['input_ids'].shape[1]:],
skip_special_tokens=True
)
return response.strip()
def _generate_remote(
self,
prompt: str,
temperature: float,
max_tokens: int
) -> str:
"""Generates a response using a remote API."""
response = self.client.chat.completions.create(
model=self.model_name,
messages=[{"role": "user", "content": prompt}],
temperature=temperature,
max_tokens=max_tokens
)
return response.choices[0].message.content.strip()
def generate_with_context(
self,
question: str,
context_documents: List[Document],
system_prompt: Optional[str] = None
) -> str:
"""Generates a response using retrieved context documents."""
context_parts = []
for i, doc in enumerate(context_documents, 1):
source = doc.metadata.get('source', 'Unknown')
page = doc.metadata.get('page', '')
page_info = f" (Page {page})" if page else ""
context_parts.append(
f"Document {i} [{source}{page_info}]:\n{doc.page_content}"
)
context = "\n\n".join(context_parts)
if system_prompt is None:
system_prompt = (
"You are a helpful assistant that answers questions based on "
"the provided context. If the answer cannot be found in the "
"context, say so clearly. Always cite which document you are "
"using for your answer."
)
prompt = f"""{system_prompt}
Context:
{context}
Question: {question}
Answer:"""
return self.generate(prompt)
class RAGSystem:
"""
Complete Retrieval-Augmented Generation system.
"""
def __init__(
self,
embedding_model: str = "all-MiniLM-L6-v2",
llm_model: str = "gpt-3.5-turbo",
use_local_embeddings: bool = True,
use_local_llm: bool = False,
vector_store_backend: str = "faiss",
persist_directory: Optional[str] = None,
api_key: Optional[str] = None,
chunk_size: int = 1000,
chunk_overlap: int = 200
):
print("Initializing RAG system...")
self.hardware_detector = HardwareDetector()
self.document_loader = MultiFormatDocumentLoader()
self.chunker = DocumentChunker(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
self.embeddings = MultiBackendEmbeddings(
model_name=embedding_model,
use_local=use_local_embeddings,
api_key=api_key,
hardware_detector=self.hardware_detector
)
self.vector_store_manager = VectorStoreManager(
embeddings=self.embeddings,
backend=vector_store_backend,
persist_directory=persist_directory
)
self.llm = MultiBackendLLM(
model_name=llm_model,
use_local=use_local_llm,
api_key=api_key,
hardware_detector=self.hardware_detector
)
self.indexed = False
print("RAG system initialized successfully")
def index_documents(
self,
file_paths: Optional[List[str]] = None,
directory_path: Optional[str] = None,
recursive: bool = True
) -> Dict[str, Any]:
"""Indexes documents from files or a directory."""
start_time = time.time()
print("Loading documents...")
documents = []
if file_paths:
for file_path in file_paths:
try:
docs = self.document_loader.load_documents(file_path)
documents.extend(docs)
except Exception as e:
print(f"Error loading {file_path}: {str(e)}")
if directory_path:
try:
docs = self.document_loader.load_directory(directory_path, recursive=recursive)
documents.extend(docs)
except Exception as e:
print(f"Error loading directory {directory_path}: {str(e)}")
if not documents:
raise ValueError("No documents loaded")
print(f"Loaded {len(documents)} document(s)")
print("Chunking documents...")
chunked_documents = self.chunker.chunk_documents(documents)
print(f"Created {len(chunked_documents)} chunk(s)")
self.vector_store_manager.create_from_documents(chunked_documents)
self.indexed = True
elapsed_time = time.time() - start_time
stats = {
"num_documents": len(documents),
"num_chunks": len(chunked_documents),
"indexing_time_seconds": elapsed_time,
"vector_store_stats": self.vector_store_manager.get_statistics()
}
print(f"Indexing completed in {elapsed_time:.2f} seconds")
return stats
def query(
self,
question: str,
num_results: int = 4,
return_sources: bool = True,
filter_metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Answers a question using the RAG pipeline."""
if not self.indexed:
raise ValueError("No index available. Run index_documents first.")
start_time = time.time()
print(f"Retrieving relevant documents for: {question}")
retrieved_docs = self.vector_store_manager.similarity_search(
query=question,
k=num_results,
filter_metadata=filter_metadata
)
if not retrieved_docs:
return {
"question": question,
"answer": "I could not find any relevant information to answer this question.",
"sources": [],
"retrieval_time_seconds": time.time() - start_time
}
retrieval_time = time.time() - start_time
print("Generating answer...")
generation_start = time.time()
answer = self.llm.generate_with_context(question=question, context_documents=retrieved_docs)
generation_time = time.time() - generation_start
response = {
"question": question,
"answer": answer,
"retrieval_time_seconds": retrieval_time,
"generation_time_seconds": generation_time,
"total_time_seconds": time.time() - start_time
}
if return_sources:
sources = []
for doc in retrieved_docs:
source_info = {
"source": doc.metadata.get('source', 'Unknown'),
"page": doc.metadata.get('page'),
"chunk_index": doc.metadata.get('chunk_index'),
"content_preview": doc.page_content[:200] + "..."
}
sources.append(source_info)
response["sources"] = sources
return response
def save_index(self, path: str) -> None:
"""Saves the vector store index to disk."""
if not self.indexed:
raise ValueError("No index to save")
self.vector_store_manager.save(path)
def load_index(self, path: str) -> None:
"""Loads a previously saved vector store index."""
self.vector_store_manager.load(path)
self.indexed = True
def get_system_info(self) -> Dict[str, Any]:
"""Returns information about the system configuration."""
return {
"hardware": {
"device_type": self.hardware_detector.get_device_type(),
"device_name": self.hardware_detector.device_name
},
"embeddings": {
"model": self.embeddings.model_name,
"local": self.embeddings.use_local,
"dimension": self.embeddings.get_embedding_dimension()
},
"llm": {
"model": self.llm.model_name,
"local": self.llm.use_local
},
"vector_store": self.vector_store_manager.get_statistics(),
"indexed": self.indexed
}
def main():
"""
Demonstration of the complete RAG system with sample usage.
"""
print("=" * 80)
print("RAG SYSTEM DEMONSTRATION")
print("=" * 80)
# Initialize the RAG system with local embeddings and remote LLM
# For production use, replace with your actual API key
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
print("\nWarning: No OpenAI API key found in environment.")
print("Set OPENAI_API_KEY environment variable to use remote LLM.")
print("Continuing with local embeddings only for demonstration.\n")
rag = RAGSystem(
embedding_model="all-MiniLM-L6-v2",
llm_model="gpt-3.5-turbo",
use_local_embeddings=True,
use_local_llm=False,
vector_store_backend="faiss",
api_key=api_key,
chunk_size=1000,
chunk_overlap=200
)
# Display system information
print("\nSystem Information:")
print("-" * 80)
system_info = rag.get_system_info()
for category, details in system_info.items():
print(f"\n{category.upper()}:")
if isinstance(details, dict):
for key, value in details.items():
print(f" {key}: {value}")
else:
print(f" {details}")
# Create sample documents for demonstration
print("\n" + "=" * 80)
print("CREATING SAMPLE DOCUMENTS")
print("=" * 80)
sample_dir = Path("sample_documents")
sample_dir.mkdir(exist_ok=True)
# Create a sample text file
with open(sample_dir / "machine_learning.txt", "w") as f:
f.write("""
Machine Learning Overview
Machine learning is a subset of artificial intelligence that focuses on
building systems that can learn from and make decisions based on data.
Unlike traditional programming where explicit instructions are provided,
machine learning algorithms identify patterns in data and use those
patterns to make predictions or decisions.
There are three main types of machine learning:
Supervised Learning: The algorithm learns from labeled training data,
making predictions based on that data. Common applications include
classification and regression tasks.
Unsupervised Learning: The algorithm works with unlabeled data to find
hidden patterns or structures. Clustering and dimensionality reduction
are typical use cases.
Reinforcement Learning: The algorithm learns through trial and error,
receiving rewards or penalties for actions taken. This approach is
commonly used in robotics and game playing.
Deep learning, a subset of machine learning, uses neural networks with
multiple layers to learn hierarchical representations of data. This
approach has achieved remarkable success in image recognition, natural
language processing, and many other domains.
""")
# Create a sample markdown file
with open(sample_dir / "neural_networks.md", "w") as f:
f.write("""
# Neural Networks
## Introduction
Neural networks are computing systems inspired by biological neural
networks in animal brains. They consist of interconnected nodes called
neurons organized in layers.
## Architecture
A typical neural network consists of:
- **Input Layer**: Receives the initial data
- **Hidden Layers**: Process information through weighted connections
- **Output Layer**: Produces the final result
## Training Process
Neural networks learn through a process called backpropagation:
1. Forward pass: Data flows through the network
2. Loss calculation: Compare output to expected result
3. Backward pass: Adjust weights to minimize loss
4. Iteration: Repeat until convergence
## Applications
Neural networks excel at:
- Image and speech recognition
- Natural language processing
- Time series prediction
- Anomaly detection
""")
print(f"\nCreated sample documents in {sample_dir}")
# Index the documents
print("\n" + "=" * 80)
print("INDEXING DOCUMENTS")
print("=" * 80)
indexing_stats = rag.index_documents(directory_path=str(sample_dir))
print("\nIndexing Statistics:")
for key, value in indexing_stats.items():
if isinstance(value, dict):
print(f"\n{key}:")
for k, v in value.items():
print(f" {k}: {v}")
else:
print(f"{key}: {value}")
# Perform sample queries
print("\n" + "=" * 80)
print("QUERYING THE SYSTEM")
print("=" * 80)
sample_questions = [
"What are the three main types of machine learning?",
"How do neural networks learn?",
"What is deep learning?"
]
for question in sample_questions:
print(f"\n{'=' * 80}")
print(f"Question: {question}")
print("=" * 80)
try:
result = rag.query(question, num_results=3)
print(f"\nAnswer:\n{result['answer']}\n")
print(f"Retrieval Time: {result['retrieval_time_seconds']:.3f}s")
print(f"Generation Time: {result['generation_time_seconds']:.3f}s")
print(f"Total Time: {result['total_time_seconds']:.3f}s")
if result.get('sources'):
print("\nSources:")
for i, source in enumerate(result['sources'], 1):
print(f"\n {i}. {source['source']}")
if source.get('page'):
print(f" Page: {source['page']}")
print(f" Preview: {source['content_preview'][:100]}...")
except Exception as e:
print(f"\nError processing query: {str(e)}")
if api_key is None:
print("Note: This error may be due to missing OpenAI API key.")
print("\n" + "=" * 80)
print("DEMONSTRATION COMPLETE")
print("=" * 80)
if __name__ == "__main__":
main()
This complete running example demonstrates all the components working together in a production-ready system. The code handles document loading from multiple formats, chunks them appropriately, generates embeddings, stores them in a vector database, retrieves relevant context for queries, and generates answers using a language model.
The system automatically detects and utilizes available GPU hardware across different manufacturers, falling back gracefully to CPU when necessary. The demonstration creates sample documents, indexes them, and answers questions to show the complete workflow from start to finish.
