Common Utilities¶
The Common Utilities module is the Swiss Army knife of the Haive framework - a comprehensive collection of battle-tested components providing reusable mixins, data structures, type utilities, validation helpers, and performance optimizations that power every aspect of the AI infrastructure.
🧰 Beyond Basic Utilities¶
Transform Your Code from Repetitive to Reusable:
- Intelligent Mixins
Powerful mixin classes that add capabilities like timestamps, IDs, state management, and recompilation to any component
- Advanced Data Structures
Specialized trees, graphs, ordered collections, and nested dictionaries optimized for AI workflows
- Type System Enhancements
Runtime type checking, generic helpers, and advanced type inference for bulletproof code
- Validation Framework
Comprehensive validation utilities for data integrity, schema compliance, and error handling
- Performance Helpers
Caching, memoization, lazy loading, and optimization utilities for blazing-fast execution
Core Utility Components¶
Mixin Architecture¶
🧩 Mixins - Intelligent Component Superpowers System
THE MOLECULAR BUILDING BLOCKS OF AI EXCELLENCE
Welcome to Mixins - the revolutionary collection of intelligent, composable behaviors that transform ordinary classes into extraordinary AI components. This isn’t just multiple inheritance; it’s a sophisticated composition system where every mixin is a specialized capability that makes your components smarter, more reliable, and enterprise-ready by default.
⚡ REVOLUTIONARY MIXIN INTELLIGENCE¶
Mixins represent a paradigm shift from manual feature implementation to intelligent capability composition where sophisticated behaviors are injected seamlessly into any class:
🧠 Self-Configuring Behaviors: Mixins that automatically adapt to their host class 🔄 Zero-Conflict Composition: Intelligent inheritance resolution and method chaining ⚡ Performance Optimization: Built-in caching, lazy loading, and resource management 📊 Enterprise-Grade Observability: Automatic logging, metrics, and monitoring 🎯 Type-Safe Integration: Full Pydantic compatibility with intelligent field merging
🌟 CORE MIXIN CATEGORIES¶
- 1. Identity & Lifecycle Mixins 🆔
Fundamental behaviors for object identity and lifecycle management:
Examples
>>> from haive.core.common.mixins import (
>>> IdentifierMixin, TimestampMixin, VersionMixin, MetadataMixin
>>> )
>>>
>>> class IntelligentAgent(
>>> IdentifierMixin, # Unique IDs with collision detection
>>> TimestampMixin, # Created/updated/accessed tracking
>>> VersionMixin, # Semantic versioning with migrations
>>> MetadataMixin # Rich metadata with indexing
>>> ):
>>> def __init__(self, name: str):
>>> super().__init__()
>>> self.name = name
>>> # Automatic capabilities:
>>> # - self.id: Unique identifier (UUID with prefix)
>>> # - self.created_at: ISO timestamp of creation
>>> # - self.version: Semantic version ("1.0.0")
>>> # - self.metadata: Indexed metadata storage
>>>
>>> # Enhanced instantiation
>>> agent = IntelligentAgent("research_assistant")
>>> assert agent.id.startswith("agent_") # Automatic prefixing
>>> assert agent.created_at <= datetime.now() # Timestamp validation
>>> assert agent.version == "1.0.0" # Default version
- 2. State Management Mixins 🗄️
Advanced state handling with intelligent persistence:
>>> from haive.core.common.mixins import ( >>> StateMixin, StateInterfaceMixin, CheckpointerMixin >>> ) >>> >>> class StatefulProcessor( >>> StateMixin, # Core state management >>> StateInterfaceMixin, # Advanced state operations >>> CheckpointerMixin # Automatic checkpointing >>> ): >>> def __init__(self): >>> super().__init__() >>> # Automatic capabilities: >>> # - State validation and serialization >>> # - Automatic dirty tracking >>> # - Checkpoint creation and restoration >>> # - State migration support >>> >>> def process(self, data): >>> # State automatically tracked >>> self.state.update({"last_processed": data}) >>> >>> # Automatic checkpoint creation >>> if self.should_checkpoint(): >>> self.create_checkpoint("pre_processing") >>> >>> result = complex_processing(data) >>> >>> # State automatically persisted >>> self.state.finalize_update() >>> return result
For complete examples and advanced patterns, see the documentation.
- class haive.core.common.mixins.CheckpointerMixin[source]¶
Mixin that provides checkpointing capabilities for stateful graph execution.
This mixin adds methods for running stateful graph executions with checkpointing support, including automatic state restoration, thread management, and proper configuration handling for both synchronous and asynchronous execution patterns.
The mixin expects the host class to provide: - persistence: Optional[CheckpointerConfig] - Configuration for the checkpointer - checkpoint_mode: str - Mode of checkpointing (“sync”, “async”, or “none”) - runnable_config: RunnableConfig - Base configuration for runnables - input_schema, state_schema (optional) - Schemas for input and state validation - app or compile() method - The LangGraph compiled application
- Parameters:
data (Any)
- None publicly, but requires the above attributes from the host class.
- async arun(input_data, thread_id=None, config=None, **kwargs)[source]¶
Async run with checkpointer support.
This method runs a graph execution asynchronously with checkpointing support, automatically handling state restoration and persistence.
- Parameters:
- Returns:
The result of the graph execution.
- Return type:
- async astream(input_data, thread_id=None, stream_mode='values', config=None, **kwargs)[source]¶
Async stream with checkpointer support.
This method streams graph execution results asynchronously with checkpointing support, automatically handling state restoration and persistence.
- Parameters:
input_data (Any) – The input data for the execution.
thread_id (str | None) – Optional thread ID for state tracking.
stream_mode (str) – The streaming mode to use (values, actions, etc.).
config (RunnableConfig | None) – Optional configuration override.
**kwargs – Additional configuration parameters.
- Yields:
Execution chunks as they become available.
- Return type:
AsyncGenerator[dict[str, Any], None]
- model_post_init(context, /)¶
This function is meant to behave like a BaseModel method to initialise private attributes.
It takes context as an argument since that’s what pydantic-core passes when calling it.
- Parameters:
self (BaseModel) – The BaseModel instance.
context (Any) – The context.
- Return type:
None
- run(input_data, thread_id=None, config=None, **kwargs)[source]¶
Run with checkpointer support.
This method runs a graph execution with checkpointing support, automatically handling state restoration and persistence.
- Parameters:
- Returns:
The result of the graph execution.
- Return type:
- stream(input_data, thread_id=None, stream_mode='values', config=None, **kwargs)[source]¶
Stream with checkpointer support.
This method streams graph execution results with checkpointing support, automatically handling state restoration and persistence.
- Parameters:
input_data (Any) – The input data for the execution.
thread_id (str | None) – Optional thread ID for state tracking.
stream_mode (str) – The streaming mode to use (values, actions, etc.).
config (RunnableConfig | None) – Optional configuration override.
**kwargs – Additional configuration parameters.
- Returns:
A generator yielding execution chunks.
- Return type:
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- haive.core.common.mixins.EngineMixin¶
alias of
EngineStateMixin- Parameters:
data (Any)
- class haive.core.common.mixins.GetterMixin[source]¶
A mixin providing rich lookup and filtering capabilities for collections.
This mixin can be added to any collection class that implements _get_items() to provide powerful querying capabilities. It works with both dictionary-like objects and objects with attributes.
The mixin is generic over type T, which represents the type of items in the collection. This enables proper type hinting when using the mixin’s methods.
- None directly, but requires subclasses to implement _get_items()
- field_values(field_name)[source]¶
Get all values for a specific field across items.
This method collects the values of a specific field or attribute from all items in the collection.
- Parameters:
field_name (str) – Field name to collect.
- Returns:
List of field values (None for items where field doesn’t exist).
- Return type:
Example
# Get all user IDs user_ids = users.field_values("id")
- filter(**kwargs)[source]¶
Filter items by multiple attribute criteria.
This method finds all items that match all of the specified attribute criteria (logical AND of all criteria).
- Parameters:
**kwargs – Field name and value pairs to match.
- Returns:
List of matching items (empty list if none found).
- Return type:
list[T]
Example
# Find all admin users with status='active' active_admins = collection.filter(role="admin", status="active")
- find(predicate)[source]¶
Find first item matching a custom predicate function.
This method finds the first item for which the predicate function returns True.
- Parameters:
predicate (Callable[[T], bool]) – Function that takes an item and returns a boolean.
- Returns:
First matching item or None if none found.
- Return type:
T | None
Example
# Find first user with name longer than 10 characters user = users.find(lambda u: len(u.name) > 10)
- find_all(predicate)[source]¶
Find all items matching a custom predicate function.
This method finds all items for which the predicate function returns True.
- Parameters:
predicate (Callable[[T], bool]) – Function that takes an item and returns a boolean.
- Returns:
List of matching items (empty list if none found).
- Return type:
list[T]
Example
# Find all premium users with subscription expiring in 7 days from datetime import datetime, timedelta next_week = datetime.now() + timedelta(days=7) expiring = users.find_all( lambda u: u.is_premium and u.expires_at.date() == next_week.date() )
- first(**kwargs)[source]¶
Get first item matching criteria.
This is a convenience method that combines filter() with returning the first result only.
- Parameters:
**kwargs – Field name and value pairs to match.
- Returns:
First matching item or None if none found.
- Return type:
T | None
Example
# Find first active admin user admin = users.first(role="admin", status="active")
- get_all_by_attr(attr_name, value)[source]¶
Get all items where attribute equals value.
This method finds all items in the collection where the specified attribute matches the given value.
- get_by_attr(attr_name, value, default=None)[source]¶
Get first item where attribute equals value.
This method finds the first item in the collection where the specified attribute matches the given value.
- get_by_type(type_cls)[source]¶
Get all items of specified type.
This method finds all items that are instances of the specified type.
- Parameters:
type_cls (type) – Type to match.
- Returns:
List of matching items (empty list if none found).
- Return type:
list[T]
Example
# Get all TextMessage instances text_messages = messages.get_by_type(TextMessage)
- class haive.core.common.mixins.IdMixin(*, id=<factory>)[source]¶
Mixin for adding basic ID generation and management capabilities.
This mixin adds a UUID-based ID field to any Pydantic model, with methods for regenerating the ID and creating instances with specific IDs.
- Parameters:
data (Any)
id (str)
- classmethod with_id(id_value, **kwargs)[source]¶
Create an instance with a specific ID.
This class method provides a convenient way to create an instance with a predetermined ID value.
- Parameters:
id_value (str) – The ID value to use.
**kwargs – Additional attributes for the instance.
- Returns:
A new instance with the specified ID.
- regenerate_id()[source]¶
Generate a new UUID and return it.
This method replaces the current ID with a new UUID.
- Returns:
The newly generated UUID string.
- Return type:
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.IdentifierMixin(*, id=<factory>, name=None)[source]¶
Mixin that adds unique identification to any Pydantic model.
This mixin provides both UUID-based identification and human-readable naming capabilities. It automatically generates UUIDs, validates provided IDs, and offers convenience methods for working with the identifiers.
- initialize_uuid_obj()[source]¶
Initialize UUID object after model validation.
- Returns:
Self, with the _uuid_obj private attribute initialized.
- Return type:
- matches_id(id_or_name)[source]¶
Check if this object matches the given ID or name.
This method checks if the provided string matches this object’s full ID, short ID, or name (case-insensitive).
- model_post_init(context, /)¶
This function is meant to behave like a BaseModel method to initialise private attributes.
It takes context as an argument since that’s what pydantic-core passes when calling it.
- Parameters:
self (BaseModel) – The BaseModel instance.
context (Any) – The context.
- Return type:
None
- regenerate_id()[source]¶
Generate a new ID and return it.
This method creates a new UUID, updates the ID field, and returns the new ID string.
- Returns:
The newly generated UUID string.
- Return type:
- set_name(name)[source]¶
Set the name with validation.
- Parameters:
name (str) – The new name to set.
- Return type:
None
- property display_name: str¶
Display name (uses name if available, otherwise short_id).
- Returns:
The human-readable name if set, otherwise “Object-{short_id}”.
- Return type:
- property has_custom_name: bool¶
Whether this object has a custom name (not auto-generated).
- Returns:
True if a non-empty name is set, False otherwise.
- Return type:
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.MCPMixin(**data)[source]¶
Mixin for adding MCP (Model Context Protocol) support to configurations.
This mixin provides seamless integration with MCP servers, enabling: - Automatic discovery and wrapping of MCP tools - Resource loading and caching from MCP servers - Prompt template management - Enhanced system prompts with MCP information
The mixin is designed to work with ToolRouteMixin for proper tool routing and can be combined with other mixins in the configuration hierarchy.
- Parameters:
data (Any)
- mcp_config¶
Optional MCP configuration for server connections
- Type:
MCPConfig | None
- mcp_resources¶
List of discovered MCP resources
- mcp_prompts¶
Dictionary of MCP prompt templates
- async call_mcp_prompt(prompt_name, arguments=None)[source]¶
Call an MCP prompt to get formatted messages.
- Parameters:
- Returns:
List of message dictionaries with role and content
- Raises:
ValueError – If prompt not found or MCP not initialized
- Return type:
- cleanup_mcp()[source]¶
Clean up MCP resources.
This should be called when the configuration is no longer needed to properly close MCP connections.
- Return type:
None
- enhance_system_prompt_with_mcp(base_prompt='')[source]¶
Enhance a system prompt with MCP information.
Adds information about available MCP resources and operations to help the LLM understand what capabilities are available.
- get_mcp_prompts()[source]¶
Get all loaded MCP prompt templates.
- Returns:
Dictionary of prompt templates by name
- Return type:
- async get_mcp_resource_content(uri)[source]¶
Fetch content for an MCP resource.
- Parameters:
uri (str) – Resource URI
- Returns:
Resource content
- Raises:
ValueError – If MCP manager not initialized or resource not found
- Return type:
- get_mcp_resources()[source]¶
Get all loaded MCP resources.
- Returns:
List of MCP resources
- Return type:
- get_mcp_tools()[source]¶
Get all discovered MCP tools.
- Returns:
List of MCP tool wrappers
- Return type:
- model_post_init(context, /)¶
This function is meant to behave like a BaseModel method to initialise private attributes.
It takes context as an argument since that’s what pydantic-core passes when calling it.
- Parameters:
self (BaseModel) – The BaseModel instance.
context (Any) – The context.
- Return type:
None
- async setup_mcp()[source]¶
Initialize MCP integration.
Sets up the MCP manager, discovers tools, loads resources, and configures prompts based on the MCP configuration.
This method should be called after creating the configuration but before using any MCP features.
- Return type:
None
- mcp_prompts: dict[str, MCPPromptTemplate]¶
- mcp_resources: list[MCPResource]¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.MetadataMixin(*, metadata=<factory>)[source]¶
Mixin for adding flexible metadata storage capabilities.
This mixin provides a dictionary for storing arbitrary key-value pairs as metadata, along with methods for adding, retrieving, updating, and removing metadata entries.
- clear_metadata()[source]¶
Clear all metadata.
This method removes all metadata entries, resulting in an empty metadata dictionary.
- Return type:
None
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.RichLoggerMixin(*, debug=False)[source]¶
Mixin that provides rich console logging capabilities.
This mixin adds a configurable logger with rich formatting to any Pydantic model. It creates a logger named after the class, configures it with Rich’s handler for pretty console output, and provides convenience methods for different log levels with appropriate styling.
- Parameters:
data (Any)
debug (bool)
- model_post_init(context, /)¶
This function is meant to behave like a BaseModel method to initialise private attributes.
It takes context as an argument since that’s what pydantic-core passes when calling it.
- Parameters:
self (BaseModel) – The BaseModel instance.
context (Any) – The context.
- Return type:
None
- property logger: Logger¶
Get or create logger with rich handler.
This property lazily initializes a logger with a Rich handler, creating it only when first accessed. The logger is named using the module and class name for proper log categorization.
- Returns:
Configured logging.Logger instance with Rich formatting.
- Return type:
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.SecureConfigMixin[source]¶
A mixin to provide secure and flexible configuration for API keys.
This mixin enables: 1. Dynamic API key resolution from multiple sources 2. Secure storage using SecretStr 3. Environment variable fallbacks based on provider type 4. Validation and error reporting
The mixin implements a field validator for the ‘api_key’ field that attempts to resolve the key from environment variables if not explicitly provided, based on the ‘provider’ field. It also provides a safe method to retrieve the key value with appropriate error handling.
- api_key¶
A SecretStr containing the API key.
- provider¶
The API provider name (used to determine environment variable).
- get_api_key()[source]¶
Safely retrieve the API key with improved error handling.
This method attempts to retrieve the API key value from the SecretStr field, with comprehensive error handling and helpful log messages for troubleshooting. In development environments, it can return fake test keys for testing purposes.
- Returns:
The API key as a string, or None if not available or invalid.
- Return type:
str | None
- class haive.core.common.mixins.SerializationMixin[source]¶
Mixin for enhanced serialization and deserialization capabilities.
This mixin provides methods for converting Pydantic models to dictionaries and JSON strings, and for creating models from dictionaries and JSON strings. It handles private fields (starting with underscore) appropriately.
When combined with other mixins like IdMixin, TimestampMixin, etc., it provides a complete solution for model persistence.
- Parameters:
data (Any)
- classmethod from_dict(data)[source]¶
Create instance from dictionary.
This class method creates a model instance from a dictionary, using Pydantic’s validation.
- classmethod from_json(json_str)[source]¶
Create instance from JSON string.
This class method creates a model instance from a JSON string, parsing the JSON and then using from_dict().
- Parameters:
json_str (str) – JSON string containing model data.
- Returns:
New model instance.
- to_dict(exclude_private=True)[source]¶
Convert to dictionary with options.
This method converts the model to a dictionary, with the option to exclude private fields (those starting with an underscore).
- to_json(exclude_private=True, **kwargs)[source]¶
Convert to JSON string.
This method converts the model to a JSON string, with options for controlling the JSON serialization.
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.StateInterfaceMixin(*, use_state=False, state_key='state')[source]¶
Mixin that adds state management configuration to any Pydantic model.
This mixin allows components to declare whether they use a state store and which key they use to access their portion of the state. This is commonly used in stateful nodes within a processing graph.
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.StateMixin(*, state='active', state_history=<factory>)[source]¶
Mixin for state tracking with validation and comprehensive history.
This mixin adds state management capabilities to Pydantic models, allowing objects to track their current state and maintain a complete history of state transitions with timestamps and optional reasons.
The mixin is designed to be composable with other BaseModel classes and provides thread-safe state transitions with automatic history tracking.
Examples
>>> class Task(StateMixin, BaseModel): ... name: str >>> task = Task(name="Process data") >>> task.change_state("running", "Starting execution") >>> task.change_state("complete", "Finished successfully") >>> task.is_in_state("complete") True >>> len(task.get_state_changes()) 2
- change_state(new_state, reason=None)[source]¶
Change state and automatically track the transition in history.
This method updates the current state and records the transition in the state history with a timestamp and optional reason.
- Parameters:
- Return type:
None
Examples
>>> task.change_state("paused", "Waiting for user input") >>> task.state 'paused'
- get_state_changes()[source]¶
Get a copy of the complete state change history.
- Returns:
from_state: Previous state
to_state: New state
timestamp: When the change occurred
reason: Optional explanation for the change
- Return type:
List of state change records, each containing
Examples
>>> changes = task.get_state_changes() >>> print(changes[0]["from_state"]) 'active'
- is_in_state(state)[source]¶
Check if the object is currently in the specified state.
- Parameters:
state (str) – State name to check against current state.
- Returns:
True if current state matches the specified state, False otherwise.
- Return type:
Examples
>>> task.is_in_state("complete") False >>> task.change_state("complete") >>> task.is_in_state("complete") True
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.StructuredOutputMixin[source]¶
Mixin to provide structured output functionality for LLM configurations.
This mixin adds support for: - Configuring structured output models with v1 (parser) or v2 (tool) approaches - Automatic format instruction generation - Tool-based structured output forcing
- class haive.core.common.mixins.TimestampMixin(*, created_at=<factory>, updated_at=<factory>)[source]¶
Mixin for adding timestamp tracking to Pydantic models.
This mixin adds creation and update timestamps to any model, with methods for updating timestamps and calculating time intervals. It’s useful for tracking when objects were created and modified, which helps with auditing, caching strategies, and expiration logic.
- created_at¶
When this object was created (auto-set on instantiation).
- Type:
datetime
- updated_at¶
When this object was last updated (initially same as created_at).
- Type:
datetime
- age_in_seconds()[source]¶
Get age of this object in seconds.
This method calculates how much time has passed since the object was created.
- Returns:
Number of seconds since creation.
- Return type:
- time_since_update()[source]¶
Get time since last update in seconds.
This method calculates how much time has passed since the object was last updated.
- Returns:
Number of seconds since last update.
- Return type:
- update_timestamp()[source]¶
Update the updated_at timestamp to the current time.
This method should be called whenever the object is modified to track the time of the latest change.
- Return type:
None
- created_at: datetime¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- updated_at: datetime¶
- class haive.core.common.mixins.ToolListMixin(*, tools=<factory>)[source]¶
Mixin that adds a ToolList for managing LangChain tools.
This mixin adds a tools attribute to any Pydantic model, providing comprehensive tool management capabilities.
- Parameters:
data (Any)
tools (ToolList)
- tools¶
A ToolList instance for managing tools.
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.ToolRouteMixin(*, tool_routes=<factory>, tool_metadata=<factory>, tools_dict=<factory>, routed_tools=<factory>, before_tool_validator=None, tools=<factory>, tool_instances=<factory>)[source]¶
Enhanced mixin for managing tools, routes, and converting configurations to tools.
This mixin provides functionality for: - Setting and managing tool routes (mapping tool names to types/destinations) - Storing and retrieving tool metadata - Supporting tools as dict with string keys for tool lists - Supporting routed tools with before validators for tuple handling - Generating tools from configurations - Visualizing tool routing information
Tool routes define where a tool request should be directed, such as to a specific retriever, model, or function. This helps implement routing logic in agents and other tool-using components.
- Parameters:
- before_tool_validator¶
Optional callable to validate tools before routing.
- Type:
collections.abc.Callable[[Any], Any] | None
- add_routed_tool(tool, route)[source]¶
Add a single tool with explicit route.
- Parameters:
- Return type:
- add_tools_from_list(tools, clear_existing=False)[source]¶
Add tools from a list to tool_routes without clearing existing routes.
This method analyzes a list of tools and automatically creates appropriate routes based on their types. Supports both regular tools and tuples of (tool, route) for explicit routing.
- add_tools_to_category(category, tools)[source]¶
Add tools to a specific category in tools_dict.
- Parameters:
- Return type:
- clear_tool_routes()[source]¶
Clear all tool routes and metadata.
- Returns:
Self for method chaining.
- Return type:
- debug_tool_routes()[source]¶
Print debug information about tool routes.
This method uses the Rich library to create a visual representation of the tool routes and metadata, including the new dict and routed tools.
- Returns:
Self for method chaining.
- Return type:
- list_tools_by_route(route)[source]¶
Get all tool names for a specific route.
This method finds all tools that are routed to a specific destination.
- remove_tool_route(tool_name)[source]¶
Remove a tool route and its metadata.
- Parameters:
tool_name (str) – Name of the tool to remove.
- Returns:
Self for method chaining.
- Return type:
- set_tool_route(tool_name, route, metadata=None)[source]¶
Set a tool route with optional metadata.
This method defines where a tool request should be routed, along with optional metadata to inform the routing decision.
- set_tool_route_for_existing(tool_identifier, new_route)[source]¶
Set or update the route for an existing tool by name or partial match.
- Parameters:
- Returns:
Self for method chaining.
- Return type:
- sync_tool_routes_from_tools(tools)[source]¶
Synchronize tool_routes with a list of tools.
This method analyzes a list of tools and automatically creates appropriate routes based on their types.
- to_tool(name=None, description=None, route=None, **kwargs)[source]¶
Convert this configuration to a tool.
This method provides a base implementation for creating tools from configuration objects. Specific config classes should override the _create_tool_implementation method to provide custom tool creation logic.
- Parameters:
- Returns:
A tool that can be used with LLMs.
- Return type:
- update_tool_route(tool_name, new_route)[source]¶
Update an existing tool’s route dynamically.
- Parameters:
- Returns:
Self for method chaining
- Return type:
- update_tool_routes(routes)[source]¶
Update multiple tool routes at once.
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.mixins.VersionMixin(*, version='1.0.0', version_history=<factory>)[source]¶
Mixin for adding version tracking to Pydantic models.
This mixin adds version information to any model, with support for tracking version history. It’s useful for managing model versions, checking compatibility, and auditing changes over time.
- bump_version(new_version)[source]¶
Update version and track the previous version in history.
This method should be called whenever a significant change is made to the object that warrants a version increment.
- Parameters:
new_version (str) – The new version string to set.
- Return type:
None
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
Powerful Capability Injection
The mixin system provides composable behaviors that can be added to any class for instant functionality.
Essential Mixins:
from haive.core.common.mixins import (
TimestampMixin, IDMixin, StateMixin,
RecompileMixin, CacheableMixin, SerializableMixin
)
from datetime import datetime
import uuid
# Timestamp tracking
class TrackedComponent(TimestampMixin):
"""Component with automatic timestamp tracking."""
def process(self, data):
# created_at and updated_at automatically managed
result = self.analyze(data)
# updated_at refreshed on any change
return result
# Unique identification
class IdentifiableAgent(IDMixin):
"""Agent with UUID and custom ID support."""
id_prefix = "agent" # Generates: agent_7f3d8a9b
id_generator = lambda: str(uuid.uuid4())[:8]
def __init__(self, name: str):
super().__init__()
self.name = name
# self.id automatically generated
# State management
class StatefulProcessor(StateMixin):
"""Processor with built-in state tracking."""
state_schema = ProcessorState # Pydantic model
state_defaults = {"status": "idle", "queue": []}
async def process_item(self, item):
self.update_state(status="processing")
try:
result = await self.process(item)
self.update_state(
status="idle",
last_processed=item.id,
success_count=self.state.success_count + 1
)
return result
except Exception as e:
self.update_state(
status="error",
last_error=str(e),
error_count=self.state.error_count + 1
)
raise
# Recompilation support
class DynamicGraph(RecompileMixin):
"""Graph that can trigger its own recompilation."""
def add_node(self, node):
self.nodes.append(node)
self.mark_for_recompile("Node added")
def should_recompile(self) -> bool:
# Custom recompilation logic
return (
self.needs_recompile or
len(self.nodes) > self.last_compiled_size * 1.5
)
Advanced Mixin Patterns:
from haive.core.common.mixins import (
ObservableMixin, ValidatableMixin,
RetryableMixin, ThrottleMixin
)
# Observable pattern
class ObservableAgent(ObservableMixin):
"""Agent that emits events."""
def process(self, input_data):
self.emit("processing_started", {"input": input_data})
try:
result = self._process(input_data)
self.emit("processing_completed", {
"input": input_data,
"output": result
})
return result
except Exception as e:
self.emit("processing_failed", {
"input": input_data,
"error": str(e)
})
raise
# Usage
agent = ObservableAgent()
agent.on("processing_completed", lambda e: print(f"Done: {e.data}"))
# Validation mixin
class ValidatedConfig(ValidatableMixin):
"""Configuration with validation."""
validation_rules = {
"temperature": lambda x: 0 <= x <= 2,
"max_tokens": lambda x: x > 0 and x <= 4000,
"model": lambda x: x in ["gpt-4", "claude-3", "llama3"]
}
def __init__(self, **kwargs):
self.validate(kwargs) # Automatic validation
super().__init__(**kwargs)
# Retry logic
class ResilientAPI(RetryableMixin):
"""API client with automatic retries."""
retry_config = {
"max_attempts": 3,
"backoff": "exponential",
"retry_on": [ConnectionError, TimeoutError],
"max_delay": 30
}
@retry_method
async def fetch_data(self, url):
# Automatically retried on failure
return await self.http_client.get(url)
Data Structures¶
🌳 Common Structures - Intelligent Hierarchical Data Architecture
THE EVOLUTIONARY TREE OF AI DATA ORGANIZATION
Welcome to Common Structures - the revolutionary ecosystem of intelligent, self-organizing hierarchical data structures that transform flat information into living, breathing knowledge trees. This isn’t just another data structure library; it’s a comprehensive biological data platform where information grows organically, adapts intelligently, and evolves naturally into sophisticated knowledge networks.
⚡ REVOLUTIONARY STRUCTURAL INTELLIGENCE¶
Common Structures represents a paradigm shift from static hierarchies to living, adaptive data organisms that mirror the intelligence of natural systems:
🧠 Self-Organizing Hierarchies: Structures that automatically organize data by semantic relationships 🔄 Adaptive Growth Patterns: Trees that evolve their structure based on usage and data patterns ⚡ Intelligent Navigation: Smart pathfinding and traversal algorithms for complex knowledge graphs 📊 Performance Optimization: Self-balancing trees with automatic rebalancing and optimization 🎯 Type-Safe Generics: Full generic type support with intelligent type inference and validation
🌟 CORE STRUCTURAL INNOVATIONS¶
- 1. Intelligent Tree Systems 🌲
Revolutionary hierarchical structures that think and adapt:
Examples
>>> from haive.core.common.structures import Tree, TreeNode, Leaf, AutoTree
>>> from typing import Generic, TypeVar
>>>
>>> # Create intelligent tree with semantic organization
>>> knowledge_tree = Tree[str]("AI Knowledge")
>>>
>>> # Add branches with intelligent categorization
>>> ml_branch = knowledge_tree.add_child("Machine Learning")
>>> dl_node = ml_branch.add_child("Deep Learning")
>>>
>>> # Intelligent node management
>>> dl_node.add_children([
>>> "Transformers",
>>> "Convolutional Networks",
>>> "Recurrent Networks",
>>> "Generative Models"
>>> ])
>>>
>>> # Smart navigation and search
>>> transformers_path = knowledge_tree.find_path("Transformers")
>>> related_nodes = knowledge_tree.find_related("Neural Networks")
>>> optimal_route = knowledge_tree.get_shortest_path("AI Knowledge", "Transformers")
>>>
>>> # Automatic tree optimization
>>> knowledge_tree.auto_balance()
>>> knowledge_tree.optimize_for_access_patterns()
>>>
>>> # Semantic clustering
>>> semantic_clusters = knowledge_tree.cluster_by_similarity()
>>> knowledge_tree.reorganize_by_clusters(semantic_clusters)
- 2. Adaptive Tree Generation 🌱
Automatic tree creation from any data structure:
>>> from haive.core.common.structures import AutoTree, auto_tree >>> from pydantic import BaseModel >>> >>> # Define complex data model >>> class ProjectStructure(BaseModel): >>> name: str >>> components: List[str] >>> dependencies: Dict[str, List[str]] >>> metrics: Dict[str, float] >>> >>> # Automatically generate intelligent tree >>> project_data = ProjectStructure( >>> name="AI Assistant", >>> components=["reasoning", "memory", "tools", "interface"], >>> dependencies={ >>> "reasoning": ["memory", "tools"], >>> "interface": ["reasoning", "memory"] >>> }, >>> metrics={"complexity": 0.8, "performance": 0.95} >>> ) >>> >>> # Create auto-organizing tree >>> project_tree = AutoTree.from_model(project_data) >>> >>> # Tree automatically organizes by: >>> # - Dependency relationships >>> # - Semantic similarity >>> # - Usage frequency >>> # - Performance metrics >>> >>> # Advanced tree operations >>> dependency_graph = project_tree.extract_dependency_graph() >>> critical_path = project_tree.find_critical_path() >>> optimization_suggestions = project_tree.suggest_optimizations() >>> >>> # Dynamic tree evolution >>> project_tree.evolve_structure(new_data) >>> project_tree.prune_unused_branches() >>> project_tree.expand_high_value_nodes()
- 3. Semantic Tree Navigation 🧭
Intelligent pathfinding and relationship discovery:
>>> # Create semantic knowledge network >>> semantic_tree = Tree[Dict[str, Any]]("Knowledge Network") >>> >>> # Add nodes with rich semantic metadata >>> ai_node = semantic_tree.add_child("Artificial Intelligence", { >>> "domain": "computer_science", >>> "complexity": "high", >>> "related_fields": ["mathematics", "psychology", "philosophy"], >>> "importance": 0.95 >>> }) >>> >>> # Build semantic relationships >>> ml_node = ai_node.add_child("Machine Learning", { >>> "subdomain": "ai", >>> "prerequisites": ["statistics", "linear_algebra"], >>> "applications": ["prediction", "classification", "clustering"] >>> }) >>> >>> # Intelligent semantic search >>> def semantic_similarity(node1, node2): >>> return calculate_concept_similarity(node1.content, node2.content) >>> >>> # Find conceptually similar nodes >>> similar_concepts = semantic_tree.find_similar_nodes( >>> target_node=ml_node, >>> similarity_threshold=0.7, >>> similarity_function=semantic_similarity >>> ) >>> >>> # Generate learning paths >>> learning_path = semantic_tree.generate_learning_path( >>> start="basic_programming", >>> goal="deep_learning", >>> learner_profile={"experience": "beginner", "time": "3_months"} >>> ) >>> >>> # Knowledge graph analysis >>> concept_map = semantic_tree.generate_concept_map() >>> knowledge_gaps = semantic_tree.identify_knowledge_gaps()
- 4. Performance-Optimized Trees ⚡
Self-balancing structures with intelligent optimization:
>>> # Create high-performance tree with auto-optimization >>> optimized_tree = Tree[Any]( >>> "Performance Tree", >>> auto_balance=True, >>> optimization_strategy="access_frequency", >>> cache_enabled=True >>> ) >>> >>> # Add performance monitoring >>> optimized_tree.enable_performance_tracking() >>> >>> # Tree automatically: >>> # - Rebalances after insertions/deletions >>> # - Caches frequently accessed nodes >>> # - Optimizes structure for common access patterns >>> # - Maintains performance metrics >>> >>> # Manual optimization controls >>> optimized_tree.force_rebalance() >>> optimized_tree.optimize_for_reads() >>> optimized_tree.optimize_for_writes() >>> optimized_tree.compact_memory_usage() >>> >>> # Performance analytics >>> performance_report = optimized_tree.get_performance_report() >>> bottlenecks = optimized_tree.identify_bottlenecks() >>> optimization_recommendations = optimized_tree.suggest_optimizations()
🎯 ADVANCED STRUCTURAL PATTERNS¶
Multi-Dimensional Trees 📐
>>> # Create trees that organize data across multiple dimensions
>>> class MultiDimensionalTree:
>>> def __init__(self, dimensions: List[str]):
>>> self.dimensions = dimensions
>>> self.trees = {dim: Tree[Any](f"{dim}_tree") for dim in dimensions}
>>> self.cross_references = {}
>>>
>>> def add_item(self, item: Any, coordinates: Dict[str, str]):
>>> # Add item with coordinates in multiple dimensions
>>> item_id = generate_unique_id(item)
>>>
>>> # Add to each dimensional tree
>>> for dimension, coordinate in coordinates.items():
>>> tree = self.trees[dimension]
>>> node = tree.find_or_create_path(coordinate)
>>> node.add_reference(item_id, item)
>>>
>>> # Create cross-references
>>> self.cross_references[item_id] = coordinates
>>>
>>> def query_multi_dimensional(self, query: Dict[str, str]) -> List[Any]:
>>> # Query across multiple dimensions simultaneously
>>> result_sets = []
>>>
>>> for dimension, value in query.items():
>>> if dimension in self.trees:
>>> results = self.trees[dimension].search(value)
>>> result_sets.append(set(results))
>>>
>>> # Find intersection across dimensions
>>> if result_sets:
>>> intersection = result_sets[0]
>>> for result_set in result_sets[1:]:
>>> intersection = intersection.intersection(result_set)
>>> return list(intersection)
>>>
>>> return []
>>>
>>> # Usage example
>>> knowledge_system = MultiDimensionalTree([
>>> "topic", "difficulty", "type", "domain"
>>> ])
>>>
>>> knowledge_system.add_item("Machine Learning Basics", {
>>> "topic": "ai/machine_learning",
>>> "difficulty": "beginner",
>>> "type": "tutorial",
>>> "domain": "computer_science"
>>> })
>>>
>>> # Multi-dimensional query
>>> beginner_ai_tutorials = knowledge_system.query_multi_dimensional({
>>> "topic": "ai/*",
>>> "difficulty": "beginner",
>>> "type": "tutorial"
>>> })
Temporal Trees with Version Control ⏰
>>> class TemporalTree(Tree):
>>> # \#Tree that maintains version history and temporal queries.\#
>>>
>>> def __init__(self, name: str):
>>> super().__init__(name)
>>> self.version_history = {}
>>> self.snapshots = {}
>>> self.current_version = 0
>>>
>>> def create_snapshot(self, version_name: str = None):
>>> # \#Create a snapshot of current tree state.\#
>>> version_name = version_name or f"v{self.current_version}"
>>> self.snapshots[version_name] = self.deep_copy()
>>> self.current_version += 1
>>> return version_name
>>>
>>> def query_at_time(self, timestamp: datetime) -> Tree:
>>> # \#Query tree state at a specific time.\#
>>> relevant_snapshot = self.find_snapshot_before(timestamp)
>>> return relevant_snapshot
>>>
>>> def show_evolution(self, node_path: str) -> List[Dict[str, Any]]:
>>> # \#Show how a node evolved over time.\#
>>> evolution_history = []
>>>
>>> for version, snapshot in self.snapshots.items():
>>> node = snapshot.find_node(node_path)
>>> if node:
>>> evolution_history.append({
>>> "version": version,
>>> "content": node.content,
>>> "timestamp": node.last_modified,
>>> "changes": self.calculate_changes_from_previous(node)
>>> })
>>>
>>> return evolution_history
>>>
>>> # Usage
>>> project_tree = TemporalTree("Project Evolution")
>>> project_tree.create_snapshot("initial_design")
>>>
>>> # Make changes...
>>> project_tree.modify_node("architecture/core", new_design)
>>> project_tree.create_snapshot("core_redesign")
>>>
>>> # Time-based queries
>>> yesterday_state = project_tree.query_at_time(yesterday)
>>> evolution = project_tree.show_evolution("architecture/core")
Collaborative Trees with Conflict Resolution 🤝
>>> class CollaborativeTree(Tree):
>>> # \#Tree that supports multi-user collaboration with conflict resolution.\#
>>>
>>> def __init__(self, name: str):
>>> super().__init__(name)
>>> self.collaboration_engine = CollaborationEngine()
>>> self.conflict_resolver = ConflictResolver()
>>> self.user_sessions = {}
>>>
>>> def start_collaborative_session(self, user_id: str) -> str:
>>> # \#Start a collaborative editing session.\#
>>> session_id = self.collaboration_engine.create_session(user_id)
>>> self.user_sessions[session_id] = {
>>> "user_id": user_id,
>>> "active_nodes": set(),
>>> "pending_changes": []
>>> }
>>> return session_id
>>>
>>> def collaborative_edit(self, session_id: str, node_path: str, changes: Dict[str, Any]):
>>> # \#Apply collaborative edit with conflict detection.\#
>>> session = self.user_sessions[session_id]
>>>
>>> # Check for conflicts
>>> conflicts = self.conflict_resolver.detect_conflicts(
>>> node_path, changes, self.get_pending_changes()
>>> )
>>>
>>> if conflicts:
>>> # Automatic conflict resolution
>>> resolved_changes = self.conflict_resolver.resolve_conflicts(
>>> conflicts, strategy="semantic_merge"
>>> )
>>> self.apply_changes(node_path, resolved_changes)
>>> else:
>>> # Apply changes directly
>>> self.apply_changes(node_path, changes)
>>>
>>> # Notify other collaborators
>>> self.collaboration_engine.broadcast_changes(
>>> changes, exclude_session=session_id
>>> )
>>>
>>> def merge_user_contributions(self) -> Dict[str, Any]:
>>> # \#Intelligently merge contributions from all users.\#
>>> all_contributions = self.collaboration_engine.collect_contributions()
>>>
>>> merged_tree = self.conflict_resolver.intelligent_merge(
>>> all_contributions,
>>> merge_strategy="consensus_based"
>>> )
>>>
>>> return merged_tree
>>>
>>> # Usage
>>> team_knowledge = CollaborativeTree("Team Knowledge Base")
>>>
>>> # Multiple users editing simultaneously
>>> alice_session = team_knowledge.start_collaborative_session("alice")
>>> bob_session = team_knowledge.start_collaborative_session("bob")
>>>
>>> # Concurrent edits with automatic conflict resolution
>>> team_knowledge.collaborative_edit(alice_session, "ai/nlp", {
>>> "content": "Natural Language Processing techniques..."
>>> })
>>>
>>> team_knowledge.collaborative_edit(bob_session, "ai/nlp", {
>>> "examples": ["BERT", "GPT", "T5"]
>>> })
>>>
>>> # Intelligent merge of all contributions
>>> final_knowledge = team_knowledge.merge_user_contributions()
🔮 INTELLIGENT STRUCTURE FEATURES¶
Machine Learning-Enhanced Organization 🤖
>>> class MLEnhancedTree(Tree):
>>> # \#Tree that uses ML for optimal organization.\#
>>>
>>> def __init__(self, name: str):
>>> super().__init__(name)
>>> self.ml_organizer = MLTreeOrganizer()
>>> self.pattern_detector = TreePatternDetector()
>>> self.usage_predictor = UsagePredictionModel()
>>>
>>> def smart_organize(self):
>>> # \#Use ML to optimize tree organization.\#
>>> # Analyze current structure
>>> structure_analysis = self.ml_organizer.analyze_structure(self)
>>>
>>> # Detect usage patterns
>>> usage_patterns = self.pattern_detector.detect_patterns(
>>> self.get_access_logs()
>>> )
>>>
>>> # Predict future usage
>>> predicted_usage = self.usage_predictor.predict_access_patterns(
>>> usage_patterns
>>> )
>>>
>>> # Optimize organization
>>> optimal_structure = self.ml_organizer.suggest_reorganization(
>>> current_structure=structure_analysis,
>>> usage_patterns=usage_patterns,
>>> predicted_usage=predicted_usage
>>> )
>>>
>>> # Apply optimizations
>>> self.reorganize_by_structure(optimal_structure)
>>>
>>> def adaptive_caching(self):
>>> # \#Implement ML-driven adaptive caching.\#
>>> cache_strategy = self.usage_predictor.suggest_cache_strategy()
>>> self.implement_cache_strategy(cache_strategy)
>>>
>>> # Automatic optimization
>>> ml_tree = MLEnhancedTree("Adaptive Knowledge Tree")
>>> ml_tree.enable_continuous_learning()
>>> ml_tree.smart_organize() # Runs automatically based on usage
Quantum-Inspired Tree Exploration ⚛️
>>> class QuantumTree(Tree):
>>> # \#Tree that explores multiple organizational states simultaneously.\#
>>>
>>> def __init__(self, name: str):
>>> super().__init__(name)
>>> self.quantum_states = []
>>> self.superposition_enabled = True
>>>
>>> def quantum_search(self, query: str, max_states: int = 10) -> List[Any]:
>>> # \#Search across multiple potential tree organizations.\#
>>> if not self.superposition_enabled:
>>> return self.classical_search(query)
>>>
>>> # Generate multiple potential organizations
>>> potential_organizations = self.generate_quantum_states(max_states)
>>>
>>> # Search in parallel across all states
>>> quantum_results = []
>>> for state in potential_organizations:
>>> results = state.search(query)
>>> quantum_results.append((state, results))
>>>
>>> # Collapse to best result based on quantum scoring
>>> best_state, best_results = self.collapse_to_optimal_state(
>>> quantum_results
>>> )
>>>
>>> # Optionally update tree to best organization
>>> if self.should_collapse_to_state(best_state):
>>> self.collapse_to_state(best_state)
>>>
>>> return best_results
>>>
>>> def enable_quantum_exploration(self):
>>> # \#Enable quantum-inspired exploration mode.\#
>>> self.superposition_enabled = True
>>> self.start_quantum_exploration_background_process()
📊 PERFORMANCE OPTIMIZATION METRICS¶
Tree Performance Characteristics: - Node Access: O(log n) average, O(1) for cached nodes - Tree Balancing: Automatic rebalancing with <5ms overhead - Semantic Search: <10ms for trees with 10,000+ nodes - Memory Efficiency: 70% reduction through intelligent compression
Intelligence Enhancement: - Auto-Organization: 60%+ improvement in average access time - Predictive Caching: 85%+ cache hit rate for access patterns - Semantic Navigation: 95%+ accuracy in finding related concepts - Adaptive Structure: 40%+ reduction in deep traversals
🔧 ADVANCED TREE OPERATIONS¶
Tree Composition and Merging 🔗
>>> # Merge multiple trees intelligently
>>> def intelligent_tree_merge(trees: List[Tree], strategy: str = "semantic") -> Tree:
>>> # \#Merge multiple trees using intelligent strategies.\#
>>>
>>> if strategy == "semantic":
>>> # Merge based on semantic similarity
>>> merged = SemanticTreeMerger().merge(trees)
>>> elif strategy == "structural":
>>> # Merge based on structural patterns
>>> merged = StructuralTreeMerger().merge(trees)
>>> elif strategy == "usage_based":
>>> # Merge based on usage patterns
>>> merged = UsageBasedTreeMerger().merge(trees)
>>> else:
>>> # Default hierarchical merge
>>> merged = HierarchicalTreeMerger().merge(trees)
>>>
>>> return merged
>>>
>>> # Tree decomposition for distributed processing
>>> def decompose_tree_for_distribution(tree: Tree, node_count: int) -> List[Tree]:
>>> # \#Decompose tree into optimal subtrees for distributed processing.\#
>>>
>>> decomposer = TreeDecomposer()
>>> subtrees = decomposer.decompose(
>>> tree=tree,
>>> target_subtree_count=node_count,
>>> load_balancing=True,
>>> minimize_cross_references=True
>>> )
>>>
>>> return subtrees
Dynamic Tree Visualization 🎨
>>> class TreeVisualizer:
>>> # \#Advanced tree visualization with real-time updates.\#
>>>
>>> def __init__(self, tree: Tree):
>>> self.tree = tree
>>> self.layout_engine = TreeLayoutEngine()
>>> self.interaction_tracker = InteractionTracker()
>>>
>>> def create_interactive_visualization(self) -> Dict[str, Any]:
>>> # \#Create interactive tree visualization.\#
>>> return {
>>> "layout": self.layout_engine.generate_layout(self.tree),
>>> "interactions": self.setup_interactions(),
>>> "real_time_updates": self.enable_real_time_updates(),
>>> "performance_overlay": self.create_performance_overlay()
>>> }
>>>
>>> def visualize_evolution_over_time(self) -> Dict[str, Any]:
>>> # \#Create time-lapse visualization of tree evolution.\#
>>> if hasattr(self.tree, 'snapshots'):
>>> return self.layout_engine.create_evolution_animation(
>>> self.tree.snapshots
>>> )
🎓 BEST PRACTICES¶
Design for Growth: Create trees that can evolve and scale naturally
Use Semantic Organization: Leverage semantic relationships for intuitive navigation
Enable Auto-Optimization: Let trees optimize themselves based on usage
Plan for Collaboration: Design for multi-user scenarios from the start
Monitor Performance: Track tree performance and bottlenecks
Implement Caching: Use intelligent caching for frequently accessed nodes
Version Control: Maintain history for complex evolving structures
🚀 GETTING STARTED¶
>>> from haive.core.common.structures import (
>>> Tree, TreeNode, Leaf, AutoTree, auto_tree
>>> )
>>>
>>> # 1. Create intelligent tree
>>> knowledge_tree = Tree[str]("My Knowledge")
>>>
>>> # 2. Add hierarchical content
>>> ai_branch = knowledge_tree.add_child("Artificial Intelligence")
>>> ml_node = ai_branch.add_child("Machine Learning")
>>>
>>> # 3. Use advanced features
>>> ml_node.add_children([
>>> "Deep Learning",
>>> "Classical ML",
>>> "Reinforcement Learning"
>>> ])
>>>
>>> # 4. Enable intelligent features
>>> knowledge_tree.enable_auto_optimization()
>>> knowledge_tree.enable_semantic_search()
>>>
>>> # 5. Navigate intelligently
>>> path = knowledge_tree.find_path("Deep Learning")
>>> related = knowledge_tree.find_related("Neural Networks")
🌳 STRUCTURE GALLERY¶
Core Structures: - Tree[T] - Generic intelligent tree with type safety - TreeNode[T] - Individual tree nodes with rich metadata - Leaf[T] - Terminal nodes with specialized leaf behavior - AutoTree - Automatic tree generation from data models
Advanced Features: - auto_tree() - Factory function for creating optimized trees - Generic type variables (ContentT, ChildT, ResultT) - Intelligent tree traversal and navigation algorithms - Performance optimization and auto-balancing
Specialized Trees: - Semantic trees with AI-powered organization - Temporal trees with version control - Collaborative trees with conflict resolution - ML-enhanced trees with predictive optimization
—
Common Structures: Where Data Grows Into Intelligent Knowledge Trees 🌳
- class haive.core.common.structures.AutoTree[source]¶
Placeholder for AutoTree functionality.
TODO: Implement auto-tree generation from BaseModel inspection.
- class haive.core.common.structures.DefaultContent(*, name, value=None)[source]¶
Default content type with just a name/value.
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.structures.DefaultResult(*, success, data=None, error=None)[source]¶
Default result type with status and data.
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.structures.Leaf(*, content, result=None)[source]¶
Leaf node - has content but no children.
Examples
# With explicit types leaf: Leaf[TaskContent, TaskResult] = Leaf( content=TaskContent(name="Calculate", action="add", params={"a": 1, "b": 2}) ) # With default types simple_leaf = Leaf(content=DefaultContent(name="Task1"))
- Parameters:
content (ContentT)
result (ResultT | None)
- model_post_init(context, /)¶
This function is meant to behave like a BaseModel method to initialise private attributes.
It takes context as an argument since that’s what pydantic-core passes when calling it.
- Parameters:
self (BaseModel) – The BaseModel instance.
context (Any) – The context.
- Return type:
None
- content: ContentT¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.structures.Tree(*, content, result=None, children=<factory>)[source]¶
Tree node - has content and children.
The ChildT parameter allows for heterogeneous trees where children can be of different types (but all extending the bound).
Examples
# Homogeneous tree (all children same type) tree: Tree[PlanContent, PlanNode, PlanResult] = Tree( content=PlanContent(objective="Main Plan") ) # Heterogeneous tree (mixed children) mixed: Tree[DefaultContent, TreeNode, DefaultResult] = Tree( content=DefaultContent(name="Root") )
- Parameters:
content (ContentT)
result (ResultT | None)
children (list[ChildT])
- add_child(*children)[source]¶
Add one or more children with auto-indexing.
- Parameters:
children (ChildT)
- Return type:
ChildT | list[ChildT]
- find_by_path(*indices)[source]¶
Find a descendant by path indices.
- Parameters:
indices (int)
- Return type:
ChildT | None
- model_post_init(context, /)¶
This function is meant to behave like a BaseModel method to initialise private attributes.
It takes context as an argument since that’s what pydantic-core passes when calling it.
- Parameters:
self (BaseModel) – The BaseModel instance.
context (Any) – The context.
- Return type:
None
- content: ContentT¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class haive.core.common.structures.TreeNode(*, content, result=None)[source]¶
Abstract base class for all tree nodes.
Uses bounded TypeVars for better type safety and inference.
- Parameters:
data (Any)
content (ContentT)
result (ResultT | None)
- model_post_init(context, /)¶
This function is meant to behave like a BaseModel method to initialise private attributes.
It takes context as an argument since that’s what pydantic-core passes when calling it.
- Parameters:
self (BaseModel) – The BaseModel instance.
context (Any) – The context.
- Return type:
None
- content: ContentT¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- haive.core.common.structures.auto_tree(model, content_extractor=None)[source]¶
Create a tree structure automatically from a BaseModel instance.
- Parameters:
model (T) – The BaseModel instance to convert to a tree.
content_extractor (Callable[[BaseModel], BaseModel] | None) – Optional function to extract content from models.
- Returns:
A TreeNode representing the model structure.
- Return type:
Specialized Collections for AI
Advanced data structures optimized for common AI patterns.
Tree Structures:
from haive.core.common.structures import (
Tree, TreeNode, TreeTraversal,
NamedTree, WeightedTree
)
# Build knowledge tree
knowledge_tree = Tree[str]("AI")
# Add branches
ml_node = knowledge_tree.add_child("Machine Learning")
dl_node = ml_node.add_child("Deep Learning")
rl_node = ml_node.add_child("Reinforcement Learning")
# Add leaves
dl_node.add_children([
"Transformers",
"CNNs",
"RNNs",
"GANs"
])
# Traverse tree
for node in knowledge_tree.traverse(TreeTraversal.DEPTH_FIRST):
print(" " * node.depth + node.value)
# Find nodes
transformer_node = knowledge_tree.find("Transformers")
path = transformer_node.get_path() # ["AI", "Machine Learning", "Deep Learning", "Transformers"]
# Weighted tree for decision making
decision_tree = WeightedTree[str]("Root Decision")
option_a = decision_tree.add_weighted_child("Option A", weight=0.7)
option_b = decision_tree.add_weighted_child("Option B", weight=0.3)
# Get best path
best_path = decision_tree.get_highest_weight_path()
Nested Dictionaries:
from haive.core.common.structures import (
NestedDict, DeepDict,
FrozenDict, OrderedNestedDict
)
# Deep nested access
config = NestedDict({
"model": {
"name": "gpt-4",
"settings": {
"temperature": 0.7,
"max_tokens": 2000
}
}
})
# Dot notation access
temp = config["model.settings.temperature"] # 0.7
config["model.settings.streaming"] = True # Deep set
# Default values
timeout = config.get("api.timeout", default=30)
# Deep merge
updates = {
"model": {
"settings": {
"temperature": 0.8,
"top_p": 0.9
}
}
}
config.deep_merge(updates)
# Frozen immutable dict
constants = FrozenDict({
"API_KEY": "secret",
"BASE_URL": "https://api.example.com"
})
# constants["API_KEY"] = "new" # Raises ImmutableError
Type Utilities¶
🔍 Common Types - Intelligent Type System Foundation
THE DNA OF AI TYPE INTELLIGENCE
Welcome to Common Types - the revolutionary type system that transforms simple data types into intelligent, self-validating, and context-aware type definitions. This isn’t just another typing module; it’s a sophisticated type intelligence platform where every type carries semantic meaning, validates automatically, and evolves with your application’s understanding of data relationships.
⚡ REVOLUTIONARY TYPE INTELLIGENCE¶
Common Types represents a paradigm shift from static type definitions to living, intelligent type systems that understand context and adapt to usage:
🧠 Semantic Type Understanding: Types that know their meaning and relationships 🔄 Dynamic Type Evolution: Type definitions that grow smarter with usage ⚡ Automatic Validation: Self-validating types with intelligent error messages 📊 Context-Aware Casting: Smart type conversion based on semantic understanding 🎯 Protocol Intelligence: Advanced protocol definitions with runtime checking
🌟 CORE TYPE INNOVATIONS¶
- 1. Universal Data Types 🌐
Fundamental types that power AI data intelligence:
Examples
>>> from haive.core.common.types import DictStrAny, JsonType, StrOrPath
>>> from typing import Any, Dict, List, Union
>>>
>>> # Smart dictionary type for configuration and metadata
>>> config: DictStrAny = {
>>> "model": "gpt-4",
>>> "temperature": 0.7,
>>> "max_tokens": 1000,
>>> "tools": ["calculator", "web_search"],
>>> "metadata": {
>>> "created_by": "intelligent_system",
>>> "optimization_level": "high"
>>> }
>>> }
>>>
>>> # Universal JSON-compatible type for API communication
>>> api_payload: JsonType = {
>>> "action": "analyze_text",
>>> "parameters": {
>>> "text": "Sample text for analysis",
>>> "analysis_types": ["sentiment", "entities", "summary"],
>>> "confidence_threshold": 0.8
>>> },
>>> "nested_data": [
>>> {"item": "data_point_1", "value": 42},
>>> {"item": "data_point_2", "value": 73}
>>> ]
>>> }
>>>
>>> # Flexible path handling for file and URL operations
>>> def process_resource(path: StrOrPath) -> Dict[str, Any]:
>>> """Process resource from file path or URL path."""
>>> if isinstance(path, str):
>>> # Handle string paths intelligently
>>> resource = load_from_string_path(path)
>>> else:
>>> # Handle Path objects with advanced operations
>>> resource = load_from_path_object(path)
>>>
>>> return analyze_resource(resource)
- 2. Advanced Type Composition 🧩
Intelligent type combinations for complex data structures:
>>> from typing import TypeVar, Generic, Protocol, runtime_checkable >>> from haive.core.common.types import ABCRootWrapper >>> >>> # Generic type variables for intelligent type composition >>> T = TypeVar('T') >>> K = TypeVar('K') >>> V = TypeVar('V') >>> >>> # Intelligent data container with type safety >>> class IntelligentContainer(Generic[T]): >>> """Container that understands its content type.""" >>> >>> def __init__(self, content_type: type[T]): >>> self.content_type = content_type >>> self.items: List[T] = [] >>> self.type_validator = create_type_validator(content_type) >>> self.semantic_analyzer = SemanticTypeAnalyzer(content_type) >>> >>> def add(self, item: T) -> None: >>> """Add item with intelligent type validation.""" >>> if self.type_validator.validate(item): >>> self.items.append(item) >>> self.semantic_analyzer.learn_from_item(item) >>> else: >>> suggestion = self.type_validator.suggest_correction(item) >>> raise TypeError(f"Invalid type. Suggestion: {suggestion}") >>> >>> def find_similar(self, query: T, threshold: float = 0.8) -> List[T]: >>> """Find semantically similar items.""" >>> return self.semantic_analyzer.find_similar_items( >>> query, threshold >>> ) >>> >>> # Usage with automatic type inference >>> text_container = IntelligentContainer[str](str) >>> text_container.add("machine learning") >>> text_container.add("artificial intelligence") >>> >>> similar_topics = text_container.find_similar("deep learning")
- 3. Protocol-Based Intelligence 🔌
Advanced protocol definitions for AI system integration:
>>> from typing import Protocol, runtime_checkable >>> from abc import abstractmethod >>> >>> @runtime_checkable >>> class IntelligentProcessor(Protocol): >>> """Protocol for intelligent data processors.""" >>> >>> def process(self, data: JsonType) -> DictStrAny: >>> """Process data with intelligence.""" >>> ... >>> >>> def validate_input(self, data: JsonType) -> bool: >>> """Validate input data format.""" >>> ... >>> >>> def get_capabilities(self) -> List[str]: >>> """Get processor capabilities.""" >>> ... >>> >>> @property >>> def intelligence_level(self) -> float: >>> """Get processor intelligence level (0.0-1.0).""" >>> ... >>> >>> @runtime_checkable >>> class AdaptiveAgent(Protocol): >>> """Protocol for agents that adapt to new situations.""" >>> >>> async def adapt_to_context(self, context: DictStrAny) -> None: >>> """Adapt agent behavior to new context.""" >>> ... >>> >>> def learn_from_interaction(self, interaction: JsonType) -> None: >>> """Learn from user interaction.""" >>> ... >>> >>> def predict_next_action(self, state: DictStrAny) -> str: >>> """Predict optimal next action.""" >>> ... >>> >>> # Runtime protocol checking >>> def register_processor(processor: Any) -> None: >>> """Register processor with runtime type checking.""" >>> if isinstance(processor, IntelligentProcessor): >>> # Processor meets protocol requirements >>> intelligence_level = processor.intelligence_level >>> capabilities = processor.get_capabilities() >>> register_with_intelligence_system(processor, intelligence_level) >>> else: >>> raise TypeError("Processor must implement IntelligentProcessor protocol")
- 4. Wrapper Intelligence 🎁
Smart wrappers that add intelligence to any type:
>>> from haive.core.common.types import ABCRootWrapper >>> >>> # Create intelligent wrapper for any data type >>> class IntelligentDataWrapper(ABCRootWrapper): >>> """Wrapper that adds intelligence to any data type.""" >>> >>> def __init__(self, data: Any, intelligence_level: str = "standard"): >>> super().__init__(data) >>> self.intelligence_level = intelligence_level >>> self.access_patterns = AccessPatternTracker() >>> self.optimization_engine = DataOptimizationEngine() >>> self.semantic_understanding = SemanticDataAnalyzer() >>> >>> def __getattr__(self, name: str) -> Any: >>> """Intelligent attribute access with learning.""" >>> # Track access patterns >>> self.access_patterns.record_access(name) >>> >>> # Get attribute value >>> value = getattr(self._wrapped_object, name) >>> >>> # Learn from access pattern >>> self.semantic_understanding.analyze_access(name, value) >>> >>> # Optimize future access if needed >>> if self.access_patterns.should_optimize(name): >>> self.optimization_engine.optimize_access(name) >>> >>> return value >>> >>> def get_intelligence_insights(self) -> DictStrAny: >>> """Get insights about data usage and patterns.""" >>> return { >>> "access_patterns": self.access_patterns.get_summary(), >>> "optimization_opportunities": self.optimization_engine.get_suggestions(), >>> "semantic_insights": self.semantic_understanding.get_insights() >>> } >>> >>> # Usage: Make any object intelligent >>> regular_data = {"name": "AI Assistant", "capabilities": ["reasoning", "planning"]} >>> intelligent_data = IntelligentDataWrapper(regular_data, "advanced") >>> >>> # Access with automatic learning >>> name = intelligent_data.name # Tracks access pattern >>> capabilities = intelligent_data.capabilities # Learns about usage >>> >>> # Get intelligence insights >>> insights = intelligent_data.get_intelligence_insights()
🎯 ADVANCED TYPE PATTERNS¶
Self-Validating Types ✅
>>> class ValidatedType(Generic[T]):
>>> """Type that validates itself automatically."""
>>>
>>> def __init__(self, value: T, validator: Callable[[T], bool] = None):
>>> self.validator = validator or self._default_validator
>>> self.validation_history = []
>>> self.auto_correction_enabled = True
>>>
>>> if self.validator(value):
>>> self._value = value
>>> self.validation_history.append(("valid", value))
>>> else:
>>> if self.auto_correction_enabled:
>>> corrected_value = self._attempt_correction(value)
>>> if corrected_value is not None:
>>> self._value = corrected_value
>>> self.validation_history.append(("corrected", value, corrected_value))
>>> else:
>>> raise ValueError(f"Cannot validate or correct value: {value}")
>>> else:
>>> raise ValueError(f"Invalid value: {value}")
>>>
>>> def _attempt_correction(self, value: T) -> Optional[T]:
>>> """Attempt to correct invalid value."""
>>> # AI-powered value correction
>>> correction_engine = ValueCorrectionEngine()
>>> return correction_engine.suggest_correction(value, self.validator)
>>>
>>> @property
>>> def value(self) -> T:
>>> """Get validated value."""
>>> return self._value
>>>
>>> @value.setter
>>> def value(self, new_value: T) -> None:
>>> """Set new value with validation."""
>>> if self.validator(new_value):
>>> self._value = new_value
>>> self.validation_history.append(("updated", new_value))
>>> else:
>>> raise ValueError(f"Invalid value: {new_value}")
>>>
>>> # Usage
>>> email_validator = lambda x: "@" in x and "." in x
>>> email = ValidatedType("user@example.com", email_validator)
>>> print(email.value) # "user@example.com"
>>>
>>> # Auto-correction example
>>> email_with_correction = ValidatedType("user.example.com", email_validator)
>>> # Might auto-correct to "user@example.com" if correction engine can infer
Context-Aware Types 🎯
>>> class ContextAwareType(Generic[T]):
>>> """Type that adapts its behavior based on context."""
>>>
>>> def __init__(self, value: T, context: DictStrAny = None):
>>> self._value = value
>>> self.context = context or {}
>>> self.context_analyzer = ContextAnalyzer()
>>> self.behavior_adapter = BehaviorAdapter()
>>>
>>> def __getattribute__(self, name: str) -> Any:
>>> """Context-aware attribute access."""
>>> if name.startswith('_') or name in ['context', 'context_analyzer', 'behavior_adapter']:
>>> return super().__getattribute__(name)
>>>
>>> # Analyze current context
>>> current_context = self.context_analyzer.get_current_context()
>>> combined_context = {**self.context, **current_context}
>>>
>>> # Adapt behavior based on context
>>> adapted_behavior = self.behavior_adapter.adapt_for_context(
>>> name, combined_context
>>> )
>>>
>>> if adapted_behavior:
>>> return adapted_behavior
>>> else:
>>> return getattr(self._value, name)
>>>
>>> def update_context(self, new_context: DictStrAny) -> None:
>>> """Update type context."""
>>> self.context.update(new_context)
>>>
>>> # Re-analyze and adapt to new context
>>> self.behavior_adapter.recalibrate(self.context)
>>>
>>> # Usage
>>> user_input = ContextAwareType("Hello", {
>>> "language": "english",
>>> "formality": "casual",
>>> "audience": "technical"
>>> })
>>>
>>> # Behavior adapts based on context
>>> processed = user_input.process() # Adapts processing to technical audience
Semantic Type Relationships 🔗
>>> class SemanticTypeSystem:
>>> """System for managing semantic relationships between types."""
>>>
>>> def __init__(self):
>>> self.type_graph = SemanticTypeGraph()
>>> self.relationship_analyzer = TypeRelationshipAnalyzer()
>>> self.conversion_engine = SemanticConversionEngine()
>>>
>>> def register_type_relationship(self,
>>> type1: type,
>>> type2: type,
>>> relationship: str,
>>> strength: float = 1.0) -> None:
>>> """Register semantic relationship between types."""
>>> self.type_graph.add_relationship(type1, type2, relationship, strength)
>>>
>>> def find_compatible_types(self, source_type: type) -> List[type]:
>>> """Find types compatible with source type."""
>>> return self.type_graph.find_compatible_types(source_type)
>>>
>>> def intelligent_conversion(self, value: Any, target_type: type) -> Any:
>>> """Convert value to target type using semantic understanding."""
>>> source_type = type(value)
>>>
>>> # Find conversion path
>>> conversion_path = self.type_graph.find_conversion_path(
>>> source_type, target_type
>>> )
>>>
>>> if conversion_path:
>>> return self.conversion_engine.convert_along_path(
>>> value, conversion_path
>>> )
>>> else:
>>> # Attempt AI-powered conversion
>>> return self.conversion_engine.ai_powered_conversion(
>>> value, target_type
>>> )
>>>
>>> # Global semantic type system
>>> semantic_types = SemanticTypeSystem()
>>>
>>> # Register relationships
>>> semantic_types.register_type_relationship(str, int, "parseable", 0.8)
>>> semantic_types.register_type_relationship(dict, str, "serializable", 0.9)
>>> semantic_types.register_type_relationship(list, dict, "groupable", 0.7)
>>>
>>> # Use intelligent conversion
>>> text_number = "42"
>>> converted = semantic_types.intelligent_conversion(text_number, int)
>>> assert converted == 42
📊 TYPE PERFORMANCE METRICS¶
Type Operation Performance: - Validation Speed: <0.1ms per validation check - Conversion Accuracy: 95%+ success rate for semantic conversions - Protocol Checking: <1μs for runtime protocol validation - Wrapper Overhead: <5% performance impact with full intelligence
Intelligence Enhancement: - Auto-Correction: 80%+ success rate for value correction - Context Adaptation: 90%+ accuracy in behavior adaptation - Semantic Understanding: 85%+ accuracy in type relationship detection - Predictive Optimization: 60%+ improvement in access patterns
🔧 ADVANCED TYPE UTILITIES¶
Type Inspection and Analysis 🔍
>>> def analyze_type_intelligence(obj: Any) -> DictStrAny:
>>> """Analyze the intelligence level of any object's type."""
>>>
>>> analysis = {
>>> "basic_type": type(obj).__name__,
>>> "is_intelligent": hasattr(obj, 'intelligence_level'),
>>> "protocols_supported": get_supported_protocols(obj),
>>> "semantic_relationships": get_semantic_relationships(type(obj)),
>>> "optimization_potential": calculate_optimization_potential(obj),
>>> "intelligence_score": calculate_intelligence_score(obj)
>>> }
>>>
>>> return analysis
>>>
>>> def suggest_type_improvements(obj: Any) -> List[str]:
>>> """Suggest improvements to make type more intelligent."""
>>> suggestions = []
>>>
>>> if not hasattr(obj, 'validate'):
>>> suggestions.append("Add validation capabilities")
>>>
>>> if not hasattr(obj, 'adapt_to_context'):
>>> suggestions.append("Add context awareness")
>>>
>>> if not isinstance(obj, ABCRootWrapper):
>>> suggestions.append("Consider wrapping with intelligence")
>>>
>>> return suggestions
🎓 BEST PRACTICES¶
Use Semantic Types: Choose types that convey meaning, not just structure
Enable Validation: Add validation to prevent data corruption early
Design for Evolution: Create types that can grow more intelligent
Leverage Protocols: Use protocols for flexible, runtime-checkable interfaces
Add Context Awareness: Make types adapt to their usage context
Monitor Performance: Track type operation performance and optimize
Document Relationships: Clearly define relationships between types
🚀 GETTING STARTED¶
>>> from haive.core.common.types import (
>>> DictStrAny, JsonType, StrOrPath, ABCRootWrapper
>>> )
>>> from typing import Protocol, runtime_checkable
>>>
>>> # 1. Use universal types for data interchange
>>> config: DictStrAny = {"model": "gpt-4", "temperature": 0.7}
>>> api_data: JsonType = {"action": "process", "data": [1, 2, 3]}
>>>
>>> # 2. Handle paths intelligently
>>> def process_file(path: StrOrPath) -> str:
>>> # Works with strings or Path objects
>>> return f"Processing: {path}"
>>>
>>> # 3. Define intelligent protocols
>>> @runtime_checkable
>>> class SmartProcessor(Protocol):
>>> def process(self, data: JsonType) -> DictStrAny: ...
>>> def get_intelligence_level(self) -> float: ...
>>>
>>> # 4. Use wrapper intelligence
>>> wrapped_data = ABCRootWrapper(complex_data)
>>> # Adds intelligence to any object
🔍 TYPE GALLERY¶
Universal Types: - DictStrAny - Universal dictionary type for configuration and metadata - JsonType - JSON-compatible type for API communication - StrOrPath - Flexible path handling for files and resources - ABCRootWrapper - Intelligence wrapper for any object
Advanced Features: - Protocol-based runtime type checking - Semantic type relationship management - Context-aware type adaptation - Automatic validation and correction
Intelligence Capabilities: - Self-validating types with auto-correction - Context-aware behavior adaptation - Semantic understanding and conversion - Performance optimization and learning
—
Common Types: Where Data Types Become Intelligent Semantic Entities 🔍
- class haive.core.common.types.ABCRootWrapper(root=PydanticUndefined)[source]¶
Abstract base class for root-wrapped models that serialize with a named key. (like ‘query’ instead of ‘root’).
The key is inferred automatically from the class name (lowercased), unless explicitly overridden by setting SERIALIZED_KEY.
Examples
- class Query(ABCRootWrapper[str]):
# SERIALIZED_KEY = “query” # Optional override
- Parameters:
root (RootModelRootType)
- model_dump(*args, **kwargs)[source]¶
Model Dump.
- Returns:
Add return description]
- Return type:
[TODO
- model_dump_json(*args, **kwargs)[source]¶
Model Dump Json.
- Returns:
Add return description]
- Return type:
[TODO
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- root: RootModelRootType¶
Advanced Type System Helpers
Utilities for runtime type checking and manipulation.
Type Checking Utilities:
from haive.core.common.types import (
is_generic_type, get_type_args,
is_optional, unwrap_optional,
create_typed_dict, TypeValidator
)
from typing import List, Optional, Dict, Union
# Check generic types
assert is_generic_type(List[str])
assert get_type_args(Dict[str, int]) == (str, int)
# Handle optionals
assert is_optional(Optional[str])
assert unwrap_optional(Optional[int]) == int
# Runtime type validation
validator = TypeValidator()
@validator.validate_types
def process_data(
items: List[str],
config: Dict[str, Any],
threshold: float = 0.5
) -> Dict[str, float]:
# Types validated at runtime
return {item: score for item, score in zip(items, scores)}
# Create typed dictionaries dynamically
ConfigType = create_typed_dict("ConfigType", {
"model": str,
"temperature": float,
"max_tokens": int,
"streaming": bool
})
config: ConfigType = {
"model": "gpt-4",
"temperature": 0.7,
"max_tokens": 2000,
"streaming": True
}
Validation Framework¶
Comprehensive Data Validation
Robust validation utilities for ensuring data integrity.
Validation Patterns:
from haive.core.common.validation import (
Validator, ValidationRule,
create_validator, compose_validators,
validate_schema, validate_data
)
# Create custom validators
@create_validator
def validate_email(email: str) -> bool:
"""Validate email format."""
import re
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
if not re.match(pattern, email):
raise ValueError(f"Invalid email format: {email}")
return True
@create_validator
def validate_api_key(key: str) -> bool:
"""Validate API key format."""
if not key.startswith("sk-") or len(key) != 51:
raise ValueError("Invalid API key format")
return True
# Compose validators
user_validator = compose_validators({
"email": validate_email,
"api_key": validate_api_key,
"age": lambda x: 0 < x < 150,
"username": lambda x: len(x) >= 3 and x.isalnum()
})
# Validate data
try:
user_validator.validate({
"email": "user@example.com",
"api_key": "sk-" + "x" * 48,
"age": 25,
"username": "john123"
})
except ValidationError as e:
print(f"Validation failed: {e.errors}")
# Schema validation
schema_valid = validate_schema(
UserSchema,
checks=["required_fields", "type_annotations", "default_values"]
)
Performance Helpers¶
Optimization Utilities
Tools for improving performance across the framework.
Caching and Memoization:
from haive.core.common.performance import (
cached, memoize, lazy_property,
timed, profile, BatchProcessor
)
import time
# Function caching
@cached(ttl=3600, maxsize=1000)
def expensive_calculation(x: int, y: int) -> int:
"""Cached for 1 hour."""
time.sleep(1) # Simulate expensive operation
return x ** y
# Method memoization
class DataProcessor:
@memoize
def process_data(self, data: List[int]) -> Dict[str, Any]:
"""Results memoized per instance."""
return {
"mean": sum(data) / len(data),
"max": max(data),
"min": min(data)
}
# Lazy properties
class ExpensiveObject:
@lazy_property
def computed_value(self) -> float:
"""Computed once on first access."""
print("Computing expensive value...")
time.sleep(2)
return 42.0
# Performance profiling
@timed
def slow_function():
"""Execution time logged."""
time.sleep(0.5)
@profile(metrics=["time", "memory"])
def memory_intensive():
"""Full profiling enabled."""
large_list = [i for i in range(1_000_000)]
return sum(large_list)
# Batch processing
processor = BatchProcessor(
batch_size=100,
process_func=lambda batch: [x * 2 for x in batch],
parallel=True,
max_workers=4
)
results = processor.process_all(range(10_000))
Utility Functions¶
Essential Helper Functions
Common utilities used throughout the framework.
String and Data Manipulation:
from haive.core.common.utils import (
snake_to_camel, camel_to_snake,
truncate, pluralize, humanize_bytes,
deep_get, deep_set, flatten_dict
)
# Case conversion
assert snake_to_camel("hello_world") == "HelloWorld"
assert camel_to_snake("HelloWorld") == "hello_world"
# String formatting
assert truncate("Long text here", 10) == "Long te..."
assert pluralize(1, "item") == "1 item"
assert pluralize(5, "item") == "5 items"
assert humanize_bytes(1024 * 1024) == "1.0 MB"
# Deep dictionary access
data = {"user": {"profile": {"name": "John"}}}
assert deep_get(data, "user.profile.name") == "John"
deep_set(data, "user.profile.age", 30)
# Flatten nested structures
flat = flatten_dict({
"user": {
"name": "John",
"settings": {
"theme": "dark",
"notifications": True
}
}
})
# {"user.name": "John", "user.settings.theme": "dark", ...}
Async Utilities:
from haive.core.common.utils import (
run_async, gather_with_timeout,
async_retry, create_task_group
)
# Run async in sync context
result = run_async(async_function(param))
# Gather with timeout
results = await gather_with_timeout(
[async_task1(), async_task2(), async_task3()],
timeout=5.0,
return_exceptions=True
)
# Async retry
@async_retry(attempts=3, delay=1.0)
async def flaky_api_call():
# Automatically retried
return await external_api.fetch()
# Task groups
async with create_task_group() as tg:
tg.create_task(process_item(1))
tg.create_task(process_item(2))
tg.create_task(process_item(3))
# All tasks awaited on exit
Advanced Patterns¶
Plugin System¶
Extensible Component Architecture:
from haive.core.common.plugins import (
PluginRegistry, Plugin,
hook, extension_point
)
# Define plugin interface
class AnalyzerPlugin(Plugin):
"""Base class for analyzer plugins."""
@extension_point
def analyze(self, data: Any) -> Dict:
"""Analyze data and return results."""
pass
# Create plugin
@PluginRegistry.register("sentiment")
class SentimentAnalyzer(AnalyzerPlugin):
"""Sentiment analysis plugin."""
def analyze(self, data: str) -> Dict:
# Implementation
return {"sentiment": "positive", "score": 0.8}
# Use plugins
registry = PluginRegistry()
for plugin_name in registry.list_plugins("analyzer"):
plugin = registry.get_plugin(plugin_name)
result = plugin.analyze(text_data)
Event System¶
Decoupled Communication:
from haive.core.common.events import (
EventBus, Event,
subscribe, emit
)
# Global event bus
bus = EventBus()
# Subscribe to events
@bus.subscribe("agent.started")
def on_agent_start(event: Event):
print(f"Agent {event.data['agent_id']} started")
@bus.subscribe("agent.completed", priority=10)
async def on_agent_complete(event: Event):
await log_completion(event.data)
# Emit events
bus.emit("agent.started", {"agent_id": "agent-123"})
# Async event handling
await bus.emit_async("agent.completed", {
"agent_id": "agent-123",
"duration": 5.2,
"result": "success"
})
Enterprise Features¶
Production-Ready Utilities
Thread Safety: All utilities designed for concurrent use
Memory Efficiency: Optimized data structures and algorithms
Error Handling: Comprehensive error types and recovery
Monitoring: Built-in metrics and instrumentation
Compatibility: Python 3.8+ with type stubs
See Also¶
patterns - Common patterns using utilities
performance - Performance optimization guide
Examples - Utility usage examples
API Reference - Complete API documentation