orkes.graph.OrkesGraph#

class orkes.graph.OrkesGraph(state, name='default_graph', description='', traced=True)[source]#

A class to represent a stateful graph for orchestrating multi-agent workflows.

The OrkesGraph allows you to define a graph of nodes, where each node is a function that operates on a shared state. The graph can have a single start and end point, and nodes can be connected with forward or conditional edges.

Parameters:

name (str)
description (str)
traced (bool)

state#

The TypedDict class that defines the shared state of the graph.

Type:: type

name#

The name of the graph.

Type:: str

description#

A description of the graph.

Type:: str

traced#

Whether to trace the graph execution.

Type:: bool

Example

>>> from typing import TypedDict, List
>>>
>>> class MyState(TypedDict):
...     messages: List[str]
...
>>> def node1(state: MyState) -> MyState:
...     state['messages'].append("Hello from node1")
...     return state
...
>>> def node2(state: MyState) -> MyState:
...     state['messages'].append("Hello from node2")
...     return state
...
>>> graph = OrkesGraph(state=MyState)
>>> graph.add_node("node1", node1)
>>> graph.add_node("node2", node2)
>>> graph.add_edge(graph.START, "node1")
>>> graph.add_edge("node1", "node2")
>>> graph.add_edge("node2", graph.END)
>>> compiled_graph = graph.compile()
>>> result = compiled_graph.run({"messages": []})
>>> print(result)
{'messages': ['Hello from node1', 'Hello from node2']}

__init__(state, name='default_graph', description='', traced=True)[source]#

Initializes an OrkesGraph.

Parameters:

state (type) – The TypedDict class that defines the shared state of the graph.
name (str, optional) – The name of the graph. Defaults to “default_graph”.
description (str, optional) – A description of the graph. Defaults to “”.
traced (bool, optional) – Whether to trace the graph execution. Defaults to True.

Raises:

TypeError – If the state is not a TypedDict class.

Methods

`__init__`(state[, name, description, traced])	Initializes an OrkesGraph.
`add_conditional_edge`(from_node, ...[, ...])	Adds a conditional edge from a node.
`add_edge`(from_node, to_node[, max_passes])	Adds a forward edge between two nodes.
`add_node`(name, func)	Adds a node to the graph.
`add_parallel_edges`(from_node, to_nodes, ...)	Adds a parallel edge that splits into multiple branches.
`can_reach_node`(start_node_name, target_node_name)	Determines if the target node is reachable from the start node.
`compile`()	Compiles the graph, making it ready for execution.
`detect_loop`()	Detects loops in the graph.