haive.core.graph.node.callable_node

Callable Node - Wrap any callable as a graph node.

from typing import Any This module provides a way to wrap any Python callable (function, method, lambda) as a proper graph node that returns Command or Send objects.

Classes

CallableNodeConfig

Configuration for wrapping a callable as a node.

Functions

as_node(**kwargs)	Decorator to turn a function into a node.
wrap_callable(func[, name])	Convenience function to wrap a callable as a node.

Module Contents

class haive.core.graph.node.callable_node.CallableNodeConfig(/, **data)

Bases: haive.core.graph.node.base_node_config.BaseNodeConfig

Configuration for wrapping a callable as a node.

This allows any function to be used as a graph node by: 1. Extracting required parameters from state 2. Calling the function 3. Wrapping the result in Command/Send

Examples

Simple boolean check:

def check_threshold(messages: List[BaseMessage], threshold: int = 100) -> bool:
    total_length = sum(len(msg.content) for msg in messages)
    return total_length > threshold

node = CallableNodeConfig(
    name="check_threshold",
    callable_func=check_threshold,
    goto_on_true="summarize",
    goto_on_false="continue"
)

State function:

def needs_summarization(state: MessagesState) -> bool:
    return state.token_count > 1000

node = CallableNodeConfig(
    name="check_summary",
    callable_func=needs_summarization,
    extract_full_state=True
)

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)

get_default_input_fields()

Get input fields based on callable signature.

Return type:

list[haive.core.schema.field_definition.FieldDefinition]

validate_config()

Validate the configuration.

Return type:

Self

haive.core.graph.node.callable_node.as_node(**kwargs)

Decorator to turn a function into a node.

Examples

@as_node(goto_on_true=”next”, goto_on_false=”retry”) def should_continue(messages: List[BaseMessage]) -> bool:

return len(messages) > 5

Return type:

Any

haive.core.graph.node.callable_node.wrap_callable(func, name=None, **kwargs)

Convenience function to wrap a callable as a node.

Parameters:

• func (collections.abc.Callable) – The function to wrap

• name (str | None) – Node name (defaults to function name)

• **kwargs – Additional CallableNodeConfig parameters

Returns:

Configured CallableNodeConfig

Return type:

CallableNodeConfig

Examples

node = wrap_callable(

check_threshold, goto_on_true=”summarize”, goto_on_false=”continue”

)