haive.core.graph.state_graph.base_graph2¶

Base graph implementation for the Haive framework.

Provides a comprehensive system for building, manipulating, and executing graphs with consistent interfaces, serialization support, and dynamic composition.

Classes¶

`BaseGraph`	Base class for graph management in the Haive framework.
`BranchType`	Types of branches for conditional routing.
`EdgeType`	Types of edges in a graph.
`Node`	Base node in a graph system.

Functions¶

`create_debug_has_tool_calls`(original_func)	Create a debug wrapper around a has_tool_calls function to help diagnose issues.
`debug_wrapper`(original_func)	Debug wrapper function for external use.
`has_tool_calls_fixed`(state)	FIXED VERSION: Check if the last AI message has tool calls.

Module Contents¶

class haive.core.graph.state_graph.base_graph2.BaseGraph(/, **data)¶

Bases: pydantic.BaseModel, haive.core.graph.state_graph.validation_mixin.ValidationMixin

Base class for graph management in the Haive framework.

Provides comprehensive graph management capabilities including: - Node management (add, remove, update) - Edge management (direct and branch-based) - Branch management - Graph validation - Serialization

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)

add_boolean_conditional_edges(source_node, condition, true_destination, false_destination=END, also_accept_strings=True)¶

Add conditional edges that explicitly handle boolean results.

This is a convenience method for the common case where you have a condition that returns True/False and you want clear routing.

Parameters:

source_node (str) – Source node name
condition (collections.abc.Callable[[Any], bool]) – Function that returns True or False
true_destination (str) – Where to go when condition returns True
false_destination (str) – Where to go when condition returns False
also_accept_strings (bool) – Whether to also accept string equivalents like ‘has_tool_calls’/’no_tool_calls’

Returns:

Self for method chaining

Return type:

BaseGraph

Example:

graph.add_boolean_conditional_edges(
    'agent_node',
    has_tool_calls,  # Function that returns True/False
    'validation',    # Go here when True
    END             # Go here when False
)

add_branch(branch_or_name, source_node=None, condition=None, routes=None, branch_type=None, **kwargs)¶

Add a branch to the graph with flexible input options.

Parameters:

branch_or_name (haive.core.graph.branches.branch.Branch | str) – Branch object or branch name
source_node (str | None) – Source node for the branch (required if branch_or_name is a string)
condition (Any | None) – Condition function or key/value for evaluation
routes (dict[bool | str, str] | None) – Mapping of condition results to target nodes
branch_type (BranchType | None) – Type of branch (determined automatically if not provided)
**kwargs – Additional parameters for branch creation

Returns:

Self for method chaining

Return type:

BaseGraph

add_conditional_edges(source_node, condition, destinations=None, default=END, create_missing_nodes=False)¶

Add conditional edges from a source node based on a condition.

This method supports multiple ways to handle True/False routing:

Boolean destinations: Use True/False as keys:

graph.add_conditional_edges(
    'agent_node',
    has_tool_calls,
    {True: 'validation', False: END}
)

String destinations with optional boolean fallbacks: String keys like ‘has_tool_calls’/’no_tool_calls’ can optionally get True/False fallbacks added by setting add_boolean_fallbacks=True:

graph.add_conditional_edges(
    'agent_node',
    has_tool_calls,
    {'has_tool_calls': 'validation', 'no_tool_calls': END},
    add_boolean_fallbacks=True
)
# With add_boolean_fallbacks=True, adds: {True: 'validation', False: END}

List format: First item = True destination, Second item = False destination:

graph.add_conditional_edges(
    'agent_node',
    has_tool_calls,
    ['validation', END]  # validation when True, END when False
)

Alternative: For simple boolean routing, consider using add_boolean_conditional_edges() which provides cleaner syntax for True/False conditions.

Parameters:

source_node (str) – Source node name
condition (haive.core.graph.branches.branch.Branch | collections.abc.Callable[[haive.core.graph.common.types.StateLike, haive.core.graph.common.types.ConfigLike | None], BranchResultType] | Any) – A function, Branch, NodeConfig or any object that can determine branching. For callables, takes (state, optional config) and returns a node name, boolean, list of nodes, list of Send objects, Send object, Command object, or a Branch object itself
destinations (str | list[str] | dict[bool | str | int, str] | None) – Target node(s) - can be: - A single node name string (which will be mapped to True) - A list of node names (which will be mapped by index) - A dictionary mapping condition results to target nodes
default (str | Literal['END'] | None) – Default destination if no condition matches (defaults to END). IMPORTANT: When destinations is a dictionary, no default is ever added.
create_missing_nodes (bool) – Whether to create missing destination nodes automatically (defaults to False)
add_boolean_fallbacks – Whether to automatically add True/False keys for string-keyed destinations (defaults to False)

Returns:

Self for method chaining

Return type:

BaseGraph

add_edge(source, target)¶

Add a direct edge to the graph.

Parameters:

source (str) – Source node name
target (str) – Target node name

Returns:

Self for method chaining

Return type:

BaseGraph

add_function_branch(source_node, condition, routes, default_route=END, name=None)¶

Add a function-based branch.

Parameters:

source_node (str) – Source node name
condition (collections.abc.Callable[[Any], Any]) – Condition function
routes (dict[bool | str, str]) – Mapping of condition results to target nodes
default_route (str) – Default destination
name (str | None) – Optional branch name

Returns:

Self for method chaining

Return type:

BaseGraph

add_intelligent_agent_routing(agents, execution_mode='infer', branches=None, prefix='agent_')¶

Add intelligent agent routing with sequence inference and branching.

Parameters:

agents (dict[str, Any]) – Dictionary of agent name -> agent instance
execution_mode (str) – Execution mode (infer, sequential, parallel, branch, conditional)
branches (dict[str, dict[str, Any]] | None) – Branch configurations for conditional routing
prefix (str) – Prefix for agent node names

Returns:

Self for method chaining

Return type:

BaseGraph

add_key_value_branch(source_node, key, value, comparison=ComparisonType.EQUALS, true_dest='continue', false_dest=END, name=None)¶

Add a key-value comparison branch.

Parameters:

source_node (str) – Source node name
key (str) – State key to check
value (Any) – Value to compare against
comparison (haive.core.graph.branches.types.ComparisonType | str) – Type of comparison
true_dest (str) – Destination if true
false_dest (str) – Destination if false
name (str | None) – Optional branch name

Returns:

Self for method chaining

Return type:

BaseGraph

add_node(node_or_name, node_like=None, **kwargs)¶

Add a node to the graph with flexible input options.

Parameters:

node_or_name (Node | dict[str, Any] | str | haive.core.graph.node.config.NodeConfig) – Node object, dictionary, node name, or NodeConfig
node_like (Any | None) – If node_or_name is a string, this is the node object, callable, or engine
**kwargs – Additional properties when creating a node from name

Returns:

Self for method chaining

Return type:

BaseGraph

add_parallel_branches(source_node, branches, branch_names=None, join_node=None, join_node_obj=None, **kwargs)¶

Add parallel branches from a source node, optionally joining at a common node.

Parameters:

source_node (str) – Name of the source node
branches (list[list[str] | list[Node] | list[dict] | list[Any]]) – List of node sequences (each sequence is a branch)
branch_names (list[str] | None) – Optional names for the branches (used in branch creation)
join_node (str | Node | dict | Any | None) – Optional node to join all branches
join_node_obj (Any | None) – If join_node is a string, this is the node object/callable
**kwargs – Additional properties for nodes

Returns:

Self for method chaining

Return type:

BaseGraph

add_postlude_node(postlude_node, node_obj=None, **kwargs)¶

Add a node at the end of the graph (before END).

Parameters:

postlude_node (str | Node | dict | Any) – Node to add at the end
node_obj (Any | None) – If postlude_node is a string, this is the node object/callable
**kwargs – Additional properties if creating from name

Returns:

Self for method chaining

Return type:

BaseGraph

add_prelude_node(prelude_node, node_obj=None, **kwargs)¶

Add a node at the beginning of the graph (after START).

Parameters:

prelude_node (str | Node | dict | Any) – Node to add at the start
node_obj (Any | None) – If prelude_node is a string, this is the node object/callable
**kwargs – Additional properties if creating from name

Returns:

Self for method chaining

Return type:

BaseGraph

add_sequence(nodes, node_objects=None, connect_start=False, connect_end=False, **kwargs)¶

Add a sequence of nodes and connect them in order.

Parameters:

nodes (list[str | Node | dict | Any]) – List of nodes to add (names, objects, or dictionaries)
node_objects (list[Any] | None) – If nodes contains strings, these are the corresponding objects/callables
connect_start (bool) – Whether to connect the first node to START
connect_end (bool) – Whether to connect the last node to END
**kwargs – Additional properties to apply to all nodes

Returns:

Self for method chaining

Return type:

BaseGraph

add_subgraph(name, subgraph, **kwargs)¶

Add a subgraph as a node.

Parameters:

name (str) – Name for the subgraph node
subgraph (BaseGraph) – Subgraph object to add
**kwargs – Additional node properties

Returns:

Self for method chaining

Return type:

BaseGraph

add_tool_node(node_name, node_type=NodeType.TOOL, **kwargs)¶

Add a tool node to the graph.

Parameters:

node_name (str) – Name of the node
node_type (haive.core.graph.common.types.NodeType) – Type of node (defaults to TOOL)
**kwargs – Additional properties for the node

Returns:

Self for method chaining

Return type:

BaseGraph

analyze_cycles()¶

Find all cycles in the graph.

Return type:: list[list[str]]

check_full_recompilation_needed(input_schema=None, output_schema=None, config_schema=None, interrupt_before=None, interrupt_after=None, **compile_kwargs)¶

Check if recompilation is needed for any reason and provide details.

Parameters:

input_schema (Any | None) – Input schema to check
output_schema (Any | None) – Output schema to check
config_schema (Any | None) – Config schema to check
interrupt_before (list | None) – Interrupt before configuration to check
interrupt_after (list | None) – Interrupt after configuration to check
**compile_kwargs – Additional compilation parameters to check

Returns:

Dictionary with recompilation status and reasons

Return type:

dict[str, Any]

check_graph_validity()¶

Validate the graph structure.

Returns:: List of validation issues (empty if graph is valid)
Return type:: Any

compile(raise_on_validation_error=False)¶

Validate and compile the graph to a runnable LangGraph StateGraph.

Parameters:: raise_on_validation_error (bool) – Whether to raise an exception on validation errors
Returns:: Compiled LangGraph StateGraph
Raises:: ValueError – If validation fails and raise_on_validation_error is True
Return type:: Any

debug_conditional_routing(source_node)¶

Debug conditional routing for a specific node.

This shows how boolean and string results will be routed for debugging purposes.

Parameters:: source_node (str) – The node to debug routing for
Return type:: None

extend_from(other_graph, prefix='')¶

Extend this graph with nodes and edges from another graph.

Parameters:

other_graph – Graph to extend from
prefix – Optional prefix for imported node names

Returns:

Self for method chaining

Return type:

Any

find_all_paths(start_node=START, end_node=END, max_depth=100, include_loops=False, debug=False)¶

Find all possible paths between two nodes.

Parameters:

start_node – Starting node (defaults to START)
end_node – Ending node (defaults to END)
max_depth – Maximum path depth to prevent infinite loops
include_loops – Whether to include paths with loops/cycles
debug – Whether to show detailed debug logging

Returns:

List of GraphPath objects

find_dangling_edges()¶

Find edges pointing to non-existent nodes.

Return type:: list[tuple[str, str]]

find_nodes_without_end_path()¶

Find nodes that can’t reach END.

Returns:: List of node names that can’t reach END
Return type:: Any | None

find_nodes_without_finish_path()¶

Find nodes that can’t reach a finish point. Alias for find_nodes_without_end_path for API consistency.

Returns:: List of node names that can’t reach a finish point
Return type:: Any | None

find_orphan_nodes()¶

Find nodes with no incoming or outgoing edges.

Return type:: list[str]

find_unreachable_nodes()¶

Find nodes that can’t be reached from START.

Returns:: List of unreachable node names
Return type:: Any | None

classmethod from_dict(data)¶

Create a graph from a dictionary.

Parameters:: data (dict[str, Any]) – Dictionary representation of the graph
Returns:: Instantiated graph
Return type:: BaseGraph

classmethod from_json(json_str)¶

Create a graph from a JSON string.

Parameters:: json_str (str) – JSON string representation of the graph
Returns:: Instantiated graph
Return type:: BaseGraph

classmethod from_langgraph(state_graph, name=None)¶

Create a BaseGraph from a LangGraph StateGraph.

Parameters:

state_graph (Any) – LangGraph StateGraph instance
name (str | None) – Optional name for the created graph

Returns:

BaseGraph instance

Return type:

BaseGraph

get_branch(branch_id)¶

Get a branch by ID.

Parameters:: branch_id (str) – ID of the branch to retrieve
Returns:: Branch object if found, None otherwise
Return type:: haive.core.graph.branches.branch.Branch | None

get_branch_by_name(name)¶

Get a branch by name.

Parameters:: name (str) – Name of the branch to retrieve
Returns:: First matching branch or None if not found
Return type:: haive.core.graph.branches.branch.Branch | None

get_branches_for_node(node_name)¶

Get all branches with a given source node.

Parameters:: node_name (str) – Name of the source node
Returns:: List of branch objects
Return type:: list[haive.core.graph.branches.branch.Branch]

get_compilation_info()¶

Get information about the compilation state.

Returns:: Dictionary with compilation state information
Return type:: dict[str, Any]

get_conditional_entries()¶

Get all conditional entry points.

Returns:: Dictionary of conditional entries indexed by ID
Return type:: dict[str, dict[str, Any]]

get_conditional_exits()¶

Get all conditional exit points.

Returns:: Dictionary of conditional exits indexed by ID
Return type:: dict[str, dict[str, Any]]

get_edges(source=None, target=None, include_branches=True)¶

Get edges matching criteria.

Parameters:

source (str | None) – Filter by source node
target (str | None) – Filter by target node
include_branches (bool) – Include edges from branches

Returns:

List of matching edges as (source, target) tuples

Return type:

list[tuple[str, str]]

get_node(node_name)¶

Get a node by name.

Parameters:: node_name (str) – Name of the node to retrieve
Returns:: Node object if found, None otherwise
Return type:: Any | None

get_node_pattern(pattern)¶

Get nodes matching a pattern.

Parameters:: pattern (str) – Pattern to match (supports * wildcard)
Returns:: List of matching nodes
Return type:: list[Node]

get_sink_nodes()¶

Get nodes that have no outgoing edges (other than to END).

Returns:: List of sink node names
Return type:: Any | None

get_source_nodes()¶

Get nodes that have no incoming edges (other than START).

Returns:: List of source node names
Return type:: Any | None

has_entry_point()¶

Check if the graph has an entry point.

Return type:: bool

has_path(source, target)¶

Check if there is a path between source and target nodes.

Parameters:

source (str) – Source node name
target (str) – Target node name

Returns:

True if path exists, False otherwise

Return type:

bool

insert_node_after(target_node, new_node, new_node_obj=None, **kwargs)¶

Insert a new node after an existing node, redirecting all outgoing connections.

Parameters:

target_node (str) – Name of the existing node
new_node (str | Node | dict | Any) – New node name, object, or dictionary
new_node_obj (Any | None) – If new_node is a string, this is the node object/callable
**kwargs – Additional properties if creating from name

Returns:

Self for method chaining

Return type:

BaseGraph

insert_node_before(target_node, new_node, new_node_obj=None, **kwargs)¶

Insert a new node before an existing node, redirecting all incoming connections.

Parameters:

target_node (str) – Name of the existing node
new_node (str | Node | dict | Any) – New node name, object, or dictionary
new_node_obj (Any | None) – If new_node is a string, this is the node object/callable
**kwargs – Additional properties if creating from name

Returns:

Self for method chaining

Return type:

BaseGraph

mark_compiled(input_schema=None, output_schema=None, config_schema=None, interrupt_before=None, interrupt_after=None, **compile_kwargs)¶

Mark the graph as compiled and reset the recompilation flag.

Parameters:

input_schema (Any | None) – Input schema used for compilation
output_schema (Any | None) – Output schema used for compilation
config_schema (Any | None) – Config schema used for compilation
interrupt_before (list | None) – Interrupt before nodes used for compilation
interrupt_after (list | None) – Interrupt after nodes used for compilation
**compile_kwargs – Additional compilation parameters

Return type:

None

needs_recompile()¶

Check if the graph needs recompilation.

Returns:: True if the graph has been modified since last compilation
Return type:: bool

needs_recompile_for_interrupts(interrupt_before=None, interrupt_after=None)¶

Check if recompilation is needed due to interrupt changes.

Parameters:

interrupt_before (list | None) – New interrupt_before list to compare
interrupt_after (list | None) – New interrupt_after list to compare

Returns:

True if interrupt configuration has changed since last compilation

Return type:

bool

needs_recompile_for_schemas(input_schema=None, output_schema=None, config_schema=None)¶

Check if recompilation is needed due to schema changes.

Parameters:

input_schema (Any | None) – New input schema to compare
output_schema (Any | None) – New output schema to compare
config_schema (Any | None) – New config schema to compare

Returns:

True if any schema has changed since last compilation

Return type:

bool

remove_branch(branch_id)¶

Remove a branch from the graph.

Parameters:: branch_id (str) – ID of the branch to remove
Returns:: Self for method chaining
Return type:: BaseGraph

remove_conditional_entry(entry_id)¶

Remove a conditional entry point.

Parameters:: entry_id (str) – ID of the conditional entry to remove
Returns:: Self for method chaining
Return type:: BaseGraph

remove_conditional_exit(exit_id)¶

Remove a conditional exit point.

Parameters:: exit_id (str) – ID of the conditional exit to remove
Returns:: Self for method chaining
Return type:: BaseGraph

remove_edge(source, target=None)¶

Remove an edge from the graph.

Parameters:

source (str) – Source node name
target (str | None) – Target node name (if None, removes all edges from source)

Returns:

Self for method chaining

Return type:

BaseGraph

remove_node(node_name)¶

Remove a node from the graph.

Parameters:: node_name (str) – Name of the node to remove
Returns:: Self for method chaining
Return type:: BaseGraph

replace_branch(branch_id, new_branch)¶

Replace a branch with a new one.

Parameters:

branch_id (str) – ID of the branch to replace
new_branch (haive.core.graph.branches.branch.Branch) – New branch to insert

Returns:

Self for method chaining

Return type:

BaseGraph

replace_node(node_name, new_node, preserve_connections=True)¶

Replace a node while optionally preserving its connections.

Parameters:

node_name (str) – Name of the node to replace
new_node (Node | dict | Any) – New node to insert
preserve_connections (bool) – Whether to preserve existing connections

Returns:

Self for method chaining

Return type:

BaseGraph

set_conditional_entry(condition, entry_node, default_entry=None)¶

Set a conditional entry point for the graph.

Parameters:

condition (collections.abc.Callable[[haive.core.graph.common.types.StateLike, haive.core.graph.common.types.ConfigLike | None], bool]) – Function that takes state and config, returns boolean
entry_node (str) – Node to enter if condition is True
default_entry (str | None) – Node to enter if condition is False (uses self.entry_point if None)

Returns:

Self for method chaining

Return type:

BaseGraph

set_conditional_exit(node_name, condition, exit_if_true=True)¶

Set a conditional exit point for the graph.

Parameters:

node_name (str) – Name of the node to set as conditional exit
condition (collections.abc.Callable[[haive.core.graph.common.types.StateLike, haive.core.graph.common.types.ConfigLike | None], bool]) – Function that takes state and config, returns boolean
exit_if_true (bool) – Whether to exit when condition is True (default) or False

Returns:

Self for method chaining

Return type:

BaseGraph

set_end_point(node_name)¶

Deprecated: Use set_finish_point instead.

Set a finish point of the graph.

Parameters:: node_name (str) – Name of the node to set as a finish point
Returns:: Self for method chaining
Return type:: BaseGraph

set_entry_point(node_name)¶

Set an entry point of the graph.

Parameters:: node_name (str) – Name of the node to set as an entry point
Returns:: Self for method chaining
Return type:: BaseGraph

set_finish_point(node_name)¶

Set a finish point of the graph.

Parameters:: node_name (str) – Name of the node to set as a finish point
Returns:: Self for method chaining
Return type:: BaseGraph

set_state_schema(schema)¶

Set the state schema and mark the graph as needing recompilation.

Parameters:: schema (type[pydantic.BaseModel] | None) – New state schema to use
Returns:: Self for method chaining
Return type:: BaseGraph

to_dict()¶

Convert the graph to a serializable dictionary.

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

to_json(**kwargs)¶

Convert graph to JSON string.

Parameters:: **kwargs – Additional parameters for JSON serialization
Returns:: JSON string representation of the graph
Return type:: str

to_langgraph(state_schema=None, input_schema=None, output_schema=None, config_schema=None, **kwargs)¶

Convert to LangGraph StateGraph with proper schema handling.

Schema Resolution Logic: 1. If state_schema provided: use it, default input/output to state_schema 2. If input_schema and output_schema provided: use them, create PassThroughState for state_schema 3. If only input_schema provided: use it for both input and state, output defaults to state 4. If only output_schema provided: use it for both output and state, input defaults to state 5. If none provided: use self.state_schema or dict

Note: This method marks the graph as compiled after successful conversion.

Parameters:

state_schema (type[pydantic.BaseModel] | None)
input_schema (type[pydantic.BaseModel] | None)
output_schema (type[pydantic.BaseModel] | None)
config_schema (type[pydantic.BaseModel] | None)

Return type:

Any

to_mermaid(include_subgraphs=True, theme='default', subgraph_mode='cluster', show_default_branches=False)¶

Generate a Mermaid graph diagram string.

Parameters:

include_subgraphs (bool) – Whether to visualize subgraphs as clusters
theme (str) – Mermaid theme name (default, forest, dark, neutral)
subgraph_mode (str) – How to render subgraphs (“cluster”, “inline”, or “separate”) - DEPRECATED
show_default_branches (bool) – Whether to show default branches

Returns:

Mermaid diagram as string

Return type:

str

update_branch(branch_id, **updates)¶

Update a branch’s properties.

Parameters:

branch_id (str) – ID of the branch to update
**updates – Properties to update

Returns:

Self for method chaining

Return type:

BaseGraph

update_node(node_name, **updates)¶

Update a node’s properties.

Parameters:

node_name (str) – Name of the node to update
**updates – Properties to update

Returns:

Self for method chaining

Return type:

BaseGraph

validate_graph()¶

Validate the graph structure.

Return type:: Self

visualize(output_path=None, include_subgraphs=True, highlight_nodes=None, highlight_paths=None, save_png=True, width='100%', theme='default', subgraph_mode='cluster', show_default_branches=False, debug=False)¶

Generate and display a visualization of the graph.

This method attempts multiple rendering approaches based on the environment, with fallbacks to ensure something is always displayed.

Parameters:

output_path (str | None) – Optional path to save the diagram
include_subgraphs (bool) – Whether to visualize subgraphs as clusters
highlight_nodes (list[str] | None) – List of node names to highlight
highlight_paths (list[list[str]] | None) – List of paths to highlight (each path is a list of node names)
save_png (bool) – Whether to save the diagram as PNG
width (str) – Width of the displayed diagram
theme (str) – Mermaid theme to use (e.g., “default”, “forest”, “dark”, “neutral”)
subgraph_mode (str) – How to render subgraphs (“cluster”, “inline”, or “separate”) - DEPRECATED
show_default_branches (bool) – Whether to show default branches
debug (bool) – Whether to enable debug output

Returns:

The generated Mermaid code

Return type:

str

property all_entry_points: dict[str, Any]¶

Property that returns all entry points (regular and conditional).

Returns:: Dictionary containing all entry points information
Return type:: dict[str, Any]

property all_exit_points: dict[str, Any]¶

Use all_finish_points instead.

Type:: Deprecated
Return type:: dict[str, Any]

property all_finish_points: dict[str, Any]¶

Property that returns all finish points (regular and conditional).

Returns:: Dictionary containing all finish points information
Return type:: dict[str, Any]

property conditional_edges: Any¶

Property for accessing branches as conditional edges (compatibility).

Returns:: Dictionary of branches indexed by ID
Return type:: Any

property entry_points_data: dict[str, Any]¶

Use all_entry_points instead.

Type:: Deprecated
Return type:: dict[str, Any]

property exit_points: dict[str, Any]¶

Use all_finish_points instead.

Type:: Deprecated
Return type:: dict[str, Any]

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

class haive.core.graph.state_graph.base_graph2.BranchType¶

Bases: str, enum.Enum

Types of branches for conditional routing.

Initialize self. See help(type(self)) for accurate signature.

class haive.core.graph.state_graph.base_graph2.EdgeType¶

Bases: str, enum.Enum

Types of edges in a graph.

Initialize self. See help(type(self)) for accurate signature.

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

Bases: pydantic.BaseModel, Generic[haive.core.graph.common.types.StateLike, haive.core.graph.common.types.ConfigLike, haive.core.graph.common.types.NodeOutput]

Base node in a graph system.

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)

abstractmethod process(state, config=None)¶

Process state and return output.

Parameters:

state (haive.core.graph.common.types.StateLike)
config (haive.core.graph.common.types.ConfigLike | None)

Return type:

haive.core.graph.common.types.NodeOutput

to_dict()¶

Convert to serializable dictionary.

Return type:: dict[str, Any]

property display_name: str¶

Get a human-readable display name.

Return type:: str

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

haive.core.graph.state_graph.base_graph2.create_debug_has_tool_calls(original_func)¶

Create a debug wrapper around a has_tool_calls function to help diagnose issues.

Parameters:: original_func – The original has_tool_calls function
Returns:: A wrapped function with detailed debugging
Return type:: Any

haive.core.graph.state_graph.base_graph2.debug_wrapper(original_func)¶

Debug wrapper function for external use.

Return type:: None

haive.core.graph.state_graph.base_graph2.has_tool_calls_fixed(state)¶

FIXED VERSION: Check if the last AI message has tool calls.

This function properly checks for tool calls in various message formats and handles edge cases that the original function missed.

Parameters:: state (dict[str, Any]) – The state object containing messages
Returns:: True if the last AI message has tool calls, False otherwise
Return type:: bool