agents.supervisor.utils.registry

Agent Registry for Haive Supervisor System.

Manages agent lifecycle and routing model synchronization using DynamicChoiceModel. Provides runtime agent registration/deregistration with automatic routing updates.

Classes

AgentRegistry

Manages agent lifecycle and routing model synchronization.

Module Contents

class agents.supervisor.utils.registry.AgentRegistry(routing_model)

Manages agent lifecycle and routing model synchronization.

This class provides a centralized registry for managing agents in a supervisor system, with automatic synchronization to a DynamicChoiceModel for routing.

Features:

• Runtime agent registration/deregistration

• Automatic routing model updates

• Agent capability tracking

• Conflict detection and resolution

• Rich visualization of registry state

Initialize agent registry.

Parameters:

routing_model (haive.core.common.models.dynamic_choice_model.DynamicChoiceModel[str]) – DynamicChoiceModel to synchronize with agent changes

clear_all()

Remove all registered agents.

Warning: This will clear the entire registry!

Return type:

None

get_agent(name)

Get agent by name.

Parameters:

name (str) – Agent name

Returns:

Agent instance or None if not found

Return type:

haive.agents.base.agent.Agent | None

get_agent_capabilities()

Get mapping of agent names to their capabilities.

Returns:

Dict mapping agent names to capability descriptions

Return type:

dict[str, str]

get_agent_capability(agent_name)

Get capability description for specific agent.

Parameters:

agent_name (str) – Name of agent

Returns:

Capability description or None if agent not found

Return type:

str | None

get_agent_count()

Get number of registered agents.

Returns:

Number of currently registered agents

Return type:

int

get_available_agents()

Get list of currently available agent names.

Returns:

List of agent names available for routing

Return type:

list[str]

get_registry_stats()

Get registry statistics.

Returns:

Dictionary with registry statistics

Return type:

dict[str, Any]

get_routing_options()

Get all routing options including END.

Returns:

List of all valid routing choices

Return type:

list[str]

is_agent_registered(agent_name)

Check if agent is registered.

Parameters:

agent_name (str) – Agent name to check

Returns:

True if agent is registered, False otherwise

Return type:

bool

mark_rebuilt()

Mark that supervisor graph has been rebuilt.

Return type:

None

needs_rebuild()

Check if supervisor graph needs rebuilding.

Returns:

True if agents have been added/removed since last check

Return type:

bool

print_registry_state()

Print comprehensive registry state information.

Return type:

None

register(agent, capability_description=None)

Register an agent and update routing model.

Parameters:

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

• capability_description (str | None) – Optional description of agent capabilities

Returns:

True if registration successful, False if agent name conflict

Return type:

bool

Raises:

ValueError – If agent has no name or invalid configuration

unregister(agent_name)

Remove an agent and update routing model.

Parameters:

agent_name (str) – Name of agent to remove

Returns:

True if removal successful, False if agent not found

Return type:

bool

validate_routing_choice(choice)

Validate if a routing choice is valid.

Parameters:

choice (str) – Routing choice to validate

Returns:

True if choice is valid, False otherwise

Return type:

bool