haive.core.schema.state¶
🌊 State System - Dynamic AI State Evolution Engine
THE LIVING MEMORY OF INTELLIGENT SYSTEMS
Welcome to the State System - the revolutionary foundation that transforms static data models into dynamic, evolving memory structures for AI agents. This isn’t just another state management library; it’s a comprehensive state evolution platform that enables AI systems to dynamically adapt their memory structures as they learn and grow.
🧬 STATE EVOLUTION REVOLUTION¶
The State System represents a paradigm shift in how AI agents manage and evolve their internal state. Every state becomes a living, breathing data structure that can:
🔄 Self-Modifying Architecture: States that evolve their own structure based on discovered patterns ⚡ Real-time Adaptation: Dynamic field addition and type evolution without restarts 🧠 Intelligent Merging: Semantic-aware state combination with conflict resolution 📊 Temporal Memory: Version history and time-travel debugging capabilities 🎯 Type-Safe Evolution: Guaranteed type safety even during dynamic schema changes
🌟 CORE INNOVATIONS¶
- 1. Dynamic State Architecture 🏗️
States that grow and evolve with your AI agents:
Examples
>>> from haive.core.schema.state import StateSchema, Field
>>>
>>> class AdaptiveAgentState(StateSchema):
>>> # Core fields that are always present
>>> messages: List[BaseMessage] = Field(default_factory=list)
>>> confidence: float = Field(default=0.0)
>>>
>>> # Fields that can be added dynamically
>>> __dynamic_fields__ = True
>>> __auto_evolve__ = True
>>>
>>> def discover_capability(self, capability: str, data: Any):
>>> """Dynamically add new state fields based on discovered capabilities."""
>>> if capability == "vision":
>>> self.add_field("visual_memory", List[Image], default=[])
>>> self.add_field("object_recognition", Dict[str, float], default={})
>>> elif capability == "planning":
>>> self.add_field("goal_stack", List[Goal], default=[])
>>> self.add_field("plan_history", List[Plan], default=[])
- 2. Intelligent State Merging 🧩
Advanced merging strategies that understand your data:
>>> # Define semantic merge strategies >>> class ConversationState(StateSchema): >>> messages: List[BaseMessage] = Field(default_factory=list) >>> knowledge: Dict[str, Any] = Field(default_factory=dict) >>> user_preferences: Dict[str, Any] = Field(default_factory=dict) >>> >>> # Custom reducer functions for intelligent merging >>> __reducer_fields__ = { >>> "messages": preserve_chronological_order, >>> "knowledge": semantic_knowledge_merge, >>> "user_preferences": preference_conflict_resolution >>> } >>> >>> # Automatic intelligent merging >>> state1 = ConversationState(messages=[msg1, msg2]) >>> state2 = ConversationState(messages=[msg3, msg4]) >>> merged = state1.merge(state2) # Intelligent chronological merge
- 3. Temporal State Management ⏰
Time-aware state with history and rollback capabilities:
>>> class TemporalState(StateSchema): >>> __enable_history__ = True >>> __snapshot_frequency__ = 10 # Snapshot every 10 updates >>> __max_history_size__ = 100 >>> >>> current_data: Dict[str, Any] = Field(default_factory=dict) >>> >>> def rollback_to(self, timestamp: datetime): >>> """Rollback state to specific point in time.""" >>> return self.restore_snapshot(timestamp) >>> >>> def get_state_at(self, timestamp: datetime): >>> """Get state at specific time without modifying current state.""" >>> return self.view_snapshot(timestamp) >>> >>> def analyze_evolution(self) -> StateEvolutionAnalysis: >>> """Analyze how state has evolved over time.""" >>> return StateEvolutionAnalysis(self.history)
- 4. Multi-Agent State Coordination 🤝
Sophisticated state sharing and synchronization:
>>> class SharedState(StateSchema): >>> # Global shared fields >>> __shared_fields__ = ["global_knowledge", "conversation_history"] >>> >>> # Agent-specific private fields >>> __private_fields__ = ["internal_thoughts", "private_memory"] >>> >>> # Coordination fields >>> global_knowledge: KnowledgeGraph = Field(default_factory=KnowledgeGraph) >>> conversation_history: List[Message] = Field(default_factory=list) >>> >>> # Private to each agent instance >>> internal_thoughts: List[str] = Field(default_factory=list, private=True) >>> private_memory: Dict[str, Any] = Field(default_factory=dict, private=True) >>> >>> def share_with_agent(self, agent_id: str) -> AgentStateView: >>> """Create filtered view for specific agent.""" >>> return self.create_view( >>> include=self.__shared_fields__, >>> exclude=self.__private_fields__, >>> agent_id=agent_id >>> )
🎯 ADVANCED FEATURES¶
Real-time Schema Evolution 🔮
>>> # Schema that adapts to new patterns
>>> class LearningState(StateSchema):
>>> __learning_mode__ = True
>>> __pattern_detection__ = True
>>>
>>> def update(self, data: Dict[str, Any]):
>>> # Detect new patterns in incoming data
>>> new_patterns = self.detect_patterns(data)
>>>
>>> # Automatically evolve schema
>>> for pattern in new_patterns:
>>> if pattern.confidence > 0.8:
>>> self.add_computed_field(
>>> pattern.field_name,
>>> pattern.computation_logic,
>>> pattern.type_hint
>>> )
>>>
>>> # Apply update with evolved schema
>>> super().update(data)
State Validation Chains ✅
>>> class ValidatedState(StateSchema):
>>> data: Dict[str, Any] = Field(default_factory=dict)
>>>
>>> @field_validator("data")
>>> @classmethod
>>> def validate_data_structure(cls, v):
>>> # Multi-level validation
>>> structural_validation(v)
>>> semantic_validation(v)
>>> consistency_validation(v)
>>> return v
>>>
>>> @model_validator(mode="after")
>>> def cross_field_validation(self):
>>> # Complex cross-field relationships
>>> if not self.validate_field_relationships():
>>> raise ValueError("Field relationships are inconsistent")
>>> return self
State Performance Analytics 📊
>>> class AnalyticsState(StateSchema):
>>> __enable_analytics__ = True
>>> __track_performance__ = True
>>>
>>> def get_performance_metrics(self) -> StateMetrics:
>>> return StateMetrics(
>>> memory_usage=self.calculate_memory_usage(),
>>> access_patterns=self.analyze_access_patterns(),
>>> evolution_rate=self.calculate_evolution_rate(),
>>> merge_efficiency=self.analyze_merge_performance()
>>> )
>>>
>>> def optimize_structure(self) -> OptimizationReport:
>>> """Automatically optimize state structure for performance."""
>>> return self.apply_optimizations([
>>> "field_reordering",
>>> "type_optimization",
>>> "compression",
>>> "lazy_loading"
>>> ])
🏗️ STATE COMPOSITION PATTERNS¶
Hierarchical State Systems 🏛️
>>> class HierarchicalState(StateSchema):
>>> # Root level data
>>> session_info: SessionInfo = Field(...)
>>>
>>> # Nested state components
>>> conversation_state: ConversationState = Field(default_factory=ConversationState)
>>> knowledge_state: KnowledgeState = Field(default_factory=KnowledgeState)
>>> planning_state: PlanningState = Field(default_factory=PlanningState)
>>>
>>> def get_component(self, component_name: str) -> StateSchema:
>>> """Access specific state component."""
>>> return getattr(self, f"{component_name}_state")
>>>
>>> def synchronize_components(self):
>>> """Ensure all components are synchronized."""
>>> shared_context = self.extract_shared_context()
>>> for component in self.get_all_components():
>>> component.update_from_shared_context(shared_context)
Plugin-based State Extensions 🔌
>>> class ExtensibleState(StateSchema):
>>> __enable_plugins__ = True
>>>
>>> # Core state
>>> base_data: Dict[str, Any] = Field(default_factory=dict)
>>>
>>> # Plugin registry
>>> plugins: Dict[str, StatePlugin] = Field(default_factory=dict)
>>>
>>> def load_plugin(self, plugin: StatePlugin):
>>> """Dynamically load state extension plugin."""
>>> plugin_fields = plugin.get_fields()
>>> plugin_methods = plugin.get_methods()
>>> plugin_validators = plugin.get_validators()
>>>
>>> # Extend state with plugin capabilities
>>> self.add_fields(plugin_fields)
>>> self.add_methods(plugin_methods)
>>> self.add_validators(plugin_validators)
>>>
>>> self.plugins[plugin.name] = plugin
🔒 ENTERPRISE FEATURES¶
State Governance: Approval workflows for schema changes
Access Control: Field-level permissions and role-based access
Audit Logging: Complete state evolution history
Multi-tenancy: Isolated state spaces per tenant
Compliance: GDPR, HIPAA compliance with automatic PII handling
Backup & Recovery: Automated state backup and disaster recovery
🎓 BEST PRACTICES¶
Design for Evolution: Plan for schema changes from day one
Use Typed Fields: Leverage Pydantic’s type system for safety
Implement Reducers: Define intelligent merge strategies for complex data
Monitor Performance: Track state size and access patterns
Version Your Schemas: Use migration strategies for breaking changes
Test State Evolution: Validate schema changes with comprehensive tests
Document State Contracts: Clear documentation for shared state fields
🚀 GETTING STARTED¶
>>> from haive.core.schema.state import StateSchema, Field
>>> from typing import List, Dict, Any
>>>
>>> # 1. Define your evolving state
>>> class MyAgentState(StateSchema):
>>> # Essential fields
>>> messages: List[str] = Field(default_factory=list)
>>> context: Dict[str, Any] = Field(default_factory=dict)
>>>
>>> # Enable dynamic evolution
>>> __dynamic_fields__ = True
>>>
>>> # Define intelligent merging
>>> __reducer_fields__ = {
>>> "messages": lambda old, new: old + new, # Append messages
>>> "context": lambda old, new: {**old, **new} # Merge contexts
>>> }
>>>
>>> # 2. Use in your agent
>>> state = MyAgentState()
>>> state.messages.append("Hello!")
>>>
>>> # 3. Dynamic evolution
>>> state.discover_capability("memory", {"type": "semantic", "size": 1000})
>>>
>>> # 4. Intelligent merging
>>> other_state = MyAgentState(messages=["World!"])
>>> merged = state.merge(other_state) # ["Hello!", "World!"]
—
State System: Where Data Structures Become Living, Intelligent Memory 🌊