haive.core.graph.state_graph.components.branch

Branch component implementation for the Haive graph system.

This module defines the Branch class which represents a decision point in a graph that routes execution based on state conditions. Branches enable the creation of dynamic, conditional flows within the graph system.

Classes:

Branch: Conditional routing component for decision branching in a graph

Typical usage:

from haive.core.graph.state_graph.components import Branch
from haive.core.graph.branches.types import BranchMode

# Create a branch for routing based on state values
branch = Branch(
    name="route_by_score",
    source_node="evaluate",
    mode=BranchMode.DIRECT,
    key="score",
    comparison="greater_than",
    value=80,
    destinations={"True": "high_score_path", "False": "low_score_path"}
)

# Evaluate state to determine the next node
next_node = branch({"score": 95})  # Returns "high_score_path"

Classes

Branch

Unified branch for dynamic routing based on state values.

Module Contents

class haive.core.graph.state_graph.components.branch.Branch(/, **data)

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

Unified branch for dynamic routing based on state values.

This class provides conditional routing functionality in a graph, handling different routing modes and comparison types. Branches act as decision points that evaluate state and determine the next node in the execution flow.

Parameters:

data (Any)

id

Unique identifier for the branch

name

Human-readable name for the branch

source_node

Name of the source node (where this branch is attached)

destinations

Mapping of condition results to target node names

default

Default target node if no condition matches

mode

Branch evaluation mode (DIRECT, FUNCTION, etc.)

key

State key to evaluate in DIRECT mode

value

Value to compare against in comparisons

comparison

Type of comparison to perform

function

Optional callable for FUNCTION mode

function_ref

Optional reference to a callable

allow_none

Whether to allow None values in comparisons

message_key

Key to use when evaluating message content

Example:

# Function-based routing
def route_by_length(state):
    if len(state.get("text", "")) > 100:
        return "long_text"
    return "short_text"

branch = Branch(
    name="text_router",
    source_node="analyze_text",
    mode=BranchMode.FUNCTION,
    function=route_by_length,
    destinations={"long_text": "process_long", "short_text": "process_short"}
)

# Direct comparison routing
branch = Branch(
    name="score_router",
    source_node="calculate_score",
    mode=BranchMode.DIRECT,
    key="score",
    comparison=ComparisonType.GREATER_THAN,
    value=75,
    destinations={True: "high_score", False: "low_score"}
)

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.

evaluate(state)

Evaluate the branch against state.

Performs the actual branch evaluation logic based on the configured mode and comparison settings.

Parameters:

state (T) – Current state to evaluate

Returns:

Routing result based on the evaluation

Return type:

O

setup_function_and_mappings()

Set up function and mappings after initialization.

Resolves function references and ensures proper function setup.

Returns:

The validated Branch instance

Return type:

Self

validate_destinations_and_default()

Validate destinations and set default if needed.

Ensures proper destination mappings and sets reasonable defaults.

Returns:

The validated Branch instance

Return type:

Self

model_config

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