agents.multi.enhanced.multi_agent_v4¶

Enhanced MultiAgent V4 - Advanced multi-agent orchestration with enhanced base agent pattern.

This module provides the MultiAgent class, which represents the recommended multi-agent coordination implementation in the Haive framework. It leverages the enhanced base agent pattern to provide sophisticated agent orchestration with clean, intuitive APIs.

Current Status: This is the most advanced and recommended MultiAgent implementation for new projects. It provides the cleanest API, best performance, and most complete feature set for multi-agent coordination.

The MultiAgent extends the base Agent class and implements the required build_graph() abstract method, enabling it to participate fully in the Haive ecosystem while providing advanced multi-agent capabilities.

Key Features:

Enhanced Base Agent Pattern: Properly extends Agent and implements build_graph()
Direct List Initialization: Simple API with agents=[agent1, agent2, …]
Multiple Execution Modes: Sequential, parallel, conditional, and manual orchestration
AgentNodeV3 Integration: Advanced state projection for clean agent isolation
MultiAgentState Management: Type-safe state handling across agents
Dynamic Graph Building: Auto, manual, and lazy build modes
Conditional Routing: Rich conditional edge support via BaseGraph2
Hot Agent Addition: Add agents dynamically with automatic recompilation

Architecture:

The MultiAgent follows a hierarchical architecture:

Agent Layer: Individual agents with their own state and logic
Orchestration Layer: Coordination logic and routing decisions
State Layer: MultiAgentState for shared and private state management
Execution Layer: AgentNodeV3 for proper state projection

Example

Basic sequential workflow:

>>> from haive.agents.multi.agent import MultiAgent
>>> from haive.agents.simple import SimpleAgent
>>> from haive.agents.react import ReactAgent
>>>
>>> # Create individual agents
>>> analyzer = ReactAgent(name="analyzer", tools=[...])
>>> formatter = SimpleAgent(name="formatter")
>>>
>>> # Create multi-agent workflow
>>> workflow = MultiAgent(
...     name="analysis_pipeline",
...     agents=[analyzer, formatter],
...     execution_mode="sequential"
... )
>>>
>>> # Execute workflow
>>> result = await workflow.arun({"task": "Analyze this data"})

Advanced conditional routing:

>>> # Create workflow with conditional execution
>>> workflow = MultiAgent(
...     name="smart_processor",
...     agents=[classifier, simple_processor, complex_processor],
...     execution_mode="conditional"
... )
>>>
>>> # Add routing logic
>>> workflow.add_conditional_edge(
...     from_agent="classifier",
...     condition=lambda state: state.get("complexity") > 0.7,
...     true_agent="complex_processor",
...     false_agent="simple_processor"
... )

Parallel execution with convergence:

>>> # Create parallel workflow
>>> workflow = MultiAgent(
...     name="parallel_analysis",
...     agents=[analyzer1, analyzer2, analyzer3, aggregator],
...     execution_mode="manual"
... )
>>>
>>> # Configure parallel execution
>>> workflow.add_edge(START, "analyzer1")
>>> workflow.add_edge(START, "analyzer2")
>>> workflow.add_edge(START, "analyzer3")
>>> workflow.add_edge("analyzer1", "aggregator")
>>> workflow.add_edge("analyzer2", "aggregator")
>>> workflow.add_edge("analyzer3", "aggregator")
>>> workflow.add_edge("aggregator", END)

Classes¶

MultiAgent

Enhanced MultiAgent V4 using enhanced base agent pattern.

Module Contents¶

class agents.multi.enhanced.multi_agent_v4.MultiAgent¶

Bases: haive.agents.base.agent.Agent

Enhanced MultiAgent V4 using enhanced base agent pattern.

This class properly extends the enhanced base Agent class and implements the build_graph() abstract method. It provides clean initialization with direct list support and flexible execution modes.

Example

>>> # Simple sequential
>>> workflow = MultiAgent(
...     name="my_workflow",
...     agents=[planner, executor, reviewer],
...     execution_mode="sequential"
... )
>>>
>>> # With conditional branching
>>> workflow = MultiAgent(
...     name="smart_workflow",
...     agents=[classifier, simple_processor, complex_processor],
...     execution_mode="conditional",
...     build_mode="manual"
... )
>>>
>>> # Add conditional edges
>>> workflow.add_conditional_edge(
...     from_agent="classifier",
...     condition=lambda state: state.get("complexity") > 0.7,
...     true_agent="complex_processor",
...     false_agent="simple_processor"
... )
>>>
>>> # Build and execute
>>> workflow.build()
>>> result = await workflow.arun({"task": "Process this data"})

add_agent(agent)¶

Add an agent dynamically to the workflow.

This method allows adding agents after initialization. If build_mode is ‘auto’, the graph will be automatically rebuilt.

Parameters:: agent (haive.agents.base.agent.Agent) – The Agent instance to add.
Raises:: ValueError – If agent lacks a name or name already exists.
Return type:: None

Example

>>> new_agent = SimpleAgent(name="validator")
>>> workflow.add_agent(new_agent)

Note

In ‘auto’ build mode, this triggers graph recompilation. In other modes, you must rebuild the graph manually.

add_conditional_edge(from_agent, condition, true_agent, false_agent=END)¶

Add a conditional edge that routes based on a boolean condition.

This method creates a branching point in the workflow where execution can take different paths based on the result of a condition function.

Parameters:

from_agent (str) – Name of the agent where the condition is evaluated.
condition (collections.abc.Callable[[Any], bool]) – Callable that takes the state and returns True or False.
true_agent (str) – Agent to route to when condition returns True.
false_agent (str) – Agent to route to when condition returns False (default: END).

Raises:

ValueError – If from_agent doesn’t exist.

Return type:

None

Example

>>> def check_complexity(state):
...     return state.get("complexity", 0) > 0.7
...
>>> workflow.add_conditional_edge(
...     from_agent="analyzer",
...     condition=check_complexity,
...     true_agent="complex_processor",
...     false_agent="simple_processor"
... )

Note

The condition function receives the full MultiAgentState and should return a boolean value. For more complex routing, use add_multi_conditional_edge.

add_edge(from_agent, to_agent)¶

Add a direct edge between two agents in the graph.

This method creates a simple connection from one agent to another, useful for building custom execution flows in manual mode.

Parameters:

from_agent (str) – Name of the source agent.
to_agent (str) – Name of the destination agent (or END for termination).

Raises:

ValueError – If from_agent doesn’t exist or to_agent is invalid.

Return type:

None

Example

>>> workflow = MultiAgent(
...     agents=[agent1, agent2, agent3],
...     execution_mode="manual"
... )
>>> workflow.add_edge("agent1", "agent2")
>>> workflow.add_edge("agent2", "agent3")
>>> workflow.add_edge("agent3", END)

Note

If the graph is already built, the edge is added immediately. Otherwise, it will be added when the graph is built.

add_multi_conditional_edge(from_agent, condition, routes, default=END)¶

Add multi-way conditional edge with multiple destinations.

This method creates a branching point where the condition function returns a string key that maps to different destination agents.

Parameters:

from_agent (str) – Name of the agent where routing decision is made.
condition (collections.abc.Callable[[Any], str]) – Callable that returns a route key string.
routes (dict[str, str]) – Dictionary mapping route keys to agent names.
default (str) – Default agent when condition returns unmatched key.

Raises:

ValueError – If from_agent doesn’t exist.

Return type:

None

Example

>>> def categorize(state):
...     return state.get("category", "other")
...
>>> workflow.add_multi_conditional_edge(
...     from_agent="categorizer",
...     condition=categorize,
...     routes={
...         "technical": "tech_agent",
...         "sales": "sales_agent",
...         "support": "support_agent"
...     },
...     default="general_agent"
... )

build_graph()¶

Build the computational graph for multi-agent orchestration.

This method implements the abstract build_graph() from the base Agent class, fulfilling the enhanced base agent pattern. It constructs a BaseGraph that defines how agents are connected and how data flows between them.

The graph structure depends on the execution_mode: - sequential: Agents execute one after another in order - parallel: All agents execute simultaneously - conditional: Agents execute based on conditional routing - manual: User must add edges manually after creation

Returns:: The constructed graph ready for compilation and execution.
Return type:: BaseGraph
Raises:: ValueError – If no agents are available to build the graph.

Note

This method is called automatically based on build_mode: - auto: Called during initialization - manual: Must be called explicitly via build() - lazy: Called on first execution

Example

>>> # Manual build mode
>>> workflow = MultiAgent(
...     name="custom",
...     agents=[agent1, agent2],
...     build_mode="manual"
... )
>>> graph = workflow.build_graph()  # Build explicitly
>>> workflow.add_edge("agent1", "agent2")  # Add custom edges

display_info()¶

Display detailed information about the workflow configuration.

This method prints a formatted summary of the workflow including: - Execution and build modes - Registered agents and their types - Number of conditional edges - Graph build status

Example

>>> workflow.display_info()
=== Enhanced MultiAgent V4: analysis_pipeline ===
Execution Mode: sequential
Build Mode: auto
Entry Point: analyzer
Agents (3):
  1. analyzer (ReactAgent)
  2. processor (SimpleAgent)
  3. formatter (SimpleAgent)
Conditional Edges: 0
Graph Built: Yes

Return type:: None

get_agent(name)¶

Retrieve an agent instance by name.

Parameters:: name (str) – The name of the agent to retrieve.
Returns:: The agent instance if found, None otherwise.
Return type:: Optional[Agent]

Example

>>> agent = workflow.get_agent("analyzer")
>>> if agent:
...     print(f"Found agent: {agent.name}")

get_agent_names()¶

Get list of all agent names in the workflow.

Returns:: Names of all registered agents.
Return type:: List[str]

Example

>>> names = workflow.get_agent_names()
>>> print(names)  # ['analyzer', 'processor', 'formatter']

setup_agent()¶

Set up multi-agent configuration before graph building.

This method is called BEFORE schema generation and graph building, allowing us to convert agents list to dict properly.

Return type:: None