Acceso a herramientas de LangChain -- InjectedState e InjectedStore

A veces las herramientas necesitan acceder a más información de contexto, como el estado de la conversación actual, los datos persistentes del usuario, etc.

LangChain, mediante el mecanismo de inyección de dependencias, permite que las funciones de herramienta obtengan esta información automáticamente.

InjectedState—Acceder al estado del Agente en herramientas

De forma predeterminada, las herramientas solo pueden recibir datos del modelo a través de parámetros. Pero a veces las herramientas necesitan conocer el contexto de la conversación actual, como el historial de conversaciones anterior, información ya confirmada por el usuario, etc.

InjectedStatePermite que las herramientas lean directamente el estado completo del Agente.

Ejemplo

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:
    """Recuerda las preferencias del usuario.

    Args:
preference: el contenido de preferencias del usuario
state: el estado actual del Agente inyectado automáticamente por el sistema
    """
    # Obtener el historial de mensajes anterior desde el estado
    messages = state.get("messages", [])
    message_count = len(messages)

    # Se puede leer cualquier campo del estado
    previous_prefs = state.get("user_preferences", "Ninguno")

    return (
        f"Preferencia recordada: {preference}."
        f"(En la conversación actual hay {message_count} mensajes,"
        f"preferencia anterior: {previous_prefs})"
    )


# InjectedState se inyecta automáticamente, el Agente no necesita pasar este parámetro
result = remember_preference.invoke({
    "preference": "tema oscuro",
    # state no se pasa, lo inyecta automáticamente el marco
})
print(result)

Resultado de la ejecución:

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

Usar InjectedState en el Agente

Ejemplo

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:
    """Obtener estadísticas de la conversación actual, como número de mensajes, longitud de la conversación, etc.

No necesita ningún parámetro, la información estadística se lee automáticamente del estado actual.
    """
    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"Estadísticas de conversación: {len(messages)} mensajes en total | "
        f"Mensajes de usuario: {len(human_msgs)} | "
        f"Respuestas de IA: {len(ai_msgs)} | "
        f"Llamadas a herramientas: {len(tool_msgs)}"
    )

InjectedState te permite acceder a todos los campos de AgentState. Si extiendes el state_schema (agregando campos personalizados), InjectedState también podrá leer esos campos.

InjectedStore—Acceder al almacenamiento persistente en herramientas

El estado del Agente (state) es a nivel de conversación; cuando la conversación termina, desaparece. Mientras queStorees un almacenamiento persistente entre sesiones, que se puede usar para guardar información a largo plazo como preferencias del usuario, progreso de aprendizaje, etc.

InjectedStorePermite que las herramientas lean y escriban directamente en Store.

Ejemplo

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


# Crear un almacenamiento persistente (en proyectos reales se puede usar una base de datos)
store = InMemoryStore()
# Precargar algunos datos
store.put(("users", "user_001"), "profile", {
    "data": {
        "name": "Xiaoming",
        "level": "Principiante",
        "completed_courses": ["Tutorial básico de HTML"]
    }
})


@tool
def get_user_profile(
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Obtener la información del perfil de aprendizaje del usuario actual.

Leer los datos del usuario desde el almacenamiento persistente.
    """
    # Leer datos de Store
    # Store usa espacios de nombres (namespace, key) para organizar los datos
    item = store.get(("users", "user_001"), "profile")

    if item is None:
        return "No se encontró el perfil del usuario"

    profile = item.value["data"]
    return (
        f"Perfil del usuario: nombre={profile['name']},"
        f"nivel={profile['level']},"
        f"cursos completados={', '.join(profile['completed_courses'])}"
    )


@tool
def save_course_progress(
    course_name: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Guardar el progreso de aprendizaje del usuario en el almacenamiento persistente.

    Args:
course_name: el nombre del curso completado
store: el almacenamiento persistente inyectado automáticamente por el sistema
    """
    # Leer datos existentes
    item = store.get(("users", "user_001"), "profile")
    profile = item.value["data"] if item else {"name": "Xiaoming", "level": "Principiante", "completed_courses": []}

    # Actualizar la lista de cursos
    if course_name not in profile["completed_courses"]:
        profile["completed_courses"].append(course_name)

    # Escribir de vuelta en Store
    store.put(("users", "user_001"), "profile", {"data": profile})

    return (
        f"¡Progreso de aprendizaje actualizado! Cursos completados: {len(profile['completed_courses'])}"
        f"{', '.join(profile['completed_courses'])}"
    )


# Probar la herramienta
print(get_user_profile.invoke({}))
print(save_course_progress.invoke({"course_name": "Tutorial básico de Python3"}))
print(get_user_profile.invoke({}))

Resultado de la ejecución:

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

Combinar Store en el Agente

Si pasas Store a create_agent(), todas las herramientas del Agente pueden acceder a él a través de InjectedStore:

Ejemplo

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

# Crear Store y precargar datos
store = InMemoryStore()
store.put(("ejemplo", "courses"), "catalog", {
    "data": {
        "Tutorial básico de Python3": {"price": "Gratis", "duration": "20 horas"},
        "Análisis de datos con Python": {"price": "Premium", "duration": "30 horas"},
        "Tutorial básico de HTML": {"price": "Gratis", "duration": "15 horas"},
    }
})


@tool
def query_course_price(
    course_name: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Consultar la información de precios del curso especificado en EJEMPLO.

    Args:
course_name: el nombre del curso
    """
    item = store.get(("ejemplo", "courses"), "catalog")
    catalog = item.value["data"] if item else {}

    if course_name in catalog:
        info = catalog[course_name]
        return f"《{course_name}》- Precio: {info['price']}, Duración: {info['duration']}"
    return f"No se encontró el curso 《{course_name}》"


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[query_course_price],
    store=store,  # Pasar Store al Agente
    system_prompt="Eres el asesor de cursos de EJEMPLO.",
)

result = agent.invoke({
    "messages": [HumanMessage(content="¿Cuánto cuestan el Tutorial básico de Python3 y Análisis de datos con Python respectivamente?")]
})
print(result["messages"][-1].content)

Resultado de la ejecución:

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

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

Comparación entre InjectedState e InjectedStore

DimensiónInjectedStateInjectedStore
AlcanceConversación actual (una sola ejecución del Agente)Entre sesiones (compartido entre múltiples ejecuciones del Agente)
Ciclo de vidaDesaparece al finalizar la conversaciónAlmacenamiento persistente
Usos típicosLeer historial de mensajes, resultados intermedios de la conversación actualPreferencias del usuario, progreso de aprendizaje, información de configuración
Forma de inyecciónInjectedState (inyección automática)InjectedStore() (requiere paréntesis)
Organización de datosDiccionario planoEstructura jerárquica de espacio de nombres + clave

InjectedToolArg—Marca parámetros de inyección genéricos

Además de los dedicados InjectedState e InjectedStore, también puedes usarInjectedToolArgpara marcar cualquier parámetro que necesite ser inyectado por el marco:

Ejemplo

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


@tool
def my_tool(
    normal_param: str,
    injected_param: Annotated[str, InjectedToolArg],
) -> str:
    """Una herramienta de ejemplo con parámetros de inyección.

    Args:
normal_param: este parámetro lo proporciona el modelo
injected_param: este parámetro lo inyecta el marco (el Agente no necesita proporcionarlo)
    """
    return f"normal={normal_param}, injected={injected_param}"

InjectedToolArg es una marca de inyección genérica; InjectedState, InjectedStore e InjectedToolCallId se implementan basados en ella. En la mayoría de los casos basta con usar las marcas de inyección específicas; InjectedToolArg se usa para extender la lógica de inyección personalizada.

Otras extensiones