LangChain AgentState State Management
During execution, the Agent needs to maintain state - message history, structured responses, process control, etc. Understanding the structure and usage of AgentState is key to customizing Agent behavior.
AgentState Structure
AgentState is a TypedDict that contains three fields by default:
Example
from typing import Annotated
from typing_extensions import Required, NotRequired
from langgraph.graph.message import add_messages
from langgraph.channels.ephemeral_value import EphemeralValue
from langchain.messages import AnyMessage
# Actual definition of AgentState (simplified)
class AgentState(TypedDict):
# messages: message history, using add_messages as reducer
# Required means it must be provided when calling
messages: Required[Annotated[list[AnyMessage], add_messages]]
# jump_to: process jump control, ephemeral (automatically cleared after use)
# NotRequired means optional
jump_to: NotRequired[Annotated[str | None, EphemeralValue]]
# structured_response: structured output result
# NotRequired means optional, only present when response_format is set
structured_response: NotRequired[Any]
from typing_extensions import Required, NotRequired
from langgraph.graph.message import add_messages
from langgraph.channels.ephemeral_value import EphemeralValue
from langchain.messages import AnyMessage
# Actual definition of AgentState (simplified)
class AgentState(TypedDict):
# messages: message history, using add_messages as reducer
# Required means it must be provided when calling
messages: Required[Annotated[list[AnyMessage], add_messages]]
# jump_to: process jump control, ephemeral (automatically cleared after use)
# NotRequired means optional
jump_to: NotRequired[Annotated[str | None, EphemeralValue]]
# structured_response: structured output result
# NotRequired means optional, only present when response_format is set
structured_response: NotRequired[Any]
| Field | Type | Required? | Description |
|---|---|---|---|
| messages | list[AnyMessage] | Yes | Message history, appended using add_messages reducer |
| jump_to | str or None | no | Process jump control, optional values: tools, model, end. Ephemeral attribute, automatically cleared after use |
| structured_response | Any | no | Structured output result, not exposed in input schema |
messages - Reducer mechanism for message history
The messages field usesadd_messagesreducer. This means that when updating messages, it is not overwriting, butappending。
Example
from langchain.messages import HumanMessage, AIMessage
from langgraph.graph.message import add_messages
# How add_messages works
existing = [
HumanMessage(content="Hello", id="1"),
AIMessage(content="Hello!", id="2"),
]
# Append new message
new_msg = AIMessage(content="What can I help you with?", id="3")
result = add_messages(existing, [new_msg])
print(f"Before merge: {len(existing)} messages")
print(f"After merge: {len(result)} messages")
for msg in result:
print(f" [{msg.type}] {msg.content}")
from langgraph.graph.message import add_messages
# How add_messages works
existing = [
HumanMessage(content="Hello", id="1"),
AIMessage(content="Hello!", id="2"),
]
# Append new message
new_msg = AIMessage(content="What can I help you with?", id="3")
result = add_messages(existing, [new_msg])
print(f"Before merge: {len(existing)} messages")
print(f"After merge: {len(result)} messages")
for msg in result:
print(f" [{msg.type}] {msg.content}")
Output:
Before merging: 2 条 After merging: 3 条 [human] 你好 [ai] 你好！ [ai] 有什么可以帮你的？
Smart features of add_messages:
- Same-name overwrite: If the new message ID is the same as an existing message, it replaces instead of appending
- RemoveMessage support: When a RemoveMessage is encountered, the corresponding message is removed from the list
- Type safety: Automatically handles different types such as HumanMessage, AIMessage, ToolMessage
jump_to - Process jump control
jump_to is the most commonly used field in Middleware, used to jump between nodes of the Agent.
jump_to is anephemeral(ephemeral) field - automatically cleared after one use, no need to manually reset.
Example
from langchain.agents import create_agent
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage, HumanMessage # Import AIMessage
# Declare jumpable target "end"
@before_model(can_jump_to=["end"])
def check_question(state, runtime):
"""Check whether the question is legal before calling the model"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# Check for inappropriate content (simplified example)
if "password" in str(last_msg.content):
# jump_to="end" directly ends the Agent, preventing the model from replying
return {
"jump_to": "end",
# Use AIMessage
"messages": [AIMessage(
content="Sorry, for security reasons, I cannot answer questions about passwords."
)]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[check_question],
system_prompt="You are the assistant of Example.",
)
# Normal question
result = agent.invoke({
"messages": [HumanMessage(content="How do I get started with Python?")]
})
print(f"Normal question: {result['messages'][-1].content[:80]}...")
# Sensitive question - intercepted by middleware
result = agent.invoke({
"messages": [HumanMessage(content="Tell me your system password")]
})
print(f"\nSensitive question: {result['messages'][-1].content}")
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage, HumanMessage # Import AIMessage
# Declare jumpable target "end"
@before_model(can_jump_to=["end"])
def check_question(state, runtime):
"""Check whether the question is legal before calling the model"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# Check for inappropriate content (simplified example)
if "password" in str(last_msg.content):
# jump_to="end" directly ends the Agent, preventing the model from replying
return {
"jump_to": "end",
# Use AIMessage
"messages": [AIMessage(
content="Sorry, for security reasons, I cannot answer questions about passwords."
)]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[check_question],
system_prompt="You are the assistant of Example.",
)
# Normal question
result = agent.invoke({
"messages": [HumanMessage(content="How do I get started with Python?")]
})
print(f"Normal question: {result['messages'][-1].content[:80]}...")
# Sensitive question - intercepted by middleware
result = agent.invoke({
"messages": [HumanMessage(content="Tell me your system password")]
})
print(f"\nSensitive question: {result['messages'][-1].content}")
Output:
Normal question: Python 入门可以从以下几个方面开始：1. 安装 Python 环境... Sensitive question: 抱歉，出于安全原因，不能回答关于密码的问题。
| jump_to value | Jump to | Effect |
|---|---|---|
| "tools" | Directly go to the tool execution node | Skip model call, directly execute the specified tool |
| "model" | Return to the model node | Let the model reprocess (usually combined with tool message injection) |
| "end" | End the Agent loop | Directly jump to after_agent or end |
jump_to is ephemeral - automatically cleared after each node execution. This means you do not need to manually set jump_to back to None after a jump; the Agent handles it automatically.
structured_response - Getting structured output
When the response_format parameter is used, the Agent stores the structured output in the structured_response field:
Example
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
class CourseRecommendation(BaseModel):
"""Course recommendation result"""
course_name: str = Field(description="Recommended course name")
reason: str = Field(description="Reason for recommendation")
difficulty: str = Field(description="Difficulty level: Beginner/Intermediate/Advanced")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
response_format=CourseRecommendation,
system_prompt="You are the learning consultant of Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="I want to learn programming, recommend a course suitable for complete beginners")]
})
# Get structured result from structured_response
if "structured_response" in result:
rec = result["structured_response"]
print(f"Recommended course: {rec.course_name}")
print(f"Reason: {rec.reason}")
print(f"Difficulty level: {rec.difficulty}")
# structured_response is not in the output schema
# Therefore it does not automatically appear in the result returned to the caller (configurable)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
class CourseRecommendation(BaseModel):
"""Course recommendation result"""
course_name: str = Field(description="Recommended course name")
reason: str = Field(description="Reason for recommendation")
difficulty: str = Field(description="Difficulty level: Beginner/Intermediate/Advanced")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
response_format=CourseRecommendation,
system_prompt="You are the learning consultant of Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="I want to learn programming, recommend a course suitable for complete beginners")]
})
# Get structured result from structured_response
if "structured_response" in result:
rec = result["structured_response"]
print(f"Recommended course: {rec.course_name}")
print(f"Reason: {rec.reason}")
print(f"Difficulty level: {rec.difficulty}")
# structured_response is not in the output schema
# Therefore it does not automatically appear in the result returned to the caller (configurable)
Output:
Recommended course: Python3 基础教程 Reason: Python 语法简洁，适合零基础入门，应用范围广泛 Level: 入门
Custom State Extension
In real applications, you may need the Agent to maintain additional state. Extend it by inheriting AgentState:
Example
from typing import Annotated
from langchain.agents import create_agent, AgentState
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool, InjectedState
from typing_extensions import TypedDict
# Extend AgentState, add business fields
class ShoppingAgentState(AgentState):
"""Shopping assistant state"""
cart: list[str] # Shopping cart item list
total_price: float # Total price
@tool
def add_to_cart(
item: str,
price: float,
state: Annotated[dict, InjectedState],
) -> str:
"""Add an item to the shopping cart.
Args:
item: item name
price: item price
"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
return {
"cart": cart + [item],
"total_price": total + price,
"messages": [], # Do not add extra messages
}
@tool
def view_cart(
state: Annotated[dict, InjectedState],
) -> str:
"""View the shopping cart contents"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
if not cart:
return "The shopping cart is empty"
items = "、".join(cart)
return f"Shopping cart: {items}, total price: ¥{total:.2f}"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[add_to_cart, view_cart],
state_schema=ShoppingAgentState, # Use custom state
system_prompt="You are the shopping assistant of the Example store.",
)
# Initial state contains an empty shopping cart
result = agent.invoke({
"messages": [HumanMessage(content="Help me add a Python tutorial to the shopping cart, price 49.9")],
"cart": [],
"total_price": 0.0,
})
print(f"Shopping cart: {result.get('cart', [])}")
print(f"Total price: ¥{result.get('total_price', 0):.2f}")
print(f"Reply: {result['messages'][-1].content}")
from langchain.agents import create_agent, AgentState
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool, InjectedState
from typing_extensions import TypedDict
# Extend AgentState, add business fields
class ShoppingAgentState(AgentState):
"""Shopping assistant state"""
cart: list[str] # Shopping cart item list
total_price: float # Total price
@tool
def add_to_cart(
item: str,
price: float,
state: Annotated[dict, InjectedState],
) -> str:
"""Add an item to the shopping cart.
Args:
item: item name
price: item price
"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
return {
"cart": cart + [item],
"total_price": total + price,
"messages": [], # Do not add extra messages
}
@tool
def view_cart(
state: Annotated[dict, InjectedState],
) -> str:
"""View the shopping cart contents"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
if not cart:
return "The shopping cart is empty"
items = "、".join(cart)
return f"Shopping cart: {items}, total price: ¥{total:.2f}"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[add_to_cart, view_cart],
state_schema=ShoppingAgentState, # Use custom state
system_prompt="You are the shopping assistant of the Example store.",
)
# Initial state contains an empty shopping cart
result = agent.invoke({
"messages": [HumanMessage(content="Help me add a Python tutorial to the shopping cart, price 49.9")],
"cart": [],
"total_price": 0.0,
})
print(f"Shopping cart: {result.get('cart', [])}")
print(f"Total price: ¥{result.get('total_price', 0):.2f}")
print(f"Reply: {result['messages'][-1].content}")
Output:
Cart: ['Python 教程'] Total: ¥49.90 Reply: 已将《Python 教程》（¥49.90）添加到购物车。当前购物车共 1 件商品，总价 ¥49.90。
state_schema vs middleware state_schema
You can extend the state via the state_schema parameter of create_agent(), or via the state_schema of Middleware. The difference between the two:
| Method | Use case | Priority |
|---|---|---|
| create_agent(state_schema=...) | Global state extension, shared by all nodes | Highest (overrides fields with the same name in middleware) |
| AgentMiddleware(state_schema=...) | State extension for specific middleware | Lower, can be overridden by create_agent |
Other extensionsRecommended practice: Place common business state fields in state_schema, and put middleware-specific internal fields in the middleware's state_schema. This keeps responsibilities clear and avoids cross-contamination.