Управление состоянием AgentState в LangChain
Во время выполнения Agent должен поддерживать состояние — историю сообщений, структурированные ответы, управление процессом и т. д. Понимание структуры и использования AgentState — ключ к настройке поведения Agent.
Структура AgentState
AgentState — это TypedDict, который по умолчанию содержит три поля:
Пример
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
# Фактическое определение AgentState (упрощённо)
class AgentState(TypedDict):
# messages: история сообщений, использующая add_messages в качестве редьюсера
# Required означает, что его необходимо указать при вызове
messages: Required[Annotated[list[AnyMessage], add_messages]]
# jump_to: обработка управления переходом, временный (автоматически очищается после использования)
# NotRequired означает, что параметр необязателен
jump_to: NotRequired[Annotated[str | None, EphemeralValue]]
# structured_response: результат структурированного вывода
# NotRequired означает, что параметр необязателен, присутствует только когда задан response_format
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
# Фактическое определение AgentState (упрощённо)
class AgentState(TypedDict):
# messages: история сообщений, использующая add_messages в качестве редьюсера
# Required означает, что его необходимо указать при вызове
messages: Required[Annotated[list[AnyMessage], add_messages]]
# jump_to: обработка управления переходом, временный (автоматически очищается после использования)
# NotRequired означает, что параметр необязателен
jump_to: NotRequired[Annotated[str | None, EphemeralValue]]
# structured_response: результат структурированного вывода
# NotRequired означает, что параметр необязателен, присутствует только когда задан response_format
structured_response: NotRequired[Any]
| Поле | Type | Обязательно? | Описание |
|---|---|---|---|
| messages | list[AnyMessage] | Да | История сообщений, добавляемая с помощью редьюсера add_messages |
| jump_to | str или None | нет | Управление переходом процесса, необязательные значения: tools, model, end. Эфемерный атрибут, автоматически очищается после использования |
| structured_response | Any | нет | Результат структурированного вывода, не отображается во входной схеме |
messages - механизм редьюсера для истории сообщений
Поле messages используетadd_messagesreducer. Это означает, что при обновлении messages происходит не перезапись, адобавление。
Пример
from langchain.messages import HumanMessage, AIMessage
from langgraph.graph.message import add_messages
# Как работает add_messages
existing = [
HumanMessage(content="Hello", id="1"),
AIMessage(content="Hello!", id="2"),
]
# Добавить новое сообщение
new_msg = AIMessage(content="What can I help you with?", id="3")
result = add_messages(existing, [new_msg])
print(f"До объединения: {len(existing)} сообщений")
print(f"После объединения: {len(result)} сообщений")
for msg in result:
print(f" [{msg.type}] {msg.content}")
from langgraph.graph.message import add_messages
# Как работает add_messages
existing = [
HumanMessage(content="Hello", id="1"),
AIMessage(content="Hello!", id="2"),
]
# Добавить новое сообщение
new_msg = AIMessage(content="What can I help you with?", id="3")
result = add_messages(existing, [new_msg])
print(f"До объединения: {len(existing)} сообщений")
print(f"После объединения: {len(result)} сообщений")
for msg in result:
print(f" [{msg.type}] {msg.content}")
Вывод:
Before merging: 2 条 After merging: 3 条 [human] 你好 [ai] 你好！ [ai] 有什么可以帮你的？
Умные возможности add_messages:
- Перезапись при совпадении имени: Если ID нового сообщения совпадает с ID существующего сообщения, новое сообщение заменяет его, а не добавляется
- Поддержка RemoveMessage: Когда встречается RemoveMessage, соответствующее сообщение удаляется из списка
- Типобезопасность: Автоматически обрабатывает различные типы, такие как HumanMessage, AIMessage, ToolMessage
jump_to — управление переходами процесса
jump_to — наиболее часто используемое поле в Middleware, используется для перехода между узлами Agent.
jump_to — этовременное(ephemeral) поле — автоматически очищается после одного использования, сбрасывать вручную не требуется.
Пример
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 # Импорт AIMessage
# Объявите цель перехода "end"
@before_model(can_jump_to=["end"])
def check_question(state, runtime):
"""Проверить, допустим ли вопрос перед вызовом модели"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# Проверка на неприемлемое содержимое (упрощённый пример)
if "password" in str(last_msg.content):
# jump_to="end" напрямую завершает Agent, не позволяя модели ответить
return {
"jump_to": "end",
# Использовать AIMessage
"messages": [AIMessage(
content="Извините, по соображениям безопасности я не могу отвечать на вопросы о паролях."
)]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[check_question],
system_prompt="Вы — ассистент Example.",
)
# Обычный вопрос
result = agent.invoke({
"messages": [HumanMessage(content="Как начать работать с Python?")]
})
print(f"Обычный вопрос: {result['messages'][-1].content[:80]}...")
# Конфиденциальный вопрос - перехвачен middleware
result = agent.invoke({
"messages": [HumanMessage(content="Скажи мне свой системный пароль")]
})
print(f"\nКонфиденциальный вопрос: {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 # Импорт AIMessage
# Объявите цель перехода "end"
@before_model(can_jump_to=["end"])
def check_question(state, runtime):
"""Проверить, допустим ли вопрос перед вызовом модели"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# Проверка на неприемлемое содержимое (упрощённый пример)
if "password" in str(last_msg.content):
# jump_to="end" напрямую завершает Agent, не позволяя модели ответить
return {
"jump_to": "end",
# Использовать AIMessage
"messages": [AIMessage(
content="Извините, по соображениям безопасности я не могу отвечать на вопросы о паролях."
)]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[check_question],
system_prompt="Вы — ассистент Example.",
)
# Обычный вопрос
result = agent.invoke({
"messages": [HumanMessage(content="Как начать работать с Python?")]
})
print(f"Обычный вопрос: {result['messages'][-1].content[:80]}...")
# Конфиденциальный вопрос - перехвачен middleware
result = agent.invoke({
"messages": [HumanMessage(content="Скажи мне свой системный пароль")]
})
print(f"\nКонфиденциальный вопрос: {result['messages'][-1].content}")
Вывод:
Normal question: Python 入门可以从以下几个方面开始：1. 安装 Python 环境... Sensitive question: 抱歉，出于安全原因，不能回答关于密码的问题。
| jump_to value | Перейти к | Эффект |
|---|---|---|
| "tools" | Непосредственно перейти к узлу выполнения инструмента | Пропустить вызов модели, непосредственно выполнить указанный инструмент |
| "model" | Вернуться к узлу модели | Позволить модели обработать повторно (обычно в сочетании с внедрением сообщения инструмента) |
| "end" | Завершить цикл агента | Непосредственно перейти к after_agent или end |
jump_to является эфемерным — автоматически очищается после выполнения каждого узла. Это означает, что вам не нужно вручную сбрасывать jump_to обратно в None после перехода; Agent делает это автоматически.
structured_response — получение структурированного вывода
Когда используется параметр response_format, Agent сохраняет структурированный вывод в поле structured_response:
Пример
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_name: str = Field(description="Рекомендуемое название курса")
reason: str = Field(description="Причина рекомендации")
difficulty: str = Field(description="Уровень сложности: начальный/средний/продвинутый")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
response_format=CourseRecommendation,
system_prompt="Вы консультант по обучению в Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Я хочу изучать программирование, порекомендуйте курс, подходящий для полных новичков")]
})
# Получение структурированного результата из structured_response
if "structured_response" in result:
rec = result["structured_response"]
print(f"Рекомендуемый курс: {rec.course_name}")
print(f"Причина: {rec.reason}")
print(f"Уровень сложности: {rec.difficulty}")
# structured_response отсутствует в выходной схеме
# Поэтому он не появляется автоматически в результате, возвращаемом вызывающей стороне (можно настроить)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
class CourseRecommendation(BaseModel):
"""Результат рекомендации курса"""
course_name: str = Field(description="Рекомендуемое название курса")
reason: str = Field(description="Причина рекомендации")
difficulty: str = Field(description="Уровень сложности: начальный/средний/продвинутый")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
response_format=CourseRecommendation,
system_prompt="Вы консультант по обучению в Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Я хочу изучать программирование, порекомендуйте курс, подходящий для полных новичков")]
})
# Получение структурированного результата из structured_response
if "structured_response" in result:
rec = result["structured_response"]
print(f"Рекомендуемый курс: {rec.course_name}")
print(f"Причина: {rec.reason}")
print(f"Уровень сложности: {rec.difficulty}")
# structured_response отсутствует в выходной схеме
# Поэтому он не появляется автоматически в результате, возвращаемом вызывающей стороне (можно настроить)
Вывод:
Recommended course: Python3 基础教程 Reason: Python 语法简洁，适合零基础入门，应用范围广泛 Level: 入门
Расширение пользовательского состояния
В реальных приложениях может потребоваться, чтобы Агент поддерживал дополнительное состояние. Расширьте его, унаследовав AgentState:
Пример
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
# Расширить AgentState, добавить бизнес-поля
class ShoppingAgentState(AgentState):
"""Состояние помощника по покупкам"""
cart: list[str] # Список товаров в корзине
total_price: float # Итоговая цена
@tool
def add_to_cart(
item: str,
price: float,
state: Annotated[dict, InjectedState],
) -> str:
"""Добавить товар в корзину.
Аргументы:
item: название товара
price: цена товара
"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
return {
"cart": cart + [item],
"total_price": total + price,
"messages": [], # Не добавляйте лишние сообщения
}
@tool
def view_cart(
state: Annotated[dict, InjectedState],
) -> str:
"""Просмотреть содержимое корзины покупок"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
if not cart:
return "Корзина покупок пуста"
items = "、".join(cart)
return f"Корзина покупок: {items}, общая стоимость: ¥{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, # Использовать пользовательское состояние
system_prompt="Вы — помощник по покупкам магазина Example.",
)
# Начальное состояние содержит пустую корзину покупок
result = agent.invoke({
"messages": [HumanMessage(content="Помогите мне добавить учебник по Python в корзину покупок, цена 49.9")],
"cart": [],
"total_price": 0.0,
})
print(f"Корзина покупок: {result.get('cart', [])}")
print(f"Итоговая цена: ¥{result.get('total_price', 0):.2f}")
print(f"Ответ: {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
# Расширить AgentState, добавить бизнес-поля
class ShoppingAgentState(AgentState):
"""Состояние помощника по покупкам"""
cart: list[str] # Список товаров в корзине
total_price: float # Итоговая цена
@tool
def add_to_cart(
item: str,
price: float,
state: Annotated[dict, InjectedState],
) -> str:
"""Добавить товар в корзину.
Аргументы:
item: название товара
price: цена товара
"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
return {
"cart": cart + [item],
"total_price": total + price,
"messages": [], # Не добавляйте лишние сообщения
}
@tool
def view_cart(
state: Annotated[dict, InjectedState],
) -> str:
"""Просмотреть содержимое корзины покупок"""
cart = state.get("cart", [])
total = state.get("total_price", 0.0)
if not cart:
return "Корзина покупок пуста"
items = "、".join(cart)
return f"Корзина покупок: {items}, общая стоимость: ¥{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, # Использовать пользовательское состояние
system_prompt="Вы — помощник по покупкам магазина Example.",
)
# Начальное состояние содержит пустую корзину покупок
result = agent.invoke({
"messages": [HumanMessage(content="Помогите мне добавить учебник по Python в корзину покупок, цена 49.9")],
"cart": [],
"total_price": 0.0,
})
print(f"Корзина покупок: {result.get('cart', [])}")
print(f"Итоговая цена: ¥{result.get('total_price', 0):.2f}")
print(f"Ответ: {result['messages'][-1].content}")
Вывод:
Cart: ['Python 教程'] Total: ¥49.90 Reply: 已将《Python 教程》（¥49.90）添加到购物车。当前购物车共 1 件商品，总价 ¥49.90.
state_schema против state_schema промежуточного ПО
Вы можете расширить состояние с помощью параметра state_schema функции create_agent() или с помощью state_schema Middleware. Разница между ними:
| Метод | Сценарий использования | Приоритет |
|---|---|---|
| create_agent(state_schema=...) | Глобальное расширение состояния, общее для всех узлов | Наивысший (переопределяет поля с тем же именем в промежуточном ПО) |
| AgentMiddleware(state_schema=...) | Расширение состояния для конкретного промежуточного ПО | Ниже, может быть переопределено с помощью create_agent |
Другие расширенияРекомендуемая практика: размещайте общие бизнес-поля состояния в state_schema, а внутренние поля, специфичные для промежуточного ПО, помещайте в state_schema промежуточного ПО. Это сохраняет чёткое разделение обязанностей и предотвращает перекрёстное загрязнение.