YouTip LogoYouTip

Langchain Agent State

During execution, Agents need to maintain stateโ€”message history, structured responses, flow 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, add_messages]] # jump_to: flow 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 appears when response_format is set structured_response: NotRequired | Field | Type | Required | Description | | --- | --- | --- | --- | | messages | list | Yes | Message history, uses add_messages reducer to append | | jump_to | str or None | No | Flow jump control, optional values: tools, model, end. Ephemeral property, automatically cleared after use | | structured_response | Any | No | Structured output result, not exposed in input schema | * * * ## messagesโ€”โ€”Message History Reducer Mechanism The messages field uses the add_messages reducer. This means when updating messages, it doesn't overwrite but appends. ## 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,) print(f"before merging: {len(existing)} records") print(f"After Merging: {len(result)} records") for msg in result: print(f" [{msg.type}] {msg.content}") Result: before merging: 2 recordsAfter Merging: 3 records Hello Hello๏ผ What can I help you with? Smart features of add_messages: * Same-name overwrite: If new message ID matches an existing message, it replaces instead of appends * RemoveMessage support: When encountering RemoveMessage, deletes the corresponding message from the list * Type safety: Automatically handles different types like HumanMessage, AIMessage, ToolMessage, etc. * * * ## jump_toโ€”โ€”Flow Jump Control jump_to is the most commonly used field in Middleware, used to jump between various nodes of the Agent. jump_to is an ephemeral (transient) fieldโ€”it is automatically cleared after one use, no manual reset needed. ## 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 HumanMessage # Use before_model middleware, control flow through jump_to @before_model def check_question(state, runtime): """Check if question is valid before model call""" messages = state.get("messages",[]) if not messages: return None last_msg = messages # Check for inappropriate content (simplified example) if"Password"in str(last_msg.content): # jump_to="end" directly ends Agent, prevents model from responding return{ "jump_to": "end", "messages": [HumanMessage( 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=, system_prompt="You are an assistant for the TUTORIAL tutorial.", ) # Normal question result = agent.invoke({ "messages": [HumanMessage(content="Python How toGetting Started?")] }) print(f"Normal question: {result['messages'].content[:80]}...") # Sensitive questionโ€”intercepted by middleware result = agent.invoke({ "messages": [HumanMessage(content="Tell me your system password")] }) print(f"\\ Sensitive question: {result['messages'].content}") Result: Normal question: Python Getting StartedYou can start from the following aspects:1. Install Python environment...Sensitive question: Sorry, for security reasons, I cannot answer questions about passwords. | jump_to Value | Jumps To | Effect | | --- | --- | --- | | "tools" | Directly enter tool execution node | Skip model call, directly execute specified tool | | "model" | Return to model node | Let model reprocess (usually combined with tool message injection) | | "end" | End Agent loop | Directly jump to after_agent or end | > jump_to is ephemeralโ€”it is automatically cleared after each node execution. This means you don't need to manually set jump_to back to None after jumping; the Agent handles this automatically. * * * ## structured_responseโ€”โ€”Getting Structured Output When using the response_format parameter, 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 results""" course_name: str= Field(description="Recommended course name") reason: str= Field(description="Recommended Reason") difficulty: str= Field(description="Difficulty level๏ผšGetting Started/Advanced/Advanced") model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0) agent = create_agent( model=model, response_format=CourseRecommendation, system_prompt="You are a learning consultant for the TUTORIAL tutorial.", ) result = agent.invoke({ "messages": [HumanMessage(content="I want to learn programming, recommend a course suitable for beginners")] }) # Get structured result from structured_response if"structured_response"in result: rec = result print(f"Recommended courses: {rec.course_name}") print(f"Recommended Reason: {rec.reason}") print(f"Difficulty level: {rec.difficulty}") # structured_response is not in output schema # So it won't automatically appear in results returned to caller (configurable) Result: Recommended courses: Python3 Basic tutorial recommendation reasons: Python Concise syntax, suitable for beginnersGetting Started๏ผŒWide range of applicationsDifficulty level: Getting Started * * * ## Custom State Extension In practical applications, you may need the Agent to maintain additional state. Extend by inheriting from 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# Cart item list total_price: float# Total price @tool def add_to_cart( item: str, price: float, state: Annotated[dict, InjectedState], ) ->str: """Add item to cart. Args: item: Product Name price: Item price """ cart = state.get("cart",[]) total = state.get("total_price",0.0) return{ "cart": cart + , "total_price": total + price, "messages": [],# Do not add extra messages } @tool def view_cart( state: Annotated[dict, InjectedState], ) ->str: """View cart contents""" cart = state.get("cart",[]) total = state.get("total_price",0.0) if not cart: return"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
โ† Langchain StreamingLangchain Create_Agent โ†’