agents.supervisor.dynamic.dynamic_supervisor

DynamicSupervisor - Advanced supervisor with runtime agent management.

This module provides the DynamicSupervisor class, which extends SupervisorAgent with dynamic agent management capabilities. It allows adding and removing agents at runtime, with automatic graph rebuilding and tool discovery.

Current Status: This is the recommended dynamic supervisor implementation. It provides the most complete feature set for runtime agent coordination. For basic routing without dynamic features, use SupervisorAgent.

The DynamicSupervisor extends ReactAgent and adds sophisticated agent lifecycle management, making it ideal for adaptive workflows where agents need to be added or removed based on runtime conditions.

Key Features:

  • Runtime agent management: Add/remove agents dynamically

  • Automatic graph rebuilding: Recompiles graph when agents change

  • Dynamic tool discovery: Aggregates tools from all registered agents

  • Agent capability tracking: Maintains descriptions of agent capabilities

  • Performance monitoring: Tracks routing decisions and agent usage

  • History tracking: Maintains routing history for analysis

  • Hot-swapping: Replace agents without restarting the supervisor

Architecture:

The DynamicSupervisor maintains a registry of agents and rebuilds its execution graph whenever agents are added or removed. Tools are dynamically discovered and prefixed to avoid conflicts.

Examples

Dynamic agent management:

>>> from haive.agents.supervisor import DynamicSupervisor
>>> from haive.agents.simple import SimpleAgent
>>> from haive.core.engine.aug_llm import AugLLMConfig
>>>
>>> # Start with empty supervisor
>>> supervisor = DynamicSupervisor(
...     name="dynamic_manager",
...     engine=AugLLMConfig(temperature=0.3)
... )
>>>
>>> # Add agents at runtime
>>> analyst = SimpleAgent(name="analyst", engine=AugLLMConfig())
>>> await supervisor.add_agent("analyst", analyst)
>>>
>>> # Agent's tools are automatically available
>>> coder = SimpleAgent(name="coder", tools=[python_repl])
>>> await supervisor.add_agent("coder", coder)
>>>
>>> # Execute task - supervisor routes appropriately
>>> result = await supervisor.arun("Analyze data and write code")
>>>
>>> # Remove agent when no longer needed
>>> await supervisor.remove_agent("analyst")
>>>
>>> # List current agents
>>> agents = supervisor.list_agents()
>>> print(f"Active agents: {agents}")

With capability descriptions:

>>> # Add agents with explicit capabilities
>>> await supervisor.register_agent(
...     "data_analyst",
...     analyst_agent,
...     capability="Performs statistical analysis and data visualization"
... )
>>>
>>> await supervisor.register_agent(
...     "ml_engineer",
...     ml_agent,
...     capability="Builds and trains machine learning models"
... )

See also

  • haive.agents.supervisor.SupervisorAgent: Basic supervisor

  • haive.agents.supervisor.SimpleSupervisor: Lightweight routing

  • haive.agents.react.agent.ReactAgent: Base class

  • haive.agents.supervisor.registry: Agent registry utilities

Classes

DynamicSupervisor

Dynamic supervisor with runtime agent management.

DynamicSupervisorState

Enhanced state for dynamic supervisor.

Module Contents

class agents.supervisor.dynamic.dynamic_supervisor.DynamicSupervisor(/, **data)

Bases: haive.agents.react.agent.ReactAgent

Dynamic supervisor with runtime agent management.

Key features: - Add/remove agents at runtime - Automatic graph rebuilding - Tool aggregation from agents - Intelligent routing with history

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)

build_graph()

Build dynamic supervisor graph.

Return type:

haive.core.graph.state_graph.base_graph2.BaseGraph

classmethod create_with_agents(agents, name='dynamic_supervisor', **kwargs)

Create dynamic supervisor with initial agents.

Parameters:

  • agents (list[tuple[str, haive.agents.base.Agent, str]]) – List of (name, agent, capability) tuples

  • name (str) – Supervisor name

  • **kwargs – Additional arguments

Returns:

Configured DynamicSupervisor

Return type:

DynamicSupervisor

register_agent(name, agent, capability)

Register an agent dynamically.

Parameters:

  • name (str) – Unique agent name

  • agent (haive.agents.base.Agent) – Agent instance

  • capability (str) – Capability description

Return type:

None

setup_agent()

Setup dynamic supervisor with management tools.

Return type:

None

unregister_agent(name)

Remove an agent dynamically.

Parameters:

name (str) – Agent name to remove

Returns:

True if removed, False if not found

Return type:

bool

class agents.supervisor.dynamic.dynamic_supervisor.DynamicSupervisorState(/, **data)

Bases: pydantic.BaseModel

Enhanced state for dynamic supervisor.

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)