Graphs¶

Build a graph over a state schema, run it, and describe it as data. See Graphs and flow and Persistence and time travel.

Name	What it is
Building and running
`Graph`	A graph of nodes that run in supersteps over a shared state.
`StateT`	The state type of a graph: the class passed as `Graph(state_schema)`.
`NodeSpec`	A registered node and its name.
`mermaid_id`	Return the Mermaid id derived from a node name.
Graph specs
`GraphSpec`	Serializable description of a graph.
`GraphNodeSpec`	Serializable description of a node and its declared goto targets.
`EdgeSpec`	Serializable description of a static edge.
`BranchSpec`	Serializable description of a conditional edge.

Building and running¶

Graph¶

Graph(state_schema, *, name='graph', max_steps=25, timeout=None, state_store=None, middleware=None, workspace=None)

A graph of nodes that run in supersteps over a shared state.

Parameters
`state_schema`	`type[StateT]`	Pydantic model, `TypedDict` or `dict`, with the reducers.
`name`	`str`	Graph name, used in errors and in node and middleware contexts.
`max_steps`	`int`	Superstep limit per call.
`timeout`	`float \| None`	Seconds one call may spend inside supersteps, summed; `None` means no limit. Time spent handling events between supersteps is not counted.
`state_store`	`StateStore \| None`	Store that enables threads, history, interrupts and resume.
`middleware`	`list[Middleware] \| None`	Hooks around the graph, nodes, model calls and tools.
`workspace`	`Workspace \| None`	Files available to tools.

Raises
`GraphConfigError`	If `name` is blank or `state_schema` is not a valid state type (see `StateSchema.from_type`).
`TypeError`	If an item of `middleware` does not subclass `Middleware`.

Methods

`flow(*items)`¶

Declare the flow; a graph has one.

flow(*items) -> Graph[StateT]

The flow's nodes are those in items plus the node objects in their goto=; a goto= name refers to the flow's node of that name. Every node needs an edge, a branch or goto targets, and must be reachable from START. Nothing is assigned unless the flow is valid.

Parameters
`*items`	`FlowStartEdge \| FlowEdge \| FlowBranchEdge`	Transitions such as `START >> a`, `a >> b` and `b >> when(...)`.

Returns
`Graph`	The same graph.

Raises
`GraphConfigError`	If the graph already has a flow; the flow is empty, has no or several `START`, two transitions from one node or two nodes with one name; or, listed together in one error, a node has no way out, is unreachable from `START`, or a `goto` target matches no node of the flow or a different one.

`async ainvoke(initial_state, ...)`¶

Run the graph until it finishes or pauses.

async ainvoke(initial_state=None, *, thread_id=None, branch_id='main', resume=None, state_store=None, context=None) -> GraphResult[StateT]

Parameters
`initial_state`	`StateT \| dict[str, Any] \| None`	Input, merged into the thread's state through the reducers and validated like a node's update; a new thread starts from the schema defaults. A state model instance replaces every field. Input starts a new turn from `START` and drops the tasks of an unfinished superstep. `None` continues the thread's unfinished superstep without rerunning the tasks that finished; it never starts a turn.
`thread_id`	`str \| None`	Required with a state store; generated when omitted without one.
`branch_id`	`str`	History branch; one other than `"main"` must come from `fork`.
`resume`	`Resume \| None`	Answers for every pending interrupt. Requires `initial_state=None`.
`state_store`	`StateStore \| None`	Store to use instead of the graph's own.
`context`	`object \| None`	Run-time object read as `ctx.context`. Never stored: pass it again on resume.

Raises
`TypeError`	If `resume` is not a `Resume` or comes with `initial_state`.
`GraphConfigError`	If the graph has no flow; with a store, if `thread_id` is missing; without one, if `initial_state` is `None`, `resume` is given or `branch_id` is not `"main"`.
`UnknownThreadError`	If `initial_state` is `None` and the thread or branch does not exist, or `branch_id` names a branch that does not exist.
`ResumeError`	If the thread is paused and gets input or no `resume`; there is nothing to continue (no unfinished superstep, or one whose writes cannot be merged); the answers do not match the pending interrupts; or the stored history does not fit the graph or its schema.
`InvalidUpdateError`	If two writes to one field in a superstep, an `update_state` made while it was paused included, cannot be merged: one has no merging reducer or is a `Replace`.
`StateUpdateError`	If the input or a node's update has an unknown key, an invalid value, a value its reducer rejects or a secret, or leaves a required field of a new thread empty.
`StateStoreError`	If a written value does not survive being stored as JSON, with or without a store, or the stored state has undeclared fields.
`GraphExecutionError`	If the run finishes or pauses with spawned sub-agents never gathered; they are cancelled first.

`invoke(initial_state, ...)`¶

Run ainvoke on a new event loop.

invoke(initial_state=None, *, thread_id=None, branch_id='main', resume=None, state_store=None, context=None) -> GraphResult[StateT]

It takes the arguments of ainvoke and returns its GraphResult.

Raises
`RuntimeError`	If an event loop is running; use `ainvoke` there.
`TypeError, GraphConfigError, UnknownThreadError, ResumeError, InvalidUpdateError, StateUpdateError, StateStoreError, GraphExecutionError`	As for `ainvoke`.

`stream(initial_state, ...)`¶

Run astream on a private event loop and yield its events.

stream(initial_state=None, *, stream_mode, thread_id=None, branch_id='main', resume=None, state_store=None, context=None) -> Iterator[StreamEvent]

It takes the arguments of astream. The graph does not run while the caller handles an event: time on an event emitted during a superstep ("custom", "tokens", the "node_input" debug event) counts toward the graph's and the running nodes' timeouts.

Raises
`RuntimeError`	If an event loop is running; use `astream` there.
`TypeError, ValueError, GraphConfigError, UnknownThreadError, ResumeError, InvalidUpdateError, StateUpdateError, StateStoreError, GraphExecutionError`	As for `astream`, while iterating.

`async astream(initial_state, ...)`¶

Run the graph and yield events as they happen.

async astream(initial_state=None, *, stream_mode, thread_id=None, branch_id='main', resume=None, state_store=None, context=None) -> AsyncGenerator[StreamEvent]

The arguments other than stream_mode are those of ainvoke.

Parameters
`initial_state`	`StateT \| dict[str, Any] \| None`
`stream_mode`	`str \| Sequence[str]`	Required. Any of `"updates"` (one event per task after each superstep), `"values"` (the merged state after each superstep), `"custom"` (`ctx.emit`), `"tokens"` (model chunks) and `"debug"`. The terminal `"final"` or `"interrupt"` event always comes, so `[]` yields only that one.
`thread_id`	`str \| None`
`branch_id`	`str`
`resume`	`Resume \| None`
`state_store`	`StateStore \| None`
`context`	`object \| None`

Raises
`TypeError`	If `stream_mode` is not a mode name or a sequence of them, or as for `ainvoke`.
`ValueError`	If a mode is unknown, `"final"` or `"interrupt"`.
`GraphConfigError, UnknownThreadError, ResumeError, InvalidUpdateError, StateUpdateError, StateStoreError`	As for `ainvoke`.
`GraphExecutionError`	As for `ainvoke`, instead of the terminal event.

`to_mermaid()`¶

Render the flow as a Mermaid diagram.

to_mermaid() -> str

`to_spec()`¶

Describe the graph as data, with the import path of each node and router.

to_spec() -> GraphSpec

Raises
`GraphConfigError`	If the graph has no flow, a node or router has no import path, or two branch keys of a node have the same string form.

`to_json()`¶

Serialize to_spec() to JSON.

to_json() -> str

`async exists(thread_id, ...)`¶

Tell whether the state store holds history for a thread branch.

async exists(thread_id, *, branch_id='main', state_store=None) -> bool

Parameters
`thread_id`	`str`
`branch_id`	`str`
`state_store`	`StateStore \| None`	Store to read instead of the graph's own.

Returns
`bool`	True once a run, `update_state(create=True)` or `fork` has written the branch.

Raises
`GraphConfigError`	If neither `state_store` nor the graph has a state store.

`async history(thread_id, ...)`¶

Return the recorded events and checkpoints of a thread branch.

async history(thread_id, *, branch_id='main', state_store=None) -> History

Parameters
`thread_id`	`str`
`branch_id`	`str`
`state_store`	`StateStore \| None`	Store to read instead of the graph's own.

Returns
`History`	Empty for a thread that has not run yet.

Raises
`GraphConfigError`	If neither `state_store` nor the graph has a state store.
`UnknownThreadError`	If `branch_id` is not `"main"` and no such branch was forked.

`async branches(thread_id, ...)`¶

Return the branches forked from a thread, oldest first.

async branches(thread_id, *, state_store=None) -> list[BranchRecord]

Parameters
`thread_id`	`str`
`state_store`	`StateStore \| None`	Store to read instead of the graph's own.

Returns
`list[BranchRecord]`	One record per `fork`; `"main"` is not listed, and a thread without forks, or one that has not run yet, has none.

Raises
`GraphConfigError`	If neither `state_store` nor the graph has a state store.

`async load(thread_id, ...)`¶

Rebuild the state of a thread from its history.

async load(thread_id, *, branch_id='main', at=None, state_store=None) -> dict[str, Any]

Parameters
`thread_id`	`str`
`branch_id`	`str`
`at`	`str \| None`	Event id to load the state at; defaults to the latest event.
`state_store`	`StateStore \| None`	Store to read instead of the graph's own.

Returns
`dict`	The state. Fields added to the schema after the thread was stored have their defaults.

Raises
`GraphConfigError`	If neither `state_store` nor the graph has a state store.
`UnknownThreadError`	If the thread or the branch does not exist.
`StateUpdateError`	If `at` is not an event of the branch.
`StateStoreError`	If the stored state has fields the state schema does not declare.
`ResumeError`	If a stored value is not valid for its field.

`async fork(thread_id, ...)`¶

Create a new branch of a thread at one of its events.

async fork(thread_id, *, from_, name=None, branch_id='main', state_store=None) -> BranchRecord

The forked branch is not changed. See Persistence and time travel for how a run continues there.

Parameters
`thread_id`	`str`
`from_`	`str`	Id of an event of the branch being forked.
`name`	`str \| None`	Id of the new branch; a random id when omitted.
`branch_id`	`str`	Branch to fork.
`state_store`	`StateStore \| None`	Store to use instead of the graph's own.

Returns
`BranchRecord`	Run on the new branch with `branch_id=record.id`.

Raises
`ValueError`	If `name` is blank.
`GraphConfigError`	If neither `state_store` nor the graph has a state store.
`UnknownThreadError`	If the thread or `branch_id` does not exist.
`StateUpdateError`	If `from_` is not an event of that branch.
`StateStoreError`	If the thread already has a branch named `name`, `"main"` included, or its stored state has fields the schema does not declare.
`ResumeError`	If a stored value is not valid for its field.

`async get_state(thread_id, ...)`¶

Return the state of a thread as the declared schema type.

async get_state(thread_id, *, branch_id='main', at=None, state_store=None) -> StateSnapshot[StateT]

Parameters
`thread_id`	`str`
`branch_id`	`str`
`at`	`str \| None`	Event id to read the state at; defaults to the latest event.
`state_store`	`StateStore \| None`	Store to read instead of the graph's own.

Returns
`StateSnapshot`	The state and the `sequence` of the event it was read at: `at`, or the latest event of the branch.

Raises
`GraphConfigError`	If neither `state_store` nor the graph has a state store.
`UnknownThreadError`	If the thread or the branch does not exist.
`StateUpdateError`	If `at` is not an event of the branch.
`StateStoreError`	If the stored state has fields the state schema does not declare.
`ResumeError`	If a stored value is not valid for its field.

`async update_state(thread_id, values, ...)`¶

Apply a manual update to a thread, for example while it is paused.

async update_state(thread_id, values, *, branch_id='main', create=False, state_store=None) -> str

See Persistence and time travel for updates to a paused thread.

Parameters
`thread_id`	`str`
`values`	`dict[str, Any]`	Update applied through the reducers and validated like a node's. `Replace(value)` bypasses a reducer; `RemoveMessage(id)` removes a message from an `add_messages` field.
`branch_id`	`str`
`create`	`bool`	Create the thread if it does not exist, from the schema defaults. Branches come only from `fork`.
`state_store`	`StateStore \| None`	Store to write instead of the graph's own.

Returns
`str`	Id of the recorded event.

Raises
`ValueError`	If `create` is true and `thread_id` is empty.
`GraphConfigError`	If neither `state_store` nor the graph has a state store.
`UnknownThreadError`	If the thread does not exist and `create` is false, or the branch does not exist.
`InvalidUpdateError`	If a finished task of the paused or unfinished superstep wrote a field of `values` and the two writes cannot be merged.
`StateUpdateError`	If a key is not a state field, a value is invalid or its reducer rejects it, a `RemoveMessage` is not in a list update to an `add_messages` field, a value holds a secret, or a new thread lacks a required field.
`StateStoreError`	If a value, or a default a new thread keeps, does not survive being stored as JSON.

StateT¶

StateT = TypeVar('StateT')

The state type of a graph: the class passed as Graph(state_schema).

NodeSpec¶

NodeSpec(name, handler, goto=None)

A registered node and its name.

Attributes
`name`	`str`
`handler`	`Node`
`goto`	`tuple[Node \| EndSentinel, ...] \| None`	Declared `goto` targets, names resolved to the flow's nodes.

mermaid_id¶

mermaid_id(name) -> str

Return the Mermaid id derived from a node name.

Characters outside [A-Za-z0-9_] become _. When a graph is rendered, an id already used by another node, by START or END, or equal to the Mermaid keyword end gets the suffix _2, _3 and so on.

Parameters
`name`	`str`	Node name.

Graph specs¶

GraphSpec¶

GraphSpec(*, version=1, name, state_ref=None, start, nodes, edges, branches)

Serializable description of a graph.

Attributes
`version`	`int`	Version of this format.
`name`	`str`
`state_ref`	`str \| None`	`module:qualname` import path of the state type; `None` for a dict state.
`start`	`str`	Name of the first node.
`nodes`	`list[GraphNodeSpec]`
`edges`	`list[EdgeSpec]`
`branches`	`list[BranchSpec]`

GraphNodeSpec¶

GraphNodeSpec(*, name, ref=None, kind='node', goto=None)

Serializable description of a node and its declared goto targets.

Attributes
`name`	`str`
`ref`	`str \| None`	`module:qualname` import path of the node function.
`kind`	`str`	Always `"node"`.
`goto`	`list[str] \| None`	Names of the declared `goto` targets, `"END"` for the end.

EdgeSpec¶

EdgeSpec(*, source, target)

Serializable description of a static edge.

Attributes
`source`	`str`	Name of the node the edge leaves.
`target`	`str`	Name of the next node, or `"END"`.

BranchSpec¶

BranchSpec(*, source, router_ref, mapping)

Serializable description of a conditional edge.

Attributes
`source`	`str`	Name of the node the edge leaves.
`router_ref`	`str`	`module:qualname` import path of the router.
`mapping`	`dict[str, str]`	Router result, as a string, to the name of the next node or `"END"`.

Graphs¶

Building and running¶

Graph¶

flow(*items)¶

async ainvoke(initial_state, ...)¶

invoke(initial_state, ...)¶

stream(initial_state, ...)¶

async astream(initial_state, ...)¶

to_mermaid()¶

to_spec()¶

to_json()¶

async exists(thread_id, ...)¶

async history(thread_id, ...)¶

async branches(thread_id, ...)¶

async load(thread_id, ...)¶

async fork(thread_id, ...)¶

async get_state(thread_id, ...)¶

async update_state(thread_id, values, ...)¶