agents.multi.experiments.implementations.clean_multi_agent¶

Clean Multi-Agent Implementation using AgentNodeV3.

from typing import Any, Dict This module provides a clean multi-agent system that: - Uses AgentNodeV3 for proper state projection - Emulates the engines dict pattern from base Agent - Supports private state passing between agents - Maintains type safety without schema flattening

Classes¶

`ConditionalAgent`	Conditional routing multi-agent.
`ContainerMultiAgentState`	Container pattern with isolated agent states.
`MinimalMultiAgentState`	Minimal state for multi-agent coordination.
`MultiAgent`	Multi-agent coordinator using AgentNodeV3 for proper state management.
`SequentialAgent`	Sequential multi-agent execution.

Module Contents¶

class agents.multi.experiments.implementations.clean_multi_agent.ConditionalAgent(agents, routing_function, **kwargs)¶

Bases: MultiAgent

Conditional routing multi-agent.

Examples

agent = ConditionalAgent(: agents={“analyzer”: analyzer, “synthesizer”: synthesizer}, routing_function=my_router

)

Initialize with agents dict and routing function.

Parameters:

agents (dict[str, haive.agents.base.agent.Agent])
routing_function (Any)

class agents.multi.experiments.implementations.clean_multi_agent.ContainerMultiAgentState(/, **data)¶

Bases: haive.core.schema.state_schema.StateSchema

Container pattern with isolated agent states.

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)

class agents.multi.experiments.implementations.clean_multi_agent.MinimalMultiAgentState¶

Bases: typing_extensions.TypedDict

Minimal state for multi-agent coordination.

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

class agents.multi.experiments.implementations.clean_multi_agent.MultiAgent¶

Bases: haive.agents.base.agent.Agent

Multi-agent coordinator using AgentNodeV3 for proper state management.

This class provides a clean implementation that: - Emulates the engines dict pattern from base Agent - Uses AgentNodeV3 for each agent with state projection - Supports both list and dict agent specifications - Maintains type safety without schema flattening

Examples

Sequential execution:

multi = MultiAgent(
    agents=[react_agent, simple_agent],
    mode="sequential"
)

With dict specification:

multi = MultiAgent(
    agents={
        "reasoner": react_agent,
        "formatter": simple_agent
    }
)

build_graph()¶

Build graph using AgentNodeV3 for each agent.

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

normalize_agents()¶

Normalize agents into registry dict, similar to engines normalization.

Return type:: MultiAgent

setup_agent()¶

Setup multi-agent specific configuration.

Return type:: None

class agents.multi.experiments.implementations.clean_multi_agent.SequentialAgent(agents, **kwargs)¶

Bases: MultiAgent

Sequential multi-agent execution.

Examples

agent = SequentialAgent([react_agent, simple_agent])

Initialize with list of agents for sequential execution.

Parameters:: agents (list[haive.agents.base.agent.Agent])