haive.core.graph.node.state_driven_node¶

State-driven nodes that get behavior from state.

This module implements nodes whose behavior is determined at runtime from the state, enabling true dynamic execution without recompilation.

Classes¶

DynamicNodeFactory

Factory for creating state-driven nodes.

StateDrivenBranch

Branch node that determines routing from state.

StateDrivenNode

Node that executes behavior from state.

StateNodeSchema

Schema for defining state-driven nodes in configuration.

Module Contents¶

class haive.core.graph.node.state_driven_node.DynamicNodeFactory¶

Factory for creating state-driven nodes.

This factory simplifies the creation of state-driven nodes and provides patterns for common use cases.

static create_aggregator_node(name, aggregation_key='results')¶

Create a node that aggregates results.

Parameters:

  • name (str) – Node name

  • aggregation_key (str) – Key in state to store aggregated results

Returns:

StateDrivenNode that aggregates parallel results

Return type:

StateDrivenNode

static create_branch(name, default=None)¶

Create a state-driven branch.

Parameters:

  • name (str) – Branch name

  • default (str) – Default route

Returns:

StateDrivenBranch instance

Return type:

StateDrivenBranch

static create_node(name, **kwargs)¶

Create a state-driven node.

Parameters:

  • name (str) – Node name

  • **kwargs – Additional node configuration

Returns:

StateDrivenNode instance

Return type:

StateDrivenNode

static create_parallel_node(name, nodes)¶

Create a node that triggers parallel execution.

Parameters:

  • name (str) – Node name

  • nodes (list[str]) – List of nodes to execute in parallel

Returns:

StateDrivenNode that sets up parallel execution

Return type:

StateDrivenNode

class haive.core.graph.node.state_driven_node.StateDrivenBranch(name, default_route=None)¶

Branch node that determines routing from state.

Unlike StateDrivenNode which executes behavior, this node specifically handles conditional routing based on state.

Examples

>>> branch = StateDrivenBranch("router")
>>>
>>> # In state, define routing logic
>>> state.branches["router"] = lambda s: "path_a" if s.score > 0.5 else "path_b"
>>>
>>> # Branch will route based on state
>>> next_node = branch(state)

Initialize state-driven branch.

Parameters:

  • name (str) – Branch identifier

  • default_route (Optional[str]) – Default route if no logic in state

class haive.core.graph.node.state_driven_node.StateDrivenNode(name, fallback=None)¶

Node that executes behavior from state.

Instead of having fixed behavior at compile time, this node retrieves its behavior from the state at runtime. This enables:

  • Runtime behavior modification

  • Dynamic routing updates

  • Hot-swappable logic

  • No recompilation needed

The state must have a ‘nodes’ dictionary containing callables and a ‘routing_table’ for dynamic routing.

Examples

>>> # In state schema
>>> class DynamicState(StateSchema):
...     nodes: dict[str, Callable] = Field(default_factory=dict)
...     routing_table: dict[str, list[str]] = Field(default_factory=dict)
>>>
>>> # Create state-driven node
>>> node = StateDrivenNode("processor")
>>>
>>> # Add behavior to state
>>> state.nodes["processor"] = lambda x: x.upper()
>>> state.routing_table["processor"] = ["validator"]
>>>
>>> # Execute - behavior comes from state!
>>> result = node(state)

Initialize state-driven node.

Parameters:

  • name (str) – Node identifier used to lookup behavior in state

  • fallback (Optional[Callable]) – Optional fallback behavior if not found in state

reset_execution_count()¶

Reset execution counter.

Return type:

None

property execution_count: int¶

Get number of times this node has executed.

Return type:

int

class haive.core.graph.node.state_driven_node.StateNodeSchema(/, **data)¶

Bases: pydantic.BaseModel

Schema for defining state-driven nodes in configuration.

This allows nodes to be defined declaratively and created from configuration rather than code.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

create_node()¶

Create the actual node from this schema.

Returns:

StateDrivenNode or StateDrivenBranch instance

Return type:

Any