agents.experiments.dynamic_supervisor

Dynamic Supervisor Agent Experiment.

from typing import Any This module contains experimental implementation of a dynamic supervisor that can select and execute agents based on runtime decisions.

Classes

AgentRegistry	Registry for managing available agents.
AgentRegistryEntry	Registry entry for an agent.
DynamicSupervisorAgent	Dynamic supervisor that selects and executes agents at runtime.
SupervisorState	State schema for dynamic supervisor agent.

Functions

create_dynamic_handoff_tool(supervisor_instance, ...)	Create a handoff tool for a specific agent.
create_forward_message_tool([supervisor_name])	Create a tool to forward agent messages.
create_list_agents_tool(supervisor_instance)	Create a tool to list available agents.
create_test_registry()	Create a test registry with some mock agents.
test_dynamic_tools()	Test dynamic tool creation and handoff functionality.
test_supervisor_basic()	Basic test of supervisor functionality.
test_supervisor_workflow()	Test a complete supervisor workflow.

Module Contents

class agents.experiments.dynamic_supervisor.AgentRegistry

Registry for managing available agents.

Init .

Returns:

Add return description]

Return type:

[TODO

instantiate_agent(name)

Instantiate an agent from the registry.

Parameters:

name (str)

Return type:

haive.agents.base.agent.Agent | None

list_agents()

List all registered agents with their metadata.

Return type:

dict[str, dict[str, Any]]

register(name, description, agent_class, config=None, capabilities=None)

Register an agent with the registry.

Parameters:

• name (str)

• description (str)

• agent_class (type[haive.agents.base.agent.Agent])

• config (dict[str, Any] | None)

• capabilities (list[str] | None)

Return type:

None

to_state_format()

Convert registry to format suitable for SupervisorState.

Return type:

dict[str, dict[str, Any]]

unregister(name)

Remove an agent from the registry.

Parameters:

name (str)

Return type:

None

class agents.experiments.dynamic_supervisor.AgentRegistryEntry(/, **data)

Bases: pydantic.BaseModel

Registry entry for an agent.

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)

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

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

Bases: haive.agents.react.agent.ReactAgent

Dynamic supervisor that selects and executes agents at runtime.

This agent inherits from ReactAgent to get the looping behavior needed for continuous agent selection and execution. It dynamically creates handoff tools for each agent in the registry.

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)

add_agent_to_registry(name, description, agent_class, config=None)

Dynamically add an agent to the registry and update tools.

Parameters:

• name (str)

• description (str)

• agent_class (type[haive.agents.base.agent.Agent])

• config (dict[str, Any] | None)

remove_agent_from_registry(name)

Dynamically remove an agent from the registry and update tools.

Parameters:

name (str)

setup_agent()

Setup supervisor with dynamic agent tools.

Return type:

None

class agents.experiments.dynamic_supervisor.SupervisorState(/, **data)

Bases: haive.core.schema.prebuilt.messages_state.MessagesState

State schema for dynamic supervisor agent.

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)

property available_agents: list[str]

Get list of available agent names from registry.

Return type:

list[str]

property is_at_max_iterations: bool

Check if we’ve reached the maximum iteration limit.

Return type:

bool

agents.experiments.dynamic_supervisor.create_dynamic_handoff_tool(supervisor_instance, agent_name)

Create a handoff tool for a specific agent.

Parameters:

agent_name (str)

agents.experiments.dynamic_supervisor.create_forward_message_tool(supervisor_name='supervisor')

Create a tool to forward agent messages.

Parameters:

supervisor_name (str)

agents.experiments.dynamic_supervisor.create_list_agents_tool(supervisor_instance)

Create a tool to list available agents.

Return type:

Any

agents.experiments.dynamic_supervisor.create_test_registry()

Create a test registry with some mock agents.

Return type:

AgentRegistry

agents.experiments.dynamic_supervisor.test_dynamic_tools()

Test dynamic tool creation and handoff functionality.

Return type:

Any

agents.experiments.dynamic_supervisor.test_supervisor_basic()

Basic test of supervisor functionality.

Return type:

Any

agents.experiments.dynamic_supervisor.test_supervisor_workflow()

Test a complete supervisor workflow.

Return type:

Any