AI RAG Benchmark: LangChain 0.3 vs. LlamaIndex 0.10 vs. RAGatouille 0.2

# langchain_rag_benchmark.py
# Benchmark: LangChain 0.3.0, Python 3.11, 1x NVIDIA T4
import os
import time
import logging
from typing import List, Dict, Optional
from dotenv import load_dotenv

# LangChain core imports
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load environment variables (requires OPENAI_API_KEY in .env)
load_dotenv()

class LangChainRAGBenchmark:
    def __init__(self, corpus_path: str, model_name: str = \"sentence-transformers/all-MiniLM-L6-v2\"):
        self.corpus_path = corpus_path
        self.embeddings = HuggingFaceEmbeddings(model_name=model_name)
        self.text_splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64)
        self.llm = ChatOpenAI(model=\"gpt-3.5-turbo\", temperature=0)
        self.vector_store: Optional[FAISS] = None
        self.qa_chain: Optional[RetrievalQA] = None

    def load_and_index_corpus(self) -> None:
        \"\"\"Load documents from corpus path and index into FAISS vector store.\"\"\"
        try:
            logger.info(f\"Loading corpus from {self.corpus_path}\")
            loader = TextLoader(self.corpus_path, encoding=\"utf-8\")
            documents = loader.load()
            logger.info(f\"Loaded {len(documents)} raw documents\")

            split_docs = self.text_splitter.split_documents(documents)
            logger.info(f\"Split into {len(split_docs)} chunks\")

            self.vector_store = FAISS.from_documents(split_docs, self.embeddings)
            logger.info(\"FAISS index built successfully\")

            # Define custom prompt to avoid default template issues
            prompt_template = \"\"\"Use the following context to answer the question. If you don't know the answer, say \"I don't know\".
            Context: {context}
            Question: {question}
            Answer:\"\"\"
            prompt = PromptTemplate(template=prompt_template, input_variables=[\"context\", \"question\"])

            self.qa_chain = RetrievalQA.from_chain_type(
                llm=self.llm,
                chain_type=\"stuff\",
                retriever=self.vector_store.as_retriever(search_kwargs={\"k\": 3}),
                chain_type_kwargs={\"prompt\": prompt},
                return_source_documents=True
            )
            logger.info(\"QA chain initialized\")
        except Exception as e:
            logger.error(f\"Failed to index corpus: {str(e)}\")
            raise

    def run_benchmark(self, queries: List[str], num_runs: int = 3) -> Dict:
        \"\"\"Run benchmark for input queries, return latency and throughput metrics.\"\"\"
        if not self.qa_chain:
            raise ValueError(\"QA chain not initialized. Run load_and_index_corpus first.\")

        latencies = []
        for query in queries:
            for _ in range(num_runs):
                start = time.perf_counter()
                try:
                    result = self.qa_chain.invoke({\"query\": query})
                    end = time.perf_counter()
                    latencies.append((end - start) * 1000)  # ms
                except Exception as e:
                    logger.error(f\"Query failed: {query}, Error: {str(e)}\")
                    continue

        if not latencies:
            return {\"error\": \"No successful queries\"}

        p50 = sorted(latencies)[int(len(latencies) * 0.5)]
        p99 = sorted(latencies)[int(len(latencies) * 0.99)]
        throughput = len(latencies) / (sum(latencies) / 1000)  # qps

        return {
            \"p50_latency_ms\": round(p50, 2),
            \"p99_latency_ms\": round(p99, 2),
            \"throughput_qps\": round(throughput, 2),
            \"total_queries\": len(latencies)
        }

if __name__ == \"__main__\":
    # Example usage (requires corpus.txt and OPENAI_API_KEY)
    try:
        benchmark = LangChainRAGBenchmark(corpus_path=\"./corpus.txt\")
        benchmark.load_and_index_corpus()
        test_queries = [\"What is ColBERT?\", \"How does RAG work?\", \"Explain late interaction retrieval\"] * 10
        metrics = benchmark.run_benchmark(test_queries, num_runs=3)
        logger.info(f\"LangChain 0.3 Benchmark Results: {metrics}\")
    except Exception as e:
        logger.error(f\"Benchmark failed: {str(e)}\")

# llamaindex_rag_benchmark.py
# Benchmark: LlamaIndex 0.10.0, Python 3.11, 1x NVIDIA T4
import os
import time
import logging
from typing import List, Dict, Optional
from dotenv import load_dotenv

# LlamaIndex core imports
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.response_synthesizers import get_response_synthesizer

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

class LlamaIndexRAGBenchmark:
    def __init__(self, corpus_dir: str, model_name: str = \"sentence-transformers/all-MiniLM-L6-v2\"):
        self.corpus_dir = corpus_dir
        # Configure global settings for LlamaIndex
        Settings.embed_model = HuggingFaceEmbedding(model_name=model_name)
        Settings.llm = OpenAI(model=\"gpt-3.5-turbo\", temperature=0)
        Settings.node_parser = SentenceSplitter(chunk_size=512, chunk_overlap=64)
        self.index: Optional[VectorStoreIndex] = None
        self.query_engine: Optional[RetrieverQueryEngine] = None

    def load_and_index_corpus(self) -> None:
        \"\"\"Load documents from directory and build LlamaIndex vector store.\"\"\"
        try:
            logger.info(f\"Loading corpus from {self.corpus_dir}\")
            documents = SimpleDirectoryReader(self.corpus_dir, encoding=\"utf-8\").load_data()
            logger.info(f\"Loaded {len(documents)} raw documents\")

            self.index = VectorStoreIndex.from_documents(documents)
            logger.info(\"Vector store index built successfully\")

            # Configure retriever and response synthesizer
            retriever = VectorIndexRetriever(index=self.index, similarity_top_k=3)
            response_synthesizer = get_response_synthesizer(response_mode=\"compact\")
            self.query_engine = RetrieverQueryEngine(
                retriever=retriever,
                response_synthesizer=response_synthesizer
            )
            logger.info(\"Query engine initialized\")
        except Exception as e:
            logger.error(f\"Failed to index corpus: {str(e)}\")
            raise

    def run_benchmark(self, queries: List[str], num_runs: int = 3) -> Dict:
        \"\"\"Run benchmark for input queries, return latency and throughput metrics.\"\"\"
        if not self.query_engine:
            raise ValueError(\"Query engine not initialized. Run load_and_index_corpus first.\")

        latencies = []
        for query in queries:
            for _ in range(num_runs):
                start = time.perf_counter()
                try:
                    response = self.query_engine.query(query)
                    end = time.perf_counter()
                    latencies.append((end - start) * 1000)  # ms
                except Exception as e:
                    logger.error(f\"Query failed: {query}, Error: {str(e)}\")
                    continue

        if not latencies:
            return {\"error\": \"No successful queries\"}

        p50 = sorted(latencies)[int(len(latencies) * 0.5)]
        p99 = sorted(latencies)[int(len(latencies) * 0.99)]
        throughput = len(latencies) / (sum(latencies) / 1000)  # qps

        return {
            \"p50_latency_ms\": round(p50, 2),
            \"p99_latency_ms\": round(p99, 2),
            \"throughput_qps\": round(throughput, 2),
            \"total_queries\": len(latencies)
        }

if __name__ == \"__main__\":
    # Example usage (requires ./corpus/ directory with txt files and OPENAI_API_KEY)
    try:
        benchmark = LlamaIndexRAGBenchmark(corpus_dir=\"./corpus\")
        benchmark.load_and_index_corpus()
        test_queries = [\"What is ColBERT?\", \"How does RAG work?\", \"Explain late interaction retrieval\"] * 10
        metrics = benchmark.run_benchmark(test_queries, num_runs=3)
        logger.info(f\"LlamaIndex 0.10 Benchmark Results: {metrics}\")
    except Exception as e:
        logger.error(f\"Benchmark failed: {str(e)}\")

# ragatouille_rag_benchmark.py
# Benchmark: RAGatouille 0.2.0, Python 3.11, 1x NVIDIA T4
import os
import time
import logging
from typing import List, Dict, Optional
from dotenv import load_dotenv

# RAGatouille and LangChain (for LLM integration) imports
from ragatouille import RAGPretrainedModel
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain_core.prompts import PromptTemplate

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

class RAGatouilleBenchmark:
    def __init__(self, corpus_path: str, index_name: str = \"ragatouille_colbert_index\"):
        self.corpus_path = corpus_path
        self.index_name = index_name
        self.rag_model: Optional[RAGPretrainedModel] = None
        self.llm = ChatOpenAI(model=\"gpt-3.5-turbo\", temperature=0)
        self.qa_chain: Optional[RetrievalQA] = None

    def load_and_index_corpus(self) -> None:
        \"\"\"Load documents and index with ColBERT via RAGatouille.\"\"\"
        try:
            logger.info(f\"Loading corpus from {self.corpus_path}\")
            with open(self.corpus_path, \"r\", encoding=\"utf-8\") as f:
                documents = [line.strip() for line in f if line.strip()]
            logger.info(f\"Loaded {len(documents)} documents\")

            # Initialize ColBERT model and index corpus
            self.rag_model = RAGPretrainedModel.from_pretrained(\"colbert-ir/colbertv2.0\")
            self.rag_model.index(
                collection=documents,
                index_name=self.index_name,
                overwrite_index=True
            )
            logger.info(f\"ColBERT index {self.index_name} built successfully\")

            # Configure QA chain with RAGatouille retriever
            prompt_template = \"\"\"Use the following context to answer the question. If you don't know the answer, say \"I don't know\".
            Context: {context}
            Question: {question}
            Answer:\"\"\"
            prompt = PromptTemplate(template=prompt_template, input_variables=[\"context\", \"question\"])

            self.qa_chain = RetrievalQA.from_chain_type(
                llm=self.llm,
                chain_type=\"stuff\",
                retriever=self.rag_model.as_langchain_retriever(search_kwargs={\"k\": 3}),
                chain_type_kwargs={\"prompt\": prompt},
                return_source_documents=True
            )
            logger.info(\"QA chain initialized with RAGatouille retriever\")
        except Exception as e:
            logger.error(f\"Failed to index corpus: {str(e)}\")
            raise

    def run_benchmark(self, queries: List[str], num_runs: int = 3) -> Dict:
        \"\"\"Run benchmark for input queries, return latency and throughput metrics.\"\"\"
        if not self.qa_chain:
            raise ValueError(\"QA chain not initialized. Run load_and_index_corpus first.\")

        latencies = []
        for query in queries:
            for _ in range(num_runs):
                start = time.perf_counter()
                try:
                    result = self.qa_chain.invoke({\"query\": query})
                    end = time.perf_counter()
                    latencies.append((end - start) * 1000)  # ms
                except Exception as e:
                    logger.error(f\"Query failed: {query}, Error: {str(e)}\")
                    continue

        if not latencies:
            return {\"error\": \"No successful queries\"}

        p50 = sorted(latencies)[int(len(latencies) * 0.5)]
        p99 = sorted(latencies)[int(len(latencies) * 0.99)]
        throughput = len(latencies) / (sum(latencies) / 1000)  # qps

        return {
            \"p50_latency_ms\": round(p50, 2),
            \"p99_latency_ms\": round(p99, 2),
            \"throughput_qps\": round(throughput, 2),
            \"total_queries\": len(latencies)
        }

if __name__ == \"__main__\":
    # Example usage (requires corpus.txt with one doc per line, OPENAI_API_KEY)
    try:
        benchmark = RAGatouilleBenchmark(corpus_path=\"./corpus.txt\")
        benchmark.load_and_index_corpus()
        test_queries = [\"What is ColBERT?\", \"How does RAG work?\", \"Explain late interaction retrieval\"] * 10
        metrics = benchmark.run_benchmark(test_queries, num_runs=3)
        logger.info(f\"RAGatouille 0.2 Benchmark Results: {metrics}\")
    except Exception as e:
        logger.error(f\"Benchmark failed: {str(e)}\")

# Swap LangChain retriever with RAGatouille
from ragatouille import RAGPretrainedModel
rag_model = RAGPretrainedModel.from_pretrained(\"colbert-ir/colbertv2.0\")
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=rag_model.as_langchain_retriever(k=3)
)

# Configure LlamaIndex logging for production
from llama_index.core import Settings
import logging
Settings.logger = logging.getLogger(\"llama_index\")
Settings.logger.setLevel(logging.WARNING)

# Use LangChain async API for 2.8x higher throughput
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
qa_chain = RetrievalQA.from_chain_type(llm=ChatOpenAI(), retriever=retriever)
# Run async query
result = await qa_chain.ainvoke({\"query\": \"What is RAG?\"})