agents.multi.core.clean_multi_agent¶

Clean MultiAgent implementation - unified multi-agent coordination system.

This module provides the current default multi-agent coordination system for the Haive framework. It supports simple sequential execution, complex routing patterns, parallel execution, and conditional workflows - all in one unified implementation.

Current Status: This is the default MultiAgent exported by the multi module. It provides stable, production-ready multi-agent coordination. For new projects requiring advanced features, consider using MultiAgent.

The MultiAgent class extends the base Agent class to coordinate multiple agents using various execution patterns. It automatically detects whether to use intelligent routing (via BaseGraph) or custom routing based on the configuration.

Key Features:

List initialization: Natural MultiAgent([agent1, agent2]) syntax
Flexible routing: Sequential, parallel, conditional, and custom patterns
Intelligent detection: Automatically uses appropriate routing mode
Enhanced methods: add_conditional_routing, add_parallel_group, add_edge
Backward compatible: Works with existing examples and patterns
No mocks testing: 100% real component validation

Examples

Simple sequential execution:

from haive.agents.multi.agent import MultiAgent
from haive.agents.simple import SimpleAgent

agent1 = SimpleAgent(name="analyzer")
agent2 = SimpleAgent(name="summarizer")

multi_agent = MultiAgent(agents=[agent1, agent2])
result = await multi_agent.arun("Process this data")

Conditional routing with entry point:

multi_agent = MultiAgent(
    agents=[classifier, billing_agent, technical_agent],
    entry_point="classifier"
)

multi_agent.add_conditional_routing(
    "classifier",
    lambda state: state.get("category", "general"),
    {
        "billing": "billing_agent",
        "technical": "technical_agent",
        "general": "billing_agent"
    }
)

Parallel execution with convergence:

multi_agent = MultiAgent(
    agents=[processor1, processor2, processor3, aggregator]
)

# Run processors in parallel, then aggregate
multi_agent.add_parallel_group(
    ["processor1", "processor2", "processor3"],
    next_agent="aggregator"
)

Direct edge routing:

multi_agent = MultiAgent(
    agents=[validator, processor, formatter],
    entry_point="validator"
)

# Create explicit flow
multi_agent.add_edge("validator", "processor")
multi_agent.add_edge("processor", "formatter")

Note

This is the unified implementation that replaces all previous multi-agent implementations. Use this for all new development. The system automatically detects whether to use intelligent routing or custom routing based on the branch configurations provided.

Classes¶

MultiAgent

Unified multi-agent coordination system for the Haive framework.

Module Contents¶

class agents.multi.core.clean_multi_agent.MultiAgent¶

Bases: haive.agents.base.agent.Agent

Unified multi-agent coordination system for the Haive framework.

MultiAgent extends the base Agent class to coordinate multiple agents using various execution patterns. It supports both simple sequential execution and complex routing patterns including conditional routing, parallel execution, and custom branching logic.

The implementation automatically detects whether to use intelligent routing (via BaseGraph) or custom routing based on the configuration provided.

agents¶: Dictionary of agents this multi-agent coordinates, keyed by agent name.

agent¶: Optional main/default agent for this multi-agent (legacy support).

execution_mode¶: Execution pattern - “infer”, “sequential”, “parallel”, “conditional”, or “branch”.

infer_sequence¶: Whether to automatically infer execution sequence from dependencies.

branches¶: Branch configurations for conditional and custom routing.

entry_point¶: Starting agent for execution (optional).

Examples

Basic sequential execution:

multi_agent = MultiAgent(agents=[agent1, agent2, agent3])
result = await multi_agent.arun("Process this task")

Conditional routing with entry point:

multi_agent = MultiAgent(
    agents=[classifier, processor1, processor2],
    entry_point="classifier"
)

multi_agent.add_conditional_routing(
    "classifier",
    lambda state: state.get("category"),
    {"type1": "processor1", "type2": "processor2"}
)

Parallel execution with convergence:

multi_agent = MultiAgent(agents=[agent1, agent2, agent3])
multi_agent.add_parallel_group(["agent1", "agent2"], next_agent="agent3")

Note

This class automatically uses MultiAgentState for state management if no custom state schema is provided. The state schema handles message passing and context sharing between agents.

See also

BaseGraph.add_intelligent_agent_routing: For automatic routing inference MultiAgentState: Default state schema for multi-agent coordination Agent: Base class with core agent functionality

add_branch(source_agent, condition, target_agents)¶

Add a branch condition for routing between agents.

Parameters:

source_agent (str) – The agent to branch from
condition (str) – The condition logic (e.g., ‘if error’ or ‘if success’)
target_agents (list[str]) – List of possible target agents

add_conditional_edges(source, path)¶

Add conditional edges for backward compatibility with examples.

This method provides compatibility with existing examples that use add_conditional_edges directly. It wraps the add_conditional_routing method with automatic route mapping.

Parameters:

source (str) – Source agent name to route from.
path (collections.abc.Callable[[dict[str, Any]], str]) – Function that takes state and returns target agent name.

Return type:

None

Examples

Basic routing function:

def route_by_category(state):
    category = state.get("category", "default")
    if category == "billing":
        return "billing_agent"
    elif category == "technical":
        return "technical_agent"
    else:
        return "general_agent"

multi_agent.add_conditional_edges("classifier", route_by_category)

add_conditional_routing(source_agent, condition_fn, routes)¶

Add conditional routing with a function that returns route keys.

This method enables dynamic routing based on state conditions. The condition function receives the current state and returns a key that maps to a target agent in the routes dictionary.

Parameters:

source_agent (str) – The agent to route from. Must exist in the agents dictionary.
condition_fn (collections.abc.Callable[[dict[str, Any]], str]) – Function that takes state dict and returns a route key. Should return a string that exists as a key in the routes dictionary.
routes (dict[str, str]) – Dictionary mapping route keys to target agent names. Keys are the possible return values from condition_fn. Values are agent names that must exist in the agents dictionary.

Raises:

ValueError – If source_agent doesn’t exist in agents dictionary.
KeyError – If routes contain agent names that don’t exist in agents.

Return type:

None

Examples

Basic conditional routing:

def route_by_priority(state):
    return "high" if state.get("priority", 0) > 5 else "normal"

multi_agent.add_conditional_routing(
    "classifier",
    route_by_priority,
    {"high": "urgent_processor", "normal": "standard_processor"}
)

Category-based routing:

multi_agent.add_conditional_routing(
    "categorizer",
    lambda state: state.get("category", "default"),
    {
        "billing": "billing_agent",
        "technical": "tech_support_agent",
        "default": "general_agent"
    }
)

Note

This method marks the MultiAgent for custom routing mode, bypassing the intelligent routing system in favor of explicit routing logic.

add_edge(source_agent, target_agent)¶

Add a direct edge between two agents.

This method creates a direct connection from one agent to another, ensuring the target agent runs after the source agent completes.

Parameters:

source_agent (str) – Source agent name. Must exist in the agents dictionary.
target_agent (str) – Target agent name. Must exist in the agents dictionary.

Raises:

ValueError – If source_agent doesn’t exist in agents dictionary.
ValueError – If target_agent doesn’t exist in agents dictionary.

Return type:

None

Examples

Sequential flow:

multi_agent.add_edge("preprocessor", "analyzer")
multi_agent.add_edge("analyzer", "postprocessor")

Branching flow:

multi_agent.add_edge("classifier", "processor_a")
multi_agent.add_edge("classifier", "processor_b")

Note

This method marks the MultiAgent for custom routing mode, bypassing the intelligent routing system in favor of explicit connections.

add_parallel_group(agent_names, next_agent=None)¶

Add a group of agents that run in parallel.

This method configures a set of agents to execute in parallel, with optional convergence to a single agent after parallel execution completes.

Parameters:

agent_names (list[str]) – List of agent names to run in parallel. All names must exist in the agents dictionary.
next_agent (str | None) – Optional next agent to run after the parallel group completes. If provided, must exist in the agents dictionary.

Raises:

ValueError – If any agent name in agent_names doesn’t exist in agents.
ValueError – If next_agent is provided but doesn’t exist in agents.

Return type:

None

Examples

Parallel processing with convergence:

multi_agent.add_parallel_group(
    ["data_processor", "image_processor", "text_processor"],
    next_agent="aggregator"
)

Parallel processing without convergence:

multi_agent.add_parallel_group(
    ["notification_sender", "logger", "metrics_collector"]
)

Note

This method marks the MultiAgent for custom routing mode. The parallel execution is managed by the underlying graph execution system.

build_graph()¶

Build the BaseGraph for this multi-agent.

Uses intelligent routing from BaseGraph for sequence inference and branching.

Return type:: haive.core.graph.state_graph.base_graph2.BaseGraph

classmethod create(agents, name='multi_agent', execution_mode='infer', **kwargs)¶

Create a multi-agent from a list of agents.

This factory method provides a convenient way to create a MultiAgent from a list of agents with optional configuration.

Parameters:

agents (list[haive.agents.base.agent.Agent]) – List of Agent instances to coordinate.
name (str) – Name for the multi-agent instance.
execution_mode (str) – Execution pattern - “infer”, “sequential”, “parallel”, “conditional”, or “branch”.
**kwargs – Additional keyword arguments passed to the MultiAgent constructor.

Returns:

Configured multi-agent instance.

Return type:

MultiAgent

Examples

Basic creation:

agents = [SimpleAgent(name="a"), SimpleAgent(name="b")]
multi_agent = MultiAgent.create(agents, name="my_workflow")

With custom execution mode:

multi_agent = MultiAgent.create(
    agents,
    name="parallel_workflow",
    execution_mode="parallel"
)

classmethod normalize_agents_and_name(values)¶

Normalize agents dict and auto-generate name - follows engines pattern.

Parameters:: values (dict[str, Any])
Return type:: dict[str, Any]

set_sequence(sequence)¶

Manually set the execution sequence of agents.

Parameters:: sequence (list[str]) – List of agent names in execution order

setup_agent()¶

Setup multi-agent - use MultiAgentState by default.

Return type:: None