agents.supervisor.state.dynamic_state

Enhanced state schema for dynamic supervisor operations.

This module provides an enhanced state management system for dynamic supervisor agents that can add/remove agents at runtime and adapt their responses based on agent configuration and execution context.

Classes

AgentExecutionConfig

Configuration for agent execution within supervisor context.

AgentExecutionResult

Result of agent execution with metadata.

DynamicSupervisorState

Enhanced state schema for dynamic supervisor operations.

SupervisorDecision

Represents a supervisor routing decision with reasoning.

Module Contents

class agents.supervisor.state.dynamic_state.AgentExecutionConfig(/, **data)

Bases: pydantic.BaseModel

Configuration for agent execution within supervisor context.

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.supervisor.state.dynamic_state.AgentExecutionResult(/, **data)

Bases: pydantic.BaseModel

Result of agent execution with metadata.

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.supervisor.state.dynamic_state.DynamicSupervisorState(/, **data)

Bases: haive.core.schema.state_schema.StateSchema

Enhanced state schema for dynamic supervisor operations.

This state schema provides comprehensive tracking of agent execution, dynamic configuration, and adaptive response handling for supervisor agents.

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)

adapt_response_for_agent(agent_name, response)

Apply adaptation rules to agent response.

Parameters:

  • agent_name (str)

  • response (Any)

Return type:

Any

add_agent_config(agent_name, config)

Add or update agent configuration.

Parameters:
Return type:

None

add_execution_result(result)

Add execution result to history.

Parameters:

result (AgentExecutionResult)

Return type:

None

add_routing_decision(decision)

Add routing decision to history.

Parameters:

decision (SupervisorDecision)

Return type:

None

cleanup_old_history(max_history=100)

Clean up old execution history to prevent memory bloat.

Parameters:

max_history (int)

Return type:

None

get_agent_config(agent_name)

Get agent configuration by name.

Parameters:

agent_name (str)

Return type:

AgentExecutionConfig | None

get_agent_performance(agent_name)

Get performance metrics for specific agent.

Parameters:

agent_name (str)

Return type:

dict[str, Any]

get_available_agents()

Get list of available agent names.

Return type:

list[str]

get_high_priority_agents()

Get agents sorted by priority (highest first).

Return type:

list[str]

get_recent_decisions(limit=5)

Get recent routing decisions.

Parameters:

limit (int)

Return type:

list[SupervisorDecision]

increment_retry_count(agent_name)

Increment retry count for agent.

Parameters:

agent_name (str)

Return type:

None

remove_agent_config(agent_name)

Remove agent configuration.

Parameters:

agent_name (str)

Return type:

bool

reset_retry_count(agent_name)

Reset retry count for agent.

Parameters:

agent_name (str)

Return type:

None

should_retry_agent(agent_name)

Determine if agent should be retried based on configuration.

Parameters:

agent_name (str)

Return type:

bool

update_agent_stats(agent_name, success, duration)

Update agent execution statistics.

Parameters:
Return type:

None

property active_agent_count: int

Number of currently registered agents.

Return type:

int

property last_execution_time: float | None

Timestamp of last agent execution.

Return type:

float | None

property most_used_agent: str | None

Name of most frequently used agent.

Return type:

str | None

property success_rate: float

Success rate of agent executions.

Return type:

float

class agents.supervisor.state.dynamic_state.SupervisorDecision(/, **data)

Bases: pydantic.BaseModel

Represents a supervisor routing decision with reasoning.

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)