LangChain: Complete Guide from Beginner to Advanced

Large Language Models are powerful in isolation — but they become transformative when you can connect them to tools, data, memory, and logic. That is exactly what LangChain does.

LangChain is an open-source framework that lets you build LLM-powered applications by composing reusable components: prompts, chains, memory, agents, and tools. It abstracts the plumbing so you can focus on what your application needs to do.

This guide takes you from zero to production-ready. We start with the fundamentals, build intuition through working code, and progress to advanced patterns like RAG pipelines, autonomous agents, and streaming deployments.

"LLMs are the reasoning engine. LangChain is the orchestration layer that connects them to the real world."

Prerequisites and Setup

Before writing a single line of LangChain code, get your environment ready.

textCopy
# Create a virtual environment (strongly recommended)
python -m venv langchain-env
source langchain-env/bin/activate  # Windows: langchain-env\Scripts\activate

# Install core packages
pip install langchain langchain-openai langchain-community python-dotenv

# For later sections (RAG, vector stores)
pip install langchain-chroma tiktoken pypdf faiss-cpu

# Verify installation
python -c "import langchain; print(langchain.__version__)"

textCopy
# .env file — never hardcode API keys
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
TAVILY_API_KEY=tvly-...

textCopy
# config.py — load environment variables at startup
from dotenv import load_dotenv
import os

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY is not set in environment variables")

Part 1: Core Concepts

1.1 Language Models — The Foundation

LangChain wraps any LLM behind a consistent interface. You swap providers by changing one import.

textCopy
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

# Initialize the model
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.7,       # 0 = deterministic, 1 = creative
    max_tokens=1024,
    timeout=30,
    max_retries=2,
)

# Simple invocation
response = llm.invoke("Explain quantum computing in one paragraph.")
print(response.content)

# Multi-turn conversation using message history
messages = [
    SystemMessage(content="You are a senior Python developer. Be concise and practical."),
    HumanMessage(content="What is the difference between a list and a tuple?"),
    AIMessage(content="Lists are mutable sequences; tuples are immutable. Use tuples for fixed data."),
    HumanMessage(content="When would you actually choose a tuple over a list in production?"),
]
response = llm.invoke(messages)
print(response.content)

textCopy
# Switching to Anthropic Claude — zero other changes needed
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)
response = llm.invoke("What is the CAP theorem?")
print(response.content)

textCopy
# Using a local model via Ollama (fully offline, free)
from langchain_community.llms import Ollama

llm = Ollama(model="llama3.2", temperature=0)
# Pull the model first: ollama pull llama3.2
response = llm.invoke("Summarize the SOLID principles.")
print(response)

1.2 Prompt Templates — Structured Inputs

Raw strings are fragile. Prompt templates are reusable, parameterized, and composable.

textCopy
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate

# Simple string prompt template
template = PromptTemplate(
    input_variables=["topic", "level"],
    template="Explain {topic} to someone with {level} programming experience. Be practical.",
)

prompt = template.format(topic="async/await", level="beginner")
print(prompt)
# Output: Explain async/await to someone with beginner programming experience. Be practical.

textCopy
# Chat prompt template — for chat models (more powerful)
chat_template = ChatPromptTemplate.from_messages([
    ("system", "You are a {persona}. Respond in {language}."),
    ("human", "{user_question}"),
])

# Format with variables
messages = chat_template.format_messages(
    persona="senior DevOps engineer",
    language="English",
    user_question="What is blue-green deployment?",
)

response = llm.invoke(messages)
print(response.content)

textCopy
# Few-shot prompt template — teach the model by example
from langchain_core.prompts import FewShotChatMessagePromptTemplate

examples = [
    {"input": "What is Docker?", "output": "Docker is a platform for packaging applications into containers — isolated environments that run consistently anywhere."},
    {"input": "What is Kubernetes?", "output": "Kubernetes is an orchestration system that automates deploying, scaling, and managing containerized applications across clusters."},
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

final_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a cloud infrastructure expert. Give concise, one-sentence definitions."),
    few_shot_prompt,
    ("human", "{input}"),
])

chain = final_prompt | llm
response = chain.invoke({"input": "What is Terraform?"})
print(response.content)

1.3 Output Parsers — Structured Responses

LLMs return text. Output parsers transform that text into structured Python objects.

textCopy
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List

# String parser — simplest, strips metadata
str_parser = StrOutputParser()
chain = llm | str_parser
result = chain.invoke("What is REST?")
print(result)  # str, not AIMessage

# JSON output parser with Pydantic schema
class TechExplanation(BaseModel):
    concept: str = Field(description="The technical concept being explained")
    definition: str = Field(description="Clear, one-paragraph definition")
    use_cases: List[str] = Field(description="3-5 real-world use cases")
    difficulty: str = Field(description="Beginner, Intermediate, or Advanced")
    related_concepts: List[str] = Field(description="Related concepts to learn next")

parser = JsonOutputParser(pydantic_object=TechExplanation)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a technical educator. Always respond in valid JSON matching the schema exactly."),
    ("human", "Explain: {concept}\n\n{format_instructions}"),
])

chain = prompt | llm | parser
result = chain.invoke({
    "concept": "WebSockets",
    "format_instructions": parser.get_format_instructions(),
})

print(f"Concept: {result['concept']}")
print(f"Difficulty: {result['difficulty']}")
print(f"Use cases: {', '.join(result['use_cases'])}")

textCopy
# Pydantic output parser — type-safe structured output
from langchain.output_parsers import PydanticOutputParser

class CodeReview(BaseModel):
    summary: str = Field(description="One-sentence summary of the code")
    issues: List[str] = Field(description="List of bugs or code quality issues found")
    suggestions: List[str] = Field(description="Actionable improvement suggestions")
    score: int = Field(description="Code quality score from 1 to 10")

parser = PydanticOutputParser(pydantic_object=CodeReview)

prompt = ChatPromptTemplate.from_template(
    "Review this Python code:\n\n```python\n{code}\n```\n\n{format_instructions}"
)

chain = prompt | llm | parser

review: CodeReview = chain.invoke({
    "code": "def add(a,b):\n    return a+b",
    "format_instructions": parser.get_format_instructions(),
})

print(f"Score: {review.score}/10")
print(f"Issues: {review.issues}")

Part 2: Chains — The LangChain Expression Language (LCEL)

Chains are the heart of LangChain. The LangChain Expression Language (LCEL) lets you compose chains using the pipe

|

2.1 Basic LCEL Chains

textCopy
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini")

# The simplest chain: prompt | model | parser
chain = (
    ChatPromptTemplate.from_template("Tell me a fact about {topic}.")
    | llm
    | StrOutputParser()
)

# Invoke synchronously
result = chain.invoke({"topic": "black holes"})
print(result)

# Invoke asynchronously
import asyncio
async def main():
    result = await chain.ainvoke({"topic": "quantum entanglement"})
    print(result)

asyncio.run(main())

# Batch multiple inputs in parallel
results = chain.batch([
    {"topic": "Python"},
    {"topic": "Rust"},
    {"topic": "Go"},
])
for r in results:
    print(r)

2.2 Sequential Chains

textCopy
# Chain of chains — output of one feeds into the next
from langchain_core.runnables import RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Step 1: Generate a technical concept
concept_chain = (
    ChatPromptTemplate.from_template("Name one important concept in {field} in 3 words or less.")
    | llm
    | StrOutputParser()
)

# Step 2: Explain the concept that was generated
explanation_chain = (
    ChatPromptTemplate.from_template("Explain '{concept}' with a concrete code example in Python.")
    | llm
    | StrOutputParser()
)

# Sequential pipeline: field → concept → explanation
full_chain = (
    {"concept": concept_chain, "field": RunnablePassthrough()}
    | explanation_chain
)

result = full_chain.invoke("distributed systems")
print(result)

2.3 Parallel Chains with RunnableParallel

textCopy
from langchain_core.runnables import RunnableParallel

llm = ChatOpenAI(model="gpt-4o-mini")

# Run multiple chains on the same input simultaneously
analysis_chain = RunnableParallel(
    pros=ChatPromptTemplate.from_template("List 3 pros of {technology} as bullet points.")
         | llm | StrOutputParser(),

    cons=ChatPromptTemplate.from_template("List 3 cons of {technology} as bullet points.")
         | llm | StrOutputParser(),

    alternatives=ChatPromptTemplate.from_template("Name 3 alternatives to {technology}, one sentence each.")
                 | llm | StrOutputParser(),
)

result = analysis_chain.invoke({"technology": "MongoDB"})
print("PROS:\n", result["pros"])
print("\nCONS:\n", result["cons"])
print("\nALTERNATIVES:\n", result["alternatives"])

2.4 Branching with RunnableBranch

textCopy
from langchain_core.runnables import RunnableBranch

# Route to different chains based on input content
code_chain = (
    ChatPromptTemplate.from_template("Explain this code step by step: {input}")
    | llm | StrOutputParser()
)

concept_chain = (
    ChatPromptTemplate.from_template("Explain this concept with an analogy: {input}")
    | llm | StrOutputParser()
)

default_chain = (
    ChatPromptTemplate.from_template("Answer this question clearly: {input}")
    | llm | StrOutputParser()
)

branch = RunnableBranch(
    (lambda x: "def " in x["input"] or "class " in x["input"], code_chain),
    (lambda x: any(w in x["input"].lower() for w in ["what is", "explain", "define"]), concept_chain),
    default_chain,
)

# Each is routed to the right chain automatically
print(branch.invoke({"input": "def fibonacci(n): return n if n <= 1 else fibonacci(n-1)+fibonacci(n-2)"}))
print(branch.invoke({"input": "What is a neural network?"}))
print(branch.invoke({"input": "Should I use Redis or Memcached?"}))

Part 3: Memory — Stateful Conversations

LLMs are stateless by default. Every call is isolated. Memory gives them context across turns.

3.1 In-Memory Conversation History

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

llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful coding assistant. Be concise."),
    ("placeholder", "{chat_history}"),  # This gets replaced with message history
    ("human", "{input}"),
])

chain = prompt | llm | StrOutputParser()

# Store per-session histories
store: dict[str, InMemoryChatMessageHistory] = {}

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

# Wrap chain with memory management
chain_with_memory = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

config = {"configurable": {"session_id": "user-123"}}

# Turn 1
response = chain_with_memory.invoke({"input": "My name is Vighnesh."}, config=config)
print(response)  # "Hi Vighnesh! How can I help you today?"

# Turn 2 — model remembers the name from turn 1
response = chain_with_memory.invoke({"input": "What's my name?"}, config=config)
print(response)  # "Your name is Vighnesh."

# Different session — separate memory
config2 = {"configurable": {"session_id": "user-456"}}
response = chain_with_memory.invoke({"input": "What's my name?"}, config=config2)
print(response)  # "I don't know your name yet. What is it?"

3.2 Persistent Memory with Redis

textCopy
from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_redis_session_history(session_id: str) -> RedisChatMessageHistory:
    return RedisChatMessageHistory(
        session_id=session_id,
        url="redis://localhost:6379",
        ttl=3600,  # Auto-expire sessions after 1 hour
    )

chain_with_redis = RunnableWithMessageHistory(
    chain,
    get_redis_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)
# Now conversation history survives server restarts and scales across instances

3.3 Summarization Memory for Long Conversations

textCopy
from langchain.memory import ConversationSummaryBufferMemory

# Keeps recent messages in full, summarizes older ones
# Prevents context window overflow in long sessions
memory = ConversationSummaryBufferMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    max_token_limit=500,         # Once exceeded, older messages are summarized
    return_messages=True,
    memory_key="chat_history",
)

# Add messages
memory.save_context(
    {"input": "I'm building a SaaS product for code review automation."},
    {"output": "That's a great idea! What tech stack are you considering?"},
)
memory.save_context(
    {"input": "Thinking about FastAPI backend and React frontend."},
    {"output": "Good choice. FastAPI is excellent for ML-heavy backends."},
)

# Retrieve with automatic summarization of old messages
history = memory.load_memory_variables({})
print(history["chat_history"])

langchain-community

Part 4: Document Loaders and Text Splitters

Most real-world LLM apps need to work with external data — PDFs, websites, databases, code repositories. LangChain provides loaders and splitters for all of these.

4.1 Document Loaders

textCopy
from langchain_community.document_loaders import (
    PyPDFLoader,
    WebBaseLoader,
    TextLoader,
    DirectoryLoader,
    CSVLoader,
    GitLoader,
)

# Load a PDF
loader = PyPDFLoader("documents/research-paper.pdf")
pages = loader.load()  # Returns list of Document objects
print(f"Loaded {len(pages)} pages")
print(pages[0].page_content[:500])       # The text
print(pages[0].metadata)                  # {"source": "research-paper.pdf", "page": 0}

# Load from a URL
loader = WebBaseLoader("https://docs.python.org/3/library/asyncio.html")
docs = loader.load()
print(docs[0].page_content[:300])

# Load all .txt files in a directory
loader = DirectoryLoader("./docs", glob="**/*.txt", loader_cls=TextLoader)
docs = loader.load()
print(f"Loaded {len(docs)} documents")

# Load CSV as structured documents
loader = CSVLoader("data/customers.csv", source_column="id")
docs = loader.load()
print(docs[0].page_content)  # Each row becomes a Document

# Load from a Git repository
loader = GitLoader(
    repo_path="./my-repo",
    branch="main",
    file_filter=lambda path: path.endswith(".py"),
)
code_docs = loader.load()
print(f"Loaded {len(code_docs)} Python files")

4.2 Text Splitters

LLMs have context window limits. Large documents must be split into chunks before embedding.

textCopy
from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    TokenTextSplitter,
    MarkdownHeaderTextSplitter,
)

# Recursive splitter — the best default choice
# Tries to split on paragraphs → sentences → words → characters
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,         # Max characters per chunk
    chunk_overlap=200,       # Overlap to preserve context across chunks
    separators=["\n\n", "\n", ". ", " ", ""],
)

with open("long-document.txt") as f:
    text = f.read()

chunks = splitter.create_documents([text])
print(f"Split into {len(chunks)} chunks")
print(f"First chunk: {chunks[0].page_content[:200]}")

# Token-based splitter — more accurate for LLM context windows
token_splitter = TokenTextSplitter(
    chunk_size=512,     # Tokens, not characters
    chunk_overlap=50,
    encoding_name="cl100k_base",  # OpenAI's encoding
)

# Markdown-aware splitter — preserves document structure
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "h1"),
        ("##", "h2"),
        ("###", "h3"),
    ],
    strip_headers=False,
)

with open("README.md") as f:
    markdown_text = f.read()

md_chunks = md_splitter.split_text(markdown_text)
# Each chunk includes the header hierarchy in metadata
for chunk in md_chunks[:3]:
    print(chunk.metadata)    # {"h1": "Getting Started", "h2": "Installation"}
    print(chunk.page_content[:100])
    print("---")

chunk_overlap

Language.PYTHON

RecursiveCharacterTextSplitter

Part 5: Embeddings and Vector Stores

This is where LangChain unlocks semantic search — finding documents by meaning, not just keywords.

5.1 Embeddings

textCopy
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings

# OpenAI embeddings — best quality, costs money
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# Generate embedding for a single text
vector = embeddings.embed_query("What is machine learning?")
print(f"Embedding dimension: {len(vector)}")  # 3072 for large model

# Generate embeddings for a batch of documents
docs = [
    "Python is a high-level programming language.",
    "JavaScript runs in the browser and on Node.js.",
    "Rust guarantees memory safety without a garbage collector.",
]
vectors = embeddings.embed_documents(docs)
print(f"Generated {len(vectors)} vectors")

# HuggingFace embeddings — free, runs locally
hf_embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2",
    model_kwargs={"device": "cpu"},
)
vector = hf_embeddings.embed_query("semantic similarity search")
print(f"HuggingFace embedding dimension: {len(vector)}")  # 384

5.2 Vector Stores

textCopy
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Full pipeline: load → split → embed → store
loader = PyPDFLoader("documents/technical-docs.pdf")
raw_docs = loader.load()

splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(raw_docs)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Create persistent vector store from documents
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",   # Persist to disk
    collection_name="technical_docs",
)

print(f"Stored {vectorstore._collection.count()} vectors")

textCopy
# Load existing vector store and query it
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="technical_docs",
)

# Similarity search — returns most relevant chunks
results = vectorstore.similarity_search(
    query="How do I configure authentication?",
    k=4,  # Return top 4 results
)
for doc in results:
    print(f"Source: {doc.metadata['source']}, Page: {doc.metadata.get('page', 'N/A')}")
    print(doc.page_content[:200])
    print("---")

# Similarity search with relevance scores
results_with_scores = vectorstore.similarity_search_with_relevance_scores(
    query="authentication configuration",
    k=4,
    score_threshold=0.7,  # Only return results above this similarity score
)
for doc, score in results_with_scores:
    print(f"Score: {score:.3f} | {doc.page_content[:100]}")

Part 6: Retrieval-Augmented Generation (RAG)

RAG is the most widely deployed LangChain pattern in production. It lets you ground LLM responses in your own documents — eliminating hallucination and keeping answers up to date.

6.1 Basic RAG Pipeline

textCopy
from langchain_openai import ChatOpenAI, 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

llm = ChatOpenAI(model="gpt-4o", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

# Create a retriever from the vector store
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5},
)

# RAG prompt — instructs the model to use only provided context
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful assistant that answers questions based ONLY on the provided context.
If the answer is not in the context, say "I don't have enough information to answer that."
Never make up information.

Context:
{context}"""),
    ("human", "{question}"),
])

def format_docs(docs):
    """Join retrieved document chunks into a single context string."""
    return "\n\n---\n\n".join(
        f"[Source: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
        for doc in docs
    )

# Full RAG chain
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("How do I set up authentication in this project?")
print(answer)

6.2 Advanced RAG with Source Citations

textCopy
from langchain_core.runnables import RunnableParallel

# Return both the answer AND the source documents
rag_chain_with_sources = RunnableParallel(
    {
        "answer": rag_chain,
        "sources": retriever,
    }
)

result = rag_chain_with_sources.invoke("What are the API rate limits?")

print("Answer:", result["answer"])
print("\nSources:")
for i, doc in enumerate(result["sources"], 1):
    source = doc.metadata.get("source", "unknown")
    page = doc.metadata.get("page", "N/A")
    print(f"  [{i}] {source} (page {page})")

6.3 Conversational RAG

textCopy
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

llm = ChatOpenAI(model="gpt-4o-mini")

# Step 1: Rephrase follow-up questions into standalone queries
# e.g., "What about its rate limits?" → "What are the OpenAI API rate limits?"
contextualize_q_prompt = ChatPromptTemplate.from_messages([
    ("system", """Given the chat history and the latest user question,
rephrase the question to be standalone (no references to chat history).
Return ONLY the rephrased question, nothing else."""),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
])

history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_q_prompt
)

# Step 2: Answer using retrieved context
qa_prompt = ChatPromptTemplate.from_messages([
    ("system", """Answer the question based only on the following context.
If you don't know, say so — don't make up answers.

Context:
{context}"""),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
])

question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)
rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

# Wrap with memory
store = {}
def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

conversational_rag = RunnableWithMessageHistory(
    rag_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="answer",
)

config = {"configurable": {"session_id": "rag-session-1"}}

# Multi-turn Q&A grounded in your documents
r1 = conversational_rag.invoke({"input": "What authentication methods does this API support?"}, config=config)
print("Turn 1:", r1["answer"])

r2 = conversational_rag.invoke({"input": "How do I implement the second one?"}, config=config)
print("Turn 2:", r2["answer"])  # "It" is correctly resolved to the right auth method

Vector database and embeddings architecture

Part 7: Tools and Agents

Agents take LLMs beyond question-answering. They give LLMs the ability to reason, decide which tools to use, call those tools, observe results, and iterate — autonomously.

7.1 Built-in Tools

textCopy
from langchain_community.tools import TavilySearchResults, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from langchain.tools import tool

# Tavily web search (requires TAVILY_API_KEY)
search_tool = TavilySearchResults(
    max_results=3,
    search_depth="advanced",
    include_raw_content=True,
)

results = search_tool.invoke("LangChain latest version 2026")
for r in results:
    print(r["title"])
    print(r["url"])
    print(r["content"][:200])
    print("---")

# Wikipedia search
wiki = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(top_k_results=2))
result = wiki.invoke("transformer architecture neural networks")
print(result[:500])

7.2 Custom Tools

textCopy
from langchain.tools import tool
from pydantic import BaseModel, Field
import requests

# Simple decorator-based tool
@tool
def get_current_weather(city: str) -> str:
    """Get the current weather for a given city. Use this when the user asks about weather."""
    # In production, call a real weather API
    return f"The current temperature in {city} is 22°C with partly cloudy skies."

@tool
def calculate_compound_interest(
    principal: float,
    rate: float,
    time_years: int,
    compounding_frequency: int = 12,
) -> str:
    """
    Calculate compound interest.
    
    Args:
        principal: Initial investment amount in USD
        rate: Annual interest rate as a decimal (e.g., 0.05 for 5%)
        time_years: Investment period in years
        compounding_frequency: Times interest compounds per year (default: monthly=12)
    """
    amount = principal * (1 + rate / compounding_frequency) ** (compounding_frequency * time_years)
    interest = amount - principal
    return f"Principal: ${principal:,.2f}, Final Amount: ${amount:,.2f}, Interest Earned: ${interest:,.2f}"

# Structured input tool with Pydantic schema
class DatabaseQueryInput(BaseModel):
    table: str = Field(description="Database table name to query")
    condition: str = Field(description="SQL WHERE clause condition")
    limit: int = Field(default=10, description="Maximum number of rows to return")

@tool(args_schema=DatabaseQueryInput)
def query_database(table: str, condition: str, limit: int = 10) -> str:
    """Query the internal database for records matching specific conditions."""
    # In production, this connects to your actual database
    return f"SELECT * FROM {table} WHERE {condition} LIMIT {limit} — returned 3 rows: [mock data]"

# Test tools independently
print(get_current_weather.invoke({"city": "Tokyo"}))
print(calculate_compound_interest.invoke({
    "principal": 10000,
    "rate": 0.07,
    "time_years": 10,
}))

7.3 ReAct Agents — Reasoning + Acting

textCopy
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

llm = ChatOpenAI(model="gpt-4o", temperature=0)

tools = [
    TavilySearchResults(max_results=3),
    get_current_weather,
    calculate_compound_interest,
    query_database,
]

# Pull the ReAct prompt from LangChain Hub
# Thought → Action → Observation → Thought → ... → Final Answer
react_prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm=llm, tools=tools, prompt=react_prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,         # Print the reasoning trace
    max_iterations=10,    # Prevent infinite loops
    handle_parsing_errors=True,
    return_intermediate_steps=True,
)

result = agent_executor.invoke({
    "input": "What is the weather in London right now, and if I invest $5000 at 6% for 5 years with monthly compounding, how much will I have?"
})

print("\nFinal Answer:", result["output"])
print("\nSteps taken:", len(result["intermediate_steps"]))

7.4 OpenAI Tools Agent (Recommended for Production)

textCopy
from langchain.agents import create_openai_tools_agent

# More reliable than ReAct for GPT-4o — uses native function calling
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful assistant with access to tools.
Use tools when needed. Think step by step before acting.
Always provide a clear final answer after using tools."""),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

agent = create_openai_tools_agent(llm=llm, tools=tools, prompt=prompt)

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=8,
    return_intermediate_steps=True,
)

# Multi-step reasoning task
result = executor.invoke({
    "input": "Search for the latest news about AI regulation in 2026, then summarize the key points.",
    "chat_history": [],
})

print(result["output"])

Part 8: Streaming — Real-Time Responses

Streaming delivers tokens as they are generated, making your app feel instant instead of making users wait.

textCopy
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o", streaming=True)

chain = (
    ChatPromptTemplate.from_template("Write a detailed explanation of {topic}.")
    | llm
    | StrOutputParser()
)

# Stream tokens as they arrive
print("Streaming response:")
for chunk in chain.stream({"topic": "how transformers work in deep learning"}):
    print(chunk, end="", flush=True)
print()  # Newline at end

# Async streaming — use in FastAPI or async frameworks
import asyncio

async def stream_async():
    async for chunk in chain.astream({"topic": "the future of edge computing"}):
        print(chunk, end="", flush=True)

asyncio.run(stream_async())

textCopy
# Streaming with agents — stream each step as it happens
async def stream_agent():
    async for event in executor.astream_events(
        {"input": "What are the top 3 AI startups in 2026?", "chat_history": []},
        version="v2",
    ):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            chunk = event["data"]["chunk"].content
            if chunk:
                print(chunk, end="", flush=True)
        elif kind == "on_tool_start":
            print(f"\n[Using tool: {event['name']}]\n")
        elif kind == "on_tool_end":
            print(f"[Tool complete]\n")

asyncio.run(stream_agent())

Part 9: Advanced Patterns

9.1 Multi-Agent Systems

textCopy
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Specialized sub-agents as tools
@tool
def research_agent_tool(query: str) -> str:
    """Research and gather information on a given topic using web search."""
    search = TavilySearchResults(max_results=5)
    results = search.invoke(query)
    return "\n".join(r["content"] for r in results)

@tool
def code_agent_tool(task: str) -> str:
    """Generate and review Python code for a given programming task."""
    code_llm = ChatOpenAI(model="gpt-4o", temperature=0)
    code_prompt = ChatPromptTemplate.from_template(
        "Write clean, production-ready Python code for: {task}\n"
        "Include docstrings, type hints, and error handling."
    )
    chain = code_prompt | code_llm | StrOutputParser()
    return chain.invoke({"task": task})

@tool
def review_agent_tool(code: str) -> str:
    """Review Python code for bugs, security issues, and best practices."""
    review_llm = ChatOpenAI(model="gpt-4o", temperature=0)
    review_prompt = ChatPromptTemplate.from_template(
        "Review this Python code for bugs, security issues, and best practices:\n\n```python\n{code}\n```"
    )
    chain = review_prompt | review_llm | StrOutputParser()
    return chain.invoke({"code": code})

# Orchestrator agent — delegates to specialists
orchestrator_tools = [research_agent_tool, code_agent_tool, review_agent_tool]

orchestrator_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an AI project orchestrator. You coordinate specialist agents:
- research_agent_tool: For gathering information from the web
- code_agent_tool: For writing Python code  
- review_agent_tool: For reviewing and critiquing code
Break complex tasks down and delegate to the right specialist."""),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

orchestrator = create_openai_tools_agent(llm, orchestrator_tools, orchestrator_prompt)
orchestrator_executor = AgentExecutor(agent=orchestrator, tools=orchestrator_tools, verbose=True)

result = orchestrator_executor.invoke({
    "input": "Research the best Python libraries for building REST APIs in 2026, then write a basic FastAPI example, and review it for issues."
})
print(result["output"])

9.2 LangGraph — Stateful Agent Workflows

LangGraph extends LangChain for complex, stateful agent workflows with cycles, branching, and human-in-the-loop.

textCopy
# pip install langgraph
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, List

llm = ChatOpenAI(model="gpt-4o-mini")

# Define the state that flows through the graph
class AgentState(TypedDict):
    question: str
    research: str
    draft: str
    feedback: str
    final_answer: str
    iteration: int

# Define nodes (each is a function that transforms state)
def research_node(state: AgentState) -> AgentState:
    """Search for relevant information."""
    search = TavilySearchResults(max_results=3)
    results = search.invoke(state["question"])
    research = "\n".join(r["content"][:300] for r in results)
    return {**state, "research": research}

def draft_node(state: AgentState) -> AgentState:
    """Write a draft answer based on research."""
    prompt = f"""Based on this research, write a comprehensive answer to: {state['question']}

Research:
{state['research']}

{'Previous feedback to address: ' + state['feedback'] if state['feedback'] else ''}"""
    
    draft = llm.invoke(prompt).content
    return {**state, "draft": draft, "iteration": state["iteration"] + 1}

def review_node(state: AgentState) -> AgentState:
    """Review the draft and provide feedback."""
    prompt = f"""Review this draft answer for accuracy, completeness, and clarity.
If it's good enough (score 8+/10), respond with "APPROVED".
Otherwise provide specific feedback.

Draft: {state['draft']}"""
    
    feedback = llm.invoke(prompt).content
    return {**state, "feedback": feedback}

def finalize_node(state: AgentState) -> AgentState:
    return {**state, "final_answer": state["draft"]}

def should_revise(state: AgentState) -> str:
    """Routing function: approve or request revision."""
    if "APPROVED" in state["feedback"] or state["iteration"] >= 3:
        return "finalize"
    return "draft"  # Revise

# Build the graph
workflow = StateGraph(AgentState)

workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.add_node("review", review_node)
workflow.add_node("finalize", finalize_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", "review")
workflow.add_conditional_edges("review", should_revise, {"draft": "draft", "finalize": "finalize"})
workflow.add_edge("finalize", END)

app = workflow.compile()

# Run the self-improving workflow
result = app.invoke({
    "question": "What are the most important AI safety considerations for 2026?",
    "research": "",
    "draft": "",
    "feedback": "",
    "final_answer": "",
    "iteration": 0,
})

print(f"Completed in {result['iteration']} iteration(s)")
print(result["final_answer"])

9.3 Evaluation and Testing

textCopy
from langchain.evaluation import load_evaluator
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# QA correctness evaluator
evaluator = load_evaluator("qa", llm=llm)

eval_result = evaluator.evaluate_strings(
    prediction="LangChain is a framework for building LLM-powered applications.",
    input="What is LangChain?",
    reference="LangChain is an open-source framework for building applications using large language models.",
)
print(f"Score: {eval_result['score']}")
print(f"Reasoning: {eval_result['reasoning']}")

# Criteria evaluator — check for specific qualities
criteria_evaluator = load_evaluator("criteria", llm=llm, criteria={
    "correctness": "Is the answer factually correct?",
    "conciseness": "Is the answer appropriately concise?",
    "helpfulness": "Is the answer genuinely helpful to the user?",
})

result = criteria_evaluator.evaluate_strings(
    prediction="Python is a programming language created in 1991 by Guido van Rossum.",
    input="Tell me about Python.",
)
print(result)

Part 10: Production Deployment

10.1 LangServe — Deploy Chains as REST APIs

textCopy
# pip install langserve[all] fastapi uvicorn

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="LangChain API",
    description="LLM-powered API built with LangChain and LangServe",
    version="1.0.0",
)

llm = ChatOpenAI(model="gpt-4o-mini")

# Chain 1: Code explainer
code_chain = (
    ChatPromptTemplate.from_template("Explain this code clearly:\n\n```\n{code}\n```")
    | llm
    | StrOutputParser()
)
add_routes(app, code_chain, path="/explain-code")

# Chain 2: RAG endpoint
rag_chain = rag_chain  # From Part 6
add_routes(app, rag_chain, path="/ask-docs")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Now you get automatic REST endpoints:
# POST /explain-code/invoke
# POST /explain-code/stream
# POST /explain-code/batch
# GET  /explain-code/playground  ← interactive UI

10.2 LangSmith — Observability and Tracing

textCopy
import os

# Enable LangSmith tracing — just set env vars
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-production-app"

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

# Every chain invocation gets:
# - Full input/output logs
# - Token usage and costs
# - Latency per step
# - Error traces with full context
# - Side-by-side comparison of different model versions

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini")
chain = ChatPromptTemplate.from_template("Explain {topic}") | llm | StrOutputParser()

# This invocation is automatically traced in LangSmith
result = chain.invoke({"topic": "neural networks"})

10.3 Caching for Cost Reduction

textCopy
from langchain_community.cache import InMemoryCache, RedisCache
from langchain.globals import set_llm_cache
import langchain

# In-memory cache — great for development
set_llm_cache(InMemoryCache())

# Redis cache — production-grade, shared across instances
from redis import Redis
set_llm_cache(RedisCache(redis_=Redis(host="localhost", port=6379)))

llm = ChatOpenAI(model="gpt-4o-mini")

# First call — hits the API, stores result in cache
result1 = llm.invoke("What is the capital of France?")
print("First call (API):", result1.content)

# Second identical call — returned from cache instantly, no API cost
result2 = llm.invoke("What is the capital of France?")
print("Second call (cache):", result2.content)

# For high-traffic apps, semantic caching avoids near-duplicate calls
from langchain_community.cache import GPTCache
# GPTCache uses embeddings to find semantically similar cached responses
# "What is the capital of France?" hits the cache for "Capital city of France?"

• Use LangSmith for tracing and monitoring from day one
• Implement LLM response caching to reduce costs 60-80%
• Set
  textCopy

  max_retries

  and
  textCopy

  timeout

  on all LLM clients
• Use
  textCopy

  max_iterations

  on all agents to prevent infinite loops
• Store all conversation history in a persistent backend (Redis, Postgres)
• Never expose raw LLM errors to end users — wrap with try/catch

Architecture Patterns at a Glance

What to Learn Next

LangChain is a deep ecosystem. Here is where to go after mastering these fundamentals:

• LangGraph — Build complex agent workflows with cycles, human-in-the-loop, and long-running tasks
• LangSmith — Set up evals, A/B testing between models, and production monitoring dashboards
• Advanced RAG — Hybrid search (dense + sparse), reranking with Cohere, HyDE query transformation
• Fine-tuning — When RAG is not enough, fine-tune with your domain data using OpenAI or Hugging Face
• Multi-modal — Vision agents that process images, PDFs with layout, and audio transcriptions

The patterns in this guide — chains, memory, tools, agents, RAG — are the foundation. Everything else is built on top of them. Once these click, the entire LLM application space opens up.