Implementation:Langchain ai Langgraph Pregel Get State History

Attribute	Value
Page Type	Implementation (API Doc)
Library	langgraph
Workflow	Persistence_and_Memory_Setup
Principle	Langchain_ai_Langgraph_Thread_Lifecycle_Management
Implementation	Pregel_Get_State_History
Source	`libs/langgraph/langgraph/pregel/main.py:L1319-1398`, `libs/checkpoint/langgraph/checkpoint/base/__init__.py:L193-215`

Overview

Pregel.get_state_history() retrieves the full execution history of a graph thread as an iterator of StateSnapshot objects. It delegates to BaseCheckpointSaver.list() on the underlying checkpointer, which performs the actual checkpoint enumeration and filtering.

Description

These two methods work together to provide thread history browsing:

get_state_history() on the compiled graph: The user-facing API that handles config normalization, subgraph routing, and snapshot preparation.
BaseCheckpointSaver.list(): The checkpointer-level API that performs the actual database query and returns raw CheckpointTuple objects.

The compiled graph converts each CheckpointTuple into a StateSnapshot, which includes higher-level information like which nodes will execute next.

Usage

Import

# get_state_history is a method on the compiled graph instance
# No separate import needed; it is available on any compiled graph with a checkpointer

# BaseCheckpointSaver.list is available via:
from langgraph.checkpoint.base import BaseCheckpointSaver

Code Reference

Source Location

Method	File	Lines
`Pregel.get_state_history`	`libs/langgraph/langgraph/pregel/main.py`	L1319-1368
`Pregel.aget_state_history`	`libs/langgraph/langgraph/pregel/main.py`	L1370-1398
`BaseCheckpointSaver.list`	`libs/checkpoint/langgraph/checkpoint/base/__init__.py`	L193-215

Pregel.get_state_history Signature

def get_state_history(
    self,
    config: RunnableConfig,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None,
) -> Iterator[StateSnapshot]:
    """Get the history of the state of the graph."""

Pregel.aget_state_history Signature

async def aget_state_history(
    self,
    config: RunnableConfig,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None,
) -> AsyncIterator[StateSnapshot]:
    """Asynchronously get the history of the state of the graph."""

BaseCheckpointSaver.list Signature

def list(
    self,
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints that match the given criteria.

    Args:
        config: Base configuration for filtering checkpoints.
        filter: Additional filtering criteria.
        before: List checkpoints created before this configuration.
        limit: Maximum number of checkpoints to return.

    Returns:
        Iterator of matching checkpoint tuples.
    """

I/O Contract

get_state_history

Parameter	Type	Default	Description
`config`	`RunnableConfig`	(required)	Must contain `{"configurable": {"thread_id": "..."}}`. Optionally include `checkpoint_ns` for subgraph history.
`filter`	None	`None`	Metadata key-value pairs that must match. Example: `{"source": "loop"}`.
`before`	None	`None`	Only return checkpoints created before the checkpoint specified in this config. Used for cursor-based pagination.
`limit`	None	`None`	Maximum number of state snapshots to return. `None` returns all.
Returns	Type		Description
	`Iterator[StateSnapshot]`		State snapshots ordered newest to oldest.

BaseCheckpointSaver.list

Parameter	Type	Default	Description
`config`	None	(required)	Base configuration. If `None`, lists across all threads (implementation-dependent).
`filter`	None	`None`	Metadata filtering criteria.
`before`	None	`None`	Cursor for pagination: only return checkpoints before this one.
`limit`	None	`None`	Maximum number of results.
Returns	Type		Description
	`Iterator[CheckpointTuple]`		Checkpoint tuples ordered by checkpoint_id descending (newest first).

Raises

Exception	Condition
`ValueError("No checkpointer set")`	Called on a graph compiled without a checkpointer.
`ValueError("Subgraph ... not found")`	Specified `checkpoint_ns` does not match any subgraph.

Usage Examples

Browse Full Thread History

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

for state in graph.get_state_history(config):
    print(f"Step {state.metadata.get('step')}: "
          f"source={state.metadata.get('source')}, "
          f"values={state.values}")

Paginated History Browsing

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

# Get the first 5 states
states = list(graph.get_state_history(config, limit=5))

# Get the next 5 states (cursor-based pagination)
if states:
    more_states = list(graph.get_state_history(
        config,
        before=states[-1].config,
        limit=5,
    ))

Filter by Metadata

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

# Only get checkpoints created during the execution loop
loop_states = list(graph.get_state_history(
    config,
    filter={"source": "loop"},
))

Time-Travel: Resume from Historical State

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

# Find a past state to resume from
for state in graph.get_state_history(config):
    if state.metadata.get("step") == 3:
        # Resume execution from step 3
        result = graph.invoke(None, state.config)
        break

Async History Browsing

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

async for state in graph.aget_state_history(config, limit=10):
    print(state.values)

Using BaseCheckpointSaver.list Directly

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
# ... after running a graph with this checkpointer ...

config = {"configurable": {"thread_id": "thread-1"}}
for checkpoint_tuple in checkpointer.list(config, limit=5):
    print(f"ID: {checkpoint_tuple.checkpoint['id']}, "
          f"Step: {checkpoint_tuple.metadata.get('step')}")

Delete a Thread

# Clean up all checkpoints for a thread
checkpointer.delete_thread("thread-1")

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment