haive.core.graph.node.agent_node_v3

Agent Node V3 - Hierarchical state projection for multi-agent systems.

This module provides AgentNodeV3 which enables sophisticated multi-agent workflows by properly handling state projection between container states (like MultiAgentState) and individual agent states, maintaining type safety and hierarchical access patterns.

Key Features:

  • Hierarchical State Management: Projects container states to agent-specific schemas

  • Direct Field Updates: Agents with output_schema update state fields directly

  • Type Safety: Maintains schema validation throughout execution

  • Backward Compatibility: Supports both structured and message-based agents

  • Dynamic Agent Lookup: Resolves agents from state at runtime

  • Recompilation Tracking: Monitors when agents need graph rebuilding

Architecture:

The AgentNodeV3 follows a projection-based approach where:

  1. Container State: MultiAgentState holds all agents and shared data

  2. State Projection: Each agent gets exactly what it expects

  3. Output Processing: Structured outputs update fields directly

  4. State Synchronization: Changes propagate back to container

Usage Patterns:

  • Self-Discover Workflows: Sequential agents building on each other’s outputs

  • Multi-Agent Coordination: Parallel agents sharing common state

  • Hierarchical Processing: Nested agent structures with state isolation

Examples

Basic agent node creation:

from haive.core.graph.node.agent_node_v3 import create_agent_node_v3
from haive.agents.simple import SimpleAgent

# Create agent with structured output
agent = SimpleAgent(
    name="analyzer",
    engine=AugLLMConfig(),
    structured_output_model=AnalysisResult
)

# Create node that updates fields directly
node = create_agent_node_v3(
    agent_name="analyzer",
    agent=agent
)

Multi-agent workflow:

# Sequential execution where agents read each other's outputs
state = MultiAgentState()
state.agents["planner"] = planner_agent
state.agents["executor"] = executor_agent

# Step 1: Planner outputs planning_result field
planner_node = create_agent_node_v3("planner")
result1 = planner_node(state, config)

# Step 2: Executor reads planning_result directly from state
executor_node = create_agent_node_v3("executor")
result2 = executor_node(state, config)

See also

Classes

AgentNodeV3Config

Agent node configuration with hierarchical state projection support.

Functions

create_agent_node_v3(agent_name[, agent, name])

Create an agent node V3 configuration.

Module Contents

class haive.core.graph.node.agent_node_v3.AgentNodeV3Config

Bases: haive.core.graph.node.base_node_config.BaseNodeConfig[TInput, TOutput]

Agent node configuration with hierarchical state projection support.

This node configuration enables sophisticated multi-agent workflows by handling execution of agents within container states (like MultiAgentState) with proper state projection, output processing, and hierarchy maintenance.

The AgentNodeV3Config is designed to work seamlessly with structured output agents, enabling direct field updates (like engine nodes) while maintaining backward compatibility with message-based agents.

Key Features:

  • Hierarchical State Management: Projects container states to agent schemas

  • Direct Field Updates: Agents with output_schema update state fields directly

  • Type Safety: Maintains schema validation throughout execution

  • Dynamic Agent Resolution: Resolves agents from state at runtime

  • Recompilation Tracking: Monitors when agents need graph rebuilding

  • Flexible Output Modes: Supports merge, replace, and isolate patterns

Architecture:

The node follows a three-phase execution pattern:

  1. State Projection: Container state → Agent-specific schema

  2. Agent Execution: Agent processes projected state

  3. Output Integration: Results → Container state updates

Key Improvements Over V2:

  • Works with MultiAgentState and similar container patterns

  • Projects state to exact agent schema (no flattening)

  • Maintains type safety throughout execution

  • Supports dynamic agent lookup from state

  • Enables Self-Discover style workflows

agent_name

Name of agent to execute (key in container’s agents dict)

Type:

str

agent

Agent instance (extracted from state if not provided)

Type:

Optional[Agent]

agent_state_field

Field in container holding agent states (default: “agent_states”)

Type:

str

agents_field

Field in container holding agent instances (default: “agents”)

Type:

str

project_state

Whether to project state to agent’s expected schema (default: True)

Type:

bool

shared_fields

Fields to share from container to agent (default: [“messages”])

Type:

List[str]

output_mode

How to handle outputs: ‘merge’, ‘replace’, or ‘isolate’ (default: “merge”)

Type:

str

update_container_state

Whether to update the container’s agent_states (default: True)

Type:

bool

track_recompilation

Whether to track agent recompilation needs (default: True)

Type:

bool

Examples

Basic configuration:

from haive.core.graph.node.agent_node_v3 import AgentNodeV3Config
from haive.agents.simple import SimpleAgent

# Create agent with structured output
agent = SimpleAgent(
    name="analyzer",
    engine=AugLLMConfig(),
    structured_output_model=AnalysisResult
)

# Create node configuration
config = AgentNodeV3Config(
    name="analysis_node",
    agent_name="analyzer",
    agent=agent
)

Custom state projection:

# Custom shared fields and output mode
config = AgentNodeV3Config(
    name="custom_node",
    agent_name="custom_agent",
    shared_fields=["messages", "context", "metadata"],
    output_mode="replace",
    project_state=True
)

Multi-agent workflow:

# Sequential agents with direct field access
planner_config = AgentNodeV3Config(
    name="planner_node",
    agent_name="planner"
)

executor_config = AgentNodeV3Config(
    name="executor_node",
    agent_name="executor"
)

# Execute in sequence
state = MultiAgentState()
result1 = planner_config(state, config)  # Updates planning fields
result2 = executor_config(state, config)  # Reads planning fields directly
Raises:

  • ValueError – If neither agent_name nor agent is provided

  • ValueError – If agent_name not found in state’s agents dict

  • AgentExecutionError – If agent execution fails

Note

For agents with structured output schemas, the node will update state fields directly (like engine nodes). For message-based agents, it uses the traditional agent_outputs pattern for backward compatibility.

See also

get_default_input_fields()

Get input fields based on container expectations.

Return type:

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

get_default_output_fields()

Get output fields.

Return type:

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

validate_agent_config()

Validate configuration.

Return type:

Self

haive.core.graph.node.agent_node_v3.create_agent_node_v3(agent_name, agent=None, name=None, **kwargs)

Create an agent node V3 configuration.

Convenience factory function for creating AgentNodeV3Config instances with sensible defaults. This function simplifies the creation of agent nodes for multi-agent workflows and handles common configuration patterns.

The function automatically: - Generates a descriptive node name if not provided - Rebuilds the model to resolve forward references - Applies default configuration suitable for most use cases - Supports both direct agent provision and runtime resolution

Parameters:

  • agent_name (str) – Name of agent to execute. This should match a key in the container state’s agents dictionary. Used for dynamic agent resolution at runtime.

  • agent (Optional[Agent]) – Optional agent instance to use directly. If provided, the agent will be used regardless of what’s in the container state. If None, agent will be resolved from state.agents[agent_name].

  • name (Optional[str]) – Optional node name for debugging and visualization. If not provided, will be auto-generated as “agent_{agent_name}”.

  • **kwargs – Additional configuration options passed to AgentNodeV3Config: - shared_fields: Fields to share from container (default: [“messages”]) - output_mode: How to handle outputs (“merge”, “replace”, “isolate”) - project_state: Whether to project state to agent schema (default: True) - track_recompilation: Whether to track recompilation needs (default: True)

Returns:

Configured agent node ready for execution in graphs.

The returned configuration can be used directly in LangGraph workflows.

Return type:

AgentNodeV3Config

Examples

Basic agent node creation:

from haive.core.graph.node.agent_node_v3 import create_agent_node_v3
from haive.agents.simple import SimpleAgent

# Create agent with structured output
agent = SimpleAgent(
    name="analyzer",
    engine=AugLLMConfig(),
    structured_output_model=AnalysisResult
)

# Create node - agent will be resolved from state
node = create_agent_node_v3("analyzer")

# Or provide agent directly
node = create_agent_node_v3("analyzer", agent=agent)

Custom configuration:

# Advanced configuration with custom settings
node = create_agent_node_v3(
    agent_name="custom_agent",
    name="custom_analysis_node",
    shared_fields=["messages", "context", "metadata"],
    output_mode="replace",
    project_state=True,
    track_recompilation=False
)

Multi-agent workflow:

# Create nodes for sequential execution
planner_node = create_agent_node_v3("planner")
executor_node = create_agent_node_v3("executor")
reviewer_node = create_agent_node_v3("reviewer")

# Build graph
from langgraph.graph import StateGraph
graph = StateGraph(MultiAgentState)
graph.add_node("plan", planner_node)
graph.add_node("execute", executor_node)
graph.add_node("review", reviewer_node)

# Chain execution
graph.add_edge("plan", "execute")
graph.add_edge("execute", "review")

Usage in LangGraph:

# Direct usage in graph execution
state = MultiAgentState()
state.agents["analyzer"] = analyzer_agent

# Execute node
node = create_agent_node_v3("analyzer")
result = node(state, {"debug": True})

# Apply updates
for key, value in result.update.items():
    if hasattr(state, key):
        setattr(state, key, value)
Raises:

  • ValueError – If agent_name is empty or None

  • ImportError – If Agent class cannot be imported for model rebuilding

Parameters:

  • agent_name (str)

  • agent (Union[haive.agents.base.agent.Agent, None])

  • name (str | None)

Return type:

AgentNodeV3Config

Note

The function automatically rebuilds the Pydantic model to resolve forward references to the Agent class. This ensures proper type validation and IDE support.

See also