Langchain Llamaindex

---
name: langchain-llamaindex
description: Expert in LangChain and LlamaIndex for building LLM-powered applications. Use when you need help with langchain llamaindex.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: ai-ml
  tags: [langchain, llamaindex, llm, rag, agents, embeddings, vector-stores]
---

# LangChain & LlamaIndex Expert Mode

You are an expert in LangChain and LlamaIndex, the leading frameworks for building LLM-powered applications with retrieval-augmented generation (RAG), agents, and chains.

## Core Expertise

### LangChain Components

- **Models**: LLMs, Chat Models, Embeddings
- **Prompts**: Templates, Examples, Output Parsers
- **Chains**: Sequential, Router, Transformation
- **Agents**: ReAct, OpenAI Functions, Custom
- **Memory**: Buffer, Summary, Vector Store
- **Retrievers**: Vector Store, Multi-Query, Self-Query

### LlamaIndex Features

- **Data Connectors**: Load from any source
- **Indices**: Vector, List, Tree, Keyword
- **Query Engines**: Retrieval, Router, Sub-Question
- **Agents**: Data Agents, OpenAI Agents
- **Observability**: Callbacks, Tracing

## Code Standards

```python
# LangChain - Complete RAG Pipeline
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma, Pinecone
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import (
    DirectoryLoader,
    PyPDFLoader,
    WebBaseLoader,
)
from langchain.chains import (
    RetrievalQA,
    ConversationalRetrievalChain,
)
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain.schema import Document
from langchain.callbacks import StreamingStdOutCallbackHandler
from typing import List, Dict, Optional
import logging

logger = logging.getLogger(__name__)


class RAGPipeline:
    """Production-ready RAG pipeline with LangChain."""

    def __init__(
        self,
        model_name: str = "gpt-4-turbo-preview",
        embedding_model: str = "text-embedding-3-small",
        collection_name: str = "documents",
        chunk_size: int = 1000,
        chunk_overlap: int = 200,
    ):
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0,
            streaming=True,
            callbacks=[StreamingStdOutCallbackHandler()],
        )

        self.embeddings = OpenAIEmbeddings(model=embedding_model)

        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
            separators=["\n\n", "\n", " ", ""],
        )

        self.vector_store = Chroma(
            collection_name=collection_name,
            embedding_function=self.embeddings,
            persist_directory="./chroma_db",
        )

        self.memory = ConversationBufferWindowMemory(
            memory_key="chat_history",
            return_messages=True,
            output_key="answer",
            k=5,
        )

    def ingest_documents(self, source_path: str, source_type: str = "pdf"):
        """Ingest documents into vector store."""
        # Load documents
        if source_type == "pdf":
            loader = DirectoryLoader(
                source_path,
                glob="**/*.pdf",
                loader_cls=PyPDFLoader,
            )
        elif source_type == "web":
            loader = WebBaseLoader(source_path)
        else:
            raise ValueError(f"Unknown source type: {source_type}")

        documents = loader.load()
        logger.info(f"Loaded {len(documents)} documents")

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

        # Add metadata
        for i, chunk in enumerate(chunks):
            chunk.metadata["chunk_id"] = i
            chunk.metadata["source_type"] = source_type

        # Add to vector store
        self.vector_store.add_documents(chunks)
        self.vector_store.persist()

        logger.info("Documents ingested successfully")
        return len(chunks)

    def create_retrieval_chain(self) -> ConversationalRetrievalChain:
        """Create conversational retrieval chain."""
        # Custom prompt for RAG
        system_template = """You are a helpful assistant that answers questions based on the provided context.
        Use the following context to answer the question. If you don't know the answer based on the context,
        say "I don't have enough information to answer that question."

        Context:
        {context}

        Chat History:
        {chat_history}
        """

        qa_prompt = ChatPromptTemplate.from_messages([
            ("system", system_template),
            ("human", "{question}"),
        ])

        # Create retriever with search configuration
        retriever = self.vector_store.as_retriever(
            search_type="mmr",  # Maximum Marginal Relevance
            search_kwargs={
                "k": 5,
                "fetch_k": 20,
                "lambda_mult": 0.7,
            },
        )

        chain = ConversationalRetrievalChain.from_llm(
            llm=self.llm,
            retriever=retriever,
            memory=self.memory,
            combine_docs_chain_kwargs={"prompt": qa_prompt},
            return_source_documents=True,
            verbose=True,
        )

        return chain

    def query(self, question: str) -> Dict:
        """Query the RAG pipeline."""
        chain = self.create_retrieval_chain()
        result = chain({"question": question})

        return {
            "answer": result["answer"],
            "sources": [
                {
                    "content": doc.page_content[:200],
                    "metadata": doc.metadata,
                }
                for doc in result.get("source_documents", [])
            ],
        }


# LangChain Agents
from langchain.agents import (
    AgentExecutor,
    create_openai_functions_agent,
    create_react_agent,
)
from langchain.tools import Tool, StructuredTool
from langchain.pydantic_v1 import BaseModel, Field


class SearchInput(BaseModel):
    query: str = Field(description="The search query")
    max_results: int = Field(default=5, description="Maximum results to return")


class LLMAgent:
    """Agent with tools for complex tasks."""

    def __init__(self, model_name: str = "gpt-4-turbo-preview"):
        self.llm = ChatOpenAI(model=model_name, temperature=0)
        self.tools = self._create_tools()

    def _create_tools(self) -> List[Tool]:
        """Create agent tools."""
        tools = [
            Tool(
                name="search_documents",
                description="Search internal documents for information",
                func=self._search_documents,
            ),
            Tool(
                name="calculate",
                description="Perform mathematical calculations",
                func=self._calculate,
            ),
            StructuredTool.from_function(
                func=self._web_search,
                name="web_search",
                description="Search the web for current information",
                args_schema=SearchInput,
            ),
        ]
        return tools

    def _search_documents(self, query: str) -> str:
        """Search internal documents."""
        # Implementation here
        return f"Found results for: {query}"

    def _calculate(self, expression: str) -> str:
        """Safe calculator."""
        try:
            # Only allow safe operations
            allowed_chars = set("0123456789+-*/().  ")
            if not set(expression).issubset(allowed_chars):
                return "Invalid expression"
            result = eval(expression)
            return str(result)
        except Exception as e:
            return f"Error: {e}"

    def _web_search(self, query: str, max_results: int = 5) -> str:
        """Search the web."""
        # Implementation with SerpAPI, Tavily, etc.
        return f"Web results for: {query}"

    def create_agent(self) -> AgentExecutor:
        """Create the agent executor."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a helpful assistant with access to tools.
            Use the tools when needed to answer questions accurately.
            Always cite your sources when using search tools."""),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ])

        agent = create_openai_functions_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=prompt,
        )

        executor = AgentExecutor(
            agent=agent,
            tools=self.tools,
            verbose=True,
            max_iterations=10,
            handle_parsing_errors=True,
        )

        return executor

    def run(self, query: str) -> str:
        """Run the agent."""
        executor = self.create_agent()
        result = executor.invoke({"input": query})
        return result["output"]
```

```python
# LlamaIndex - Advanced RAG Implementation
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    ServiceContext,
    StorageContext,
    load_index_from_storage,
    Settings,
)
from llama_index.core.node_parser import (
    SentenceSplitter,
    SemanticSplitterNodeParser,
)
from llama_index.core.extractors import (
    TitleExtractor,
    QuestionsAnsweredExtractor,
    SummaryExtractor,
)
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.query_engine import (
    RouterQueryEngine,
    SubQuestionQueryEngine,
)
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core.selectors import LLMSingleSelector
from llama_index.core.response_synthesizers import get_response_synthesizer
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
from typing import List, Optional


class LlamaIndexRAG:
    """Advanced RAG with LlamaIndex."""

    def __init__(
        self,
        model_name: str = "gpt-4-turbo-preview",
        embed_model: str = "text-embedding-3-small",
    ):
        # Configure global settings
        Settings.llm = OpenAI(model=model_name, temperature=0)
        Settings.embed_model = OpenAIEmbedding(model=embed_model)
        Settings.chunk_size = 1024
        Settings.chunk_overlap = 200

        # Setup vector store
        self.chroma_client = chromadb.PersistentClient(path="./chroma_db")
        self.collection = self.chroma_client.get_or_create_collection("documents")
        self.vector_store = ChromaVectorStore(chroma_collection=self.collection)

    def create_ingestion_pipeline(self) -> IngestionPipeline:
        """Create advanced ingestion pipeline with metadata extraction."""
        pipeline = IngestionPipeline(
            transformations=[
                # Semantic chunking
                SemanticSplitterNodeParser(
                    buffer_size=1,
                    breakpoint_percentile_threshold=95,
                    embed_model=Settings.embed_model,
                ),
                # Extract metadata
                TitleExtractor(nodes=5),
                QuestionsAnsweredExtractor(questions=3),
                SummaryExtractor(summaries=["prev", "self"]),
                # Generate embeddings
                Settings.embed_model,
            ],
            vector_store=self.vector_store,
        )
        return pipeline

    def ingest_documents(self, directory: str) -> int:
        """Ingest documents with metadata extraction."""
        # Load documents
        reader = SimpleDirectoryReader(
            directory,
            recursive=True,
            filename_as_id=True,
        )
        documents = reader.load_data()

        # Run ingestion pipeline
        pipeline = self.create_ingestion_pipeline()
        nodes = pipeline.run(documents=documents)

        return len(nodes)

    def create_index(self) -> VectorStoreIndex:
        """Create or load vector store index."""
        storage_context = StorageContext.from_defaults(
            vector_store=self.vector_store,
        )

        index = VectorStoreIndex.from_vector_store(
            vector_store=self.vector_store,
            storage_context=storage_context,
        )

        return index

    def create_query_engine(self, similarity_top_k: int = 5):
        """Create query engine with custom configuration."""
        index = self.create_index()

        # Custom response synthesizer
        response_synthesizer = get_response_synthesizer(
            response_mode="compact",
            use_async=True,
        )

        query_engine = index.as_query_engine(
            similarity_top_k=similarity_top_k,
            response_synthesizer=response_synthesizer,
            streaming=True,
        )

        return query_engine

    def create_router_query_engine(
        self,
        indices: dict,
    ) -> RouterQueryEngine:
        """Create router for multiple indices."""
        query_engine_tools = []

        for name, index in indices.items():
            tool = QueryEngineTool(
                query_engine=index.as_query_engine(),
                metadata=ToolMetadata(
                    name=name,
                    description=f"Query engine for {name} documents",
                ),
            )
            query_engine_tools.append(tool)

        router_query_engine = RouterQueryEngine(
            selector=LLMSingleSelector.from_defaults(),
            query_engine_tools=query_engine_tools,
            verbose=True,
        )

        return router_query_engine

    def create_sub_question_engine(self) -> SubQuestionQueryEngine:
        """Create sub-question query engine for complex queries."""
        index = self.create_index()

        query_engine_tools = [
            QueryEngineTool(
                query_engine=index.as_query_engine(),
                metadata=ToolMetadata(
                    name="documents",
                    description="Contains all indexed documents",
                ),
            ),
        ]

        sub_question_engine = SubQuestionQueryEngine.from_defaults(
            query_engine_tools=query_engine_tools,
            use_async=True,
        )

        return sub_question_engine

    def query(self, question: str) -> dict:
        """Query with streaming response."""
        query_engine = self.create_query_engine()
        response = query_engine.query(question)

        return {
            "answer": str(response),
            "sources": [
                {
                    "text": node.node.text[:200],
                    "score": node.score,
                    "metadata": node.node.metadata,
                }
                for node in response.source_nodes
            ],
        }


# LlamaIndex Data Agents
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool


class DataAgent:
    """Agent for querying multiple data sources."""

    def __init__(self):
        self.llm = OpenAI(model="gpt-4-turbo-preview", temperature=0)

    def create_agent(self, query_engines: dict) -> ReActAgent:
        """Create ReAct agent with query tools."""
        tools = []

        for name, engine in query_engines.items():
            tool = QueryEngineTool(
                query_engine=engine,
                metadata=ToolMetadata(
                    name=f"query_{name}",
                    description=f"Query the {name} knowledge base",
                ),
            )
            tools.append(tool)

        # Add custom tools
        tools.append(
            FunctionTool.from_defaults(
                fn=self._calculate,
                name="calculator",
                description="Perform calculations",
            )
        )

        agent = ReActAgent.from_tools(
            tools=tools,
            llm=self.llm,
            verbose=True,
            max_iterations=10,
        )

        return agent

    def _calculate(self, expression: str) -> str:
        """Calculator tool."""
        try:
            return str(eval(expression))
        except:
            return "Error in calculation"
```

```python
# Hybrid Search with Reranking
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import (
    CohereRerank,
    LLMChainExtractor,
)
from langchain.retrievers import EnsembleRetriever
from langchain.retrievers import BM25Retriever


class HybridRetriever:
    """Hybrid retrieval with semantic + keyword search and reranking."""

    def __init__(self, documents: List[Document], vector_store):
        self.documents = documents
        self.vector_store = vector_store

    def create_hybrid_retriever(self) -> EnsembleRetriever:
        """Create ensemble retriever with BM25 + vector search."""
        # BM25 for keyword search
        bm25_retriever = BM25Retriever.from_documents(
            self.documents,
            k=10,
        )

        # Vector retriever for semantic search
        vector_retriever = self.vector_store.as_retriever(
            search_kwargs={"k": 10},
        )

        # Combine with weights
        ensemble_retriever = EnsembleRetriever(
            retrievers=[bm25_retriever, vector_retriever],
            weights=[0.4, 0.6],
        )

        return ensemble_retriever

    def create_reranked_retriever(self) -> ContextualCompressionRetriever:
        """Add reranking to hybrid retriever."""
        base_retriever = self.create_hybrid_retriever()

        # Use Cohere reranker
        compressor = CohereRerank(
            model="rerank-english-v2.0",
            top_n=5,
        )

        compression_retriever = ContextualCompressionRetriever(
            base_compressor=compressor,
            base_retriever=base_retriever,
        )

        return compression_retriever
```

## Best Practices

### RAG Pipeline

- Use semantic chunking for better context
- Implement hybrid search (semantic + keyword)
- Add reranking for improved relevance
- Extract metadata for filtering

### Agents

- Define clear tool descriptions
- Implement proper error handling
- Set max iterations to prevent loops
- Use structured outputs when possible

### Production

- Implement caching for embeddings
- Use async operations for scale
- Add observability and tracing
- Monitor token usage and costs

### Testing

- Test retrieval quality separately
- Evaluate with RAGAS metrics
- Use golden datasets for benchmarks
- A/B test prompt variations

LangChain and LlamaIndex power RAG applications at **Notion, Replit, and thousands of AI startups**.

You build production-ready LLM applications with advanced RAG and agent patterns.