Implementation:Langchain ai Langgraph PregelProtocol

Attribute	Value
Source	`libs/langgraph/langgraph/pregel/protocol.py` (164 lines)
Domain	Core, Protocol
Principle	Graph_Execution
Library	langgraph
Import	`from langgraph.pregel.protocol import PregelProtocol`

Overview

The `protocol.py` module defines `PregelProtocol`, the abstract protocol class that specifies the complete interface for a compiled LangGraph graph runtime. It also defines `StreamProtocol`, a lightweight class for routing stream chunks to consumer callbacks. Together they establish the contract that all graph implementations must satisfy.

Description

`PregelProtocol(Runnable[InputT, Any], Generic[StateT, ContextT, InputT, OutputT])` is an abstract class that extends LangChain's `Runnable` and parameterizes over four type variables: `StateT` (graph state), `ContextT` (run context), `InputT` (graph input), and `OutputT` (graph output).

It defines the following abstract methods (each with both sync and async variants):

Execution methods:

• `invoke` / `ainvoke` -- Run the graph to completion on a given input and return the final output. Accepts optional `context`, `interrupt_before`, and `interrupt_after` parameters.
• `stream` / `astream` -- Run the graph and yield intermediate results as an iterator. Supports `stream_mode` (e.g., `"values"`, `"updates"`, `"debug"`), `subgraphs` flag, and interrupt configuration.

State inspection methods:

• `get_state` / `aget_state` -- Retrieve the current `StateSnapshot` for a given config, optionally including subgraph state.
• `get_state_history` / `aget_state_history` -- Iterate over historical `StateSnapshot` entries with optional `filter`, `before`, and `limit` parameters.

State mutation methods:

• `update_state` / `aupdate_state` -- Apply values to the state as if they were returned by a specific node (`as_node`).
• `bulk_update_state` / `abulk_update_state` -- Apply multiple batches of `StateUpdate` sequences atomically.

Configuration and introspection methods:

• `with_config` -- Return a copy of the graph bound to a specific `RunnableConfig`.
• `get_graph` / `aget_graph` -- Return a `DrawableGraph` representation for visualization, with optional `xray` depth for subgraph expansion.

`StreamProtocol` is a slot-based class that bundles a callable and a set of `StreamMode` values. It is used internally to dispatch `StreamChunk` tuples (namespace tuple, mode string, data) to the appropriate stream consumers.

Usage

from langgraph.pregel.protocol import PregelProtocol

# PregelProtocol is not instantiated directly. It is the base protocol
# implemented by compiled graph objects such as CompiledStateGraph.
# Use it for type annotations:

def process_graph(graph: PregelProtocol) -> dict:
    return graph.invoke({"messages": [("user", "Hello")]})

Code Reference

PregelProtocol Methods

Method	Signature	Description
`invoke`	Any`	Execute graph to completion synchronously.
`ainvoke`	Any`	Execute graph to completion asynchronously.
`stream`	`(input, config, *, context, stream_mode, interrupt_before, interrupt_after, subgraphs) -> Iterator`	Stream intermediate outputs synchronously.
`astream`	`async (input, config, *, context, stream_mode, interrupt_before, interrupt_after, subgraphs) -> AsyncIterator`	Stream intermediate outputs asynchronously.
`get_state`	`(config, *, subgraphs) -> StateSnapshot`	Get current state snapshot.
`aget_state`	`async (config, *, subgraphs) -> StateSnapshot`	Get current state snapshot asynchronously.
`get_state_history`	`(config, *, filter, before, limit) -> Iterator[StateSnapshot]`	Iterate over historical state snapshots.
`aget_state_history`	`async (config, *, filter, before, limit) -> AsyncIterator[StateSnapshot]`	Iterate over historical state snapshots asynchronously.
`update_state`	`(config, values, as_node) -> RunnableConfig`	Apply state update as if from a node.
`aupdate_state`	`async (config, values, as_node) -> RunnableConfig`	Apply state update asynchronously.
`bulk_update_state`	`(config, updates) -> RunnableConfig`	Apply multiple state update batches.
`abulk_update_state`	`async (config, updates) -> RunnableConfig`	Apply multiple state update batches asynchronously.
`with_config`	`(config, **kwargs) -> Self`	Return graph bound to config.
`get_graph`	`(config, *, xray) -> DrawableGraph`	Get drawable graph representation.
`aget_graph`	`async (config, *, xray) -> DrawableGraph`	Get drawable graph representation asynchronously.

StreamProtocol

Member	Type	Description
`modes`	`set[StreamMode]`	Set of stream modes this protocol handles.
`__call__`	`Callable[[StreamChunk], None]`	Callback invoked with each stream chunk.

StreamChunk

Element	Type	Description
Namespace	`tuple[str, ...]`	Hierarchical namespace identifying the source (e.g., subgraph path).
Mode	`str`	The stream mode for this chunk.
Data	`Any`	The actual streamed data.

I/O Contract

Aspect	Detail
Input	Command | None` plus optional `RunnableConfig` and `ContextT`. State methods: `RunnableConfig` identifying a thread/checkpoint.
Output	`invoke`: final state dict or output value. `stream`: iterator of state dicts or streamed chunks. State methods: `StateSnapshot` or `RunnableConfig`.
Side Effects	Executes graph nodes, writes to checkpointer, streams events to `StreamProtocol` consumers.
Errors	May raise `GraphRecursionError`, `InvalidUpdateError`, `GraphInterrupt`, or `EmptyInputError`.

Usage Examples

Type-Annotating a Graph Parameter

from langgraph.pregel.protocol import PregelProtocol

def run_agent(graph: PregelProtocol, user_input: str) -> dict:
    return graph.invoke(
        {"messages": [("user", user_input)]},
        {"recursion_limit": 50},
    )

Streaming with Mode Selection

from langgraph.pregel.protocol import PregelProtocol

def stream_updates(graph: PregelProtocol, user_input: str):
    for chunk in graph.stream(
        {"messages": [("user", user_input)]},
        stream_mode="updates",
    ):
        print(chunk)

Inspecting State History

from langgraph.pregel.protocol import PregelProtocol

def show_history(graph: PregelProtocol, config: dict):
    for snapshot in graph.get_state_history(config, limit=5):
        print(f"Step: {snapshot.metadata.get('step')}, State: {snapshot.values}")

Related Pages

• Langchain_ai_Langgraph_Typing_Module -- Type variables (`StateT`, `ContextT`, `InputT`, `OutputT`) used to parameterize this protocol.
• Langchain_ai_Langgraph_Public_Constants -- `START` and `END` constants defining graph topology.
• Langchain_ai_Langgraph_Error_Classes -- Exceptions raised during execution.
• Langchain_ai_Langgraph_Runtime_Class -- Runtime context injected into nodes during execution.