Доступ к инструментам LangChain -- InjectedState и InjectedStore

Иногда инструментам нужен доступ к более контекстной информации, такой как текущее состояние диалога, постоянные данные пользователя и т. д.

LangChain использует механизм внедрения зависимостей, чтобы функции инструментов могли автоматически получать эту информацию.

InjectedState — Доступ к состоянию агента в инструментах

По умолчанию инструменты могут получать только предоставленные моделью данные через параметры. Но иногда инструментам нужно знать контекст текущего диалога — например, предыдущую историю диалога, информацию, уже подтверждённую пользователем, и т. д.

InjectedStateПозволяет инструментам напрямую читать полное состояние агента.

Пример

from typing import Annotated, Any
from dotenv import load_dotenv
load_dotenv()

from langchain.tools import tool, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


@tool
def remember_preference(
    preference: str,
    state: Annotated[dict[str, Any], InjectedState],
) -> str:
    """Запомнить настройки предпочтений пользователя.

    Аргументы:
preference: содержимое предпочтения пользователя
state: текущее состояние Agent, автоматически внедряемое системой
    """
    # предыдущая история сообщений из state
    messages = state.get("messages", [])
    message_count = len(messages)

    # Можно читать любое поле в state
    previous_prefs = state.get("user_preferences", "none")

    return (
        f"Предпочтение запомнено: {preference}."
        f"(В текущем разговоре {message_count} сообщений,"
        f"Предыдущее предпочтение: {previous_prefs})"
    )


# InjectedState внедряется автоматически, Agent не нужно передавать этот параметр
result = remember_preference.invoke({
    "preference": "тёмная тема",
    # state не нужно передавать, он автоматически внедряется фреймворком
})
print(result)

Результат выполнения:

已记住偏好: 暗色主题。(当前对话共 0 条消息，之前偏好: 无)

Использование InjectedState в агенте

Пример

from typing import Annotated, Any
from langchain.tools import tool, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


@tool
def conversation_stats(
    state: Annotated[dict[str, Any], InjectedState],
) -> str:
    """Получить статистическую информацию о текущем разговоре, например количество сообщений, длину разговора и т. д.

Параметры не требуются; статистика автоматически считывается из текущего состояния.
    """
    messages = state.get("messages", [])
    human_msgs = [m for m in messages if m.type == "human"]
    ai_msgs = [m for m in messages if m.type == "ai"]
    tool_msgs = [m for m in messages if m.type == "tool"]

    return (
        f"Статистика разговора: {len(messages)} сообщений всего | "
        f"Сообщения человека: {len(human_msgs)} | "
        f"Ответы ИИ: {len(ai_msgs)} | "
        f"Вызовы инструментов: {len(tool_msgs)}"
    )

InjectedState позволяет получить доступ ко всем полям в AgentState. Если вы расширите state_schema (добавив пользовательские поля), эти поля также могут быть прочитаны с помощью InjectedState.

InjectedStore — Доступ к постоянному хранилищу в инструментах

Состояние агента существует на уровне разговора; оно исчезает при завершении разговора. Тогда какStore— это межсессионное постоянное хранилище, которое можно использовать для сохранения долгосрочной информации, такой как предпочтения пользователя, прогресс обучения и т. д.

InjectedStoreПозволяет инструментам напрямую читать и записывать в Store.

Пример

from typing import Annotated
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


# Создайте постоянное хранилище (в реальных проектах можно использовать базу данных)
store = InMemoryStore()
# Предварительно задайте некоторые данные
store.put(("users", "user_001"), "profile", {
    "data": {
        "name": "Сяо Мин",
        "level": "Новичок",
        "completed_courses": ["Учебник по основам HTML"]
    }
})


@tool
def get_user_profile(
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Получить информацию об учебном профиле текущего пользователя.

Прочитать данные пользователя из постоянного хранилища.
    """
    # Прочитать данные из Store
    # Store использует namespace (namespace, key) для организации данных
    item = store.get(("users", "user_001"), "profile")

    if item is None:
        return "Профиль пользователя не найден"

    profile = item.value["data"]
    return (
        f"Профиль пользователя: имя={profile['name']},"
        f"уровень={profile['level']},"
        f"завершённые курсы={', '.join(profile['completed_courses'])}"
    )


@tool
def save_course_progress(
    course_name: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Сохранить прогресс обучения пользователя в постоянном хранилище.

    Аргументы:
course_name: название завершённого курса
store: постоянное хранилище, автоматически внедряемое системой
    """
    # Прочитать существующие данные
    item = store.get(("users", "user_001"), "profile")
    profile = item.value["data"] if item else {"name": "Сяо Мин", "level": "Начинающий", "completed_courses": []}

    # Обновить список курсов
    if course_name not in profile["completed_courses"]:
        profile["completed_courses"].append(course_name)

    # Записать обратно в Store
    store.put(("users", "user_001"), "profile", {"data": profile})

    return (
        f"Прогресс обучения обновлён! Пройдено {len(profile['completed_courses'])} курсов:"
        f"{', '.join(profile['completed_courses'])}"
    )


# Протестировать инструмент
print(get_user_profile.invoke({}))
print(save_course_progress.invoke({"course_name": "Базовое руководство по Python3"}))
print(get_user_profile.invoke({}))

Результат выполнения:

用户档案：姓名=小明，水平=入门，已完成课程=HTML 基础教程
学习进度已更新！已完成 2 门课程：HTML 基础教程, Python3 基础教程
用户档案：姓名=小明，水平=入门，已完成课程=HTML 基础教程, Python3 基础教程

Использование Store в агенте

Передайте Store в create_agent(), и все инструменты в агенте смогут получить к нему доступ через InjectedStore:

Пример

from typing import Annotated
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

# Создать хранилище и предустановленные данные
store = InMemoryStore()
store.put(("example", "courses"), "catalog", {
    "data": {
        "Базовое руководство по Python3": {"price": "Бесплатно", "duration": "20 часов"},
        "Анализ данных на Python": {"price": "Для участников", "duration": "30 часов"},
        "Базовое руководство по HTML": {"price": "Бесплатно", "duration": "15 часов"},
    }
})


@tool
def query_course_price(
    course_name: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Запросить информацию о цене указанного курса на EXAMPLE.

    Аргументы:
course_name: название курса
    """
    item = store.get(("example", "courses"), "catalog")
    catalog = item.value["data"] if item else {}

    if course_name in catalog:
        info = catalog[course_name]
        return f"{course_name} - Цена: {info['price']}, Продолжительность обучения: {info['duration']}"
    return f"Курс не найден: {course_name}"


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[query_course_price],
    store=store,  # Передайте Store в Agent
    system_prompt="Вы консультант по курсам в EXAMPLE.",
)

result = agent.invoke({
    "messages": [HumanMessage(content="Сколько стоят Python3 Basic Tutorial и Python Data Analysis соответственно?")]
})
print(result["messages"][-1].content)

Результат выполнения:

根据查询结果：
- 《Python3 基础教程》是免费的，学习时长约 20 小时
- 《Python 数据分析》是会员课程，学习时长约 30 小时

建议先从免费的 Python3 基础教程开始学习。

Сравнение InjectedState и InjectedStore

ИзмерениеInjectedStateInjectedStore
Область действияТекущий диалог (один запуск Agent)Межсессионный (общий для нескольких запусков Agent)
Жизненный циклИсчезает при завершении разговораПостоянное хранилище
Типичные способы использованияЧтение истории сообщений, промежуточных результатов текущего разговораПользовательские предпочтения, прогресс обучения, информация о конфигурации
Способ внедренияInjectedState (внедряется автоматически)InjectedStore() (обязательны скобки)
Организация данныхПлоский словарьПространство имён + иерархия ключей

InjectedToolArg — Пометка обобщённых внедряемых параметров

Помимо специальных InjectedState и InjectedStore, вы также можете использоватьInjectedToolArgчтобы пометить любой параметр, который должен быть внедрён фреймворком:

Пример

from typing import Annotated
from langchain.tools import tool, InjectedToolArg


@tool
def my_tool(
    normal_param: str,
    injected_param: Annotated[str, InjectedToolArg],
) -> str:
    """Пример инструмента, содержащего внедрённый параметр.

    Аргументы:
normal_param: этот параметр предоставляется моделью
injected_param: этот параметр внедряется фреймворком (Агент не должен его предоставлять)
    """
    return f"normal={normal_param}, injected={injected_param}"

InjectedToolArg — это универсальный маркер внедрения; InjectedState, InjectedStore и InjectedToolCallId — все они реализованы на его основе. В большинстве случаев достаточно использовать специальные маркеры внедрения; InjectedToolArg используется для расширения пользовательской логики внедрения.

Другие расширения