Internals: How LangChain 0.2.0 Handles RAG Pipelines vs. LlamaIndex 0.10.0

import os
import sys
from typing import List, Dict, Optional
import logging
from dotenv import load_dotenv

# LangChain 0.2.0 core imports
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.output_parsers import StrOutputParser

# Community integrations for LangChain 0.2.0
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.chat_models import ChatOpenAI
from langchain_community.document_loaders import TextLoader
from langchain_community.text_splitters import RecursiveCharacterTextSplitter

# Configure logging for error tracking
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(name)s - %(levelname)s - %(message)s\"
)
logger = logging.getLogger(__name__)

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

class LangChainRAGPipeline:
    \"\"\"Production-ready RAG pipeline using LangChain 0.2.0 Runnable interface\"\"\"

    def __init__(self, data_dir: str, chunk_size: int = 512, chunk_overlap: int = 64):
        self.data_dir = data_dir
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.embeddings = None
        self.vectorstore = None
        self.pipeline = None
        self._validate_env()

    def _validate_env(self) -> None:
        \"\"\"Check required environment variables are set\"\"\"
        required_vars = [\"OPENAI_API_KEY\"]
        missing = [var for var in required_vars if not os.getenv(var)]
        if missing:
            raise EnvironmentError(f\"Missing required env vars: {missing}\")

    def _load_and_split_docs(self) -> List[Document]:
        \"\"\"Load text files from data directory and split into chunks\"\"\"
        docs = []
        try:
            for filename in os.listdir(self.data_dir):
                if filename.endswith(\".txt\"):
                    filepath = os.path.join(self.data_dir, filename)
                    loader = TextLoader(filepath, encoding=\"utf-8\")
                    docs.extend(loader.load())
            logger.info(f\"Loaded {len(docs)} raw documents from {self.data_dir}\")
        except Exception as e:
            logger.error(f\"Failed to load documents: {str(e)}\")
            raise

        # Split into chunks
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            length_function=len,
        )
        split_docs = text_splitter.split_documents(docs)
        logger.info(f\"Split into {len(split_docs)} chunks\")
        return split_docs

    def initialize_pipeline(self) -> None:
        \"\"\"Initialize embeddings, vector store, and RAG pipeline\"\"\"
        try:
            # Initialize embeddings (LangChain 0.2.0 uses langchain-community wrappers)
            self.embeddings = OpenAIEmbeddings(model=\"text-embedding-3-small\")
            logger.info(\"Initialized OpenAI embeddings\")

            # Load and split documents
            docs = self._load_and_split_docs()

            # Create FAISS vector store
            self.vectorstore = FAISS.from_documents(docs, self.embeddings)
            logger.info(\"FAISS vector store initialized\")

            # Define RAG prompt
            prompt = ChatPromptTemplate.from_messages([
                (\"system\", \"You are a helpful assistant. Use the following context to answer the question. If you don't know the answer, say you don't know. Context: {context}\"),
                (\"human\", \"{question}\")
            ])

            # Initialize LLM
            llm = ChatOpenAI(model=\"gpt-3.5-turbo\", temperature=0)

            # Create retriever
            retriever = self.vectorstore.as_retriever(search_kwargs={\"k\": 3})

            # Build LangChain 0.2.0 Runnable pipeline
            # RunnableParallel to fetch context and pass question
            self.pipeline = (
                RunnableParallel({
                    \"context\": retriever | (lambda docs: \"\n\n\".join(d.page_content for d in docs)),
                    \"question\": RunnablePassthrough()
                })
                | prompt
                | llm
                | StrOutputParser()
            )
            logger.info(\"LangChain 0.2.0 RAG pipeline initialized successfully\")
        except Exception as e:
            logger.error(f\"Pipeline initialization failed: {str(e)}\")
            raise

    def query(self, question: str) -> str:
        \"\"\"Run a query through the RAG pipeline with error handling\"\"\"
        if not self.pipeline:
            raise RuntimeError(\"Pipeline not initialized. Call initialize_pipeline() first.\")
        try:
            logger.info(f\"Processing query: {question[:50]}...\")
            response = self.pipeline.invoke(question)
            return response
        except Exception as e:
            logger.error(f\"Query failed: {str(e)}\")
            return f\"Error processing query: {str(e)}\"

if __name__ == \"__main__\":
    # Example usage
    try:
        pipeline = LangChainRAGPipeline(data_dir=\"./data\")
        pipeline.initialize_pipeline()
        response = pipeline.query(\"What is the return policy for electronics?\")
        print(f\"Response: {response}\")
    except Exception as e:
        logger.error(f\"Main execution failed: {str(e)}\")
        sys.exit(1)

import os
import sys
from typing import List, Dict, Optional
import logging
from dotenv import load_dotenv

# LlamaIndex 0.10.0 core imports
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.node_parser import SimpleNodeParser
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.postprocessor import SimilarityPostprocessor
from llama_index.core.prompts import ChatPromptTemplate
from llama_index.core.response_synthesizers import CompactAndRefine

# LlamaIndex 0.10.0 integrations
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.vector_stores.faiss import FaissVectorStore
import faiss

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(name)s - %(levelname)s - %(message)s\"
)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

class LlamaIndexRAGPipeline:
    \"\"\"Production-ready RAG pipeline using LlamaIndex 0.10.0\"\"\"

    def __init__(self, data_dir: str, chunk_size: int = 512, chunk_overlap: int = 64):
        self.data_dir = data_dir
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.index = None
        self.query_engine = None
        self._validate_env()
        self._configure_settings()

    def _validate_env(self) -> None:
        \"\"\"Check required environment variables\"\"\"
        required_vars = [\"OPENAI_API_KEY\"]
        missing = [var for var in required_vars if not os.getenv(var)]
        if missing:
            raise EnvironmentError(f\"Missing required env vars: {missing}\")

    def _configure_settings(self) -> None:
        \"\"\"Configure global LlamaIndex 0.10.0 settings\"\"\"
        try:
            # Set embeddings
            Settings.embed_model = OpenAIEmbedding(model=\"text-embedding-3-small\")
            # Set LLM
            Settings.llm = OpenAI(model=\"gpt-3.5-turbo\", temperature=0)
            # Set node parser
            Settings.node_parser = SimpleNodeParser.from_defaults(
                chunk_size=self.chunk_size,
                chunk_overlap=self.chunk_overlap
            )
            logger.info(\"LlamaIndex 0.10.0 global settings configured\")
        except Exception as e:
            logger.error(f\"Failed to configure settings: {str(e)}\")
            raise

    def initialize_pipeline(self) -> None:
        \"\"\"Initialize vector store, index, and query engine\"\"\"
        try:
            # Load documents
            logger.info(f\"Loading documents from {self.data_dir}\")
            documents = SimpleDirectoryReader(
                self.data_dir,
                required_exts=[\".txt\"],
                encoding=\"utf-8\"
            ).load_data()
            logger.info(f\"Loaded {len(documents)} raw documents\")

            # Initialize FAISS vector store (LlamaIndex 0.10.0 native integration)
            faiss_index = faiss.IndexFlatL2(1536)  # 1536 dims for text-embedding-3-small
            vector_store = FaissVectorStore(faiss_index=faiss_index)
            logger.info(\"FAISS vector store initialized\")

            # Create VectorStoreIndex (LlamaIndex 0.10.0 core abstraction)
            self.index = VectorStoreIndex.from_documents(
                documents,
                vector_store=vector_store,
                show_progress=True
            )
            logger.info(\"VectorStoreIndex created\")

            # Configure retriever
            retriever = VectorIndexRetriever(
                index=self.index,
                similarity_top_k=3,
            )

            # Configure response synthesizer
            response_synthesizer = CompactAndRefine(
                verbose=True,
                streaming=False,
            )

            # Build query engine with custom prompt
            qa_prompt_str = (
                \"You are a helpful assistant. Use the following context to answer the question. \"
                \"If you don't know the answer, say you don't know.\n\"
                \"Context: {context_str}\n\"
                \"Question: {query_str}\n\"
                \"Answer: \"
            )
            qa_prompt = ChatPromptTemplate(qa_prompt_str)

            self.query_engine = RetrieverQueryEngine(
                retriever=retriever,
                response_synthesizer=response_synthesizer,
                node_postprocessors=[SimilarityPostprocessor(similarity_cutoff=0.7)]
            )
            # Update query engine prompt
            self.query_engine.update_prompts({\"response_synthesizer:text_qa_template\": qa_prompt})
            logger.info(\"LlamaIndex 0.10.0 RAG query engine initialized successfully\")
        except Exception as e:
            logger.error(f\"Pipeline initialization failed: {str(e)}\")
            raise

    def query(self, question: str) -> str:
        \"\"\"Run a query through the RAG pipeline with error handling\"\"\"
        if not self.query_engine:
            raise RuntimeError(\"Pipeline not initialized. Call initialize_pipeline() first.\")
        try:
            logger.info(f\"Processing query: {question[:50]}...\")
            response = self.query_engine.query(question)
            return str(response)
        except Exception as e:
            logger.error(f\"Query failed: {str(e)}\")
            return f\"Error processing query: {str(e)}\"

if __name__ == \"__main__\":
    # Example usage
    try:
        pipeline = LlamaIndexRAGPipeline(data_dir=\"./data\")
        pipeline.initialize_pipeline()
        response = pipeline.query(\"What is the return policy for electronics?\")
        print(f\"Response: {response}\")
    except Exception as e:
        logger.error(f\"Main execution failed: {str(e)}\")
        sys.exit(1)

import time
import tracemalloc
from typing import List, Dict
import statistics
import logging
import sys
import traceback

# Import pipeline classes (assumes above code is in respective modules)
# For demo purposes, we inline simplified imports
from langchain_core.runnables import RunnableRetry

class RAGBenchmark:
    \"\"\"Benchmark utility to compare LangChain 0.2.0 and LlamaIndex 0.10.0 RAG pipelines\"\"\"

    def __init__(self, langchain_pipeline, llamaindex_pipeline, num_iterations: int = 100):
        self.langchain_pipeline = langchain_pipeline
        self.llamaindex_pipeline = llamaindex_pipeline
        self.num_iterations = num_iterations
        self.questions = [
            \"What is the return policy for electronics?\",
            \"How long is the warranty on laptops?\",
            \"Can I return an opened product?\",
            \"What is the shipping time for international orders?\",
            \"How do I track my order?\"
        ]

    def _run_latency_benchmark(self, pipeline, pipeline_name: str) -> Dict[str, float]:
        \"\"\"Measure query latency for a given pipeline\"\"\"
        latencies = []
        tracemalloc.start()
        start_mem = tracemalloc.get_traced_memory()[0]

        for i in range(self.num_iterations):
            question = self.questions[i % len(self.questions)]
            start = time.perf_counter()
            try:
                pipeline.query(question)
            except Exception as e:
                logging.error(f\"{pipeline_name} query {i} failed: {str(e)}\")
                continue
            end = time.perf_counter()
            latencies.append((end - start) * 1000)  # Convert to ms

        end_mem = tracemalloc.get_traced_memory()[0]
        tracemalloc.stop()

        mem_overhead = (end_mem - start_mem) / 1024 / 1024  # MB

        return {
            \"pipeline\": pipeline_name,
            \"avg_latency_ms\": statistics.mean(latencies) if latencies else 0,
            \"p99_latency_ms\": sorted(latencies)[int(0.99 * len(latencies))] if latencies else 0,
            \"min_latency_ms\": min(latencies) if latencies else 0,
            \"max_latency_ms\": max(latencies) if latencies else 0,
            \"memory_overhead_mb\": mem_overhead,
            \"success_rate\": len(latencies) / self.num_iterations * 100
        }

    def run_full_benchmark(self) -> List[Dict[str, float]]:
        \"\"\"Run full benchmark suite on both pipelines\"\"\"
        results = []

        # Benchmark LangChain 0.2.0
        logging.info(\"Running LangChain 0.2.0 benchmark...\")
        langchain_results = self._run_latency_benchmark(self.langchain_pipeline, \"LangChain 0.2.0\")
        results.append(langchain_results)

        # Benchmark LlamaIndex 0.10.0
        logging.info(\"Running LlamaIndex 0.10.0 benchmark...\")
        llamaindex_results = self._run_latency_benchmark(self.llamaindex_pipeline, \"LlamaIndex 0.10.0\")
        results.append(llamaindex_results)

        return results

    def print_results(self, results: List[Dict[str, float]]) -> None:
        \"\"\"Print formatted benchmark results\"\"\"
        print(\"\n=== RAG Pipeline Benchmark Results ===\")
        print(f\"Iterations per pipeline: {self.num_iterations}\")
        print(f\"Dataset: 100 512-token text documents\")
        print(f\"Hardware: AWS c7g.2xlarge (8 vCPU, 16GB RAM)\")
        print(f\"Python: 3.11.4\")
        print(\"=====================================\n\")

        for res in results:
            print(f\"--- {res['pipeline']} ---\")
            print(f\"Avg Latency: {res['avg_latency_ms']:.2f} ms\")
            print(f\"P99 Latency: {res['p99_latency_ms']:.2f} ms\")
            print(f\"Min Latency: {res['min_latency_ms']:.2f} ms\")
            print(f\"Max Latency: {res['max_latency_ms']:.2f} ms\")
            print(f\"Memory Overhead: {res['memory_overhead_mb']:.2f} MB\")
            print(f\"Success Rate: {res['success_rate']:.1f}%\")
            print(\"-------------------------------------\n\")

if __name__ == \"__main__\":
    try:
        # Initialize pipelines (simplified for demo)
        # langchain_pipeline = LangChainRAGPipeline(\"./data\")
        # llamaindex_pipeline = LlamaIndexRAGPipeline(\"./data\")
        # benchmark = RAGBenchmark(langchain_pipeline, llamaindex_pipeline, num_iterations=100)
        # results = benchmark.run_full_benchmark()
        # benchmark.print_results(results)
        print(\"Benchmark script ready for execution with initialized pipelines\")
    except Exception as e:
        logging.error(f\"Benchmark failed: {str(e)}\")
        traceback.print_exc()
        sys.exit(1)

from langchain_core.runnables import RunnableRetry

# Wrap entire RAG pipeline in retry logic
retriable_pipeline = RunnableRetry(
    bound=self.pipeline,
    max_retries=3,
    backoff_factor=2,
    retry_exception_types=(Exception,)  # Retry all transient errors
)

# Use retriable_pipeline instead of self.pipeline
response = retriable_pipeline.invoke(question)

from llama_index.core.postprocessor import LLMRerank, SimilarityPostprocessor

# Add postprocessors to query engine
self.query_engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=response_synthesizer,
    node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7),
        LLMRerank(top_n=3, llm=Settings.llm)
    ]
)

import tracemalloc

# Measure memory overhead of pipeline
tracemalloc.start()
pipeline = LangChainRAGPipeline(data_dir=\"./data\")
pipeline.initialize_pipeline()
current, peak = tracemalloc.get_traced_memory()
print(f\"Current memory: {current / 1024 / 1024:.2f} MB\")
print(f\"Peak memory: {peak / 1024 / 1024:.2f} MB\")
tracemalloc.stop()