haive.core.registry.dynamic_registry¶

Dynamic Registry System for Component Activation.

This module provides a generic registry system for managing activatable components (tools, agents, services) with full type safety and activation tracking.

Based on the Dynamic Activation Pattern: @project_docs/active/patterns/dynamic_activation_pattern.md

Classes¶

`DynamicRegistry`	Generic registry for managing activatable components.
`RegistryItem`	Base class for registry items with activation state.

Module Contents¶

class haive.core.registry.dynamic_registry.DynamicRegistry(/, **data)[source]¶

Bases: pydantic.BaseModel, Generic[T]

Generic registry for managing activatable components.

This registry provides a type-safe way to manage components that can be dynamically activated and deactivated. It supports limits on active components and tracks activation history.

Parameters:

items – Dictionary of component ID to RegistryItem
active_items – Set of currently active component IDs
max_active – Maximum number of components that can be active simultaneously
activation_history – List of activation/deactivation events
data (Any)

Examples

Create a tool registry:

from langchain_core.tools import Tool

registry = DynamicRegistry[Tool]()

# Register a tool
item = RegistryItem(
    id="calc",
    name="calculator",
    description="Math operations",
    component=calculator_tool
)
registry.register(item)

# Activate the tool
success = registry.activate("calc")
assert success is True

# Get active tools
active_tools = registry.get_active_components()

Create an agent registry with limits:

from haive.agents.base import Agent

registry = DynamicRegistry[Agent](max_active=3)

# Register multiple agents
for i, agent in enumerate(agents):
    item = RegistryItem(
        id=f"agent_{i}",
        name=f"Agent {i}",
        description=f"Agent {i} description",
        component=agent
    )
    registry.register(item)

# Activate agents up to limit
for i in range(5):  # Only first 3 will activate
    success = registry.activate(f"agent_{i}")
    print(f"Agent {i} activated: {success}")

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.

activate(item_id)[source]¶

Activate a component by ID.

Parameters:: item_id (str) – ID of component to activate
Returns:: True if activation succeeded, False otherwise
Return type:: bool

Examples

Activate a component:

success = registry.activate("my_tool")
if success:
    print("Tool activated successfully")
else:
    print("Failed to activate tool")

Check activation status:

registry.activate("tool_1")
registry.activate("tool_2")

active_count = len(registry.active_items)
print(f"Active components: {active_count}")

clear_inactive()[source]¶

Remove all inactive components from registry.

Returns:: Number of components removed
Return type:: int

Examples

Clean up registry:

removed_count = registry.clear_inactive()
print(f"Removed {removed_count} inactive components")

Selective cleanup:

# Only remove old inactive components
cutoff_time = datetime.now() - timedelta(hours=24)
removed = 0
for item_id, item in list(registry.items.items()):
    if (not item.is_active and
        item.last_activated and
        item.last_activated < cutoff_time):
        del registry.items[item_id]
        removed += 1
print(f"Removed {removed} old components")

deactivate(item_id)[source]¶

Deactivate a component by ID.

Parameters:: item_id (str) – ID of component to deactivate
Returns:: True if deactivation succeeded, False otherwise
Return type:: bool

Examples

Deactivate a component:

success = registry.deactivate("my_tool")
if success:
    print("Tool deactivated successfully")

Deactivate all components:

active_ids = list(registry.active_items)
for item_id in active_ids:
    registry.deactivate(item_id)

get_active_components()[source]¶

Get all active component instances.

Returns:: List of active component instances
Return type:: list[T]

Examples

Get active tools:

active_tools = registry.get_active_components()
for tool in active_tools:
    print(f"Active tool: {tool.name}")

Use active components:

active_agents = registry.get_active_components()
for agent in active_agents:
    result = await agent.arun("Hello")
    print(f"Agent {agent.name} response: {result}")

get_active_items()[source]¶

Get all active registry items.

Returns:: List of active RegistryItem instances
Return type:: list[RegistryItem[T]]

Examples

Get active items with metadata:

active_items = registry.get_active_items()
for item in active_items:
    print(f"Active: {item.name}")
    print(f"Activated: {item.activation_count} times")
    print(f"Last used: {item.last_activated}")
    print(f"Metadata: {item.metadata}")

get_component(item_id)[source]¶

Get a component instance by ID.

Parameters:: item_id (str) – ID of component to retrieve
Returns:: Component instance if found, None otherwise
Return type:: T | None

Examples

Get and use component:

tool = registry.get_component("calculator")
if tool:
    result = tool.invoke({"expression": "2 + 2"})
    print(f"Result: {result}")

get_inactive_items()[source]¶

Get all inactive registry items.

Returns:: List of inactive RegistryItem instances
Return type:: list[RegistryItem[T]]

Examples

Show available but inactive components:

inactive_items = registry.get_inactive_items()
print("Available components:")
for item in inactive_items:
    print(f"- {item.name}: {item.description}")

get_item(item_id)[source]¶

Get a registry item by ID.

Parameters:: item_id (str) – ID of item to retrieve
Returns:: RegistryItem if found, None otherwise
Return type:: RegistryItem[T] | None

Examples

Get item information:

item = registry.get_item("my_tool")
if item:
    print(f"Tool: {item.name}")
    print(f"Description: {item.description}")
    print(f"Active: {item.is_active}")
    print(f"Used: {item.activation_count} times")

get_stats()[source]¶

Get registry statistics.

Returns:: Dictionary with registry statistics
Return type:: dict[str, Any]

Examples

Show registry status:

stats = registry.get_stats()
print(f"Total components: {stats['total_components']}")
print(f"Active components: {stats['active_components']}")
print(f"Utilization: {stats['utilization_rate']:.1%}")
print(f"Most used: {stats['most_used_component']}")

is_active(item_id)[source]¶

Check if a component is currently active.

Parameters:: item_id (str) – ID of component to check
Returns:: True if component is active, False otherwise
Return type:: bool

Examples

Check activation status:

if registry.is_active("my_tool"):
    print("Tool is ready to use")
else:
    print("Tool needs to be activated first")

list_components()[source]¶

List all registered component IDs.

Returns:: List of all component IDs in the registry
Return type:: list[str]

Examples

List all components:

component_ids = registry.list_components()
for comp_id in component_ids:
    item = registry.get_item(comp_id)
    status = "ACTIVE" if item.is_active else "INACTIVE"
    print(f"{comp_id}: {item.name} [{status}]")

register(item)[source]¶

Parameters:: item (RegistryItem[T]) – RegistryItem to register
Raises:: ValueError – If item ID already exists in registry
Return type:: None

Examples

item = RegistryItem(
    id="my_tool",
    name="My Tool",
    description="Does useful things",
    component=tool_instance
)
registry.register(item)

item = RegistryItem(
    id="special_tool",
    name="Special Tool",
    description="Special functionality",
    component=tool_instance,
    metadata={
        "category": "utility",
        "version": "1.0.0",
        "author": "Developer"
    }
)
registry.register(item)

validate_active_items()[source]¶

Validate that active_items are consistent with items.

Return type:: Self

classmethod validate_max_active(v)[source]¶

Validate max_active is positive if provided.

Parameters:: v (int | None)
Return type:: int | None

class haive.core.registry.dynamic_registry.RegistryItem(/, **data)[source]¶

Bases: pydantic.BaseModel, Generic[T]

Base class for registry items with activation state.

This class wraps any component type with activation tracking, metadata, and usage statistics.

Parameters:

id – Unique identifier for the component
name – Human-readable name for the component
description – Brief description of component capabilities
component – The actual component instance (tool, agent, etc.)
is_active – Whether the component is currently active
metadata – Additional metadata for the component
activation_count – Number of times component has been activated
last_activated – Timestamp of last activation
data (Any)

Examples

Create a tool registry item:

from langchain_core.tools import tool

@tool
def calculator(expression: str) -> float:
    '''Calculate mathematical expression.'''
    return eval(expression)

item = RegistryItem(
    id="calc_001",
    name="calculator",
    description="Basic math operations",
    component=calculator,
    metadata={"category": "math"}
)

Create an agent registry item:

from haive.agents.simple import SimpleAgent

agent = SimpleAgent(name="helper")
item = RegistryItem(
    id="agent_001",
    name="helper_agent",
    description="General purpose assistant",
    component=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.

classmethod validate_id(v)[source]¶

Validate ID format.

Parameters:: v (str)
Return type:: str

classmethod validate_name(v)[source]¶

Validate name format.

Parameters:: v (str)
Return type:: str