haive.core.graph.state_graph.components.node¶

Node component implementation for the Haive graph system.

This module defines the Node class which represents a processing unit in a graph. Nodes are responsible for transforming state and producing outputs that can be used by subsequent nodes in the graph flow.

Classes:

Node: Base node class for state processing in a graph

Typical usage:

from haive.core.graph.state_graph.components import Node from haive.core.graph.common.types import NodeType

# Create a simple processing node node = Node(

name=”transform_data”, node_type=NodeType.CALLABLE, metadata={“callable”: lambda state: {“output”: state[“input”] * 2}}

)

# Process state through the node result = node.process({“input”: 5}) assert result[“output”] == 10

Classes¶

Node

Base node in a graph system.

Module Contents¶

class haive.core.graph.state_graph.components.node.Node(/, **data)¶

Bases: pydantic.BaseModel, Generic[T, C, O]

Base node in a graph system.

A node represents a processing unit in a graph that can receive a state, optionally a configuration, and produce an output. Nodes are the fundamental building blocks for computational graphs in the Haive framework.

Parameters:: data (Any)

id¶: Unique identifier for the node

name¶: Human-readable name for the node

node_type¶: Type of node (CALLABLE, LLM, TOOL, etc.)

input_mapping¶: Optional mapping of input keys

output_mapping¶: Optional mapping of output keys

command_goto¶: Optional command for routing control

retry_policy¶: Optional policy for retrying node execution

description¶: Optional human-readable description

metadata¶: Additional metadata for the node

created_at¶: Creation timestamp

Examples

node = Node(: name=”process_data”, node_type=NodeType.CALLABLE, description=”Processes input data by doubling it”, metadata={“callable”: lambda state: {“output”: state[“input”] * 2}}

)

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.

abstractmethod process(state, config=None)¶

Process state and return output.

This method processes the input state according to the node’s configuration and produces an output. Subclasses must implement this method to define the specific processing logic.

Parameters:

state (T) – Input state to be processed
config (C | None) – Optional configuration parameters for processing

Returns:

Process output based on the node’s implementation

Raises:

NotImplementedError – Subclasses must implement this method

Return type:

to_dict()¶

Convert to serializable dictionary.

Creates a dictionary representation of the node suitable for serialization.

Returns:: Dictionary representation of the node
Return type:: dict[str, Any]

property display_name: str¶

Get a human-readable display name.

Returns:: Node display name (defaults to node name)
Return type:: str

model_config¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].