Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Langchain ai Langgraph InjectedStore For Memory

From Leeroopedia
Attribute Value
Page Type Implementation (API Doc)
Library langgraph (prebuilt, langgraph)
Workflow Persistence_and_Memory_Setup
Principle Langchain_ai_Langgraph_Store_Access_In_Nodes
Implementation InjectedStore_For_Memory
Source libs/prebuilt/langgraph/prebuilt/tool_node.py:L1673-1746, libs/langgraph/langgraph/pregel/main.py:L1235-1275 (get_state), libs/langgraph/langgraph/pregel/main.py:L2309-2320 (update_state)

Overview

InjectedStore is an annotation class that enables dependency injection of the persistent store into tool arguments. Pregel.get_state() and Pregel.update_state() provide external programmatic access to the graph's checkpointed state.

Description

InjectedStore extends InjectedToolArg and acts as a marker annotation. When applied to a tool parameter using typing.Annotated, it instructs the ToolNode to inject the store instance automatically while hiding the parameter from the tool schema presented to language models.

get_state and update_state are methods on the compiled graph (Pregel) that interact with the checkpointer to retrieve and modify graph state externally.

Usage

Import

from langgraph.prebuilt import InjectedStore

Code Reference

Source Location

Component File Lines
InjectedStore libs/prebuilt/langgraph/prebuilt/tool_node.py L1673-1746
Pregel.get_state libs/langgraph/langgraph/pregel/main.py L1235-1275
Pregel.update_state libs/langgraph/langgraph/pregel/main.py L2309-2320

InjectedStore Class

class InjectedStore(InjectedToolArg):
    """Annotation for injecting persistent store into tool arguments.

    This annotation enables tools to access LangGraph's persistent storage system
    without exposing storage details to the language model. Tools annotated with
    InjectedStore receive the store instance automatically during execution while
    remaining invisible to the model's tool-calling interface.
    """

Pregel.get_state Signature

def get_state(
    self, config: RunnableConfig, *, subgraphs: bool = False
) -> StateSnapshot:
    """Get the current state of the graph."""

Pregel.update_state Signature

def update_state(
    self,
    config: RunnableConfig,
    values: dict[str, Any] | Any | None,
    as_node: str | None = None,
    task_id: str | None = None,
) -> RunnableConfig:
    """Update the state of the graph with the given values, as if they came from
    node `as_node`. If `as_node` is not provided, it will be set to the last node
    that updated the state, if not ambiguous.
    """

I/O Contract

InjectedStore

Aspect Description
Annotation Target Tool function parameter
Usage Syntax param: Annotated[Any, InjectedStore()]
Injected Value The BaseStore instance from the compiled graph
Schema Visibility Hidden from the language model's tool schema
Requirement langchain-core >= 0.3.8

get_state

Parameter Type Description
config RunnableConfig Must contain {"configurable": {"thread_id": "..."}}
subgraphs bool Whether to include subgraph states. Default: False
Returns Type Description
StateSnapshot Current state snapshot including values, next nodes, config, and metadata

update_state

Parameter Type Description
config RunnableConfig Must contain {"configurable": {"thread_id": "..."}}
values Any | None State values to apply
as_node None Node to attribute the update to. Defaults to last updating node.
task_id None Optional task identifier for the update.
Returns Type Description
RunnableConfig Updated config with the new checkpoint ID

Usage Examples

Injecting Store into Tools

from typing import Any
from typing_extensions import Annotated
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedStore

@tool
def save_preference(
    key: str,
    value: str,
    store: Annotated[Any, InjectedStore()]
) -> str:
    """Save user preference to persistent storage."""
    store.put(("preferences",), key, {"value": value})
    return f"Saved {key} = {value}"

@tool
def get_preference(
    key: str,
    store: Annotated[Any, InjectedStore()]
) -> str:
    """Retrieve user preference from persistent storage."""
    result = store.get(("preferences",), key)
    return result.value["value"] if result else "Not found"

Using Tools with ToolNode and Store

from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
tool_node = ToolNode([save_preference, get_preference])

builder = StateGraph(State)
builder.add_node("tools", tool_node)
# ... add other nodes and edges ...

graph = builder.compile(store=store)

External State Access with get_state

config = {"configurable": {"thread_id": "thread-1"}}

# Get current state
state = graph.get_state(config)
print(state.values)  # Current channel values
print(state.next)    # Next nodes to execute

External State Modification with update_state

config = {"configurable": {"thread_id": "thread-1"}}

# Inject human feedback into the graph state
updated_config = graph.update_state(
    config,
    values={"messages": [("human", "Approved")]},
    as_node="human_review",
)

# Resume execution with the updated state
result = graph.invoke(None, updated_config)

Namespaced Store Access with User Context

@tool
def save_user_memory(
    memory: str,
    config: Annotated[Any, InjectedToolArg()],
    store: Annotated[Any, InjectedStore()]
) -> str:
    """Save a memory associated with the current user."""
    user_id = config["configurable"].get("user_id", "default")
    store.put(("users", user_id, "memories"), memory[:16], {"text": memory})
    return f"Memory saved for user {user_id}"

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment