LangChain Complete Guide 2026: Build AI Applications with LLMs — Chains, Agents, RAG & Production Tips

What Is LangChain?

LangChain is an open-source framework for building applications powered by large language models (LLMs). It abstracts LLM calls, prompt management, external data retrieval, tool use, and conversation memory into composable modules so developers can focus on application logic instead of low-level integrations. LangChain supports Python and JavaScript, integrating with 100+ LLM providers and tools.

The LangChain Ecosystem

# LangChain Ecosystem Architecture
#
# +-------------------+     +-------------------+
# |    LangChain      |     |    LangGraph      |
# |  (Core Framework) |     | (Stateful Graphs) |
# |  Models, Prompts, |     | Cycles, Branching |
# |  Chains, Agents,  |<--->| Multi-Agent,      |
# |  Memory, RAG      |     | Human-in-the-Loop |
# +-------------------+     +-------------------+
#          |                          |
#          v                          v
# +-------------------+     +-------------------+
# |    LangSmith      |     |    LangServe      |
# |  (Observability)  |     |   (Deployment)    |
# |  Tracing, Evals,  |     |  FastAPI, REST,   |
# |  Monitoring, Debug|     |  Playground, Docs |
# +-------------------+     +-------------------+

Installation & Setup

LangChain uses a modular package structure with the core package and integrations installed separately.

# Core installation
pip install langchain langchain-core

# LLM provider integrations
pip install langchain-openai        # ChatGPT / GPT-4
pip install langchain-anthropic     # Claude
pip install langchain-google-genai  # Gemini
pip install langchain-community     # 100+ community integrations

# Vector stores & embeddings
pip install langchain-chroma        # ChromaDB
pip install langchain-pinecone      # Pinecone
pip install faiss-cpu               # FAISS (CPU)

# Additional ecosystem tools
pip install langgraph               # Stateful workflows
pip install langserve               # API deployment
pip install langsmith               # Tracing & evaluation

# Environment variables
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="ls__..."   # LangSmith

Core Concepts

Chat Models

Chat models are the foundation of LangChain, providing a unified interface to call OpenAI, Anthropic, Ollama, and other LLMs. All models implement the same BaseChatModel interface with support for streaming, async calls, and structured output.

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_community.chat_models import ChatOllama
from langchain_core.messages import HumanMessage, SystemMessage

# OpenAI GPT-4o
llm_openai = ChatOpenAI(model="gpt-4o", temperature=0.7)

# Anthropic Claude
llm_claude = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0)

# Local model via Ollama
llm_local = ChatOllama(model="llama3.1:8b")

# All models share the same interface
messages = [
    SystemMessage(content="You are a helpful coding assistant."),
    HumanMessage(content="Write a Python function to merge two sorted lists."),
]
response = llm_openai.invoke(messages)
print(response.content)

# Streaming output
for chunk in llm_openai.stream(messages):
    print(chunk.content, end="", flush=True)

Prompt Templates & Output Parsers

Prompt templates parameterize reusable prompts, while output parsers convert LLM text responses into structured data. Together with LCEL pipelines, they form type-safe LLM call chains.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel, Field

# Simple prompt template
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a {role} expert."),
    ("human", "{question}"),
])

# String output parser (most common)
chain = prompt | llm_openai | StrOutputParser()
result = chain.invoke({"role": "Python", "question": "Explain decorators"})

# Structured output with Pydantic
class CodeReview(BaseModel):
    issues: list[str] = Field(description="List of code issues found")
    score: int = Field(description="Code quality score 1-10")
    suggestion: str = Field(description="Main improvement suggestion")

structured_llm = llm_openai.with_structured_output(CodeReview)
review = structured_llm.invoke("Review this code: def f(x): return x+1")
print(review.score, review.issues)

LCEL — LangChain Expression Language

LCEL is LangChain's declarative composition language that chains Runnable components with the | pipe operator. Every component implements invoke, stream, batch, and ainvoke methods. LCEL chains automatically support streaming, concurrent batching, and LangSmith tracing.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel

# Basic LCEL chain: prompt | model | parser
chain = (
    ChatPromptTemplate.from_template("Summarize this text: {text}")
    | llm_openai
    | StrOutputParser()
)

# Invoke, stream, or batch
result = chain.invoke({"text": "LangChain is a framework..."})
for chunk in chain.stream({"text": "LangChain is a framework..."}):
    print(chunk, end="")
results = chain.batch([{"text": t} for t in texts])  # parallel

# Parallel branches with RunnableParallel
analysis = RunnableParallel(
    summary=ChatPromptTemplate.from_template("Summarize: {text}") | llm_openai | StrOutputParser(),
    sentiment=ChatPromptTemplate.from_template("Sentiment of: {text}") | llm_openai | StrOutputParser(),
    keywords=ChatPromptTemplate.from_template("Keywords from: {text}") | llm_openai | StrOutputParser(),
)
output = analysis.invoke({"text": "Great product, fast shipping!"})

RAG — Retrieval-Augmented Generation

RAG enables LLMs to answer questions based on your private data. The pipeline has an indexing phase (offline) and a query phase (online), covering five steps: load, split, embed, store, and retrieve-then-generate.

# RAG Pipeline Architecture
#
# INDEXING (offline):
#   Documents --> Loader --> Splitter --> Embeddings --> Vector Store
#   (PDF/Web/DB)  (chunks)  (RecursiveChar) (OpenAI)    (Chroma/FAISS)
#
# QUERYING (online):
#   User Query --> Embedding --> Similarity Search --> Context + Query --> LLM --> Answer
#                                (top-k chunks)       (prompt template)

from langchain_community.document_loaders import (
    PyPDFLoader, WebBaseLoader, TextLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# Step 1: Load documents
loader = PyPDFLoader("company_handbook.pdf")
docs = loader.load()  # List[Document]

# Step 2: Split into chunks
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_documents(docs)

# Step 3-4: Embed and store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./db")
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# Step 5: Retrieve and generate
rag_prompt = ChatPromptTemplate.from_template(
    """Answer based on the following context only.
Context: {context}
Question: {question}
Answer:"""
)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm_openai
    | StrOutputParser()
)

answer = rag_chain.invoke("What is the vacation policy?")

Agents & Tools

Agents turn LLMs into reasoning engines that autonomously decide which tools to call and in what order. LangChain uses the ReAct (Reasoning + Acting) pattern: the LLM thinks, selects a tool, executes it, observes the result, and continues reasoning until it reaches a final answer.

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

# Define custom tools with the @tool decorator
@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    # Replace with actual search API (Tavily, SerpAPI, etc.)
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression. Input should be a valid Python expression."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # Replace with actual weather API
    return f"Weather in {city}: 22C, sunny"

# Create a ReAct agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search_web, calculate, get_weather]
agent = create_react_agent(llm, tools)

# Run the agent
result = agent.invoke(
    {"messages": [("human", "What is the weather in Tokyo and convert 72F to Celsius?")]}
)
print(result["messages"][-1].content)

# Agent reasoning flow (ReAct pattern):
# Thought: I need to get Tokyo weather and convert 72F to C
# Action: get_weather("Tokyo")
# Observation: Weather in Tokyo: 22C, sunny
# Thought: Now convert 72F to Celsius: (72 - 32) * 5/9
# Action: calculate("(72 - 32) * 5 / 9")
# Observation: 22.22
# Final Answer: Tokyo is 22C and sunny. 72F = 22.22C

Conversation Memory

LangChain provides multiple memory types to maintain conversation context. The choice depends on context window limits, cost, and semantic retrieval needs. In LCEL and LangGraph, message history management is preferred over legacy Memory classes.

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# Modern approach: RunnableWithMessageHistory
store = {}  # session_id -> ChatMessageHistory

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

chain_with_history = RunnableWithMessageHistory(
    chain,  # your LCEL chain
    get_session_history,
    input_messages_key="question",
    history_messages_key="history",
)

# Each session maintains its own conversation history
config = {"configurable": {"session_id": "user-123"}}
chain_with_history.invoke({"question": "My name is Alice"}, config=config)
chain_with_history.invoke({"question": "What is my name?"}, config=config)
# -> "Your name is Alice"

# Memory types comparison:
# | Type                         | Strategy            | Best For            |
# |------------------------------|---------------------|---------------------|
# | ConversationBufferMemory     | Store everything    | Short conversations |
# | ConversationWindowMemory     | Keep last K turns   | Cost-sensitive apps |
# | ConversationSummaryMemory    | LLM summarizes      | Long conversations  |
# | ConversationTokenBufferMemory| Truncate by tokens  | Fixed-budget calls  |
# | VectorStoreMemory            | Semantic retrieval  | Large knowledge base|

LangGraph: Complex Workflows & State Machines

LangGraph models AI workflows as stateful graphs. Nodes are functions, edges are transition conditions. It supports cycles (agent loops), conditional branching, persistent checkpoints, and human-in-the-loop, making it ideal for multi-step, multi-actor AI applications.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

# Define the state schema
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    next_step: str

# Define node functions
def classifier(state: AgentState) -> AgentState:
    """Classify the user query into a category."""
    last_msg = state["messages"][-1].content
    # LLM classifies the query
    category = llm.invoke(f"Classify this query: {last_msg}")
    return {"next_step": category.content}

def handle_technical(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"] + [("system", "You are a tech expert.")])
    return {"messages": [response]}

def handle_general(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"] + [("system", "You are a helpful assistant.")])
    return {"messages": [response]}

def route(state: AgentState) -> str:
    if "technical" in state["next_step"].lower():
        return "technical"
    return "general"

# Build the graph
graph = StateGraph(AgentState)
graph.add_node("classifier", classifier)
graph.add_node("technical", handle_technical)
graph.add_node("general", handle_general)

graph.add_edge(START, "classifier")
graph.add_conditional_edges("classifier", route, {
    "technical": "technical",
    "general": "general",
})
graph.add_edge("technical", END)
graph.add_edge("general", END)

# Compile and run
app = graph.compile()
result = app.invoke({"messages": [("human", "How do I fix a segfault in C?")]})

LangSmith: Tracing, Evaluation & Debugging

LangSmith is LangChain's companion observability platform providing end-to-end tracing, automated evaluation, dataset management, and prompt versioning. Enable tracing by setting environment variables and all LangChain calls are reported automatically.

# Enable LangSmith tracing (set in environment)
# export LANGCHAIN_TRACING_V2="true"
# export LANGCHAIN_API_KEY="ls__..."
# export LANGCHAIN_PROJECT="my-project"

# All LangChain calls are now traced automatically!
# View traces at https://smith.langchain.com

# Programmatic evaluation with LangSmith
from langsmith import Client
from langsmith.evaluation import evaluate

client = Client()

# Create a dataset for evaluation
dataset = client.create_dataset("qa-test")
client.create_examples(
    inputs=[
        {"question": "What is LangChain?"},
        {"question": "How does RAG work?"},
    ],
    outputs=[
        {"answer": "LangChain is an LLM framework"},
        {"answer": "RAG combines retrieval with generation"},
    ],
    dataset_id=dataset.id,
)

# Define your target function and evaluator
def predict(inputs: dict) -> dict:
    return {"answer": chain.invoke(inputs["question"])}

# Run evaluation
results = evaluate(
    predict,
    data="qa-test",
    evaluators=["qa"],  # built-in QA correctness evaluator
    experiment_prefix="v1",
)

Deployment: LangServe + FastAPI

LangServe deploys LangChain Runnables as REST APIs with auto-generated OpenAPI docs, an interactive Playground, and streaming support. Built on FastAPI, it integrates seamlessly with existing Python web infrastructure.

# server.py — Deploy a chain as an API
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

app = FastAPI(title="My LLM API", version="1.0")

# Define your chain
chain = (
    ChatPromptTemplate.from_template("Translate to {language}: {text}")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# Add routes (creates /translate/invoke, /translate/stream, etc.)
add_routes(app, chain, path="/translate")

# Run: uvicorn server:app --host 0.0.0.0 --port 8000
# Playground: http://localhost:8000/translate/playground
# Docs: http://localhost:8000/docs

# --- Client-side usage ---
from langserve import RemoteRunnable

remote_chain = RemoteRunnable("http://localhost:8000/translate")
result = remote_chain.invoke({"language": "French", "text": "Hello world"})

# Streaming from client
for chunk in remote_chain.stream({"language": "Spanish", "text": "Good morning"}):
    print(chunk, end="")

Advanced RAG Techniques

A basic RAG pipeline works for simple use cases, but production environments need more refined techniques to improve retrieval quality and answer accuracy. Here are five key optimization strategies.

Multi-Query Retriever

Use an LLM to rewrite the user question from multiple angles, then merge retrieval results to improve recall.

from langchain.retrievers.multi_query import MultiQueryRetriever

# Generates 3 variations of the query for broader retrieval
multi_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.3),
)
# "What is the refund policy?" generates:
#   - "How do I get a refund?"
#   - "What are the return and refund rules?"
#   - "Refund process and eligibility requirements"
docs = multi_retriever.invoke("What is the refund policy?")

Contextual Compression & Re-ranking

After retrieval, compress and re-rank documents to keep only the most relevant passages, reducing noise sent to the LLM.

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_community.document_compressors import CohereRerank

# Option 1: LLM-based extraction (keeps only relevant sentences)
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10}),
)

# Option 2: Cohere Re-rank (fast, production-grade)
reranker = CohereRerank(model="rerank-v3.5", top_n=4)
rerank_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20}),
)

Parent Document Retriever & Hybrid Search

The parent document retriever indexes small chunks but returns larger parent chunks, solving the context loss problem. Hybrid search combines vector similarity with BM25 keyword retrieval to capture both semantic and exact matches.

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# Parent Document Retriever: index small, return big
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
parent_retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=InMemoryStore(),
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)
parent_retriever.add_documents(docs)

# Hybrid Search: combine vector + BM25 keyword search
bm25 = BM25Retriever.from_documents(chunks, k=4)
vector_ret = vectorstore.as_retriever(search_kwargs={"k": 4})
hybrid = EnsembleRetriever(
    retrievers=[bm25, vector_ret],
    weights=[0.4, 0.6],  # keyword 40%, semantic 60%
)

Production Architecture Reference

Deploying LangChain applications to production requires caching, rate limiting, monitoring, and fault tolerance. Below is a typical production architecture overview and key configurations.

# Production Architecture
#
# +----------+     +-------------+     +--------------+     +----------+
# |  Client  | --> |  FastAPI /   | --> |  LangChain   | --> |  LLM API |
# | (Web/App)|     |  LangServe  |     |  Chain/Agent |     | (OpenAI) |
# +----------+     +-------------+     +--------------+     +----------+
#                        |                    |                    |
#                        v                    v                    v
#                  +----------+        +-----------+        +----------+
#                  |  Redis   |        | LangSmith |        |  Vector  |
#                  |  Cache   |        |  Tracing  |        |   Store  |
#                  +----------+        +-----------+        +----------+

# Caching: avoid repeated LLM calls for identical inputs
from langchain_core.globals import set_llm_cache
from langchain_community.cache import RedisCache
import redis

set_llm_cache(RedisCache(redis_=redis.Redis(host="localhost", port=6379)))

# Fallback chain: switch providers on failure
main_chain = prompt | ChatOpenAI(model="gpt-4o")
fallback_chain = prompt | ChatAnthropic(model="claude-sonnet-4-20250514")
robust_chain = main_chain.with_fallbacks([fallback_chain])

# Retry with exponential backoff
from langchain_core.runnables import RunnableConfig
config = RunnableConfig(max_concurrency=5)
results = await chain.abatch(inputs, config=config)

# Token counting callback for cost monitoring
from langchain_community.callbacks import get_openai_callback
with get_openai_callback() as cb:
    result = chain.invoke({"question": "What is LangChain?"})
    print(f"Tokens: {cb.total_tokens}, Cost: ${cb.total_cost:.4f}")

LangChain vs LlamaIndex vs Haystack

The three major LLM frameworks each have distinct strengths. Your choice depends on core needs: LangChain for general agents and chain orchestration, LlamaIndex for RAG and data indexing, Haystack for enterprise NLP pipelines.

Best Practices & Common Pitfalls

• Start simple: Validate your idea with a single LLM call before introducing chains, agents, and RAG complexity
• Always enable LangSmith tracing: Enable tracing during development; debugging LLM applications becomes dramatically easier
• Optimize prompts, not code: 80% of improvements in LLM apps come from better prompts, not more complex architectures
• Tune RAG chunk size: Chunks too large introduce noise; too small lose context. 500-1500 chars with 10-20% overlap is a good starting point
• Use structured output: Use with_structured_output() instead of manually parsing LLM output for reliability and type safety
• Implement fallbacks: Use chain.with_fallbacks([fallback_chain]) to handle LLM provider failures gracefully
• Avoid over-agentifying: If the workflow is deterministic, use chains or LangGraph instead. Agent non-determinism increases debugging difficulty and cost
• Version your prompts: Use LangSmith Hub or your repository to version prompts for A/B testing and rollbacks
• Evaluation-driven development: Build automated evaluation datasets and run them after every change to prevent regressions
• Monitor token usage: Agent loops can consume many tokens. Set max_iterations limits and use callbacks to track costs

Common Pitfalls

# Pitfall 1: Not handling rate limits
from langchain_core.rate_limiters import InMemoryRateLimiter
rate_limiter = InMemoryRateLimiter(requests_per_second=1, max_bucket_size=10)
llm = ChatOpenAI(model="gpt-4o", rate_limiter=rate_limiter)

# Pitfall 2: Ignoring document metadata in RAG
# Always include metadata for filtering and source attribution
chunks = splitter.split_documents(docs)
for chunk in chunks:
    chunk.metadata["source"] = "handbook_v2"
    chunk.metadata["date"] = "2026-01"

# Pitfall 3: Not using async for I/O-bound workloads
import asyncio
async def process_queries(queries):
    tasks = [chain.ainvoke({"question": q}) for q in queries]
    return await asyncio.gather(*tasks)

# Pitfall 4: Forgetting to set temperature=0 for deterministic tasks
# Use temperature=0 for extraction, classification, structured output
# Use temperature=0.7-1.0 for creative tasks

Frequently Asked Questions