Implementation:Langchain ai Langgraph Pregel Get State

Property	Value
API	Pregel.get_state(self, config, *, subgraphs=False) and Pregel.update_state(self, config, values, as_node=None, task_id=None)
Type	API Doc
Workflow	Human_in_the_Loop_Agent
Pipeline Stage	State Management
Repository	Langchain_ai_Langgraph
Source	libs/langgraph/langgraph/pregel/main.py:L1235-1275 (get_state), L2309-2320 (update_state)
Also covers	HumanResponse TypedDict at libs/prebuilt/langgraph/prebuilt/interrupt.py:L87-106

Overview

Pregel.get_state() retrieves a StateSnapshot of the current graph execution for a given thread, including channel values, pending tasks, interrupts, and metadata. Pregel.update_state() modifies the graph state by applying values as if they came from a specified node. Together, these methods enable human operators to inspect a paused graph and modify its state before resuming.

The HumanResponse TypedDict provides a structured format for human responses to interrupts, with four response types: accept, ignore, response, and edit.

Description

get_state()

The method performs the following steps:

1. Resolves the checkpointer from the config or the graph's default checkpointer.
2. If the config specifies a subgraph namespace and the checkpointer is not already overridden, it locates the correct subgraph instance and delegates to its get_state().
3. Merges the graph's base config with the provided config.
4. Calls checkpointer.get_tuple(config) to retrieve the latest checkpoint for the thread.
5. Calls _prepare_state_snapshot() to build a StateSnapshot from the checkpoint data, optionally recursing into subgraphs.

If no checkpoint_id is specified in the config, pending writes are applied to the snapshot, giving the "current" state. If a specific checkpoint_id is provided, the snapshot reflects the state at that exact checkpoint.

update_state()

The method delegates to bulk_update_state() with a single StateUpdate. It applies the given values to the graph state as if they came from the node specified by as_node. If as_node is not provided, it defaults to the last node that updated the state. The task_id parameter allows targeting a specific task within a super-step.

HumanResponse TypedDict

A structured response format for human-in-the-loop interactions:

• type: One of "accept", "ignore", "response", or "edit"
• args: None (for accept/ignore), str (for text responses), or ActionRequest (for edit actions)

Usage

from langgraph.prebuilt import HumanResponse

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

# Inspect state after an interrupt
state = graph.get_state(config)
print(state.values)       # Current channel values
print(state.next)         # Nodes that would execute next
print(state.interrupts)   # Pending interrupts

# Modify state before resuming
graph.update_state(config, {"messages": [new_message]}, as_node="agent")

# Create a structured human response
response = HumanResponse(type="accept", args=None)

Code Reference

Source Location

Item	Path
Pregel.get_state()	libs/langgraph/langgraph/pregel/main.py, Lines 1235-1275
Pregel.aget_state()	libs/langgraph/langgraph/pregel/main.py, Lines 1277-1317
Pregel.update_state()	libs/langgraph/langgraph/pregel/main.py, Lines 2309-2320
Pregel.aupdate_state()	libs/langgraph/langgraph/pregel/main.py, Lines 2322-2335
StateSnapshot NamedTuple	libs/langgraph/langgraph/types.py, Lines 268-287
HumanResponse TypedDict	libs/prebuilt/langgraph/prebuilt/interrupt.py, Lines 87-106

Signature

def get_state(
    self, config: RunnableConfig, *, subgraphs: bool = False
) -> StateSnapshot:

def update_state(
    self,
    config: RunnableConfig,
    values: dict[str, Any] | Any | None,
    as_node: str | None = None,
    task_id: str | None = None,
) -> RunnableConfig:

class HumanResponse(TypedDict):
    type: Literal["accept", "ignore", "response", "edit"]
    args: None | str | ActionRequest

Import

from langgraph.prebuilt import HumanResponse

I/O Contract

get_state()

Parameter	Type	Default	Description
config	RunnableConfig	(required)	Must contain configurable.thread_id. Optionally checkpoint_id to retrieve a specific checkpoint.
subgraphs	bool	False	If True, recursively includes state from subgraph executions.

Returns: StateSnapshot -- A named tuple with the following fields:

Field	Type	Description
values	Any	Current values of state channels.
next	tuple[str, ...]	Names of nodes to execute in the next step.
config	RunnableConfig	Config used to fetch this snapshot (includes checkpoint_id).
metadata	None	Metadata about the checkpoint (source, step, writes).
created_at	None	Timestamp of checkpoint creation.
parent_config	None	Config for the parent checkpoint, for history traversal.
tasks	tuple[PregelTask, ...]	Pending tasks with their IDs, names, errors, and interrupts.
interrupts	tuple[Interrupt, ...]	Pending interrupts awaiting resolution.

Raises: ValueError -- If no checkpointer is configured.

update_state()

Parameter	Type	Default	Description
config	RunnableConfig	(required)	Must contain configurable.thread_id.
values	Any | None	(required)	The values to apply to the state.
as_node	None	None	The node name to attribute the update to. Affects routing logic.
task_id	None	None	Optional task ID to target a specific task.

Returns: RunnableConfig -- Updated config pointing to the new checkpoint created by the update.

HumanResponse TypedDict

Field	Type	Description
type	Literal["accept", "ignore", "response", "edit"]	The type of response: accept (approve), ignore (skip), response (text feedback), edit (modify content).
args	str | ActionRequest	Response payload. None for accept/ignore, str for text responses, ActionRequest for edits.

Usage Examples

Inspecting state after an interrupt

import uuid
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START
from langgraph.types import interrupt, Command

class State(TypedDict):
    messages: list
    approved: bool | None

def review_node(state):
    answer = interrupt({"action": "approve_tool_call", "tool": "search"})
    return {"approved": answer == "yes"}

builder = StateGraph(State)
builder.add_node("review", review_node)
builder.add_edge(START, "review")

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# Run until interrupt
list(graph.stream({"messages": [], "approved": None}, config))

# Inspect the paused state
snapshot = graph.get_state(config)
print(f"Next nodes: {snapshot.next}")
# Next nodes: ('review',)
print(f"Interrupts: {snapshot.interrupts}")
# Interrupts: (Interrupt(value={'action': 'approve_tool_call', 'tool': 'search'}, id='...'),)
print(f"Current values: {snapshot.values}")

Modifying state before resuming

# Edit tool call args in the state before allowing execution
graph.update_state(
    config,
    {"messages": [edited_tool_call_message]},
    as_node="agent",
)

# Resume execution
for event in graph.stream(None, config):
    print(event)

Using HumanResponse for structured responses

from langgraph.prebuilt import HumanResponse

# Accept the proposed action
response = HumanResponse(type="accept", args=None)
graph.stream(Command(resume=response), config)

# Provide text feedback
response = HumanResponse(type="response", args="Please use a different search query")
graph.stream(Command(resume=response), config)

# Edit the action
response = HumanResponse(
    type="edit",
    args=ActionRequest(action="search", args={"query": "updated query"})
)
graph.stream(Command(resume=response), config)

Retrieving subgraph state

# Get state with subgraph information
snapshot = graph.get_state(config, subgraphs=True)
for task in snapshot.tasks:
    if task.state is not None:
        print(f"Subgraph state for task {task.name}: {task.state}")

Related Pages

• Langchain_ai_Langgraph_State_Inspection
• Langchain_ai_Langgraph_InMemorySaver_Init
• Langchain_ai_Langgraph_Command_Resume
• Langchain_ai_Langgraph_Interrupt_Function