Graph¶

FlowStateGraph ¶

Bases: StateGraph

Methods:

Name	Description
`add_node`
`add_edge`	Add a directed edge from the start node (or list of start nodes) to the end node.
`add_conditional_edges`	Add a conditional edge from the starting node to any number of destination nodes.
`add_sequence`	Add a sequence of nodes that will be executed in the provided order.
`set_entry_point`	Specifies the first node to be called in the graph.
`set_finish_point`	Marks a node as a finish point of the graph.
`set_conditional_entry_point`	Sets a conditional entry point in the graph.
`compile`
`validate`
`process_flow`

add_node ¶

add_node(name: str, node: Node) -> Self

add_edge ¶

add_edge(start_key: Union[str, list[str]], end_key: str) -> Self

Add a directed edge from the start node (or list of start nodes) to the end node.

When a single start node is provided, the graph will wait for that node to complete before executing the end node. When multiple start nodes are provided, the graph will wait for ALL of the start nodes to complete before executing the end node.

Parameters:

Name	Type	Description	Default
`start_key`	`Union[str, list[str]]`	The key(s) of the start node(s) of the edge.	required
`end_key`	`str`	The key of the end node of the edge.	required

Raises:

Type	Description
`ValueError`	If the start key is 'END' or if the start key or end key is not present in the graph.

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the state graph, allowing for method chaining.

add_conditional_edges ¶

add_conditional_edges(source: str, path: Union[Callable[..., Union[Hashable, list[Hashable]]], Callable[..., Awaitable[Union[Hashable, list[Hashable]]]], Runnable[Any, Union[Hashable, list[Hashable]]]], path_map: Optional[Union[dict[Hashable, str], list[str]]] = None, then: Optional[str] = None) -> Self

Add a conditional edge from the starting node to any number of destination nodes.

Parameters:

Name	Type	Description	Default
`source`	`str`	The starting node. This conditional edge will run when exiting this node.	required
`path`	`Union[Callable[..., Union[Hashable, list[Hashable]]], Callable[..., Awaitable[Union[Hashable, list[Hashable]]]], Runnable[Any, Union[Hashable, list[Hashable]]]]`	The callable that determines the next node or nodes. If not specifying `path_map` it should return one or more nodes. If it returns END, the graph will stop execution.	required
`path_map`	`Optional[Union[dict[Hashable, str], list[str]]]`	Optional mapping of paths to node names. If omitted the paths returned by `path` should be node names.	`None`
`then`	`Optional[str]`	The name of a node to execute after the nodes selected by `path`.	`None`

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the graph, allowing for method chaining.

Without typehints on the path function's return value (e.g., -> Literal["foo", "__end__"]:)

or a path_map, the graph visualization assumes the edge could transition to any node in the graph.

add_sequence ¶

add_sequence(nodes: Sequence[Union[RunnableLike, tuple[str, RunnableLike]]]) -> Self

Add a sequence of nodes that will be executed in the provided order.

Parameters:

Name	Type	Description	Default
`nodes`	`Sequence[Union[RunnableLike, tuple[str, RunnableLike]]]`	A sequence of RunnableLike objects (e.g. a LangChain Runnable or a callable) or (name, RunnableLike) tuples. If no names are provided, the name will be inferred from the node object (e.g. a runnable or a callable name). Each node will be executed in the order provided.	required

Raises:

Type	Description
`ValueError`	if the sequence is empty.
`ValueError`	if the sequence contains duplicate node names.

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the state graph, allowing for method chaining.

set_entry_point ¶

set_entry_point(key: str) -> Self

Specifies the first node to be called in the graph.

Equivalent to calling add_edge(START, key).

Parameters:

Name	Type	Description	Default
`key`	`str`	The key of the node to set as the entry point.	required

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the graph, allowing for method chaining.

set_finish_point ¶

set_finish_point(key: str) -> Self

Marks a node as a finish point of the graph.

If the graph reaches this node, it will cease execution.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key of the node to set as the finish point.	required

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the graph, allowing for method chaining.

set_conditional_entry_point ¶

set_conditional_entry_point(path: Union[Callable[..., Union[Hashable, list[Hashable]]], Callable[..., Awaitable[Union[Hashable, list[Hashable]]]], Runnable[Any, Union[Hashable, list[Hashable]]]], path_map: Optional[Union[dict[Hashable, str], list[str]]] = None, then: Optional[str] = None) -> Self

Sets a conditional entry point in the graph.

Parameters:

Name	Type	Description	Default
`path`	`Union[Callable[..., Union[Hashable, list[Hashable]]], Callable[..., Awaitable[Union[Hashable, list[Hashable]]]], Runnable[Any, Union[Hashable, list[Hashable]]]]`	The callable that determines the next node or nodes. If not specifying `path_map` it should return one or more nodes. If it returns END, the graph will stop execution.	required
`path_map`	`Optional[Union[dict[Hashable, str], list[str]]]`	Optional mapping of paths to node names. If omitted the paths returned by `path` should be node names.	`None`
`then`	`Optional[str]`	The name of a node to execute after the nodes selected by `path`.	`None`

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the graph, allowing for method chaining.

compile ¶

compile(checkpointer=None, *, flow=None, **kwargs)

validate ¶

validate(interrupt: Optional[Sequence[str]] = None) -> None

process_flow ¶

process_flow(flow: Sequence[Node]) -> Self

FunctionStateGraph ¶

Bases: StateGraph

Methods:

Name	Description
`add_node`
`add_edge`	Add a directed edge from the start node (or list of start nodes) to the end node.
`add_conditional_edges`	Add a conditional edge from the starting node to any number of destination nodes.
`add_sequence`	Add a sequence of nodes that will be executed in the provided order.
`set_entry_point`	Specifies the first node to be called in the graph.
`set_finish_point`	Marks a node as a finish point of the graph.
`set_conditional_entry_point`	Sets a conditional entry point in the graph.
`compile`
`validate`
`process_flow`

add_node ¶

add_node(name: str, node: any) -> Self

add_edge ¶

add_edge(start_key: Union[str, list[str]], end_key: str) -> Self

Add a directed edge from the start node (or list of start nodes) to the end node.

When a single start node is provided, the graph will wait for that node to complete before executing the end node. When multiple start nodes are provided, the graph will wait for ALL of the start nodes to complete before executing the end node.

Parameters:

Name	Type	Description	Default
`start_key`	`Union[str, list[str]]`	The key(s) of the start node(s) of the edge.	required
`end_key`	`str`	The key of the end node of the edge.	required

Raises:

Type	Description
`ValueError`	If the start key is 'END' or if the start key or end key is not present in the graph.

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the state graph, allowing for method chaining.

add_conditional_edges ¶

add_conditional_edges(source: str, path: Union[Callable[..., Union[Hashable, list[Hashable]]], Callable[..., Awaitable[Union[Hashable, list[Hashable]]]], Runnable[Any, Union[Hashable, list[Hashable]]]], path_map: Optional[Union[dict[Hashable, str], list[str]]] = None, then: Optional[str] = None) -> Self

Add a conditional edge from the starting node to any number of destination nodes.

Parameters:

Name	Type	Description	Default
`source`	`str`	The starting node. This conditional edge will run when exiting this node.	required
`path`	`Union[Callable[..., Union[Hashable, list[Hashable]]], Callable[..., Awaitable[Union[Hashable, list[Hashable]]]], Runnable[Any, Union[Hashable, list[Hashable]]]]`	The callable that determines the next node or nodes. If not specifying `path_map` it should return one or more nodes. If it returns END, the graph will stop execution.	required
`path_map`	`Optional[Union[dict[Hashable, str], list[str]]]`	Optional mapping of paths to node names. If omitted the paths returned by `path` should be node names.	`None`
`then`	`Optional[str]`	The name of a node to execute after the nodes selected by `path`.	`None`

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the graph, allowing for method chaining.

Without typehints on the path function's return value (e.g., -> Literal["foo", "__end__"]:)

or a path_map, the graph visualization assumes the edge could transition to any node in the graph.

add_sequence ¶

add_sequence(nodes: Sequence[Union[RunnableLike, tuple[str, RunnableLike]]]) -> Self

Add a sequence of nodes that will be executed in the provided order.

Parameters:

Name	Type	Description	Default
`nodes`	`Sequence[Union[RunnableLike, tuple[str, RunnableLike]]]`	A sequence of RunnableLike objects (e.g. a LangChain Runnable or a callable) or (name, RunnableLike) tuples. If no names are provided, the name will be inferred from the node object (e.g. a runnable or a callable name). Each node will be executed in the order provided.	required

Raises:

Type	Description
`ValueError`	if the sequence is empty.
`ValueError`	if the sequence contains duplicate node names.

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the state graph, allowing for method chaining.

set_entry_point ¶

set_entry_point(key: str) -> Self

Specifies the first node to be called in the graph.

Equivalent to calling add_edge(START, key).

Parameters:

Name	Type	Description	Default
`key`	`str`	The key of the node to set as the entry point.	required

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the graph, allowing for method chaining.

set_finish_point ¶

set_finish_point(key: str) -> Self

Marks a node as a finish point of the graph.

If the graph reaches this node, it will cease execution.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key of the node to set as the finish point.	required

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the graph, allowing for method chaining.

set_conditional_entry_point ¶

set_conditional_entry_point(path: Union[Callable[..., Union[Hashable, list[Hashable]]], Callable[..., Awaitable[Union[Hashable, list[Hashable]]]], Runnable[Any, Union[Hashable, list[Hashable]]]], path_map: Optional[Union[dict[Hashable, str], list[str]]] = None, then: Optional[str] = None) -> Self

Sets a conditional entry point in the graph.

Parameters:

Name	Type	Description	Default
`path`	`Union[Callable[..., Union[Hashable, list[Hashable]]], Callable[..., Awaitable[Union[Hashable, list[Hashable]]]], Runnable[Any, Union[Hashable, list[Hashable]]]]`	The callable that determines the next node or nodes. If not specifying `path_map` it should return one or more nodes. If it returns END, the graph will stop execution.	required
`path_map`	`Optional[Union[dict[Hashable, str], list[str]]]`	Optional mapping of paths to node names. If omitted the paths returned by `path` should be node names.	`None`
`then`	`Optional[str]`	The name of a node to execute after the nodes selected by `path`.	`None`

Returns:

Name	Type	Description
`Self`	`Self`	The instance of the graph, allowing for method chaining.

compile ¶

compile(checkpointer=None, *, flow=None, **kwargs)

validate ¶

validate(interrupt: Optional[Sequence[str]] = None) -> None

process_flow ¶

process_flow(flow: Sequence[any]) -> Self