LangChain Patterns That Scale

You've built chains. You've gotten LLMs to do useful things. Now comes the hard part: making those things reliable, affordable, and maintainable in production.

This document codifies our team's standards for writing LangChain code that scales. Every recommendation here is opinionated, actionable, and backed by real production experience. Treat this as a living standard — not a suggestion list.

---

Why Best Practices Matter

Before diving into patterns, understand the three forces that make discipline non-negotiable:

Force	What Happens Without Discipline	Cost of Failure
Cost	Unbounded token usage, redundant calls, no caching	$10s become $1000s per day
Reliability	Flaky chains, silent failures, no fallbacks	Users get broken experiences
Maintainability	Monolithic chains, inline prompts, no tests	Nobody touches the code after you leave

LangChain makes it easy to write code that works once. Our job is to write code that works every time, at scale, for months.

---

Modular Design

The single most important principle for maintainable LangChain code: compose small, focused chains instead of building monolithic ones.

Single Responsibility Per Component

Each chain component should do one thing. A prompt template structures input. A model generates output. A parser extracts structure. A lambda transforms data. When a component does two things, it becomes impossible to test, replace, or reason about.

Do this:

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

model = init_chat_model("gpt-4o")

# --- Each component has one job ---
classification_prompt = ChatPromptTemplate.from_messages([
    ("system", "Classify the intent of this customer message. Respond with one word: billing, technical, account, or general."),
    ("human", "{message}"),
])

response_prompt = ChatPromptTemplate.from_messages([
    ("system", "Draft a response to a {intent} inquiry. Be professional and concise."),
    ("human", "Customer message: {message}\nYour response:"),
])

def extract_intent(response: str) -> str:
    return response.strip().lower()

# --- Compose from focused pieces ---
classifier = classification_prompt | model | StrOutputParser() | RunnableLambda(extract_intent)
responder = response_prompt | model | StrOutputParser()

# Top-level chain wires them together
support_chain = classifier | {"intent": lambda x: x, "message": lambda x: x} | responder

Don't do this:

# Anti-pattern: one chain does everything
monolithic_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a support agent. First classify the intent of the "
     "customer message as billing, technical, account, or general. "
     "Then draft a professional response. Return both the classification "
     "and response as JSON with 'intent' and 'response' keys."),
    ("human", "{message}"),
])

# This couples classification logic into the response prompt,
# makes it impossible to swap the classifier, and bloats every call.
monolithic_chain = monolithic_prompt | model | json_parser

Composing Small Chains Over Monolithic Ones

Break complex workflows into named, testable stages. If you find yourself writing a chain that spans more than three pipe operators, extract intermediate steps:

# Good: named stages that are individually testable
def build_analysis_pipeline(model: ChatModel) -> Runnable:
    extract_entities = entity_prompt | model | entity_parser
    analyze_sentiment = sentiment_prompt | model | sentiment_parser
    generate_summary = summary_prompt | model | StrOutputParser()

    return (
        extract_entities
        | {"entities": lambda x: x, "text": lambda x: x}
        | analyze_sentiment
        | {"analysis": lambda x: x, "text": lambda x: x}
        | generate_summary
    )

Using RunnableLambda and RunnablePassthrough Wisely

These utilities are powerful but easily abused:

• RunnableLambda — use for data transformation (cleaning, formatting, routing logic). Never use it to hide business logic that should be a named function.
• RunnablePassthrough — use to inject additional context into a chain without losing existing data.

from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# Good: RunnablePassthrough enriches context
chain = (
    RunnablePassthrough.assign(
        current_date=lambda _: datetime.now().isoformat(),
        user_tier=lambda x: lookup_tier(x["user_id"]),
    )
    | prompt
    | model
    | StrOutputParser()
)

# Good: RunnableLambda wraps a simple transformation
chain = (
    RunnableLambda(lambda x: x.strip().lower())
    | prompt
    | model
)

# Bad: RunnableLambda hides complex logic
chain = (
    RunnableLambda(
        lambda x: some_complex_database_query(x)
        and then_formatting
        and then_validation
    )
    | prompt
    | model
)
# Fix: extract to a named function with tests

Separating Prompt Templates from Chain Logic

Prompt templates belong in their own module or file. Chain composition belongs in orchestration code. This separation lets you:

• A/B test prompts without touching chain logic.
• Version prompts independently.
• Review prompt changes in isolation.

prompts/
  classification.py
  response.py
  summarization.py

chains/
  support_chain.py
  analysis_chain.py

---

Avoiding Over-Engineering

LangChain is a toolkit, not a requirement. The best code is often the simplest code.

Don't Use LangChain When a Simple API Call Suffices

If your task is a single model call with no preprocessing, no postprocessing, and no composition, skip the framework:

# Simple task: just call the API directly
from openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Summarize this: " + text}],
)
summary = response.choices[0].message.content

# LangChain adds overhead here with no benefit.
# Use LangChain when you need: prompt templating, output parsing,
# chaining, tools, memory, or provider-agnostic abstractions.

Avoid Deep Nesting of Chains

Nesting chains more than two levels deep makes debugging nearly impossible. Flatten where you can:

# Bad: deeply nested chains obscure the data flow
outer = prompt_a | model_a | RunnableLambda(
    lambda x: inner_chain.invoke({"input": x})
) | model_b | parser

# Good: flat, explicit stages
stage_one = prompt_a | model_a | StrOutputParser()
stage_two = prompt_b | model_b | StrOutputParser()
flat_chain = stage_one | {"input": lambda x: x} | stage_two | parser

When to Reach for LangGraph

LangChain chains are great for linear, deterministic pipelines. When your workflow needs any of these, switch to LangGraph:

Your Need	Use
Linear pipeline (A → B → C)	LangChain LCEL chains
Conditional branching	LangGraph
Loops with state	LangGraph
Human-in-the-loop approval	LangGraph (with checkpointer)
Multi-agent coordination	LangGraph
Persistence across failures	LangGraph (with checkpointer)

# LangChain is fine for linear work
chain = prompt | model | parser

# LangGraph for stateful, looping workflows
from langgraph.graph import StateGraph, START, END

graph = StateGraph(AnalysisState)
graph.add_node("extract", extract_node)
graph.add_node("validate", validate_node)
graph.add_conditional_edges("validate", should_reroute)
graph.add_edge(START, "extract")
graph.add_edge("extract", "validate")
graph.add_edge("validate", END)
app = graph.compile()

The "Less Is More" Principle

Modern models are capable. You often don't need:

• Complex few-shot chains when a well-written system prompt suffices.
• Multiple sequential model calls when one call with a structured output schema works.
• Custom output parsers when with_structured_output() handles the job.

# Over-engineered: three calls to get structured data
classify_chain = classify_prompt | model | parser
detail_chain = detail_prompt | model | parser
combine_chain = combine_prompt | model | parser

# Simpler: one call with structured output
from pydantic import BaseModel, Field

class Analysis(BaseModel):
    category: str = Field(description="The category of the text")
    key_points: list[str] = Field(description="Main points extracted")
    summary: str = Field(description="One-sentence summary")

simple_chain = prompt | model.with_structured_output(Analysis)

---

Memory Management

Memory is not free. Every token in context costs money, adds latency, and can introduce noise. Choose the right memory strategy for your use case — and often, the right strategy is no memory at all.

Stateless Is the Default

Start stateless. Add memory only when you have a proven need for conversation history. Stateless chains are:

• Cheaper (no context bloat).
• Faster (smaller prompts).
• Easier to test and debug.
• Naturally parallelizable.

# Stateless: each call is independent and cheap
chain = prompt | model | parser
result = chain.invoke({"question": "What is the capital of France?"})

Buffer Memory vs. Buffer Window Memory

When you do need memory, choose the right type:

Memory Type	Best For	Token Impact
BufferMemory (full history)	Short conversations (< 5 turns) where every message matters	Grows unbounded — dangerous
BufferWindowMemory (last N messages)	Conversations where recent context is sufficient	Bounded by message count
Token-aware truncation	Long conversations with strict budget	Bounded by token count
Summarization	Very long conversations where themes matter more than details	Bounded by summary length

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

# Token-aware summarization: summarize when context gets too large
agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",  # Cheap model for summaries
            trigger=("tokens", 4000),  # Summarize at 4k tokens
            keep=("messages", 10),     # Keep last 10 messages intact
        ),
    ],
)

Token-Aware Memory Trucation

Never let context grow without bounds. Set explicit limits:

# Good: explicit token budget
agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",
            trigger=("fraction", 0.7),  # Summarize at 70% of context window
            keep=("fraction", 0.2),     # Keep 20% of context untouched
        ),
    ],
)

# Bad: no limits — context grows until you hit the model's max tokens
# and get a hard failure

Summarization-Based Memory for Long Conversations

For conversations that span many turns, summarization is the only viable strategy. Use a cheap model for summaries to keep costs down:

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",  # Use a cheaper model for summarization
            trigger=[
                ("tokens", 5000),       # OR: summarize at 5k tokens
                ("messages", 30),       # OR: summarize at 30 messages
            ],
            keep=("messages", 15),
        ),
    ],
)

When NOT to Use Memory

Skip memory entirely when:

• Each request is self-contained (Q&A, classification, extraction).
• You batch-process documents.
• Conversation continuity is not a product requirement.

# Stateless extraction: no memory needed
extraction_chain = (
    extraction_prompt
    | model.with_structured_output(ExtractionSchema)
)
results = [extraction_chain.invoke({"text": doc}) for doc in documents]

---

Error Handling and Retries

LLM APIs are unreliable. Rate limits, timeouts, and transient errors are normal, not exceptional. Your code must handle them gracefully.

Using with_retry for Automatic Retries

LangChain models retry automatically with exponential backoff. Configure retries explicitly — don't rely on defaults in production:

from langchain.chat_models import init_chat_model

# Production: explicit retry configuration
model = init_chat_model(
    "gpt-4o",
    max_retries=6,    # Default; increase for unreliable networks
    timeout=30,       # Seconds
)

# For long-running agent tasks on unreliable networks
resilient_model = init_chat_model(
    "gpt-4o",
    max_retries=10,
    timeout=120,
)

Model Retry Middleware

For agents, use the built-in retry middleware for fine-grained control:

from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        ModelRetryMiddleware(
            max_retries=4,
            backoff_factor=2.0,
            initial_delay=1.0,
            max_delay=60.0,
            jitter=True,
            on_failure="continue",  # Return error message, don't crash
        ),
    ],
)

Handling Rate Limits and API Errors

Different errors need different responses:

Error Type	Strategy
429 Rate Limit	Retry with exponential backoff + jitter
5xx Server Error	Retry with exponential backoff
401 Unauthorized	Fail fast — don't retry
Timeout	Retry once, then fail
Content filter	Handle in application logic

from langchain.agents.middleware import ModelRetryMiddleware

def should_retry(error: Exception) -> bool:
    # Only retry on transient errors
    if hasattr(error, "status_code"):
        return error.status_code in (429, 502, 503, 504)
    if isinstance(error, (TimeoutError, ConnectionError)):
        return True
    return False

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        ModelRetryMiddleware(
            max_retries=3,
            retry_on=should_retry,
            on_failure="continue",
        ),
    ],
)

Fallback Strategies

When the primary model is unavailable, fall back to an alternative. Use the built-in fallback middleware:

from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        ModelFallbackMiddleware(
            "gpt-4o-mini",           # First fallback: cheaper, faster
            "claude-sonnet-4-6",     # Second fallback: different provider
        ),
    ],
)

Graceful Degradation Patterns

Design chains that can operate at reduced capability rather than failing completely:

from langchain_core.runnables import RunnableWithFallbacks

# Primary: structured output from GPT-4o
primary = model.with_structured_output(AnalysisSchema)

# Fallback: parse free-text from a cheaper model
fallback_chain = cheaper_model | StrOutputParser() | RunnableLambda(parse_analysis)

resilient_chain = RunnableWithFallbacks(
    runnable=primary,
    fallbacks=[fallback_chain],
)

Tool Retry Middleware

External tool calls (APIs, databases) are also unreliable. Add retry logic at the tool level:

from langchain.agents.middleware import ToolRetryMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, database_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,
            initial_delay=1.0,
            retry_on=(ConnectionError, TimeoutError),
            on_failure="return_message",  # Let the LLM handle the failure
        ),
    ],
)

---

Prompt Management

Prompts are not code, but they deserve code-level discipline. Treat prompts as first-class artifacts with versioning, testing, and review processes.

Externalizing Prompts to Files

Never inline prompt strings in chain code. Store them in dedicated files:

prompts/
  __init__.py
  classification.py
  response.py
  summarization.py

# prompts/classification.py
INTENT_CLASSIFICATION_SYSTEM = """\
You are an intent classifier. Classify the user message into one of these categories:
- billing: payment issues, invoices, refunds
- technical: bugs, errors, feature requests
- account: login, profile, subscription changes
- general: greetings, feedback, other

Respond with exactly one category name, lowercase.
"""

INTENT_CLASSIFICATION_HUMAN = "Classify: {message}"

# chains/support_chain.py — imports prompt, composes chain
from prompts.classification import INTENT_CLASSIFICATION_SYSTEM, INTENT_CLASSIFICATION_HUMAN
from langchain_core.prompts import ChatPromptTemplate

classification_prompt = ChatPromptTemplate.from_messages([
    ("system", INTENT_CLASSIFICATION_SYSTEM),
    ("human", INTENT_CLASSIFICATION_HUMAN),
])

Versioning Prompts

Track prompt changes with explicit version markers. When you change a prompt, bump the version:

# prompts/classification.py
PROMPT_VERSION = "2.1"  # Bump on every change

INTENT_CLASSIFICATION_SYSTEM_V2_1 = """\
You are an intent classifier. Classify the user message into one of these categories:
- billing: payment issues, invoices, refunds, pricing questions
- technical: bugs, errors, feature requests, integration issues
- account: login, profile, subscription changes, password reset
- general: greetings, feedback, complaints, other

Respond with exactly one category name, lowercase.
"""
# Change log:
# v2.1 — Added "pricing questions" to billing, "integration issues" to technical
# v2.0 — Rewrote categories for better distinction between account and technical
# v1.0 — Initial version

A/B Testing Prompts

When evaluating prompt changes, test them side by side before committing:

from langchain_core.prompts import ChatPromptTemplate
from langchain.chat_models import init_chat_model

# Two prompt variants
prompt_v1 = ChatPromptTemplate.from_messages([("system", SYSTEM_V1), ("human", "{text}")])
prompt_v2 = ChatPromptTemplate.from_messages([("system", SYSTEM_V2), ("human", "{text}")])

model = init_chat_model("gpt-4o")

# Run both on the same inputs and compare
for test_input in evaluation_set:
    result_v1 = (prompt_v1 | model | parser).invoke({"text": test_input})
    result_v2 = (prompt_v2 | model | parser).invoke({"text": test_input})
    compare_results(result_v1, result_v2, test_input.expected)

Using Prompt Hubs or Centralized Prompt Stores

For teams with many prompts, consider a centralized store:

Approach	Best For
Local files (our default)	Small teams, single repo
LangSmith Prompt Hub	Teams already using LangSmith, need collaboration
Database-backed store	Large teams, multi-project sharing
Feature flags + prompts	Gradual rollouts, canary testing

# Example: environment-aware prompt loading
import os

ENV = os.getenv("PROMPT_ENV", "production")

if ENV == "experiment":
    system_prompt = EXPERIMENTAL_SYSTEM_PROMPT
else:
    system_prompt = PRODUCTION_SYSTEM_PROMPT

---

Performance and Cost Optimization

Every token costs money. Every API call adds latency. Optimize aggressively.

Caching Responses

Cache repeated responses to eliminate redundant API calls:

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache
from langchain_openai import ChatOpenAI

# In-memory cache (good for development and single-process apps)
set_llm_cache(InMemoryCache())

# For production, use a persistent cache backend
# from langchain_community.cache import RedisCache, SQLiteCache
# set_llm_cache(RedisCache(redis_url="redis://localhost:6379"))

model = ChatOpenAI(model="gpt-4o")
# First call: hits the API
result1 = model.invoke("What is the capital of France?")
# Second call: returns cached result instantly
result2 = model.invoke("What is the capital of France?")

Streaming Responses

For user-facing applications, stream responses to reduce perceived latency:

from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-4o")

# Stream tokens as they arrive
for chunk in model.stream("Explain quantum computing in simple terms."):
    print(chunk.text, end="", flush=True)

# For agents, use astream_events for granular control
async for event in agent.astream_events({"messages": [...]}):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].text, end="", flush=True)

Parallel Execution

When processing independent inputs, run in parallel:

from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-4o")

# Batch: process multiple inputs in parallel
questions = [
    "What is machine learning?",
    "Explain neural networks.",
    "What is reinforcement learning?",
]

responses = model.batch(questions, config={"max_concurrency": 5})
for question, response in zip(questions, responses):
    print(f"Q: {question}\nA: {response.text}\n")

# For streaming batch results as they complete
for response in model.batch_as_completed(questions):
    print(response)

Monitoring Token Usage

Track token consumption to identify cost hotspots:

from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-4o")

# Use include_raw to access token counts
structured_model = model.with_structured_output(AnalysisSchema, include_raw=True)
result = structured_model.invoke("Analyze this text: ...")

# Access token usage from the raw response
raw_message = result["raw"]
print(f"Input tokens: {raw_message.usage_metadata['input_tokens']}")
print(f"Output tokens: {raw_message.usage_metadata['output_tokens']}")
print(f"Total tokens: {raw_message.usage_metadata['total_tokens']}")

Cost Optimization Checklist

• Use smaller/cheaper models for summarization and intermediate steps.
• Cache repeated or deterministic responses.
• Set max_tokens to prevent runaway generation.
• Use ModelCallLimitMiddleware to cap API calls per conversation.
• Monitor token usage with LangSmith tracing.
• Prefer structured output over post-processing (fewer round trips).

# Example: cheap model for intermediate work, expensive model for final output
summary_model = init_chat_model("gpt-4o-mini")   # Cheap
primary_model = init_chat_model("gpt-4o")        # Capable

chain = (
    summary_prompt | summary_model | StrOutputParser()  # Cheap summarization
    | {"summary": lambda x: x}
    | analysis_prompt | primary_model | analysis_parser  # Capable analysis
)

---

Testing LangChain Code

Untested LangChain code is a production time bomb. Here is how to test effectively.

Mocking LLM Calls

Never call real models in tests. Mock them:

from unittest.mock import MagicMock
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage

def mock_model():
    """Return a mock chat model for testing."""
    model = MagicMock(spec=BaseChatModel)
    model.invoke.return_value = AIMessage(content="Mocked response")
    return model

# Usage in tests
def test_classification_chain():
    model = mock_model()
    chain = classification_prompt | model | StrOutputParser()
    result = chain.invoke({"message": "My bill is wrong"})
    assert model.invoke.called

Snapshot Testing for Prompts

Verify that prompt templates produce the expected message structure:

def test_classification_prompt_structure():
    messages = classification_prompt.invoke({"message": "I need help"})
    assert len(messages) == 2
    assert messages[0].role == "system"
    assert messages[1].role == "human"
    assert "I need help" in messages[1].content

Unit Testing Individual Chain Components

Test each component in isolation before testing the full chain:

def test_entity_extraction_parser():
    parser = entity_parser
    raw_response = '{"entities": ["John", "New York"], "types": ["person", "location"]}'
    result = parser.parse(raw_response)
    assert result.entities == ["John", "New York"]
    assert result.types == ["person", "location"]

def test_preprocessing_lambda():
    preprocess = RunnableLambda(lambda x: x.strip().lower())
    assert preprocess.invoke("  HELLO World  ") == "hello world"

Integration Testing with Fake Models

For end-to-end chain tests, use LangChain's FakeListChatModel:

from langchain_community.chat_models import FakeListChatModel

def test_full_support_chain():
    fake_model = FakeListChatModel(responses=["billing", "We understand your concern..."])
    chain = build_support_chain(fake_model)
    result = chain.invoke({"message": "My charge is wrong"})
    assert "billing" in result.lower() or "charge" in result.lower()

Testing Checklist

• Every prompt template has a structure test.
• Every output parser has a parse test with valid and invalid input.
• Every chain has a mocked integration test.
• No test calls a real LLM API.
• Error paths are tested (model failure, parse failure, timeout).

---

Summary

Area	Rule
Modular Design	One responsibility per component. Compose small chains.
Over-Engineering	Start simple. Use LangChain only when it adds value.
Memory	Stateless by default. Bounded context always.
Error Handling	Explicit retries. Fallback models. Graceful degradation.
Prompts	Externalized. Versioned. Tested.
Performance	Cache what you can. Stream to users. Parallelize independent work.
Testing	Mock LLMs. Test prompts, parsers, and chains in isolation.

Follow these patterns and your LangChain code will be reliable, affordable, and maintainable. Break them only with a documented reason.

---

Further Reading

• LangChain Official Docs — stay current with API changes.
• LangGraph Docs — when you need stateful, looping workflows.
• LangSmith Observability — trace, debug, and evaluate in production.
• Getting Started — our team's introduction to LangChain fundamentals.