LangChain Code Snippets: 25+ Production Python Patterns

Q: How do you enable token streaming in LCEL chains?

Developers enable streaming by replacing synchronous .invoke() calls with the .stream() async generator method. This pushes partial token chunks to clients immediately, reducing time-to-first-token by up to 65% across web applications.

Building generative AI applications with Python requires composable chains, reliable error boundaries, and low-latency token streaming. The LangChain 0.3 release refactors monolithic dependencies into focused packages while standardizing on LangChain Expression Language (LCEL) for declarative runtime orchestration.

Key Takeaways

LangChain 0.3 modularizes integrations into dedicated packages reducing runtime dependency overhead by 35%.

LCEL pipe syntax (prompt | model | parser) enables native async streaming with up to 65% faster time-to-first-token.

Model fallbacks and retry mechanisms provide zero-downtime resiliency against provider API rate limits.

Stateful and cyclical workflows migrate from legacy AgentExecutor to LangGraph graphs for durable state persistence.

Engineers deploy LangChain across diverse architectures including chatbots, RAG pipelines, and multi-agent workflows. This curated collection gathers production-tested patterns that work seamlessly with LangChain 0.3 in Python environments.

Twenty-five code snippets provide the foundation for production implementation. We examine core primitives across LCEL composition, vector retrieval, structured outputs, and operational tracing.

Before You Start: Version Notes

These snippets target the LangChain v0.3 Official Documentation. For engineering teams upgrading from earlier versions, key architectural shifts include:

Imports moved: from langchain.llms import OpenAI is replaced by from langchain_openai import ChatOpenAI
invoke() is standard: The legacy __call__ pattern is deprecated in favor of chain.invoke() and chain.ainvoke()
Provider packages separated: Provider SDKs are decoupled into langchain-openai and langchain-anthropic

When starting fresh, dependency management remains lightweight. If you are upgrading existing codebases, inspect your package imports first: that is where most breaking changes reside.

Minimum installs for these examples:

pip install langchain langchain-core langchain-openai langchain-community

Setup & Models

Connecting LangChain to frontier language model providers requires standardizing on partner SDK packages. For developers comparing low-level requests against framework runtimes, our companion guide on OpenAI API code snippets details raw HTTP request payloads.

Snippet 1: Install and Basic Setup

# pip install langchain langchain-openai langchain-anthropic

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# Initialize models
gpt = ChatOpenAI(model="gpt-5.6-sol")  # Reads OPENAI_API_KEY
claude = ChatAnthropic(model="claude-sonnet-5")  # Reads ANTHROPIC_API_KEY

LangChain 0.3 uses separate packages for each provider. Installing only required adapters keeps container footprints minimal.

Why separate packages matter. In earlier LangChain releases, installing the base package pulled in heavy dependencies for every provider. The modern architecture keeps dependencies lean, ensuring an OpenAI project does not require Anthropic libraries.

Environment variable conventions. LangChain adheres to standard environment keys: OPENAI_API_KEY, ANTHROPIC_API_KEY, and GOOGLE_API_KEY. You can also pass keys explicitly during constructor instantiation.

Async support. Every model in LangChain supports synchronous and asynchronous calls. While snippets illustrate synchronous invoke(), production web backends should substitute ainvoke() to avoid blocking the asyncio event loop.

Snippet 2: Model Configuration

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-5.6-sol",
    temperature=0.7,
    max_tokens=2000,
    timeout=30,
    max_retries=2,
    api_key="sk-..."  # Optional override
)

# Test execution
response = llm.invoke("Say hello")
print(response.content)

The invoke() method provides the standard entry point for language models in LangChain 0.3.

Snippet 3: Using Open Source Models (Ollama)

from langchain_community.llms import Ollama

# Requires Ollama running locally
llm = Ollama(model="llama3.3")

response = llm.invoke("What is machine learning?")
print(response)

Local models through Ollama provide a private, cost-effective development environment.

Snippet 4: Model Fallback Chain

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# Primary model with fallback
primary = ChatOpenAI(model="gpt-5.6-sol")
fallback = ChatAnthropic(model="claude-sonnet-5")

llm = primary.with_fallbacks([fallback])

# If primary fails with rate limits, automatically tries fallback
response = llm.invoke("Hello!")

Fallback chains safeguard application uptime when primary providers experience regional degradation or rate limits.

Prompt Templates

Structured prompts ensure prompt variables remain clean, maintainable, and version-controlled.

Snippet 5: Basic Prompt Template

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant specialized in {topic}."),
    ("human", "{question}")
])

# Format with variables
formatted = prompt.invoke({
    "topic": "Python programming",
    "question": "How do I read a CSV file?"
})

# Use with model
response = llm.invoke(formatted)

Templates decouple static system instructions from dynamic user inputs.

Snippet 6: Few-Shot Prompting

from langchain_core.prompts import FewShotChatMessagePromptTemplate, ChatPromptTemplate

examples = [
    {"input": "2+2", "output": "4"},
    {"input": "5*3", "output": "15"},
]

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 calculator. Only output the number."),
    few_shot_prompt,
    ("human", "{input}")
])

response = llm.invoke(final_prompt.invoke({"input": "10/2"}))

Providing verified in-context examples enforces deterministic output formatting.

Snippet 7: Dynamic Example Selection

from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

examples = [
    {"input": "happy", "output": "sad"},
    {"input": "tall", "output": "short"},
    {"input": "bright", "output": "dark"},
]

selector = SemanticSimilarityExampleSelector.from_examples(
    examples,
    OpenAIEmbeddings(),
    FAISS,
    k=1  # Select most similar example
)

# Dynamically picks relevant example
relevant = selector.select_examples({"input": "joyful"})

Semantic selectors pick contextual examples tailored to user input query vectors.

Snippet 8: Prompt from File

from langchain_core.prompts import ChatPromptTemplate

with open("prompts/system.txt", "r") as f:
    system_prompt = f.read()

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}")
])

Persisting prompt strings in external files simplifies code reviews and semantic versioning.

LangChain Expression Language (LCEL)

LCEL provides a declarative pipeline interface that replaces legacy LLMChain classes. As documented in the LangChain GitHub Repository, runnables natively implement Python async protocols, reducing time-to-first-token by up to 65% when streaming responses.

Snippet 9: The Pipe Operator Explained

In standard Python, a | b signifies bitwise OR. In LCEL, it passes the output of runnable a directly into runnable b.

# The LCEL Way
chain = prompt | llm | parser
result = chain.invoke("topic")

Under the hood, components implement the __or__ method. The execution flow processes inputs linearly:

Dictionary input passes into prompt
PromptValue passes into llm
ChatMessage passes into parser
String payload returns to caller

Snippet 10: Runnables and Coercion

from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

def length_function(text):
    return len(text)

# Functions become chain steps
chain = prompt | llm | StrOutputParser() | length_function

# Dictionaries execute in parallel
map_chain = RunnableParallel({
    "original": RunnablePassthrough(),
    "length": length_function
})

In architectural debates regarding agno vs langchain, Agno emphasizes minimal abstraction overhead while LangChain provides extensive multi-provider connector catalogs and composable runnables.

Snippet 11: Binding Runtime Arguments

# Bind stop sequences directly to the model
model_with_stop = llm.bind(stop=["\nEND"])

chain = prompt | model_with_stop | parser
chain.invoke({"input": "Write a poem"})

Runtime argument binding attaches fixed parameters to models without altering shared base instances.

Chains

Chains combine prompts, models, and parsers into functional workflows.

Snippet 12: Simple Chain with LCEL

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

prompt = ChatPromptTemplate.from_template("Tell me a technical fact about {topic}")
model = ChatOpenAI(model="gpt-5.6-sol")
output_parser = StrOutputParser()

chain = prompt | model | output_parser
result = chain.invoke({"topic": "distributed databases"})
print(result)

LCEL chains provide clear readability and built-in type inference across inputs.

Snippet 13: Sequential Chain

outline_prompt = ChatPromptTemplate.from_template("Create an outline for: {topic}")
article_prompt = ChatPromptTemplate.from_template("Draft an article based on this outline:\n{outline}")

llm = ChatOpenAI(model="gpt-5.6-sol")
parser = StrOutputParser()

chain = (
    {"outline": outline_prompt | llm | parser}
    | article_prompt
    | llm
    | parser
)

article = chain.invoke({"topic": "AI in healthcare"})

Composing sequential chains allows outputs from earlier steps to feed subsequent prompt templates.

Snippet 14: Parallel Chains

from langchain_core.runnables import RunnableParallel

parallel = RunnableParallel(
    summary=ChatPromptTemplate.from_template("Summarize: {text}") | llm | parser,
    keywords=ChatPromptTemplate.from_template("Extract 5 keywords: {text}") | llm | parser,
    sentiment=ChatPromptTemplate.from_template("Analyze sentiment: {text}") | llm | parser
)

result = parallel.invoke({"text": "Production telemetry records demonstrate zero dropped packets."})

RunnableParallel executes multiple model calls concurrently across thread pools.

Snippet 15: Conditional Chain

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "code" in x["topic"].lower(), 
     ChatPromptTemplate.from_template("Write code for: {topic}") | llm | parser),
    (lambda x: "explain" in x["topic"].lower(),
     ChatPromptTemplate.from_template("Explain simply: {topic}") | llm | parser),
    ChatPromptTemplate.from_template("Answer: {topic}") | llm | parser
)

Conditional branches direct execution dynamically based on incoming input characteristics.

RAG & Retrieval

Retrieval-Augmented Generation represents a primary use case for enterprise LLMs. For developers seeking an end-to-end walkthrough beyond these snippets, our langchain rag tutorial workflows in the complete RAG chatbot tutorial provide full vector database integration details. For ready-to-use driver syntax across Pinecone, Weaviate, and ChromaDB, see our vector database code snippets.

Snippet 16: Basic RAG Setup

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

texts = [
    "LangChain is a framework for building AI applications.",
    "RAG stands for Retrieval-Augmented Generation.",
    "Vector databases store embeddings for semantic search."
]
vectorstore = FAISS.from_texts(texts, OpenAIEmbeddings())
retriever = vectorstore.as_retriever()

template = """Answer based on the context below:
Context: {context}
Question: {question}"""
prompt = ChatPromptTemplate.from_template(template)

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-5.6-sol")
    | StrOutputParser()
)

answer = chain.invoke("What is LangChain?")

This represents the core RAG pattern: retrieve relevant documents and inject them into the prompt.

Understanding retrieval quality. The most common RAG failure does not stem from the LLM; it stems from retrieval quality. If the retriever returns irrelevant documents, the LLM has no choice but to generate incorrect answers or hallucinate. Always evaluate your retriever separately from your full RAG chain.

Retrieval tuning strategies:

Adjust k (the number of retrieved documents): higher counts introduce token noise
Experiment with domain-specific embedding models to improve semantic resolution
Add metadata filtering to narrow search spaces before vector distance calculations
Implement hybrid search combining sparse keyword indexes with dense vector embeddings

Snippet 17: Document Loading

from langchain_community.document_loaders import TextLoader, PDFLoader, WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

text_docs = TextLoader("document.txt").load()
pdf_docs = PDFLoader("document.pdf").load()

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

Chunk documents appropriately: 1,000 characters with 200-character overlap provides a reliable default.

Snippet 18: Conversational RAG

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

chain = ConversationalRetrievalChain.from_llm(
    llm=ChatOpenAI(model="gpt-5.6-sol"),
    retriever=vectorstore.as_retriever(),
    memory=memory
)

Conversational RAG chains query vector indexes while maintaining conversation history context.

Snippet 19: Hybrid Search (Dense + Sparse)

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
bm25_retriever = BM25Retriever.from_texts(texts)
bm25_retriever.k = 3

ensemble = EnsembleRetriever(
    retrievers=[dense_retriever, bm25_retriever],
    weights=[0.6, 0.4]
)

Hybrid search combines keyword precision with semantic embeddings to prevent search misses.

Agents & Tools

Constructing a langchain python agent in LangChain 0.3 combines tool binding with structured reasoning loops. For a deep architectural comparison of single-loop executors versus cyclic graph state machines, review our LangChain agents tutorial.

Snippet 20: Simple ReAct Agent

from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.tools import DuckDuckGoSearchRun

tools = [DuckDuckGoSearchRun()]

prompt = ChatPromptTemplate.from_template("""Answer the following question:
{input}
Tools: {tools}
Tool names: {tool_names}
{agent_scratchpad}""")

agent = create_react_agent(llm=ChatOpenAI(model="gpt-5.6-sol"), tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "What is the capital of France?"})

ReAct agents reason through iterative thought, action, and observation cycles.

Snippet 21: Custom Tool

from langchain.tools import tool

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

tools = [calculator]

The @tool decorator automatically parses Python docstrings into model tool schemas.

Snippet 22: Agent with Memory

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
agent = create_openai_functions_agent(llm=ChatOpenAI(model="gpt-5.6-sol"), tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, memory=memory, verbose=True)

Agents maintain conversational coherence across multi-turn interactions using conversation buffers.

Snippet 23: Structured Output Tools

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="Search query")
    max_results: int = Field(default=5, description="Max results")

search_tool = StructuredTool.from_function(
    func=lambda query, max_results=5: f"Results for: {query}",
    name="search",
    description="Search database",
    args_schema=SearchInput
)

Pydantic schemas enforce type validation on model-generated tool arguments.

Memory Patterns for Conversations

Memory utilities preserve state across asynchronous dialogue turns.

Snippet 24: Conversation Buffer Memory

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory()
memory.save_context({"input": "Hi"}, {"output": "Hello!"})
history = memory.load_memory_variables({})

Buffer memory persists raw dialogue turns without summarization.

Snippet 25: Summary Memory

from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI

summary_memory = ConversationSummaryMemory(llm=ChatOpenAI(model="gpt-5.6-sol"))

Summary memory condenses long dialogues to fit within model context windows.

Snippet 26: Token Buffer Memory

from langchain.memory import ConversationTokenBufferMemory

token_memory = ConversationTokenBufferMemory(llm=ChatOpenAI(model="gpt-5.6-sol"), max_token_limit=1000)

Token buffer memory evicts oldest turns once token thresholds are reached.

Snippet 27: Entity Memory

from langchain.memory import ConversationEntityMemory

entity_memory = ConversationEntityMemory(llm=ChatOpenAI(model="gpt-5.6-sol"))

Entity memory builds key-value dictionaries of recognized entities.

Snippet 28: Vector Store Memory

from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

vectorstore = FAISS.from_texts(["Initial context"], OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs=dict(k=1))
memory = VectorStoreRetrieverMemory(retriever=retriever)

Vector store memory indexes past conversations for semantic retrieval. To master semantic distance and embedding dimension trade-offs, explore our guide to vector embeddings.

Output Parsing

Output parsers convert raw model string responses into validated Python dictionaries and lists.

Snippet 29: JSON Output Parser

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class ResponseFormat(BaseModel):
    answer: str = Field(description="Direct answer")
    confidence: float = Field(description="Confidence score 0-1")

parser = JsonOutputParser(pydantic_object=ResponseFormat)
prompt = ChatPromptTemplate.from_template("Answer {query}.\n{format_instructions}")
chain = prompt | llm | parser

Pydantic JSON parsers guarantee structured payloads for downstream API consumption.

Snippet 30: List Output Parser

from langchain_core.output_parsers import CommaSeparatedListOutputParser

list_parser = CommaSeparatedListOutputParser()
chain = prompt | llm | list_parser

Comma-separated parsers transform delimited model strings into typed Python lists.

LangSmith Debugging and Tracing

Observability instrumentation captures latency profiles and token expenses across production chains.

Snippet 31: Configuring LangSmith

export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="ls__..."
export LANGCHAIN_PROJECT="production-app"

Configuration requires zero code modifications: export the standard environment variables to stream telemetry directly to LangSmith dashboards.

Snippet 32: Evaluation with LangSmith

from langsmith import Client

client = Client()
# View logged traces and evaluation datasets

LangSmith traces inputs, model hyperparameters, token counts, and execution latency across distributed runnables.

Production Deployment Patterns

Deploying reliable applications requires handling edge cases, streaming tokens, and runtime user contexts.

Snippet 33: Streaming with Timeout

import asyncio

async def stream_response(chain, input_data):
    try:
        async for chunk in chain.astream(input_data):
            print(chunk, end="", flush=True)
    except asyncio.TimeoutError:
        print("\n[Error: Stream timed out]")

Async generator streaming delivers continuous token outputs to frontend clients without blocking backend processes.

Snippet 34: User-Per-Request Configuration

config = {"configurable": {"session_id": "user_123"}}
chain.invoke({"input": "Hello"}, config=config)

Runtime configurations pass dynamic session identifiers and tenant keys through chain runnables. When comparing langchain vs pydantic ai, PydanticAI specializes in type-safe single-agent validations while LangChain offers broad ecosystem integrations.

Troubleshooting LangChain Issues

“No module named ‘langchain_openai’”

Root Cause: Provider packages in LangChain 0.3 are decoupled from the core library.

Fix: Install the missing provider package directly via pip install langchain-openai.

Chain Returns Empty or None

Root Cause: Template variable mismatches or misaligned parser keys.

Fix: Inspect intermediate payloads using pipe logging: chain = prompt | RunnableLambda(lambda x: print(x) or x) | model | parser.

Vector Store Search Returns Irrelevant Results

Root Cause: Chunk size is too large or the embedding model lacks domain specificity.

Fix: Reduce chunk sizes to 500 characters and implement BM25 hybrid reranking.

Frequently Asked Questions

What Changed Between LangChain 0.2 and LangChain 0.3?

LangChain 0.3 decoupled third-party integrations into separate provider packages like langchain-openai and standardized on LCEL runnables. This architectural refactoring reduced overall installation footprint by 35% while establishing LangGraph as the default runtime for stateful agent workflows.

Should developers use LangChain or raw model APIs?

Developers should use LangChain when applications require composable RAG pipelines, dynamic model fallbacks, and multi-provider switching. For simple single-prompt completions without external tools, invoking raw SDKs as demonstrated in the OpenAI Function Calling Cookbook offers lower architectural overhead.

How do developers debug LangChain chains in production?

Teams debug production chains by enabling OpenTelemetry logging or attaching LangSmith tracing variables to inspect input-output payloads at every step. Implementing structured fallback handlers and circuit breakers as detailed in our AI error handling guide prevents runtime exceptions from crashing client sessions.

Can LangChain run with local models via Ollama?

LangChain connects natively to local inference runtimes using the langchain-community Ollama integration. Swapping the model initialization class preserves 90% of downstream LCEL chain logic while keeping inference data fully private.

What is the relationship between LangChain and LangGraph in 2026?

In modern architectures comparing langgraph vs langchain 2026 standards, LangChain provides foundational LCEL runnables for linear pipelines while LangGraph manages stateful, cyclic multi-agent graphs. Complex autonomous workflows require LangGraph checkpointing to support human-in-the-loop approvals and resilient node retries.

How do you enable token streaming in LCEL chains?

Developers enable streaming by replacing synchronous .invoke() calls with the .stream() async generator method. This pushes partial token chunks to clients immediately, reducing time-to-first-token by up to 65% across web applications.

Why does RAG return irrelevant context chunks?

Irrelevant context typically results from oversized chunk boundaries or semantic mismatches in domain-specific embedding representations. Tuning splitter chunk sizes to 500 characters and pairing vector retrieval with BM25 keyword reranking in hybrid search pipelines resolves retrieval drift.

Can developers test LangChain without active API keys?

Developers can run automated unit tests using the FakeListLLM class or local Ollama instances without incurring commercial API costs. These testing mocks simulate token streaming and structured output parsing inside CI/CD validation pipelines.

Build Your AI Application

These code snippets demonstrate how modern LCEL pipelines replace legacy chain abstractions with clean, maintainable Python code. For developers seeking an architectural evaluation between typed single-agent frameworks and cyclic state graphs, explore our PydanticAI vs LangChain vs LangGraph comparison.

LangChain’s primary strength lies in composition: combining these modular primitives enables robust production applications. These snippets provide reusable building blocks: the core primitives required for scalable AI systems. Regardless of whether your team builds a simple summarizer or a distributed multi-agent graph, prioritizing system reliability over abstraction complexity ensures long-term maintainability.