agents.dynamic_supervisor.agent

Dynamic Supervisor Agent implementation.

This module contains the main DynamicSupervisorAgent class that extends SimpleAgent to provide dynamic agent management capabilities.

Classes:

DynamicSupervisorAgent: Main supervisor implementation

Functions:

create_dynamic_supervisor: Factory function for creating supervisors

Example

Creating a dynamic supervisor:

from haive.agents.dynamic_supervisor import DynamicSupervisorAgent
from haive.core.engine import AugLLMConfig

supervisor = DynamicSupervisorAgent(
    name="task_router",
    engine=supervisor_engine,
    enable_agent_builder=True
)

# Run with initial agents
state = supervisor.create_initial_state()
state.add_agent("search", search_agent, "Search expert")

result = await supervisor.arun(
    "Find information about Paris and translate to French",
    state=state
)

Classes

DynamicSupervisorAgent

Dynamic supervisor agent that manages other agents at runtime.

Functions

create_dynamic_supervisor([name, model, temperature, ...])

Factory function to create a configured dynamic supervisor.

Module Contents

class agents.dynamic_supervisor.agent.DynamicSupervisorAgent(/, **data)

Bases: haive.agents.react.agent.ReactAgent

Dynamic supervisor agent that manages other agents at runtime.

Extends ReactAgent to get the looping behavior needed for continuous agent selection and execution. The supervisor can add, remove, activate, and deactivate agents while running, and generates handoff tools dynamically.

Architecture:

  • Inherits ReactAgent’s reasoning + acting loop

  • Tools execute agents directly (no separate node)

  • Uses SupervisorStateWithTools for dynamic tool generation

  • Handoff tools execute agents and return to supervisor loop

Parameters:

data (Any)

enable_agent_builder

Whether to include agent request capability

state_schema_override

Force use of supervisor state schema

auto_sync_tools

Whether to sync tools automatically

Example

Basic supervisor setup:

supervisor = DynamicSupervisorAgent(
    name="coordinator",
    engine=AugLLMConfig(
        model="gpt-4",
        force_tool_use=True,
        system_message="Route tasks to appropriate agents"
    )
)

# Add agents
state = supervisor.create_initial_state()
state.add_agent("coder", code_agent, "Python expert")

# Run task
result = await supervisor.arun("Write a Python function", state=state)

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.

add_default_agent(name, agent, description)

Add a default agent that’s always available.

Default agents are automatically added to new states.

Parameters:

  • name (str) – Agent identifier

  • agent (Any) – Agent instance

  • description (str) – Agent description

Return type:

None

async arun(input_data, state=None, **kwargs)

Run the supervisor asynchronously.

Extends SimpleAgent.arun to handle supervisor state and dynamic tool synchronization.

Parameters:

  • input_data (str | dict[str, Any] | list[langchain_core.messages.BaseMessage]) – Input message(s) or task

  • state (haive.agents.dynamic_supervisor.state.SupervisorStateWithTools | None) – Optional supervisor state with agents

  • **kwargs – Additional arguments for execution

Returns:

Execution result with updated state

Return type:

Any

Example

Running with state:

state = supervisor.create_initial_state()
state.add_agent("math", math_agent, "Math expert")

result = await supervisor.arun(
    "Calculate the square root of 144",
    state=state
)
build_graph()

Build supervisor graph with direct agent execution via tools.

Uses the base ReactAgent graph where handoff tools execute agents directly and the ReAct loop handles multi-step coordination. No separate agent_execution node needed.

Returns:

Configured supervisor graph with ReAct loop

Return type:

haive.core.graph.state_graph.base_graph2.BaseGraph

create_initial_state()

Create initial supervisor state.

Convenience method to create a properly initialized state with the supervisor’s configuration.

Returns:

Initialized supervisor state

Return type:

haive.agents.dynamic_supervisor.state.SupervisorStateWithTools

Example

Creating initial state:

state = supervisor.create_initial_state()
state.add_agent("search", agent, "Search expert")
run(input_data, state=None, **kwargs)

Run the supervisor synchronously.

Synchronous version of arun.

Parameters:

  • input_data (str | dict[str, Any] | list[langchain_core.messages.BaseMessage]) – Input message(s) or task

  • state (haive.agents.dynamic_supervisor.state.SupervisorStateWithTools | None) – Optional supervisor state with agents

  • **kwargs – Additional arguments for execution

Returns:

Execution result with updated state

Return type:

Any

setup_agent()

Setup the supervisor with custom state schema.

Overrides SimpleAgent setup to use SupervisorStateWithTools and configure the supervisor-specific settings.

Return type:

None

agents.dynamic_supervisor.agent.create_dynamic_supervisor(name='supervisor', model='gpt-4', temperature=0.0, force_tool_use=True, enable_agent_builder=False, **kwargs)

Factory function to create a configured dynamic supervisor.

Parameters:

  • name (str) – Supervisor name

  • model (str) – LLM model to use

  • temperature (float) – LLM temperature (0.0 for deterministic)

  • force_tool_use (bool) – Whether to force tool usage

  • enable_agent_builder (bool) – Enable agent request capability

  • **kwargs – Additional arguments for supervisor

Returns:

Configured DynamicSupervisorAgent instance

Return type:

DynamicSupervisorAgent

Example

Quick supervisor creation:

supervisor = create_dynamic_supervisor(
    name="task_coordinator",
    model="gpt-4",
    enable_agent_builder=True
)