haive.core.common ================= .. py:module:: haive.core.common .. autoapi-nested-parse:: 🧰 Haive Core Common - The Universal Toolkit **THE SWISS ARMY KNIFE OF AI INFRASTRUCTURE** Welcome to the Common module - an extraordinary collection of battle-tested utilities, intelligent mixins, and advanced data structures that form the backbone of every component in the Haive ecosystem. This isn't just a utilities folder; it's a treasure trove of engineering patterns that make AI systems more robust, maintainable, and intelligent. ðŸŽŊ PHILOSOPHY ------------- The Common module embodies the principle of "Write Once, Use Everywhere" with a twist - these aren't just reusable components, they're intelligent building blocks that adapt to their context, optimize themselves, and make your AI systems smarter by default. **Intelligent Mixins** ðŸ§Đ Composable behaviors that add superpowers to any class: - Self-documenting components with automatic metadata - Time-aware objects that track their own evolution - Version-controlled entities with migration support - Observable components with event emission **Advanced Data Structures** ðŸŒģ Beyond basic collections - structures that think: - Self-balancing trees with semantic organization - Graph structures with pathfinding algorithms - Nested dictionaries with deep merge capabilities - Observable collections with change tracking **Type System Enhancements** 🔍 Runtime type safety and intelligent inference: - Generic type manipulation and inspection - Protocol-based type checking - Dynamic type generation from data - Type-guided serialization **Performance Utilities** ⚡ Make everything faster without thinking about it: - Automatic caching with semantic similarity - Lazy evaluation with memoization - Batch processing with optimal chunking - Resource pooling and management 🏗ïļ CORE COMPONENTS ------------------ **Mixin Architecture** .. rubric:: Examples >>> from haive.core.common.mixins import ( >>> IdentifierMixin, # Unique IDs with collision detection >>> TimestampMixin, # Automatic created/updated tracking >>> MetadataMixin, # Rich metadata with indexing >>> VersionMixin, # Semantic versioning with migrations >>> ObservableMixin, # Event emission and subscription >>> CacheableMixin, # Intelligent result caching >>> ValidatableMixin, # Runtime validation hooks >>> SerializationMixin, # Type-safe serialization >>> RetryworthyMixin, # Automatic retry with backoff >>> ThrottlableMixin # Rate limiting and quotas >>> ) >>> >>> # Compose behaviors like LEGO blocks >>> class IntelligentComponent( >>> IdentifierMixin, >>> TimestampMixin, >>> ObservableMixin, >>> CacheableMixin >>> ): >>> def __init__(self): >>> super().__init__() # All mixins initialized automatically >>> self.id # Unique identifier generated >>> self.created_at # Timestamp recorded >>> self.on("change", self.invalidate_cache) # Event wiring **Tree Structures** >>> from haive.core.common.structures import Tree, TreeNode, Leaf >>> >>> # Build semantic knowledge trees >>> knowledge_tree = Tree[Concept]("AI") >>> ml_branch = knowledge_tree.add_child("Machine Learning") >>> dl_node = ml_branch.add_child("Deep Learning") >>> dl_node.add_children([ >>> Concept("Transformers", importance=0.9), >>> Concept("CNNs", importance=0.7), >>> Concept("RNNs", importance=0.6) >>> ]) >>> >>> # Intelligent tree operations >>> most_important = knowledge_tree.find_by_attribute( >>> lambda node: node.value.importance > 0.8 >>> ) >>> path_to_transformers = knowledge_tree.find_path("Transformers") >>> pruned_tree = knowledge_tree.prune_by_importance(threshold=0.5) **Type Utilities** >>> from haive.core.common.types import ( >>> JsonType, # Recursive JSON type definition >>> DictStrAny, # Common dict type alias >>> StrOrPath, # Union type for paths >>> Protocol, # Runtime protocol checking >>> TypeGuard, # Type narrowing functions >>> get_type_hints_with_extras # Enhanced type introspection >>> ) >>> >>> # Runtime type validation >>> def process_data(data: JsonType) -> DictStrAny: >>> if not is_json_compatible(data): >>> raise TypeError("Data must be JSON-serializable") >>> >>> # Type-safe processing >>> return transform_with_type_preservation(data) 🚀 USAGE PATTERNS ----------------- **1. Building Intelligent Components** >>> from haive.core.common.mixins import * >>> from haive.core.common.structures import * >>> from haive.core.common.types import * >>> >>> class SmartAgent( >>> IdentifierMixin, >>> TimestampMixin, >>> VersionMixin, >>> ObservableMixin, >>> SerializationMixin >>> ): >>> version = "1.0.0" >>> >>> def __init__(self, name: str): >>> super().__init__() >>> self.name = name >>> self.knowledge_tree = Tree[str]("root") >>> self.emit("created", {"name": name}) >>> >>> def learn(self, concept: str, parent: str = "root"): >>> node = self.knowledge_tree.add_child_to_parent(concept, parent) >>> self.emit("learned", {"concept": concept, "parent": parent}) >>> self.bump_version("patch") # Auto-increment version >>> >>> def to_dict(self) -> DictStrAny: >>> # SerializationMixin provides this automatically >>> return { >>> **super().to_dict(), >>> "knowledge": self.knowledge_tree.to_dict() >>> } **2. Advanced Data Management** >>> from haive.core.common.models import DynamicChoiceModel, NamedList >>> >>> # Dynamic enums that can grow >>> class AgentCapabilities(DynamicChoiceModel): >>> REASONING = "reasoning" >>> PLANNING = "planning" >>> EXECUTION = "execution" >>> >>> @classmethod >>> def add_capability(cls, name: str, value: str): >>> # Dynamically add new capabilities >>> setattr(cls, name.upper(), value) >>> cls._choices[name] = value >>> >>> # Named lists with attribute access >>> tools = NamedList("AvailableTools") >>> tools.append("calculator", importance=0.8) >>> tools.append("web_search", importance=0.9) >>> tools.sort_by_attribute("importance", reverse=True) **3. Performance Optimization** >>> from haive.core.common.mixins import CacheableMixin >>> from haive.core.common.decorators import memoize, rate_limit, retry >>> >>> class OptimizedProcessor(CacheableMixin): >>> @memoize(maxsize=1000, ttl=3600) >>> def expensive_computation(self, input_data: str) -> float: >>> # Automatically cached for 1 hour >>> return complex_algorithm(input_data) >>> >>> @rate_limit(calls=10, period=60) >>> async def api_call(self, endpoint: str) -> JsonType: >>> # Maximum 10 calls per minute >>> return await fetch_data(endpoint) >>> >>> @retry(attempts=3, backoff="exponential") >>> def unreliable_operation(self) -> bool: >>> # Auto-retry with exponential backoff >>> return external_service.process() **4. Event-Driven Architecture** >>> from haive.core.common.mixins import ObservableMixin >>> >>> class EventDrivenWorkflow(ObservableMixin): >>> def __init__(self): >>> super().__init__() >>> >>> # Wire up event handlers >>> self.on("data_received", self.process_data) >>> self.on("processing_complete", self.generate_report) >>> self.on("error", self.handle_error) >>> >>> async def run(self, data: Any): >>> self.emit("data_received", data) >>> >>> try: >>> result = await self.process(data) >>> self.emit("processing_complete", result) >>> except Exception as e: >>> self.emit("error", {"exception": e, "data": data}) ðŸŽĻ ADVANCED FEATURES -------------------- **1. Semantic Type Inference** 🧠 >>> from haive.core.common.types import infer_type_from_data >>> >>> # Automatically infer types from data >>> data = {"name": "John", "age": 30, "scores": [95, 87, 91]} >>> schema = infer_type_from_data(data) >>> # Result: TypedDict with proper field types **2. Deep Object Comparison** 🔍 >>> from haive.core.common.utils import deep_diff, deep_merge >>> >>> # Intelligent object diffing >>> old_state = {"a": 1, "b": {"c": 2, "d": 3}} >>> new_state = {"a": 1, "b": {"c": 4, "d": 3, "e": 5}} >>> diff = deep_diff(old_state, new_state) >>> # Result: {"b": {"c": {"old": 2, "new": 4}, "e": {"added": 5}}} >>> >>> # Smart merging with conflict resolution >>> merged = deep_merge( >>> old_state, >>> new_state, >>> conflict_resolver=lambda k, v1, v2: max(v1, v2) >>> ) **3. Resource Management** 🔄 >>> from haive.core.common.resources import ResourcePool, managed_resource >>> >>> # Connection pooling >>> db_pool = ResourcePool( >>> factory=create_db_connection, >>> max_size=10, >>> timeout=30 >>> ) >>> >>> async with db_pool.acquire() as conn: >>> # Connection automatically returned to pool >>> await conn.execute(query) >>> >>> # Automatic resource cleanup >>> @managed_resource >>> class TempFileHandler: >>> def __enter__(self): >>> self.file = create_temp_file() >>> return self.file >>> >>> def __exit__(self, *args): >>> cleanup_temp_file(self.file) **4. Intelligent Logging** 📝 >>> from haive.core.common.mixins import RichLoggerMixin >>> >>> class SmartComponent(RichLoggerMixin): >>> def process(self, data): >>> self.log.info("Processing started", extra={ >>> "data_size": len(data), >>> "component_id": self.id >>> }) >>> >>> with self.log.timed("processing"): >>> # Automatically logs execution time >>> result = expensive_operation(data) >>> >>> self.log.success("Processing complete", metrics={ >>> "records_processed": len(result), >>> "success_rate": 0.95 >>> }) 🛠ïļ UTILITY REFERENCE -------------------- **Mixins Available**: - `IdentifierMixin`: Unique ID generation - `TimestampMixin`: Created/updated tracking - `VersionMixin`: Semantic versioning - `MetadataMixin`: Arbitrary metadata storage - `ObservableMixin`: Event emission/subscription - `CacheableMixin`: Result caching - `ValidatableMixin`: Runtime validation - `SerializationMixin`: JSON/dict conversion - `RetryworthyMixin`: Automatic retries - `ThrottlableMixin`: Rate limiting **Data Structures**: - `Tree`: Hierarchical data organization - `Graph`: Network structures - `NamedList`: Lists with attribute access - `OrderedSet`: Sets that maintain order - `FrozenDict`: Immutable dictionaries - `NestedDict`: Deep key access **Type Utilities**: - Type inference from data - Runtime protocol checking - Generic type manipulation - Type-safe serialization **Performance Tools**: - Memoization decorators - Resource pooling - Lazy evaluation - Batch processing 📊 PERFORMANCE IMPACT -------------------- - **Mixin Overhead**: < 0.1ms per class initialization - **Tree Operations**: O(log n) for balanced trees - **Type Checking**: < 1Ξs for simple types - **Event Emission**: < 10Ξs per event - **Caching**: 100x speedup for repeated calls ðŸ”Ū FUTURE EVOLUTION ------------------- The Common module continues to grow: - **Quantum-Inspired Types**: Superposition types - **Neural Mixins**: Self-learning behaviors - **Distributed Structures**: Cross-machine data structures - **Temporal Utilities**: Time-aware components --- **Common Module: Where Every Line of Code Gets Superpowers** 🧰 Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/haive/core/common/logging_config/index /autoapi/haive/core/common/mixins/index /autoapi/haive/core/common/models/index /autoapi/haive/core/common/structures/index /autoapi/haive/core/common/types/index