Gestión de estado de LangChain AgentState
Durante la ejecución, el Agente necesita mantener estado: historial de mensajes, respuestas estructuradas, control de flujo, etc. Entender la estructura y el uso de AgentState es clave para personalizar el comportamiento del Agente.
Estructura de AgentState
AgentState es un TypedDict que incluye tres campos por defecto:
Ejemplo
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
# La definición real de AgentState (versión simplificada)
class AgentState(TypedDict):
# messages: historial de mensajes, usando add_messages como reducer
# Required indica que es obligatorio proporcionarlo en la llamada
messages: Required[Annotated[list[AnyMessage], add_messages]]
# jump_to: control de salto de flujo, ephemeral (se limpia automáticamente después de usarse)
# NotRequired indica que es opcional
jump_to: NotRequired[Annotated[str | None, EphemeralValue]]
# structured_response: resultado de salida estructurada
# NotRequired indica que es opcional, solo aparece cuando response_format está configurado
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
# La definición real de AgentState (versión simplificada)
class AgentState(TypedDict):
# messages: historial de mensajes, usando add_messages como reducer
# Required indica que es obligatorio proporcionarlo en la llamada
messages: Required[Annotated[list[AnyMessage], add_messages]]
# jump_to: control de salto de flujo, ephemeral (se limpia automáticamente después de usarse)
# NotRequired indica que es opcional
jump_to: NotRequired[Annotated[str | None, EphemeralValue]]
# structured_response: resultado de salida estructurada
# NotRequired indica que es opcional, solo aparece cuando response_format está configurado
structured_response: NotRequired[Any]
| Campo | Tipo | Requerido | Descripción |
|---|---|---|---|
| messages | list[AnyMessage] | Sí | Historial de mensajes, se agrega mediante el reducer add_messages |
| jump_to | str o None | no | Control de salto de flujo, valores posibles: tools, model, end. Atributo ephemeral, se limpia automáticamente después de usarse |
| structured_response | Any | no | Resultado de salida estructurada, no se expone en el input schema |
messages: mecanismo de Reducer para el historial de mensajes
El campo messages usaadd_messagesreducer. Esto significa que al actualizar messages no se sobrescribe, sino que seagrega。
Ejemplo
from langchain.messages import HumanMessage, AIMessage
from langgraph.graph.message import add_messages
# funcionamiento de add_messages
existing = [
HumanMessage(content="Hola", id="1"),
AIMessage(content="¡Hola!", id="2"),
]
# añadir un nuevo mensaje
new_msg = AIMessage(content="¿En qué puedo ayudarte?", id="3")
result = add_messages(existing, [new_msg])
print(f"Antes de combinar: {len(existing)} mensajes")
print(f"Después de combinar: {len(result)} mensajes")
for msg in result:
print(f" [{msg.type}] {msg.content}")
from langgraph.graph.message import add_messages
# funcionamiento de add_messages
existing = [
HumanMessage(content="Hola", id="1"),
AIMessage(content="¡Hola!", id="2"),
]
# añadir un nuevo mensaje
new_msg = AIMessage(content="¿En qué puedo ayudarte?", id="3")
result = add_messages(existing, [new_msg])
print(f"Antes de combinar: {len(existing)} mensajes")
print(f"Después de combinar: {len(result)} mensajes")
for msg in result:
print(f" [{msg.type}] {msg.content}")
Resultado de ejecución:
Antes de fusionar: 2 条 Después de fusionar: 3 条 [human] 你好 [ai] 你好！ [ai] 有什么可以帮你的？
Características inteligentes de add_messages:
- Sobrescritura por mismo nombre: si el ID del nuevo mensaje coincide con uno existente, se reemplaza en lugar de agregar
- Soporte de RemoveMessage: al encontrar un RemoveMessage, elimina el mensaje correspondiente de la lista
- Seguridad de tipos: maneja automáticamente diferentes tipos como HumanMessage, AIMessage, ToolMessage, etc.
jump_to: control de salto de flujo
jump_to es el campo más utilizado en Middleware, que se usa para saltar entre los distintos nodos del Agente.
jump_to es unephemeralcampo (transitorio): se limpia automáticamente después de usarlo, no es necesario restablecerlo manualmente.
Ejemplo
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 # importar AIMessage
# declarar el destino de salto "end"
@before_model(can_jump_to=["end"])
def check_question(state, runtime):
"""Verificar si la pregunta es válida antes de la llamada al modelo"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# verificar si contiene contenido inapropiado (ejemplo simplificado)
if "contraseña" in str(last_msg.content):
# jump_to="end" finaliza directamente el Agent, evita que el modelo responda
return {
"jump_to": "end",
# usar AIMessage
"messages": [AIMessage(
content=Lo siento, por razones de seguridad, no puedo responder preguntas sobre contraseñas.
)]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[check_question],
system_prompt=Eres el asistente de EJEMPLO (el tutorial para novatos).,
)
# Pregunta normal
result = agent.invoke({
"messages": [HumanMessage(content=¿Cómo empezar con Python?)]
})
print(fPregunta normal: {result['messages'][-1].content[:80]}...)
# Pregunta sensible —— interceptada por el middleware
result = agent.invoke({
"messages": [HumanMessage(content=Dime la contraseña de tu sistema)]
})
print(f"\nPregunta sensible: {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 # importar AIMessage
# declarar el destino de salto "end"
@before_model(can_jump_to=["end"])
def check_question(state, runtime):
"""Verificar si la pregunta es válida antes de la llamada al modelo"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# verificar si contiene contenido inapropiado (ejemplo simplificado)
if "contraseña" in str(last_msg.content):
# jump_to="end" finaliza directamente el Agent, evita que el modelo responda
return {
"jump_to": "end",
# usar AIMessage
"messages": [AIMessage(
content=Lo siento, por razones de seguridad, no puedo responder preguntas sobre contraseñas.
)]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[check_question],
system_prompt=Eres el asistente de EJEMPLO (el tutorial para novatos).,
)
# Pregunta normal
result = agent.invoke({
"messages": [HumanMessage(content=¿Cómo empezar con Python?)]
})
print(fPregunta normal: {result['messages'][-1].content[:80]}...)
# Pregunta sensible —— interceptada por el middleware
result = agent.invoke({
"messages": [HumanMessage(content=Dime la contraseña de tu sistema)]
})
print(f"\nPregunta sensible: {result['messages'][-1].content})
Resultado de ejecución:
Pregunta normal: Python 入门可以从以下几个方面开始：1. 安装 Python 环境... Pregunta sensible: 抱歉，出于安全原因，不能回答关于密码的问题。
| Valor de jump_to | Saltar a | Efecto |
|---|---|---|
| "tools" | Entrar directamente al nodo de ejecución de herramientas | Omitir la llamada al modelo y ejecutar directamente la herramienta especificada |
| "model" | Volver al nodo del modelo | Hacer que el modelo procese de nuevo (normalmente junto con la inyección de mensajes de herramienta) |
| "end" | Terminar el bucle del Agente | Saltar directamente a after_agent o finalizar |
jump_to es efímero: se limpia automáticamente después de cada ejecución de nodo. Esto significa que no necesitas restablecer jump_to a None manualmente después del salto; el Agente lo maneja automáticamente.
structured_response: obtención de salida estructurada
Cuando se usa el parámetro response_format, el Agente almacena la salida estructurada en el campo structured_response:
Ejemplo
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):
Resultado de la recomendación de cursos
course_name: str = Field(description=Nombre del curso recomendado)
reason: str = Field(description=Razón de la recomendación)
difficulty: str = Field(description=Nivel de dificultad：principiante/intermedio/avanzado)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
response_format=CourseRecommendation,
system_prompt=Eres el asesor de aprendizaje de EJEMPLO (el tutorial para novatos).,
)
result = agent.invoke({
"messages": [HumanMessage(content=Quiero aprender a programar, recomienda un curso adecuado para principiantes absolutos)]
})
# Obtener el resultado estructurado de structured_response
if "structured_response" in result:
rec = result["structured_response"]
print(fCurso recomendado: {rec.course_name})
print(fRazón de la recomendación: {rec.reason})
print(fNivel de dificultad: {rec.difficulty})
# structured_response no está en el output schema
# Por lo tanto, no aparecerá automáticamente en el resultado devuelto al llamador (configurable)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
class CourseRecommendation(BaseModel):
Resultado de la recomendación de cursos
course_name: str = Field(description=Nombre del curso recomendado)
reason: str = Field(description=Razón de la recomendación)
difficulty: str = Field(description=Nivel de dificultad：principiante/intermedio/avanzado)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
response_format=CourseRecommendation,
system_prompt=Eres el asesor de aprendizaje de EJEMPLO (el tutorial para novatos).,
)
result = agent.invoke({
"messages": [HumanMessage(content=Quiero aprender a programar, recomienda un curso adecuado para principiantes absolutos)]
})
# Obtener el resultado estructurado de structured_response
if "structured_response" in result:
rec = result["structured_response"]
print(fCurso recomendado: {rec.course_name})
print(fRazón de la recomendación: {rec.reason})
print(fNivel de dificultad: {rec.difficulty})
# structured_response no está en el output schema
# Por lo tanto, no aparecerá automáticamente en el resultado devuelto al llamador (configurable)
Resultado de ejecución:
Curso recomendado: Python3 基础教程 Razón: Python 语法简洁，适合零基础入门，应用范围广泛 Nivel: 入门
Extensión de State personalizado
En aplicaciones reales, es posible que necesites que el Agente mantenga estado adicional. Puedes extenderlo heredando de AgentState:
Ejemplo
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
# Extiende AgentState y agrega campos de negocio
class ShoppingAgentState(AgentState):
"""Estado del asistente de compras"""
cart: list[str] # Lista de productos del carrito
total_price: float # Precio total
@tool
def add_to_cart(
item: str,
price: float,
state: Annotated[dict, InjectedState],
) -> str:
"""Agregar el producto al carrito.
Args:
item: nombre del producto
price: precio del producto
"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
return {
"cart": cart + [item],
"total_price": total + price,
"messages": [], # No agregar mensajes adicionales
}
@tool
def view_cart(
state: Annotated[dict, InjectedState],
) -> str:
"""Ver el contenido del carrito"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
if not cart:
return "El carrito está vacío"
items = "、".join(cart)
return f"Carrito: {items}, Precio total: ¥{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, # Usar estado personalizado
system_prompt="Eres el asistente de compras de la tienda EJEMPLO.",
)
# El estado inicial contiene un carrito vacío
result = agent.invoke({
"messages": [HumanMessage(content="Ayúdame a agregar un tutorial de Python al carrito, precio 49.9")],
"cart": [],
"total_price": 0.0,
})
print(f"Carrito: {result.get('cart', [])}")
print(f"Precio total: ¥{result.get('total_price', 0):.2f}")
print(f"Respuesta: {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
# Extiende AgentState y agrega campos de negocio
class ShoppingAgentState(AgentState):
"""Estado del asistente de compras"""
cart: list[str] # Lista de productos del carrito
total_price: float # Precio total
@tool
def add_to_cart(
item: str,
price: float,
state: Annotated[dict, InjectedState],
) -> str:
"""Agregar el producto al carrito.
Args:
item: nombre del producto
price: precio del producto
"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
return {
"cart": cart + [item],
"total_price": total + price,
"messages": [], # No agregar mensajes adicionales
}
@tool
def view_cart(
state: Annotated[dict, InjectedState],
) -> str:
"""Ver el contenido del carrito"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
if not cart:
return "El carrito está vacío"
items = "、".join(cart)
return f"Carrito: {items}, Precio total: ¥{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, # Usar estado personalizado
system_prompt="Eres el asistente de compras de la tienda EJEMPLO.",
)
# El estado inicial contiene un carrito vacío
result = agent.invoke({
"messages": [HumanMessage(content="Ayúdame a agregar un tutorial de Python al carrito, precio 49.9")],
"cart": [],
"total_price": 0.0,
})
print(f"Carrito: {result.get('cart', [])}")
print(f"Precio total: ¥{result.get('total_price', 0):.2f}")
print(f"Respuesta: {result['messages'][-1].content}")
Resultado de ejecución:
Carrito: ['Python 教程'] Total: ¥49.90 Respuesta: 已将《Python 教程》（¥49.90）添加到购物车。当前购物车共 1 件商品，总价 ¥49.90.
state_schema vs middleware state_schema
Se puede extender el estado tanto con el parámetro state_schema de create_agent() como con state_schema del Middleware. La diferencia entre ambos:
| Método | Escenario de uso | Prioridad |
|---|---|---|
| create_agent(state_schema=...) | Extensión de estado global, compartida por todos los nodos | Más alta (sobrescribe los campos del middleware con el mismo nombre) |
| AgentMiddleware(state_schema=...) | Extensión de estado específica de un middleware | Más baja, puede ser sobrescrita por create_agent |
Otras extensionesRecomendación: coloque los campos de estado comerciales comunes en `state_schema` y los campos internos específicos del middleware en el `state_schema` del middleware. De esta manera, las responsabilidades quedan claras y no se contaminan entre sí.