agents.multi.enhanced_dynamic_supervisor

Enhanced DynamicSupervisor implementation extending SupervisorAgent.

DynamicSupervisor = SupervisorAgent + dynamic worker management + adaptive strategies.

Classes

DynamicSupervisor	Enhanced DynamicSupervisor with adaptive worker management.
MockWorkerTemplate	Template for creating workers.

Module Contents

class agents.multi.enhanced_dynamic_supervisor.DynamicSupervisor

Bases: haive.agents.multi.enhanced_supervisor_agent.SupervisorAgent

Enhanced DynamicSupervisor with adaptive worker management.

DynamicSupervisor extends SupervisorAgent with: 1. Dynamic worker addition/removal during execution 2. Worker performance tracking 3. Adaptive delegation strategies 4. Worker pooling and recycling

max_workers

Maximum number of concurrent workers

min_workers

Minimum number of workers to maintain

worker_performance

Track worker performance metrics

auto_scale

Enable automatic scaling based on load

recycling_enabled

Enable worker recycling

Examples

Dynamic scaling based on load:

supervisor = DynamicSupervisor(
    name="auto_scaling_manager",
    min_workers=2,
    max_workers=10,
    auto_scale=True
)

# Starts with min workers, scales up as needed
supervisor.add_worker("base_analyst", AnalystAgent())
supervisor.add_worker("base_researcher", ResearchAgent())

# During high load, automatically adds workers
result = supervisor.run("Process 100 documents")

Worker performance tracking:

supervisor = DynamicSupervisor(
    name="performance_manager",
    track_performance=True
)

# After execution, check performance
metrics = supervisor.get_worker_metrics()
# Shows success rate, average time, task count per worker

add_worker_from_template(template_name, worker_name)

Create and add a worker from template.

Parameters:

• template_name (str) – Name of the template to use

• worker_name (str) – Name for the new worker

Returns:

True if worker was added successfully

Return type:

bool

assign_task(task_id, worker_name=None)

Assign a task to a worker.

Parameters:

• task_id (str) – Unique task identifier

• worker_name (str | None) – Specific worker to assign to (optional)

Returns:

Name of assigned worker or None

Return type:

str | None

can_add_worker()

Check if more workers can be added.

Return type:

bool

complete_task(task_id, success=True, duration=0.0)

Mark a task as completed.

Parameters:

• task_id (str) – Task identifier

• success (bool) – Whether task succeeded

• duration (float) – Task duration in seconds

Return type:

None

get_best_worker()

Get the best performing idle worker.

Return type:

str | None

get_worker_metrics()

Get performance metrics for all workers.

Return type:

dict[str, dict[str, Any]]

remove_idle_worker()

Remove an idle worker.

Returns:

Name of removed worker or None

Return type:

str | None

should_scale_down()

Determine if should scale down workers.

Return type:

bool

should_scale_up()

Determine if should scale up workers.

Return type:

bool

classmethod validate_min_workers(v, info)

Ensure min_workers <= max_workers.

Parameters:

v (int)

Return type:

int

class agents.multi.enhanced_dynamic_supervisor.MockWorkerTemplate

Template for creating workers.

Init .