Gestion d'état de LangChain AgentState
Pendant l'exécution, l'Agent doit maintenir un état - historique des messages, réponses structurées, contrôle du processus, etc. Comprendre la structure et l'utilisation d'AgentState est essentiel pour personnaliser le comportement de l'Agent.
Structure d'AgentState
AgentState est un TypedDict qui contient trois champs par défaut :
Exemple
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
# Définition réelle d'AgentState (simplifiée)
class AgentState(TypedDict):
# messages : historique des messages, en utilisant add_messages comme réducteur
# Required signifie qu'il doit être fourni lors de l'appel
messages: Required[Annotated[list[AnyMessage], add_messages]]
# jump_to : traite le contrôle de saut, éphémère (automatiquement effacé après utilisation)
# NotRequired signifie facultatif
jump_to: NotRequired[Annotated[str | None, EphemeralValue]]
# structured_response : résultat de sortie structuré
# NotRequired signifie facultatif, présent uniquement lorsque response_format est défini
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
# Définition réelle d'AgentState (simplifiée)
class AgentState(TypedDict):
# messages : historique des messages, en utilisant add_messages comme réducteur
# Required signifie qu'il doit être fourni lors de l'appel
messages: Required[Annotated[list[AnyMessage], add_messages]]
# jump_to : traite le contrôle de saut, éphémère (automatiquement effacé après utilisation)
# NotRequired signifie facultatif
jump_to: NotRequired[Annotated[str | None, EphemeralValue]]
# structured_response : résultat de sortie structuré
# NotRequired signifie facultatif, présent uniquement lorsque response_format est défini
structured_response: NotRequired[Any]
| Champ | Type | Obligatoire ? | Description |
|---|---|---|---|
| messages | list[AnyMessage] | Oui | Historique des messages, ajouté à l'aide du réducteur add_messages |
| jump_to | str ou None | non | Contrôle de saut de processus, valeurs facultatives : tools, model, end. Attribut éphémère, automatiquement effacé après utilisation |
| structured_response | Any | non | Résultat de sortie structuré, non exposé dans le schéma d'entrée |
messages - Mécanisme de réduction pour l'historique des messages
Le champ messages utiliseadd_messagescomme réducteur. Cela signifie que lors de la mise à jour des messages, il ne les écrase pas, maisles ajoute。
Exemple
from langchain.messages import HumanMessage, AIMessage
from langgraph.graph.message import add_messages
# Comment fonctionne add_messages
existing = [
HumanMessage(content="Bonjour", id="1"),
AIMessage(content="Bonjour !", id="2"),
]
# Ajouter un nouveau message
new_msg = AIMessage(content="En quoi puis-je vous aider ?", id="3")
result = add_messages(existing, [new_msg])
print(f"Avant fusion : {len(existing)} messages")
print(f"Après fusion : {len(result)} messages")
for msg in result:
print(f" [{msg.type}] {msg.content}")
from langgraph.graph.message import add_messages
# Comment fonctionne add_messages
existing = [
HumanMessage(content="Bonjour", id="1"),
AIMessage(content="Bonjour !", id="2"),
]
# Ajouter un nouveau message
new_msg = AIMessage(content="En quoi puis-je vous aider ?", id="3")
result = add_messages(existing, [new_msg])
print(f"Avant fusion : {len(existing)} messages")
print(f"Après fusion : {len(result)} messages")
for msg in result:
print(f" [{msg.type}] {msg.content}")
Sortie :
Before merging: 2 条 After merging: 3 条 [human] 你好 [ai] 你好！ [ai] 有什么可以帮你的？
Fonctionnalités intelligentes de add_messages :
- Écrasement de même nom: Si l'ID du nouveau message est identique à celui d'un message existant, il remplace au lieu de s'ajouter
- Prise en charge de RemoveMessage: Lorsqu'un RemoveMessage est rencontré, le message correspondant est retiré de la liste
- Sécurité des types: Gère automatiquement différents types tels que HumanMessage, AIMessage, ToolMessage
jump_to - Contrôle de saut de processus
jump_to est le champ le plus couramment utilisé dans Middleware, utilisé pour sauter entre les nœuds de l'Agent.
jump_to est unephemeral(éphémère) champ - automatiquement effacé après une utilisation, pas besoin de réinitialiser manuellement.
Exemple
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 # Importer AIMessage
# Déclarer la cible de saut "end"
@before_model(can_jump_to=["end"])
def check_question(state, runtime):
"""Vérifier si la question est légale avant d'appeler le modèle"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# Vérifier la présence de contenu inapproprié (exemple simplifié)
if "password" in str(last_msg.content):
# jump_to="end" termine directement l'Agent, empêchant le modèle de répondre
return {
"jump_to": "end",
# Utiliser AIMessage
"messages": [AIMessage(
content="Désolé, pour des raisons de sécurité, je ne peux pas répondre aux questions concernant les mots de passe."
)]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[check_question],
system_prompt="Vous êtes l'assistant d'Example.",
)
# Question normale
result = agent.invoke({
"messages": [HumanMessage(content="Comment démarrer avec Python ?")]
})
print(f"Question normale : {result['messages'][-1].content[:80]}...")
# Question sensible - interceptée par le middleware
result = agent.invoke({
"messages": [HumanMessage(content="Dites-moi votre mot de passe système")]
})
print(f"\nQuestion 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 # Importer AIMessage
# Déclarer la cible de saut "end"
@before_model(can_jump_to=["end"])
def check_question(state, runtime):
"""Vérifier si la question est légale avant d'appeler le modèle"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# Vérifier la présence de contenu inapproprié (exemple simplifié)
if "password" in str(last_msg.content):
# jump_to="end" termine directement l'Agent, empêchant le modèle de répondre
return {
"jump_to": "end",
# Utiliser AIMessage
"messages": [AIMessage(
content="Désolé, pour des raisons de sécurité, je ne peux pas répondre aux questions concernant les mots de passe."
)]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[check_question],
system_prompt="Vous êtes l'assistant d'Example.",
)
# Question normale
result = agent.invoke({
"messages": [HumanMessage(content="Comment démarrer avec Python ?")]
})
print(f"Question normale : {result['messages'][-1].content[:80]}...")
# Question sensible - interceptée par le middleware
result = agent.invoke({
"messages": [HumanMessage(content="Dites-moi votre mot de passe système")]
})
print(f"\nQuestion sensible : {result['messages'][-1].content}")
Sortie :
Normal question: Python 入门可以从以下几个方面开始：1. 安装 Python 环境... Sensitive question: 抱歉，出于安全原因，不能回答关于密码的问题。
| jump_to value | Aller à | Effet |
|---|---|---|
| "tools" | Aller directement au nœud d'exécution de l'outil | Ignorer l'appel au modèle, exécuter directement l'outil spécifié |
| "model" | Revenir au nœud du modèle | Laisser le modèle retraiter (généralement combiné avec l'injection de message d'outil) |
| "end" | Mettre fin à la boucle de l'Agent | Aller directement à after_agent ou end |
jump_to est éphémère - automatiquement effacé après chaque exécution de nœud. Cela signifie que vous n'avez pas besoin de remettre manuellement jump_to à None après un saut ; l'Agent le gère automatiquement.
structured_response - Obtenir une sortie structurée
Lorsque le paramètre response_format est utilisé, l'Agent stocke la sortie structurée dans le champ structured_response :
Exemple
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):
"""Résultat de recommandation de cours"""
course_name: str = Field(description="Nom du cours recommandé")
reason: str = Field(description="Raison de la recommandation")
difficulty: str = Field(description="Niveau de difficulté : Débutant/Intermédiaire/Avancé")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
response_format=CourseRecommendation,
system_prompt="Vous êtes le conseiller pédagogique d'Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Je veux apprendre à programmer, recommandez un cours adapté aux débutants complets")]
})
# Obtenir un résultat structuré à partir de structured_response
if "structured_response" in result:
rec = result["structured_response"]
print(f"Cours recommandé : {rec.course_name}")
print(f"Raison : {rec.reason}")
print(f"Niveau de difficulté : {rec.difficulty}")
# structured_response n'est pas dans le schéma de sortie
# Par conséquent, il n'apparaît pas automatiquement dans le résultat renvoyé à l'appelant (configurable)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
class CourseRecommendation(BaseModel):
"""Résultat de recommandation de cours"""
course_name: str = Field(description="Nom du cours recommandé")
reason: str = Field(description="Raison de la recommandation")
difficulty: str = Field(description="Niveau de difficulté : Débutant/Intermédiaire/Avancé")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
response_format=CourseRecommendation,
system_prompt="Vous êtes le conseiller pédagogique d'Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Je veux apprendre à programmer, recommandez un cours adapté aux débutants complets")]
})
# Obtenir un résultat structuré à partir de structured_response
if "structured_response" in result:
rec = result["structured_response"]
print(f"Cours recommandé : {rec.course_name}")
print(f"Raison : {rec.reason}")
print(f"Niveau de difficulté : {rec.difficulty}")
# structured_response n'est pas dans le schéma de sortie
# Par conséquent, il n'apparaît pas automatiquement dans le résultat renvoyé à l'appelant (configurable)
Sortie :
Recommended course: Python3 基础教程 Reason: Python 语法简洁，适合零基础入门，应用范围广泛 Level: 入门
Extension d'état personnalisée
Dans les applications réelles, vous pouvez avoir besoin que l'Agent maintienne un état supplémentaire. Étendez-le en héritant de AgentState :
Exemple
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
# Étendre AgentState, ajouter des champs métier
class ShoppingAgentState(AgentState):
"""État de l'assistant d'achat"""
cart: list[str] # Liste des articles du panier
total_price: float # Prix total
@tool
def add_to_cart(
item: str,
price: float,
state: Annotated[dict, InjectedState],
) -> str:
"""Ajouter un article au panier.
Arguments :
item: nom de l'article
price: prix de l'article
"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
return {
"cart": cart + [item],
"total_price": total + price,
"messages": [], # Ne pas ajouter de messages supplémentaires
}
@tool
def view_cart(
state: Annotated[dict, InjectedState],
) -> str:
"""Afficher le contenu du panier"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
if not cart:
return "Le panier est vide"
items = "、".join(cart)
return f"Panier : {items}, prix 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, # Utiliser un état personnalisé
system_prompt="Vous êtes l'assistant d'achat du magasin Example.",
)
# L'état initial contient un panier vide
result = agent.invoke({
"messages": [HumanMessage(content="Aidez-moi à ajouter un tutoriel Python au panier, prix 49.9")],
"cart": [],
"total_price": 0.0,
})
print(f"Panier : {result.get('cart', [])}")
print(f"Prix total : ¥{result.get('total_price', 0):.2f}")
print(f"Réponse : {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
# Étendre AgentState, ajouter des champs métier
class ShoppingAgentState(AgentState):
"""État de l'assistant d'achat"""
cart: list[str] # Liste des articles du panier
total_price: float # Prix total
@tool
def add_to_cart(
item: str,
price: float,
state: Annotated[dict, InjectedState],
) -> str:
"""Ajouter un article au panier.
Arguments :
item: nom de l'article
price: prix de l'article
"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
return {
"cart": cart + [item],
"total_price": total + price,
"messages": [], # Ne pas ajouter de messages supplémentaires
}
@tool
def view_cart(
state: Annotated[dict, InjectedState],
) -> str:
"""Afficher le contenu du panier"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
if not cart:
return "Le panier est vide"
items = "、".join(cart)
return f"Panier : {items}, prix 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, # Utiliser un état personnalisé
system_prompt="Vous êtes l'assistant d'achat du magasin Example.",
)
# L'état initial contient un panier vide
result = agent.invoke({
"messages": [HumanMessage(content="Aidez-moi à ajouter un tutoriel Python au panier, prix 49.9")],
"cart": [],
"total_price": 0.0,
})
print(f"Panier : {result.get('cart', [])}")
print(f"Prix total : ¥{result.get('total_price', 0):.2f}")
print(f"Réponse : {result['messages'][-1].content}")
Sortie :
Cart: ['Python 教程'] Total: ¥49.90 Reply: 已将《Python 教程》（¥49.90）添加到购物车。当前购物车共 1 件商品，总价 ¥49.90.
state_schema vs state_schema du middleware
Vous pouvez étendre l'état via le paramètre state_schema de create_agent(), ou via le state_schema de Middleware. La différence entre les deux :
| Méthode | Cas d'utilisation | Priorité |
|---|---|---|
| create_agent(state_schema=...) | Extension d'état globale, partagée par tous les nœuds | Priorité la plus élevée (remplace les champs du même nom dans le middleware) |
| AgentMiddleware(state_schema=...) | Extension d'état pour un middleware spécifique | Priorité inférieure, peut être surchargée par create_agent |
Autres extensionsPratique recommandée : Placez les champs d'état métier communs dans state_schema, et placez les champs internes spécifiques au middleware dans le state_schema du middleware. Cela clarifie les responsabilités et évite les contaminations croisées.