agents.multi.enhanced_clean_multi_agent¶

Enhanced Clean Multi-Agent Implementation using Agent[AugLLMConfig].

MultiAgent = Agent[AugLLMConfig] + agent coordination + state management.

This combines the enhanced agent pattern with the clean multi-agent approach: - Uses Agent[AugLLMConfig] as the base - Supports AgentNodeConfig for proper agent execution - Maintains the engines dict pattern - Provides multiple state management strategies

Classes¶

`ContainerMultiAgentState`	Container pattern with isolated agent states and MetaStateSchema support.
`EnhancedMultiAgent`	Enhanced Multi-Agent coordinator with flexible state management.
`MinimalMultiAgentState`	Minimal state for multi-agent coordination.

Module Contents¶

class agents.multi.enhanced_clean_multi_agent.ContainerMultiAgentState(/, **data)¶

Bases: haive.core.schema.state_schema.StateSchema

Container pattern with isolated agent states and MetaStateSchema support.

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.enhanced_clean_multi_agent.EnhancedMultiAgent¶

Bases: haive.agents.simple.enhanced_simple_real.EnhancedAgentBase

Enhanced Multi-Agent coordinator with flexible state management.

MultiAgent = Agent[AugLLMConfig] + agent coordination + state projection.

Key features: 1. Uses AgentNodeConfig for proper agent execution in graphs 2. Supports multiple execution modes (sequential, parallel, conditional) 3. Flexible state management strategies 4. Compatible with MetaStateSchema for meta-capabilities

agents¶: List or dict of agents to coordinate

mode¶: Execution mode (sequential, parallel, conditional)

state_strategy¶: State management approach

shared_fields¶: Fields shared between agents

state_transfer_map¶: Rules for transferring state between agents

Examples

Sequential with state transfer:

multi = EnhancedMultiAgent(
    name="pipeline",
    agents=[planner, executor, reviewer],
    mode="sequential",
    state_transfer_map={
        ("planner", "executor"): {"plan": "task_plan"},
        ("executor", "reviewer"): {"result": "execution_result"}
    }
)

Parallel with aggregation:

multi = EnhancedMultiAgent(
    name="ensemble",
    agents={"expert1": agent1, "expert2": agent2},
    mode="parallel",
    state_strategy="container"
)

With MetaStateSchema:

from haive.core.schema.prebuilt.meta_state import MetaStateSchema

meta_multi = MetaStateSchema.from_agent(
    agent=multi,
    initial_state={"shared_context": {"project": "AI"}}
)

build_graph()¶

Build multi-agent execution graph.

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

get_agent(name)¶

Get agent by name.

Parameters:: name (str)
Return type:: haive.agents.simple.enhanced_simple_real.EnhancedAgentBase | None

get_agent_names()¶

Get list of agent names.

Return type:: list[str]

setup_agent()¶

Setup multi-agent coordinator.

Return type:: None

setup_state_schema()¶

Setup appropriate state schema based on strategy.

Return type:: EnhancedMultiAgent

classmethod validate_agents(v)¶

Validate and normalize agents.

Parameters:: v (list[haive.agents.simple.enhanced_simple_real.EnhancedAgentBase] | dict[str, haive.agents.simple.enhanced_simple_real.EnhancedAgentBase])
Return type:: list[haive.agents.simple.enhanced_simple_real.EnhancedAgentBase] | dict[str, haive.agents.simple.enhanced_simple_real.EnhancedAgentBase]

class agents.multi.enhanced_clean_multi_agent.MinimalMultiAgentState¶

Bases: typing_extensions.TypedDict

Minimal state for multi-agent coordination.

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