Graph Engine

The core workflow engine for Maticlib.

MaticGraph

maticlib.graph.graph.MaticGraph

MaticGraph(stateful=True, state_schema=None, max_workers=4)

A fast, pure-Python graph workflow engine with optional state management. Supports dict, TypedDict, dataclass, and Pydantic BaseModel states.

Parameters:

Name	Type	Description	Default
`stateful`	`bool`	If True, maintains and merges state across nodes.	`True`
`state_schema`	`Optional[type]`	Optional Pydantic BaseModel, dataclass, or TypedDict class	`None`
`max_workers`	`int`	Maximum number of parallel workers (default: 4)	`4`

Initializes the MaticGraph engine.

Parameters:

Name	Type	Description	Default
`stateful`	`bool`	If True, state is preserved and merged between nodes.	`True`
`state_schema`	`type`	A Pydantic model, dataclass, or TypedDict class to use as the state container.	`None`
`max_workers`	`int`	Maximum number of parallel threads for `parallel_group` execution. Default is 4.	`4`

Source code in maticlib/graph/graph.py

def __init__(
    self,
    stateful: bool = True,
    state_schema: Optional[type] = None,
    max_workers: int = 4,
):
    """
    Initializes the MaticGraph engine.

    Args:
        stateful (bool): If True, state is preserved and merged between nodes.
        state_schema (type, optional): A Pydantic model, dataclass, or
            TypedDict class to use as the state container.
        max_workers (int): Maximum number of parallel threads for `parallel_group`
            execution. Default is 4.
    """
    self.stateful = stateful
    self.state_schema = state_schema
    self.max_workers = max_workers
    self.nodes: Dict[str, Node] = {}
    self.entry_node: Optional[str] = None
    self.exit_nodes: List[str] = []
    self._execution_log: List[Dict[str, Any]] = []
    self._parallel_groups: Dict[str, List[str]] = (
        {}
    )  # Maps trigger node to parallel nodes

    # Detect if state_schema is a Pydantic model
    self._is_pydantic = False
    if state_schema is not None and PYDANTIC_AVAILABLE:
        self._is_pydantic = isinstance(state_schema, type) and issubclass(
            state_schema, BaseModel
        )

add_conditional_edge

add_conditional_edge(
    from_node, condition, routes, readable_names=None
)

Adds a conditional edge that routes execution based on a condition function.

Parameters:

Name	Type	Description	Default
`from_node`	`str`	The node from which to route.	required
`condition`	`Callable`	A function that takes the current state and returns a string key matching one of the routes.	required
`routes`	`Dict[str, str]`	A mapping of condition keys to target node names.	required
`readable_names`	`dict`	Mapping of keys to human-friendly names for documentation/visualization.	`None`

Returns:

Name	Type	Description
`MaticGraph`	`MaticGraph`	The graph instance (for chaining).

Raises:

Type	Description
`ValueError`	If from_node or route targets are not found.

Source code in maticlib/graph/graph.py

def add_conditional_edge(
    self,
    from_node: str,
    condition: Callable[[Any], str],
    routes: Dict[str, str],
    readable_names: Optional[Dict[str, str]] = None,
) -> "MaticGraph":
    """
    Adds a conditional edge that routes execution based on a condition function.

    Args:
        from_node (str): The node from which to route.
        condition (Callable): A function that takes the current state and
            returns a string key matching one of the routes.
        routes (Dict[str, str]): A mapping of condition keys to target node names.
        readable_names (dict, optional): Mapping of keys to human-friendly
            names for documentation/visualization.

    Returns:
        MaticGraph: The graph instance (for chaining).

    Raises:
        ValueError: If from_node or route targets are not found.
    """
    if from_node not in self.nodes:
        raise ValueError(f"Node '{from_node}' not found")

    for route_key, target in routes.items():
        if target not in self.nodes and target != "END":
            raise ValueError(f"Route target '{target}' not found")

    node = self.nodes[from_node]
    node.condition_func = condition
    node.condition_map = routes
    node.readable_names = readable_names or {}

    return self

add_edge

add_edge(from_node, to_node)

Adds a directed edge between two nodes.

Parameters:

Name	Type	Description	Default
`from_node`	`str`	The name of the starting node.	required
`to_node`	`str`	The name of the target node, or 'END' to exit.	required

Returns:

Name	Type	Description
`MaticGraph`	`MaticGraph`	The graph instance (for chaining).

Raises:

Type	Description
`ValueError`	If either node name is not found in the graph.

Source code in maticlib/graph/graph.py

def add_edge(self, from_node: str, to_node: str) -> "MaticGraph":
    """
    Adds a directed edge between two nodes.

    Args:
        from_node (str): The name of the starting node.
        to_node (str): The name of the target node, or 'END' to exit.

    Returns:
        MaticGraph: The graph instance (for chaining).

    Raises:
        ValueError: If either node name is not found in the graph.
    """
    if from_node not in self.nodes:
        raise ValueError(f"Source node '{from_node}' not found")
    if to_node not in self.nodes and to_node != "END":
        raise ValueError(f"Destination node '{to_node}' not found")

    self.nodes[from_node].next_nodes.append(to_node)
    return self

add_node

add_node(name, function)

Adds a processing node to the graph.

Parameters:

Name	Type	Description	Default
`name`	`str`	Unique name for the node.	required
`function`	`Callable`	Function to execute. Should accept the current state and return a dict or model update.	required

Returns:

Name	Type	Description
`MaticGraph`	`MaticGraph`	The graph instance (for chaining).

Raises:

Type	Description
`ValueError`	If a node with the same name already exists.

Source code in maticlib/graph/graph.py

def add_node(self, name: str, function: Callable) -> "MaticGraph":
    """
    Adds a processing node to the graph.

    Args:
        name (str): Unique name for the node.
        function (Callable): Function to execute. Should accept the current
            state and return a dict or model update.

    Returns:
        MaticGraph: The graph instance (for chaining).

    Raises:
        ValueError: If a node with the same name already exists.
    """
    if name in self.nodes:
        raise ValueError(f"Node '{name}' already exists")

    self.nodes[name] = Node(name=name, function=function)
    return self

get_execution_log

get_execution_log()

Get detailed execution log from last run.

Source code in maticlib/graph/graph.py

def get_execution_log(self) -> List[Dict[str, Any]]:
    """Get detailed execution log from last run."""
    return self._execution_log

parallel_group

parallel_group(
    from_node,
    parallel_nodes,
    join_node=None,
    condition=None,
)

Execute multiple nodes in parallel after a specific node.

Parameters:

Name	Type	Description	Default
`from_node`	`str`	Node that triggers parallel execution	required
`parallel_nodes`	`List[str]`	List of nodes to execute in parallel	required
`join_node`	`Optional[str]`	Optional node to execute after all parallel nodes complete	`None`
`condition`	`Optional[Callable[[Any], bool]]`	Optional function to decide whether to parallelize If returns False, executes first parallel_node only	`None`

Example

Always parallel

graph.parallel_group( "analyze", ["sentiment", "entities", "summary"], join_node="combine_results" )

Conditional parallel

graph.parallel_group( "check_size", ["process_large_a", "process_large_b"], join_node="merge", condition=lambda state: state.get("size") > 1000 )

Source code in maticlib/graph/graph.py

def parallel_group(
    self,
    from_node: str,
    parallel_nodes: List[str],
    join_node: Optional[str] = None,
    condition: Optional[Callable[[Any], bool]] = None,
) -> "MaticGraph":
    """
    Execute multiple nodes in parallel after a specific node.

    Args:
        from_node: Node that triggers parallel execution
        parallel_nodes: List of nodes to execute in parallel
        join_node: Optional node to execute after all parallel nodes complete
        condition: Optional function to decide whether to parallelize
                  If returns False, executes first parallel_node only

    Example:
        # Always parallel
        graph.parallel_group(
            "analyze",
            ["sentiment", "entities", "summary"],
            join_node="combine_results"
        )

        # Conditional parallel
        graph.parallel_group(
            "check_size",
            ["process_large_a", "process_large_b"],
            join_node="merge",
            condition=lambda state: state.get("size") > 1000
        )
    """
    # Validate nodes exist
    if from_node not in self.nodes:
        raise ValueError(f"Trigger node '{from_node}' not found")

    for node in parallel_nodes:
        if node not in self.nodes:
            raise ValueError(f"Parallel node '{node}' not found")

    if join_node and join_node not in self.nodes:
        raise ValueError(f"Join node '{join_node}' not found")

    # Store parallel group configuration
    self._parallel_groups[from_node] = {
        "nodes": parallel_nodes,
        "join_node": join_node,
        "condition": condition,
    }

    # Mark the trigger node as having parallel execution
    self.nodes[from_node].parallel_group = parallel_nodes
    self.nodes[from_node].parallel_join = join_node
    self.nodes[from_node].parallel_condition = condition

    return self

run

run(initial_state=None, max_iterations=1000, verbose=False)

Executes the graph workflow dynamically.

Parameters:

Name	Type	Description	Default
`initial_state`	`dict \| BaseModel`	Starting data for the workflow.	`None`
`max_iterations`	`int`	Safety limit on total node executions to prevent infinite loops. Default is 1000.	`1000`
`verbose`	`bool`	If True, prints execution trace to stdout.	`False`

Returns:

Name	Type	Description
`Any`	`Union[Dict[str, Any], BaseModel]`	The final accumulated state of the workflow.

Raises:

Type	Description
`RuntimeError`	If no entry node is set, or if an execution error occurs in a node.

Source code in maticlib/graph/graph.py

def run(
    self,
    initial_state: Optional[Union[Dict[str, Any], BaseModel]] = None,
    max_iterations: int = 1000,
    verbose: bool = False,
) -> Union[Dict[str, Any], BaseModel]:
    """
    Executes the graph workflow dynamically.

    Args:
        initial_state (dict | BaseModel, optional): Starting data for the workflow.
        max_iterations (int): Safety limit on total node executions to prevent
            infinite loops. Default is 1000.
        verbose (bool): If True, prints execution trace to stdout.

    Returns:
        Any: The final accumulated state of the workflow.

    Raises:
        RuntimeError: If no entry node is set, or if an execution error
            occurs in a node.
    """
    if self.entry_node is None:
        raise RuntimeError("No entry node set. Call set_entry() first.")

    self._execution_log = []

    # Initialize state based on schema
    if initial_state is None:
        if self._is_pydantic:
            state = self.state_schema()
        elif self.state_schema:
            state = self.state_schema()
        else:
            state = {}
    else:
        if self._is_pydantic and isinstance(initial_state, dict):
            state = self.state_schema(**initial_state)
        else:
            state = initial_state

    current_node = self.entry_node
    iteration = 0

    if verbose:
        print(f"Starting execution at node: {current_node}")
        print(f"Initial state: {state}\n")

    while current_node is not None and iteration < max_iterations:
        iteration += 1

        if verbose:
            print(f"[{iteration}] Executing node: {current_node}")

        # Execute current node
        state = self._execute_node(current_node, state)

        if verbose:
            print(f"    State after: {state}")

        # Get next node(s)
        next_node = self._get_next_node(current_node, state)

        # Check if next is a parallel group
        if isinstance(next_node, tuple) and next_node[0] == "PARALLEL":
            _, parallel_nodes, join_node = next_node

            # Execute parallel nodes
            state = self._execute_parallel_group(parallel_nodes, state, verbose)

            # Continue to join node if specified
            if join_node:
                current_node = join_node
                if verbose:
                    print(
                        f"  [JOIN] Parallel execution complete, continuing to: {join_node}\n"
                    )
            else:
                # No join node, end execution
                current_node = None
        else:
            # Regular sequential execution
            if verbose and next_node:
                print(f"    Next: {next_node}\n")
            elif verbose:
                print(f"    Workflow complete\n")

            current_node = next_node

    if iteration >= max_iterations:
        raise RuntimeError(
            f"Maximum iterations ({max_iterations}) reached. "
            "Possible infinite loop in graph."
        )

    return state

set_entry

set_entry(node_name)

Sets the starting node for the graph execution.

Parameters:

Name	Type	Description	Default
`node_name`	`str`	The name of the entry node.	required

Returns:

Name	Type	Description
`MaticGraph`	`MaticGraph`	The graph instance (for chaining).

Raises:

Type	Description
`ValueError`	If the node_name does not exist in the graph.

Source code in maticlib/graph/graph.py

def set_entry(self, node_name: str) -> "MaticGraph":
    """
    Sets the starting node for the graph execution.

    Args:
        node_name (str): The name of the entry node.

    Returns:
        MaticGraph: The graph instance (for chaining).

    Raises:
        ValueError: If the node_name does not exist in the graph.
    """
    if node_name not in self.nodes:
        raise ValueError(f"Node '{node_name}' not found")
    self.entry_node = node_name
    return self

set_exit

set_exit(node_name)

Marks a node as an explicit exit point for the workflow.

Parameters:

Name	Type	Description	Default
`node_name`	`str`	The name of the node.	required

Returns:

Name	Type	Description
`MaticGraph`	`MaticGraph`	The graph instance (for chaining).

Raises:

Type	Description
`ValueError`	If the node_name does not exist in the graph.

Source code in maticlib/graph/graph.py

def set_exit(self, node_name: str) -> "MaticGraph":
    """
    Marks a node as an explicit exit point for the workflow.

    Args:
        node_name (str): The name of the node.

    Returns:
        MaticGraph: The graph instance (for chaining).

    Raises:
        ValueError: If the node_name does not exist in the graph.
    """
    if node_name not in self.nodes:
        raise ValueError(f"Node '{node_name}' not found")
    if node_name not in self.exit_nodes:
        self.exit_nodes.append(node_name)
    return self

visualize

visualize()

Generate a text-based visualization of the graph.

Source code in maticlib/graph/graph.py

def visualize(self) -> str:
    """Generate a text-based visualization of the graph."""
    lines = ["Graph Structure:", "=" * 50]

    if self.state_schema:
        schema_name = getattr(self.state_schema, "__name__", str(self.state_schema))
        state_type = "Pydantic Model" if self._is_pydantic else "Schema"
        lines.append(f"State {state_type}: {schema_name}")

    lines.append(f"Max Workers: {self.max_workers}")
    lines.append("")

    for node_name, node in self.nodes.items():
        marker = ">" if node_name == self.entry_node else "•"
        exit_marker = " [EXIT]" if node_name in self.exit_nodes else ""

        # Check if node has parallel group
        parallel_marker = ""
        if hasattr(node, "parallel_group") and node.parallel_group:
            parallel_marker = f" [PARALLEL->{len(node.parallel_group)} nodes]"

        lines.append(f"{marker} {node_name}{exit_marker}{parallel_marker}")

        # Show parallel group details
        if hasattr(node, "parallel_group") and node.parallel_group:
            lines.append(f"  |- Parallel nodes:")
            for pnode in node.parallel_group:
                lines.append(f"  |  L- {pnode}")
            if hasattr(node, "parallel_join") and node.parallel_join:
                lines.append(f"  L- Join at: {node.parallel_join}")

        # Show conditional routing
        if node.condition_func:
            lines.append(f"  └─ Conditional routing:")
            for route_key, target in node.condition_map.items():
                readable = node.readable_names.get(route_key, route_key)
                lines.append(f"     [{route_key}] → {target} ({readable})")
        # Show regular edges
        elif not hasattr(node, "parallel_group"):
            for next_node in node.next_nodes:
                lines.append(f"  └─→ {next_node}")

    return "\n".join(lines)

when

when(from_node, **routes)

Simplified conditional routing using state['next'] or state.next Works with both dict and Pydantic models.

Source code in maticlib/graph/graph.py

def when(self, from_node: str, **routes: str) -> "MaticGraph":
    """
    Simplified conditional routing using state['next'] or state.next
    Works with both dict and Pydantic models.
    """

    def route_by_next(state: Any) -> str:
        # Handle Pydantic models
        if self._is_pydantic and isinstance(state, BaseModel):
            next_key = getattr(state, "next", None)
        # Handle dicts and TypedDict
        elif isinstance(state, dict):
            next_key = state.get("next")
        else:
            # Handle dataclasses
            next_key = getattr(state, "next", None)

        if next_key not in routes:
            available = ", ".join(routes.keys())
            raise ValueError(
                f"Invalid route '{next_key}'. Available routes: {available}"
            )
        return next_key

    return self.add_conditional_edge(
        from_node,
        route_by_next,
        routes,
        readable_names={k: k.replace("_", " ").title() for k in routes.keys()},
    )

Node

maticlib.graph.node.Node `dataclass`

Node(
    name,
    function,
    next_nodes=list(),
    condition_func=None,
    condition_map=None,
    readable_names=None,
    parallel_group=None,
    parallel_join=None,
    parallel_condition=None,
)

Represents a processing node within a MaticGraph workflow.

A node encapsulates a function to be executed and metadata for routing the workflow after execution, including support for conditional branching and parallel execution.

Attributes:

Name	Type	Description
`name`	`str`	Unique identifier for the node.
`function`	`Callable`	The Python function to execute. Receives the current state and returns an update (dict or model).
`next_nodes`	`List[str]`	List of possible next nodes for sequential or parallel flow.
`condition_func`	`Optional[Callable]`	A function that determines which route to take next based on the returned key.
`condition_map`	`Optional[Dict[str, str]]`	Maps keys from `condition_func` to target node names.
`readable_names`	`Optional[Dict[str, str]]`	Human-readable names for routes (useful for visualization).
`parallel_group`	`Optional[List[str]]`	List of nodes to execute in parallel after this node.
`parallel_join`	`Optional[str]`	A node where parallel execution groups re-converge.
`parallel_condition`	`Optional[Callable]`	A condition to decide whether to trigger parallel execution.

Graph Engine

MaticGraph

maticlib.graph.graph.MaticGraph

add_conditional_edge

add_edge

add_node

get_execution_log

parallel_group

Always parallel

Conditional parallel

run

set_entry

set_exit

visualize

when

Node

maticlib.graph.node.Node dataclass

maticlib.graph.node.Node `dataclass`