When you build an AI Agent with LangChain, the first thing you learn is how to define tools and write prompts. But once your agent starts doing real work — handling users, managing conversation state, dealing with failures — you quickly hit a wall: how do you peek inside the agent’s brain and control its behavior without rewriting everything?
That’s where Runtime and Middleware come in.
Runtime gives you a window into the agent’s internal state — what it remembers, what it knows, and who it’s talking to. Middleware lets you hook into the agent’s execution loop and inject custom logic at every stage — before and after model calls, tool calls, and more.
Together, they turn a fragile script into a production-ready system.
What
What is Runtime?
Runtime is the execution context of a LangChain agent. It bundles three core concepts that together represent everything the agent “knows” and “has” at any given moment:
Long-term memory — user preferences, domain knowledge, past experience
Cross-session
Context
Runtime context — user identity, config parameters, request metadata
Single invocation
Think of it this way:
1 2 3
State = "What am I doing right now?" Store = "What do I know in general?" Context = "Who asked me and under what conditions?"
What is Middleware?
Middleware is a mechanism that lets you intercept and control the agent’s execution loop. It registers hooks at specific points in the agent’s lifecycle — before the model is called, after a tool finishes, and so on.
Without Middleware: With Middleware: ┌──────────────┐ ┌──────────────┐ │ User Input │ │ User Input │ └──────┬───────┘ └──────┬───────┘ ▼ ▼ ┌──────────────┐ ┌──────────────┐ │ Model Call │ │ Hook: before│ └──────┬───────┘ │ model call │ ▼ │ ┌──────────────┐ └──────┬───────┘ │ Tool Call │ ▼ └──────┬───────┘ ┌──────────────┐ ▼ │ Model Call │ ┌──────────────┐ └──────┬───────┘ │ Response │ ▼ └──────────────┘ ┌──────────────┐ │ Hook: after │ │ model call │ └──────┬───────┘ ▼ ┌──────────────┐ │ Tool Call │ └──────┬───────┘ ▼ ┌──────────────┐ │ Hook: wrap │ │ tool call │ └──────┬───────┘ ▼ ┌──────────────┐ │ Response │ └──────────────┘
Why
Why Runtime Matters
Without Runtime, your agent is a stateless function. It can’t remember what happened in the conversation, it can’t recall user preferences from last week, and it has no idea who’s making the request.
Runtime solves three real problems:
State tracking — Count model calls, track task progress, store intermediate results
Knowledge grounding — Inject external data (user profiles, domain docs) into the agent’s decision-making
Request awareness — Know which user is asking, what permissions they have, what config to apply
Why Middleware Matters
Without Middleware, every cross-cutting concern — logging, retries, PII redaction, human approval — ends up tangled inside your tools and prompts. You copy-paste the same try/catch block into every tool. You add the same retry logic to every agent call. Your code becomes a mess.
Middleware solves this by separating concerns. You write the logic once, attach it as a hook, and it runs automatically at the right point — no invasion of your business code.
How
Runtime Deep Dive
State — Short-term Memory
State stores everything about the current invocation: the message history, task counters, temporary flags, and so on. It resets when the invocation ends.
By default, an agent uses AgentState, which only contains messages. To track anything else, you define a custom state:
1 2 3 4 5 6 7
from langchain.agents import AgentState from typing import NotRequired
classCustomState(AgentState): """Extended agent state with custom fields.""" model_call_count: NotRequired[int] # How many times the model was called session_start: NotRequired[str] # When this session started
Accessing State in a Tool
LangChain provides a reserved runtime parameter in tools. Through it, you can read state:
1 2 3 4 5 6 7
from langchain.tools import tool, ToolRuntime
@tool defmy_tool(runtime: ToolRuntime): """A tool that reads agent state.""" count = runtime.state.get("model_call_count", 0) returnf"Model has been called {count} times so far."
Modifying State in a Tool
To update state, return a Command with an update dict:
# Record session start time on the first interaction if message_count <= 2: updates["session_start"] = datetime.now()
return Command(update=updates)
Registering Custom State
When creating the agent, point it to your custom state schema:
1 2 3 4 5 6 7 8 9 10
from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent( "deepseek-chat", tools=[update_state], state_schema=CustomState, # Tell the agent to use our state checkpointer=InMemorySaver(), system_prompt="You are a helpful assistant. Always call update_state to track your invocation." )
Inspecting State
After an invocation, you can snapshot the full state:
1 2 3 4 5 6 7 8 9
config = {"configurable": {"thread_id": "1"}} response = agent.invoke( {"messages": [HumanMessage(content="Hi, my name is Alex")]}, config )
# View the full state snapshot snapshot = agent.get_state(config) print(snapshot.values)
Store — Long-term Memory
While State is ephemeral, Store persists across sessions. It’s where you keep user preferences, domain knowledge, failure logs, and anything that should survive beyond a single conversation.
For semantic search, you can back the store with an embedding model. This lets you search by meaning, not just exact field matches:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
from langgraph_cli.schemas import IndexConfig from langchain_community.embeddings import DashScopeEmbeddings import os
# Initialize embedding model embedding_model = DashScopeEmbeddings( model="text-embedding-v4", dashscope_api_key=os.getenv("DASHSCOPE_API_KEY") )
# Create vector-backed store memory_store = InMemoryStore(index=IndexConfig( embed=embedding_model, dims=1024 ))
With a vector-backed store, you can search semantically:
1 2 3 4 5 6 7
results = memory_store.search( ("users",), query="technical staff", # semantic search limit=5 ) for item in results: print(f"Score: {item.score:.3f} | Data: {item.value}")
Higher score means higher similarity.
Accessing Store in a Tool
Just like State, Store is accessible through runtime:
1 2 3 4 5 6 7 8 9 10 11
@tool defget_user_info(user_id: str, runtime: ToolRuntime) -> str: """Look up user info from the store.""" if runtime.store isNone: return"Store not available"
user_info = runtime.store.get(("users",), user_id) if user_info isNone: return"User not found"
returnf"User info: {user_info.value}"
Registering Store with the Agent
1 2 3 4 5
agent = create_agent( model="deepseek-chat", tools=[get_user_info], store=memory_store # Attach the store )
Context — Runtime Context
Context carries request-level metadata: who the user is, what permissions they have, config overrides, and so on. It lives only for the duration of a single invocation.
Defining a Context Schema
There are two ways to define it:
1 2 3 4 5 6 7 8 9 10 11 12 13
from dataclasses import dataclass from typing_extensions import TypedDict
# Option 1: dataclass @dataclass classUserContext: """Runtime context for the agent.""" user_id: str = ""
# Option 2: TypedDict classUserContext2(TypedDict): """Runtime context for the agent.""" user_id: str
@tool defget_users(runtime: ToolRuntime[UserContext]): """Query all users — requires clearance level 3+.""" store = runtime.store if store isNone: return"Store not available"
# Who is making the request? user_id = runtime.context.user_id if user_id isNoneor user_id == "": return"User not logged in."
# Check permissions user = store.get(("users",), user_id) if user isNone: return"User not found."
user_info = dict(user.value) if user_info.get("clearance_level", 0) < 3: return"Insufficient permissions!"
results = store.search(("users",)) return [item.value for item in results]
@tool defget_user_preferences(runtime: ToolRuntime[UserContext]): """Get preferences for the current user.""" user_id = runtime.context.user_id pref = runtime.store.get(("preferences",), user_id) if pref isNone: return"No preferences found." return pref.value
Passing Context at Invocation
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
agent = create_agent( model="deepseek-chat", tools=[get_users, get_user_preferences], store=memory_store, context_schema=UserContext, system_prompt=""" # identity You are a helpful assistant that can look up user info and preferences. # instruction Always format results according to the user's preference style. """ )
response = agent.invoke( {"messages": [HumanMessage("Hello, show me all users")]}, context=UserContext(user_id="user_001") # Pass context here )
Middleware Deep Dive
Prebuilt Middleware
LangChain ships with several ready-to-use middleware. We’ll cover three essential ones.
PIIMiddleware
PIIMiddleware automatically detects and sanitizes Personally Identifiable Information in model inputs and outputs — email addresses, phone numbers, ID numbers, and more.
agent = create_agent( model="deepseek-chat", tools=[transfer_money], middleware=[hitl], checkpointer=InMemorySaver(), system_prompt="You are an account management assistant. Help users transfer money." )
When the agent tries to call transfer_money, it pauses and emits an interrupt:
Logging, state updates, conditional logic at lifecycle boundaries
Wrap-style
wrap_model_call, wrap_tool_call
Intercepting and modifying request/response pairs (retries, transforms, overrides)
Node-style Hooks
Node-style hooks run at a specific point in the agent lifecycle. They receive the current state and runtime, and return a dict of state updates.
For example, counting model calls — a task we previously hacked together with a dedicated tool — becomes trivial with an after_model hook:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
from langgraph.runtime import Runtime from langchain.agents import AgentState from typing import NotRequired, Any from langchain.agents.middleware import after_model
# 1. Define custom state with a counter classCustomAgentState(AgentState): """Extended state with a model call counter.""" model_call_count: NotRequired[int]
# 2. Define the middleware @after_model(state_schema=CustomAgentState) defincrement_counter(state: CustomAgentState, runtime: Runtime) -> dict[str, Any]: """After each model call, bump the counter.""" current = state.get("model_call_count", 0) return {"model_call_count": current + 1}
The function signature must follow (state, runtime) -> dict[str, Any].
The counter increments automatically — no extra tool calls, no wasted tokens.
Wrap-style Hooks
Wrap-style hooks wrap a call (model or tool). They receive the request and a handler function that performs the actual call. This lets you retry on failure, transform inputs/outputs, or short-circuit entirely.
from langchain.agents.middleware import ( wrap_model_call, ModelRequest, ModelResponse, ) from typing importCallable
@wrap_model_call defretry_model( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: """Retry model calls up to 3 times on failure.""" for attempt inrange(3): try: return handler(request) except Exception as e: print(f"Retry {attempt + 1}/3 after error: {e}") if attempt == 2: raise returnNone# unreachable, but satisfies the type checker
from langchain.agents.middleware import AgentMiddleware from langchain.agents.middleware.types import ModelCallResult, ToolCallRequest from langgraph.types import Command from typing importCallable
for chunk, metadata in agent.stream( {"messages": [HumanMessage("How's the weather in Hangzhou?")]}, stream_mode="messages" ): if chunk and chunk.content: print(chunk.content, end="", flush=True)
Output:
1 2 3 4 5 6 7
=== Calling model with 1 messages === Let me check the weather in Hangzhou for you. === Calling tool: get_weather === Args: {'location': 'Hangzhou'}
=== Tool call succeeded === The weather in Hangzhou is sunny, 25°C — perfect for going out! 🌞
Advanced Usage
Dynamic Request Modification
Inside a wrap_model_call hook, you can override any request parameter — model, tools, system prompt — using request.override():
from langchain.agents.middleware import wrap_model_call from collections.abc importCallable from pydantic.dataclasses import dataclass from langchain.chat_models import init_chat_model
# Initialize two models reasoning_model = init_chat_model(model="deepseek-reasoner") chat_model = init_chat_model(model="deepseek-chat")
# Context to control which model to use @dataclass classUserContext: reasoning: bool = False
@wrap_model_call defdynamic_model_selector( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: """Pick a model based on the runtime context.""" is_reasoning = getattr(request.runtime.context, 'reasoning', False) selected = reasoning_model if is_reasoning else chat_model print(f"Selected model: {selected.model_name} (reasoning={is_reasoning})")
# Use the reasoning model for this invocation response = agent.invoke( {"messages": [HumanMessage("Solve this math problem: ...")]}, context=UserContext(reasoning=True) )
Conditional Jump with jump_to
Inside a node-style hook, you can return a jump_to directive to skip to a specific agent node — most commonly "end" to terminate the loop.
from langgraph.runtime import Runtime from typing import NotRequired, Any from langchain.agents.middleware import AgentMiddleware, hook_config, AgentState
classCustomAgentState(AgentState): """State with a call counter.""" model_call_count: NotRequired[int]
# ── 6. Create Agent ────────────────────────────────────────────── agent = create_agent( model="deepseek-chat", tools=[get_balance, get_preferences], store=store, state_schema=BankingState, context_schema=BankingContext, middleware=[rate_limiter, pii_email], checkpointer=InMemorySaver(), system_prompt="You are a banking assistant. Always match the user's preferred style." )
# ── 7. Run ─────────────────────────────────────────────────────── config = {"configurable": {"thread_id": "bank-001"}} response = agent.invoke( { "messages": [HumanMessage("Hi, what's my balance? My email is alex@bank.com")], "model_call_count": 0, }, config, context=BankingContext(user_id="user_001") )
for msg in response["messages"]: msg.pretty_print()
Key takeaways from this example:
State tracks model call count for rate limiting
Store holds user preferences and account balances
Context carries the logged-in user ID
Middleware provides automatic PII redaction and rate limiting — zero changes to tool logic
Best Practices
Area
Recommendation
State
Keep it minimal. Only store what the agent truly needs to track within a session. Large state slows down checkpointing.
Store
Use namespaces to organize data logically: ("preferences",), ("accounts",), ("history",). Never dump everything into one namespace.
Context
Pass user identity and config via Context, not State. Context is typed and doesn’t pollute the agent’s memory.
Middleware
One concern per middleware. Don’t mix PII redaction with rate limiting in the same class.
Middleware order
Middleware runs in registration order. Put protective middleware (PII, rate limits) first, logging last.
HITL
Always pair HumanInTheLoopMiddleware with a checkpointer. Without checkpointing, you can’t resume after an interrupt.
Testing
Test middleware in isolation first. Mock state and runtime objects before integrating with the full agent.
Retry logic
In wrap_model_call retries, add exponential backoff. Bare loops hammer the API and burn tokens.
FAQ
Q: Can I access State and Store without Runtime? A: In tools, no — runtime is the only interface. But in middleware, you receive state and runtime directly as function parameters.
Q: What happens if a middleware raises an exception? A: The agent execution halts and the exception propagates to the caller. Wrap your middleware logic in try/except if you want graceful degradation.
Q: Can middleware modify the system prompt? A: Yes, via request.override(system_prompt="...") inside a wrap_model_call hook.
Q: Is Store shared across all threads? A: Yes. Store is agent-level, not thread-level. Use different namespaces to isolate data per user or per tenant.
Q: When should I use State vs. Store? A: If the data should survive past the current invocation (e.g., user preferences), use Store. If it’s only relevant to the current session (e.g., call counters, temporary flags), use State.
Q: Can I use multiple wrap_model_call middleware together? A: Yes — they compose as a stack. The first registered middleware is the outermost wrapper. Each one calls handler(request), which invokes the next middleware in the chain.
Q: runtime is a reserved parameter name in @tool — can I use it for my own arguments? A: No. LangChain intercepts runtime: ToolRuntime to inject the runtime context. If you name your own parameter runtime, you’ll get a conflict. Choose a different name.
Summary
LangChain’s Runtime and Middleware are the two pillars of production-grade agent engineering:
State answers “what’s happening now?” Store answers “what do we know?” Context answers “who’s asking?” Middleware answers “what should happen in between?”
Master these four concepts, and you can build agents that are observable, controllable, and safe — not just clever but intelligent!