LangChain Cheat Sheet: Components, Patterns, and Production Rules

This cheat sheet is a practical reference for using LangChain in real systems. It is not a theory article. The goal is to help you make cleaner architectural choices, avoid common abstraction traps, and ship something that remains debuggable after the demo phase.

Installation

pip install langchain langchain-openai langchain-community langchain-chroma faiss-cpu tiktoken

1. LCEL — LangChain Expression Language

LCEL is the composition interface. Chains are built by piping (|) runnables together.

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

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise assistant."),
    ("human", "{question}")
])
chain = prompt | model | StrOutputParser()   # RunnableSequence
result = chain.invoke({"question": "What is RAG?"})

Runnable Types

Type	Use Case	Example
RunnableSequence	Serial pipeline A → B → C	prompt \| model \| parser
RunnableParallel	Run branches concurrently	{"a": chain_a, "b": chain_b}
RunnablePassthrough	Pass input unchanged	RunnablePassthrough()
RunnableLambda	Wrap any Python function	RunnableLambda(my_fn)
RunnableBranch	Conditional routing	RunnableBranch((cond, branch), default)

from langchain_core.runnables import (
    RunnableParallel, RunnablePassthrough,
    RunnableLambda, RunnableBranch
)

# Parallel: run two chains at the same time
parallel = RunnableParallel(
    question=RunnablePassthrough(),
    context=RunnableLambda(lambda x: retrieve(x["question"]))
)

# Branch: route based on a condition
branch = RunnableBranch(
    (lambda x: x["topic"] == "code",  code_chain),
    (lambda x: x["topic"] == "legal", legal_chain),
    default_chain                       # fallback
)

# Assign: add a key to the dict mid-chain
chain = (
    RunnablePassthrough.assign(
        context=RunnableLambda(lambda x: retrieve(x["question"]))
    )
    | prompt | model | parser
)

2. Prompt Templates

from langchain_core.prompts import (
    ChatPromptTemplate,
    PromptTemplate,
    FewShotChatMessagePromptTemplate,
    MessagesPlaceholder,
)

# Chat prompt with variables and history placeholder
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a {role}. Answer in {language}."),
    MessagesPlaceholder("chat_history"),   # inject list of messages
    ("human", "{question}")
])

# Few-shot chat prompt
examples = [
    {"input": "2+2",  "output": "4"},
    {"input": "3*3",  "output": "9"},
]
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"), ("ai", "{output}")
])
few_shot = FewShotChatMessagePromptTemplate(
    examples=examples,
    example_prompt=example_prompt
)
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a math tutor."),
    few_shot,
    ("human", "{input}")
])

# Partial: pre-fill some variables
prompt_with_role = prompt.partial(role="legal assistant", language="English")

3. Chat Models — Configuration & Methods

from langchain_openai import ChatOpenAI
from langchain_community.chat_models import ChatOllama

# OpenAI
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.2,        # 0 = deterministic, 2 = very creative
    max_tokens=1024,
    timeout=30,
    max_retries=3,
    streaming=True,
)

# Local via Ollama (no API key)
llm_local = ChatOllama(model="llama3.2", temperature=0)

# Invoke methods
response  = llm.invoke("Explain transformers")          # single call
responses = llm.batch(["What is RAG?", "What is LoRA?"]) # parallel batch
async_r   = await llm.ainvoke("Explain KV cache")       # async

# Streaming token by token
for chunk in llm.stream("Explain tokenization"):
    print(chunk.content, end="", flush=True)

# Bind: attach stop sequences or forced tool use
llm_with_stop = llm.bind(stop=["\n###", "END"])
llm_with_tools = llm.bind_tools(tools)   # forces tool-calling mode

4. Output Parsers

from langchain_core.output_parsers import (
    StrOutputParser,
    JsonOutputParser,
    CommaSeparatedListOutputParser,
)
from langchain.output_parsers import (
    PydanticOutputParser,
    RetryOutputParser,
    OutputFixingParser,
)
from langchain_core.pydantic_v1 import BaseModel, Field

# 1. String (default — returns raw text)
parser = StrOutputParser()

# 2. JSON
json_parser = JsonOutputParser()

# 3. Pydantic schema — typed, validated output
class Article(BaseModel):
    title:   str       = Field(description="Article title")
    summary: str       = Field(description="One-paragraph summary")
    tags:    list[str] = Field(description="List of topic tags")

pydantic_parser = PydanticOutputParser(pydantic_object=Article)

# Inject format instructions into the prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "Extract article metadata.\n{format_instructions}"),
    ("human",  "{text}")
]).partial(format_instructions=pydantic_parser.get_format_instructions())

chain = prompt | llm | pydantic_parser
article: Article = chain.invoke({"text": "..."})

# 4. Auto-fix: re-prompt on parse failure
fixing_parser = OutputFixingParser.from_llm(parser=pydantic_parser, llm=llm)

# 5. Retry: re-run the full chain on failure
retry_parser = RetryOutputParser.from_llm(parser=pydantic_parser, llm=llm)

5. Retrievers

from langchain_community.vectorstores import Chroma, FAISS
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import (
    MultiQueryRetriever,
    ContextualCompressionRetriever,
    EnsembleRetriever,
)
from langchain.retrievers.document_compressors import (
    LLMChainExtractor,
    CrossEncoderReranker,
)
from langchain_community.retrievers import BM25Retriever

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(docs, embeddings)

# Basic retriever — search_type options
retriever = vectorstore.as_retriever(
    search_type="mmr",          # "similarity" | "mmr" | "similarity_score_threshold"
    search_kwargs={
        "k": 5,                 # number of docs to return
        "fetch_k": 20,          # mmr candidate pool size
        "lambda_mult": 0.5,     # mmr: 0 = max diversity, 1 = max relevance
        "score_threshold": 0.7, # for similarity_score_threshold mode only
        "filter": {"source": "hr_docs"},  # metadata filter
    }
)

# Multi-query: generate N query variants, deduplicate, merge results
mq_retriever = MultiQueryRetriever.from_llm(
    retriever=retriever, llm=llm, include_original=True
)

# Contextual compression: re-rank and keep only relevant sentences
compressor = LLMChainExtractor.from_llm(llm)
compressed_retriever = ContextualCompressionRetriever(
    base_compressor=compressor, base_retriever=retriever
)

# Hybrid BM25 + vector search
bm25 = BM25Retriever.from_documents(docs, k=5)
ensemble = EnsembleRetriever(
    retrievers=[bm25, retriever],
    weights=[0.4, 0.6]          # keyword vs. semantic balance
)

6. Document Loaders & Text Splitters

from langchain_community.document_loaders import (
    PyPDFLoader, TextLoader, WebBaseLoader,
    UnstructuredMarkdownLoader, DirectoryLoader,
)
from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    TokenTextSplitter,
    MarkdownHeaderTextSplitter,
    CharacterTextSplitter,
)

# Loaders
pdf_docs  = PyPDFLoader("report.pdf").load()
web_docs  = WebBaseLoader("https://example.com").load()
dir_docs  = DirectoryLoader("./docs/", glob="**/*.md").load()

# RecursiveCharacterTextSplitter — recommended default
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,          # target chunk size in characters
    chunk_overlap=64,        # ~10–15% of chunk_size
    separators=["\n\n", "\n", ". ", " ", ""],  # split priority order
    length_function=len,
    add_start_index=True,    # adds chunk position as metadata
)
chunks = splitter.split_documents(pdf_docs)

# Token-based — use when API billing is token-based
token_splitter = TokenTextSplitter(
    chunk_size=256,
    chunk_overlap=32,
    encoding_name="cl100k_base",   # GPT-4 tokenizer
)

# Structure-aware — preserves headers as metadata
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("##", "section"),
        ("###", "subsection"),
    ],
    strip_headers=False,
)

7. Memory (Conversation History)

from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryMemory,
    ConversationBufferWindowMemory,
    ConversationSummaryBufferMemory,
)

# Buffer: keeps ALL messages — use for short conversations
memory = ConversationBufferMemory(
    return_messages=True,
    memory_key="chat_history",
    output_key="answer",       # which chain output key to save
)

# Window: keeps last k exchanges — good default
window_memory = ConversationBufferWindowMemory(k=5, return_messages=True)

# Summary: compresses history via LLM — use for long sessions
summary_memory = ConversationSummaryMemory(llm=llm, return_messages=True)

# SummaryBuffer: hybrid — buffer recent, summarize older
hybrid_memory = ConversationSummaryBufferMemory(
    llm=llm, max_token_limit=1000, return_messages=True
)

# Inject into a chain
from langchain_core.runnables import RunnablePassthrough
chain_with_history = (
    RunnablePassthrough.assign(
        chat_history=lambda x: memory.load_memory_variables({})["chat_history"]
    )
    | prompt | llm | parser
)

8. Tools & Agents

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import tool, StructuredTool
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from pydantic import BaseModel, Field

# Custom tool via decorator
@tool
def get_weather(city: str) -> str:
    """Get current weather for a city. Input must be a city name."""
    return f"Sunny, 22°C in {city}"   # replace with real API call

# Structured tool with typed input schema
class StockInput(BaseModel):
    ticker: str = Field(description="Stock ticker symbol, e.g. AAPL")
    period: str = Field(description="Period: '1d', '5d', '1mo'")

def get_stock_price(ticker: str, period: str) -> str:
    return f"{ticker}: $150 ({period})"

stock_tool = StructuredTool.from_function(
    func=get_stock_price,
    name="get_stock_price",
    description="Get historical stock price data",
    args_schema=StockInput,
)

tools = [get_weather, stock_tool, DuckDuckGoSearchRun()]

# Tool-calling agent (recommended — uses native function calling)
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,              # cap to prevent infinite loops
    max_execution_time=30,         # seconds
    handle_parsing_errors=True,    # recover from parse errors
    return_intermediate_steps=True # expose tool calls in output
)

result = executor.invoke({"input": "What is the weather in Paris?"})
print(result["output"])
print(result["intermediate_steps"])   # list of (AgentAction, observation)

9. Vector Store Operations

from langchain_community.vectorstores import Chroma, FAISS

# Create from documents
vs = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
    collection_name="my_collection",
)

# Add documents later
vs.add_documents(new_chunks)

# Similarity search with score
results = vs.similarity_search_with_score(query, k=5)
for doc, score in results:
    print(f"Score: {score:.3f} | {doc.page_content[:80]}")

# Persist and reload
vs.persist()
vs_loaded = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="my_collection",
)

# FAISS (in-memory, fast)
vs_faiss = FAISS.from_documents(chunks, embeddings)
vs_faiss.save_local("./faiss_index")
vs_faiss = FAISS.load_local("./faiss_index", embeddings)

10. Callbacks & Observability

from langchain.callbacks import StdOutCallbackHandler
from langchain.callbacks.base import BaseCallbackHandler

# Per-call verbose logging
chain.invoke({"question": "..."}, config={"callbacks": [StdOutCallbackHandler()]})

# Custom callback
class LatencyTracker(BaseCallbackHandler):
    def on_llm_start(self, *args, **kwargs):
        self.start = time.time()
    def on_llm_end(self, response, **kwargs):
        print(f"LLM latency: {time.time() - self.start:.2f}s")

# LangSmith tracing — set once, traces all chains automatically
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"]     = "ls__your_key"
os.environ["LANGCHAIN_PROJECT"]     = "my-project"

Component Decision Guide

Need	Use
Switch providers without rewriting	ChatOpenAI / ChatOllama shared interface
Repeatable, versioned prompt construction	ChatPromptTemplate
Type-safe machine-consumable outputs	PydanticOutputParser
Current or private domain knowledge	Retriever + VectorStore
Multiple query strategies	EnsembleRetriever or MultiQueryRetriever
Model must call external systems	@tool + create_tool_calling_agent
Recover from parse failures	OutputFixingParser
Trace and debug failures	LangSmith callbacks

Conclusion

LangChain is most useful as an orchestration layer, not as an architecture. This cheat sheet covers the components that appear most often in production systems: LCEL chains, prompt templates, output parsers, retrievers, agents, and observability hooks. Use it as a quick reference to avoid configuration errors and abstraction traps, and treat each component as a building block you control rather than a black box you trust.

---

11. RAG Pipeline — Complete Example

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 1. Ingest documents
loader   = PyPDFLoader("company_policy.pdf")
docs     = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64)
chunks   = splitter.split_documents(docs)

# 2. Create vector store
embeddings   = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore  = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db")
retriever    = vectorstore.as_retriever(search_type="mmr", search_kwargs={"k": 5, "fetch_k": 20})

# 3. Build RAG chain
RAG_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful assistant.
Answer based ONLY on the provided context.
If the context doesn't contain the answer, say: "I don't have information about that."

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

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

def format_docs(docs) -> str:
    return "\n\n".join(f"[{i+1}] {d.page_content}" for i, d in enumerate(docs))

rag_chain = (
    {
        "context":  retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | RAG_PROMPT
    | llm
    | parser
)

# 4. Invoke
answer = rag_chain.invoke("What is the remote work policy?")
print(answer)

---

12. Conversational RAG with Memory

from langchain.memory import ConversationSummaryBufferMemory
from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import HumanMessage

CONV_RAG_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful assistant with access to company documentation.
Use the context to answer factually. If uncertain, say so.

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

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=1000,
    memory_key="history",
    return_messages=False,
)

def load_memory(input_dict):
    return memory.load_memory_variables({})["history"]

conv_rag_chain = (
    RunnablePassthrough.assign(
        context  = lambda x: format_docs(retriever.invoke(x["question"])),
        history  = load_memory,
    )
    | CONV_RAG_PROMPT
    | llm
    | parser
)

# Multi-turn conversation
for q in ["What is the PTO policy?", "And for international employees?"]:
    response = conv_rag_chain.invoke({"question": q})
    memory.save_context({"input": q}, {"output": response})
    print(f"Q: {q}\nA: {response}\n")

---

13. Structured Extraction Chain

from pydantic import BaseModel, Field
from langchain.output_parsers import PydanticOutputParser
from typing import List, Optional

class ProductMention(BaseModel):
    product_name: str           = Field(description="Name of the product mentioned")
    sentiment:    str           = Field(description="positive, negative, or neutral")
    issues:       List[str]     = Field(description="List of issues mentioned if any")
    price_mentioned: Optional[float] = Field(default=None, description="Price if mentioned")

parser = PydanticOutputParser(pydantic_object=ProductMention)

extraction_prompt = ChatPromptTemplate.from_messages([
    ("system", "Extract product review information.\n{format_instructions}"),
    ("human",  "{review_text}"),
]).partial(format_instructions=parser.get_format_instructions())

extraction_chain = extraction_prompt | llm | parser

result: ProductMention = extraction_chain.invoke({
    "review_text": "The new iPhone 16 Pro is amazing but the battery life is terrible. Paid $1299."
})
print(result.product_name, result.sentiment, result.price_mentioned)
# → iPhone 16 Pro  negative  1299.0

---

14. Streaming and Async

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

llm    = ChatOpenAI(model="gpt-4o-mini", streaming=True)
parser = StrOutputParser()
chain  = prompt | llm | parser

# Synchronous streaming
for chunk in chain.stream({"question": "Explain attention mechanism"}):
    print(chunk, end="", flush=True)

# Async streaming
async def async_stream():
    async for chunk in chain.astream({"question": "Explain attention"}):
        print(chunk, end="", flush=True)
asyncio.run(async_stream())

# Batch async (parallel requests)
async def batch_async():
    questions = ["What is RAG?", "What is LoRA?", "What is RLHF?"]
    results   = await chain.abatch([{"question": q} for q in questions])
    for q, r in zip(questions, results):
        print(f"Q: {q}\nA: {r[:100]}...\n")
asyncio.run(batch_async())

---

15. Error Handling and Retry

from langchain_core.runnables import RunnableRetry
from langchain_core.exceptions import OutputParserException

# Add automatic retry with backoff
chain_with_retry = chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
    retry_if_exception_type=(OutputParserException, ValueError),
)

# Fallback to alternative model on failure
fallback_chain = (
    ChatOpenAI(model="gpt-4o")   | parser
)

chain_with_fallback = chain | fallback_chain.with_fallbacks(
    fallbacks=[fallback_chain],
    exceptions_to_handle=(Exception,),
)

# Configurable chain — override parameters per request
configurable_llm = ChatOpenAI(model="gpt-4o-mini").configurable_fields(
    model_name=ConfigurableField(id="model", name="Model", description="LLM model to use"),
    temperature=ConfigurableField(id="temperature"),
)

chain_configurable = prompt | configurable_llm | parser
# Override per request:
result = chain_configurable.invoke(
    {"question": "Explain RLHF"},
    config={"configurable": {"model": "gpt-4o", "temperature": 0.5}},
)

---

16. LangSmith Evaluation Integration

from langsmith import Client
from langsmith.evaluation import evaluate
import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"]    = "ls__your_key"
os.environ["LANGCHAIN_PROJECT"]    = "my-rag-project"

client = Client()

# Create evaluation dataset
dataset = client.create_dataset("rag-qa-v1")
client.create_examples(
    inputs=[
        {"question": "What is the PTO policy?"},
        {"question": "How do I request equipment?"},
    ],
    outputs=[
        {"answer": "15 days PTO per year"},
        {"answer": "Submit an IT request form"},
    ],
    dataset_id=dataset.id,
)

# Evaluator functions
def correctness_evaluator(run, example):
    score = llm.invoke(
        f"Score 0-1: does '{run.outputs['output']}' correctly answer '{example.inputs['question']}'? "
        f"Reference: '{example.outputs['answer']}'. Return only a number."
    ).content
    return {"key": "correctness", "score": float(score.strip())}

# Run evaluation experiment
results = evaluate(
    lambda inputs: {"output": rag_chain.invoke(inputs["question"])},
    data=dataset.name,
    evaluators=[correctness_evaluator],
    experiment_prefix="rag-v1-test",
    max_concurrency=3,
)
print(f"Mean correctness: {results.to_pandas()['feedback.correctness'].mean():.3f}")

---

Quick Reference: When to Use Each Component

Need	LangChain Component	Alternative
Simple LLM call	ChatOpenAI.invoke()	Direct OpenAI client
Composable pipelines	LCEL (\| operator)	Custom Python functions
Repeatable prompts	ChatPromptTemplate	F-strings (simple)
Type-safe outputs	PydanticOutputParser	instructor library
Document Q&A	Retriever + VectorStore	LlamaIndex
Multi-tool agents	create_tool_calling_agent	LangGraph (stateful)
Multi-turn with memory	ConversationSummaryBufferMemory	LangGraph state
Tracing + eval	LangSmith	Langfuse, Phoenix
Stateful complex flows	LangGraph	Direct StateGraph

---

LangChain Ecosystem Summary

LangChain’s value is not in any single component — it is in the composability:

Prompt → LLM → Parser  (simplest chain)
  ↓ add retriever
Retriever → Prompt → LLM → Parser  (RAG chain)
  ↓ add memory
Memory + Retriever → Prompt → LLM → Parser  (conversational RAG)
  ↓ add tools
Classifier → Tool Router → Tool → LLM Synthesizer  (agent)
  ↓ add state
→ LangGraph  (stateful multi-step agent)

Use LangChain as an orchestration toolkit, not as an architecture. The chains, retrievers, and output parsers are building blocks — your application logic and evaluation pipeline are what determine whether the system is reliable.

Core LCEL debugging commands:

# Visualize chain structure
print(chain.get_graph().draw_ascii())

# Validate input schema
print(chain.input_schema.schema())

# Run with full debug output
import langchain
langchain.debug = True
result = chain.invoke({...})

---

LangChain Conclusion

LangChain is most useful as an orchestration toolkit, not as an architecture. Its value is in composability—connecting prompts, models, retrievers, parsers, and tools into testable pipelines with a unified interface. The LCEL pipe operator makes chains readable and reusable. The retriever ecosystem abstracts over ten vector stores with the same API. The output parsers provide structured extraction with automatic retry. Use LangChain where its abstractions genuinely save time—and reach for LangGraph when you need stateful control flow, or direct SDK calls when you need minimal overhead. Always evaluate; always trace; always iterate with evidence.

LangChain Resources

• LangChain Documentation — Official LangChain Python docs
• LCEL Docs — LangChain Expression Language reference
• LangChain Hub — Community prompt templates
• LangSmith — Tracing, evaluation, and experiment management
• Cookbook — Production-ready examples
• LangChain YouTube — Official tutorials and deep dives

LangChain: Key Takeaways

• LCEL’s | operator produces composable, testable, observable chains
• Always prefer ChatPromptTemplate over f-strings for prompt management
• PydanticOutputParser with get_format_instructions() is the minimum for typed outputs
• EnsembleRetriever (BM25 + dense) outperforms pure vector search for most production use cases
• Use with_retry() for resilience; use .with_fallbacks() for multi-provider setups
• LangSmith tracing costs almost nothing to enable and is invaluable when you need it
• When the workflow needs branching, loops, or stateful memory: switch to LangGraph

---

LangChain Common Errors and Fixes

Error	Cause	Fix
OutputParserException	Model returned invalid JSON/schema	Use OutputFixingParser or retry with correction message
ContextWindowExceededError	Prompt too long	Reduce retrieval k; compress history; use smaller chunks
RateLimitError	Too many API calls	Add time.sleep(); use exponential backoff with with_retry()
KeyError in chain	Missing input variable	Check chain.input_schema for required keys
Callback not firing	Callbacks in wrong scope	Pass callbacks in config={"callbacks": [...]}
Memory not persisting	Wrong memory type	For multi-turn, use ConversationSummaryBufferMemory

# Inspect chain requirements
from langchain_core.runnables import RunnableSequence

print(chain.input_schema.schema())   # shows required input keys
print(chain.output_schema.schema())  # shows output structure

# Debug a chain step by step
for step_output in chain.steps:      # only for RunnableSequence
    print(type(step_output).__name__)

# Test with mock LLM
from langchain_core.language_models.fake import FakeListChatModel
fake = FakeListChatModel(responses=['{"sentiment": "positive"}'])
test_chain = prompt | fake | parser
assert test_chain.invoke({"text": "test"}).sentiment == "positive"

---

17. Multi-Provider Setup

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_community.chat_models import ChatOllama
from langchain_core.runnables import RunnableWithFallbacks

# Primary: OpenAI
primary_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Fallback 1: Anthropic
fallback_anthropic = ChatAnthropic(model="claude-3-haiku-20240307")

# Fallback 2: Local Ollama
fallback_local = ChatOllama(model="llama3.2:3b")

# Resilient chain: tries primary, then fallbacks
llm_with_fallback = primary_llm.with_fallbacks(
    [fallback_anthropic, fallback_local],
    exceptions_to_handle=(Exception,),
)

# All chains using llm_with_fallback automatically degrade gracefully
rag_chain = prompt | llm_with_fallback | parser

---

18. Caching LLM Responses

from langchain.globals import set_llm_cache
from langchain.cache import InMemoryCache, SQLiteCache, RedisCache

# In-memory cache (dev)
set_llm_cache(InMemoryCache())

# SQLite cache (single-process production)
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# Redis cache (distributed production)
import redis as redis_lib
set_llm_cache(RedisCache(redis_=redis_lib.Redis.from_url("redis://localhost:6379")))

# After setting cache, all LLM calls are automatically cached
llm = ChatOpenAI(model="gpt-4o-mini")
result1 = llm.invoke("What is RAG?")   # calls API
result2 = llm.invoke("What is RAG?")   # returns cached result

---

19. Debugging and Verbose Mode

from langchain.globals import set_debug, set_verbose

# Show all internal chain operations
set_debug(True)    # very verbose — shows every runnable execution
set_verbose(True)  # moderate — shows inputs/outputs of each step

# Per-chain verbose
result = chain.invoke({"question": "What is LLM?"}, config={"verbose": True})

# Inspect chain structure
print(chain.get_graph().draw_ascii())   # ASCII diagram of the chain

# Test with mock LLM (no API calls)
from langchain_core.language_models.fake import FakeListChatModel

fake_llm = FakeListChatModel(responses=["Paris", "42", "Machine learning"])
test_chain = prompt | fake_llm | parser
result     = test_chain.invoke({"question": "What is the capital of France?"})
assert result == "Paris"

---

20. Performance Best Practices

# 1. Always use async for concurrent requests
import asyncio
from langchain_openai import ChatOpenAI

async def batch_classify(texts: list[str]) -> list[str]:
    llm    = ChatOpenAI(model="gpt-4o-mini")
    tasks  = [llm.ainvoke(f"Classify: {t}") for t in texts]
    return [r.content for r in await asyncio.gather(*tasks)]

# 2. Use RunnableParallel for independent branches
from langchain_core.runnables import RunnableParallel

parallel_chain = RunnableParallel(
    sentiment = sentiment_chain,
    summary   = summary_chain,
    topics    = topics_chain,
)
# All three run concurrently:
result = parallel_chain.invoke({"text": long_article})

# 3. Batch process for throughput
llm     = ChatOpenAI(model="gpt-4o-mini")
results = llm.batch(
    [{"messages": [{"role": "user", "content": q}]} for q in questions],
    config={"max_concurrency": 10},
)

# 4. Stream for user-facing latency
async def stream_response(question: str):
    async for chunk in chain.astream({"question": question}):
        yield chunk   # send each token to client immediately

---

LangChain vs LangGraph vs Direct SDK Decision Guide

Use Case	Best Choice	Reason
Single LLM call	Direct SDK (OpenAI/Anthropic)	Lowest overhead
Prompt + parse pipeline	LangChain LCEL	Composable, testable
RAG application	LangChain	Retriever ecosystem
Simple agents	LangChain create_tool_calling_agent	Quick setup
Multi-turn with history	LangChain + memory	Built-in conversation management
Complex branching agent	LangGraph	Explicit state machine
Long-running background job	LangGraph + SqliteSaver	Checkpoint persistence
Multi-agent collaboration	LangGraph	Shared state architecture