agents.memory.unified_memory_api¶
Unified Memory API - Complete Memory System Integration.
This module provides a unified, easy-to-use API for the complete memory system, integrating all components including classification, storage, retrieval, knowledge graph generation, and multi-agent coordination.
Classes¶
Comprehensive configuration for the unified memory system. |
|
Comprehensive result from memory system operations with metrics and analysis. |
|
Unified Memory System - Complete memory management solution with intelligent coordination. |
Functions¶
|
Create a unified memory system with sensible default configuration. |
Comprehensive demonstration of the unified memory system capabilities. |
Module Contents¶
- class agents.memory.unified_memory_api.MemorySystemConfig(/, **data)¶
Bases:
pydantic.BaseModelComprehensive configuration for the unified memory system.
This configuration class defines all settings needed to create and customize a UnifiedMemorySystem, including store settings, LLM configuration, feature toggles, and performance parameters.
- Parameters:
data (Any)
- store_type¶
Type of store backend (“memory”, “postgres”, “redis”, etc.)
- collection_name¶
Name of the collection/table for storing memories
- default_namespace¶
Default namespace tuple for memory organization
- llm_config¶
LLM configuration for classification, analysis, and generation
- enable_auto_classification¶
Whether to automatically classify stored memories
- classification_confidence_threshold¶
Minimum confidence for auto-classification
- enable_enhanced_retrieval¶
Whether to enable enhanced retrieval features
- enable_graph_rag¶
Whether to enable Graph RAG retrieval capabilities
- enable_multi_agent_coordination¶
Whether to enable multi-agent coordination
- max_concurrent_operations¶
Maximum number of concurrent memory operations
- operation_timeout_seconds¶
Timeout for individual memory operations
- enable_memory_consolidation¶
Whether to enable automatic memory consolidation
- consolidation_interval_hours¶
Hours between automatic consolidation runs
Examples
Basic configuration for development:
config = MemorySystemConfig( store_type="memory", collection_name="dev_memories", default_namespace=("user", "development") )
Production configuration with PostgreSQL:
config = MemorySystemConfig( store_type="postgres", collection_name="prod_memories", default_namespace=("company", "production"), llm_config=AugLLMConfig( model="gpt-4", temperature=0.1, max_tokens=1000 ), enable_auto_classification=True, enable_graph_rag=True, enable_multi_agent_coordination=True, max_concurrent_operations=10, operation_timeout_seconds=600 )
Performance-optimized configuration:
config = MemorySystemConfig( store_type="redis", collection_name="fast_memories", default_namespace=("user", "cache"), enable_auto_classification=False, # Disable for speed enable_enhanced_retrieval=True, enable_graph_rag=False, # Disable for speed enable_multi_agent_coordination=False, max_concurrent_operations=20, operation_timeout_seconds=30 )
Create a new model by parsing and validating input data from keyword arguments.
Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.
self is explicitly positional-only to allow self as a field name.
- model_config¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agents.memory.unified_memory_api.MemorySystemResult(/, **data)¶
Bases:
pydantic.BaseModelComprehensive result from memory system operations with metrics and analysis.
This class encapsulates all information returned from memory system operations, including success status, operation results, performance metrics, quality scores, and metadata for analysis and monitoring.
- Parameters:
data (Any)
- success¶
Whether the operation completed successfully
- operation¶
Type of operation performed (store_memory, retrieve_memories, etc.)
- result¶
The actual result data from the operation (varies by operation type)
- error¶
Error message if the operation failed (None if successful)
- execution_time_ms¶
Time taken to complete the operation in milliseconds
- agent_used¶
Name of the agent/component that handled the operation
- confidence_score¶
Confidence in the result quality (0.0-1.0)
- completeness_score¶
How complete the result is (0.0-1.0)
- timestamp¶
UTC timestamp when the operation completed
- metadata¶
Additional metadata specific to the operation
Examples
Checking operation success:
result = await memory_system.store_memory("Important information") if result.success: memory_id = result.result["memory_id"] print(f"Memory stored successfully with ID: {memory_id}") print(f"Operation took {result.execution_time_ms:.1f}ms") else: print(f"Storage failed: {result.error}")
Analyzing retrieval results:
result = await memory_system.retrieve_memories("machine learning") if result.success: memories = result.result["memories"] count = result.result["count"] print(f"Retrieved {count} memories in {result.execution_time_ms:.1f}ms") print(f"Confidence: {result.confidence_score:.2f}") print(f"Completeness: {result.completeness_score:.2f}") print(f"Agent used: {result.agent_used}") for memory in memories: print(f"- {memory['content'][:100]}...")
Performance monitoring:
results = [] # Perform multiple operations for query in ["Python", "AI", "databases"]: result = await memory_system.retrieve_memories(query) results.append(result) # Analyze performance avg_time = sum(r.execution_time_ms for r in results) / len(results) success_rate = sum(1 for r in results if r.success) / len(results) print(f"Average response time: {avg_time:.1f}ms") print(f"Success rate: {success_rate:.1%}")
Create a new model by parsing and validating input data from keyword arguments.
Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.
self is explicitly positional-only to allow self as a field name.
- class agents.memory.unified_memory_api.UnifiedMemorySystem(config)¶
Unified Memory System - Complete memory management solution with intelligent coordination.
The UnifiedMemorySystem provides a single, comprehensive interface to all memory system capabilities, automatically coordinating between multiple specialized agents to provide optimal memory storage, retrieval, and analysis. It combines traditional vector search with knowledge graph traversal, multi-agent coordination, and intelligent classification for superior memory management.
- Key Features:
Unified API: Single interface for all memory operations
Multi-Agent Coordination: Automatic routing to best-suited agents
Graph RAG: Knowledge graph enhanced retrieval
Auto-Classification: Intelligent memory type detection
Performance Monitoring: Built-in metrics and diagnostics
Flexible Storage: Support for multiple backend stores
Async Operations: Full async/await support for scalability
- Architecture:
Memory Store Manager: Handles storage and basic retrieval
Memory Classifier: Analyzes and categorizes memory content
KG Generator Agent: Builds and maintains knowledge graphs
Graph RAG Retriever: Enhanced retrieval with graph context
Agentic RAG Coordinator: Intelligent retrieval strategy selection
Multi-Agent Coordinator: Orchestrates all agents for complex tasks
- config¶
System configuration settings
- memory_store¶
Core memory storage manager
- classifier¶
Memory classification engine
- kg_generator¶
Knowledge graph generation agent
- retrievers¶
Dictionary of specialized retrieval systems
- agentic_rag¶
Intelligent retrieval coordinator
- coordinator¶
Multi-agent coordination system (optional)
Examples
Basic usage with default configuration:
# Create system with all features enabled memory_system = await create_memory_system() # Store memories result = await memory_system.store_memory( "Alice works at TechCorp as a software engineer" ) if result.success: print(f"Memory stored: {result.result['memory_id']}") # Retrieve memories result = await memory_system.retrieve_memories( "Who works at TechCorp?", limit=5 ) if result.success: for memory in result.result["memories"]: print(f"Found: {memory['content']}")
Advanced usage with custom configuration:
config = MemorySystemConfig( store_type="postgres", collection_name="company_knowledge", default_namespace=("company", "engineering"), enable_graph_rag=True, enable_multi_agent_coordination=True, llm_config=AugLLMConfig(model="gpt-4", temperature=0.1) ) memory_system = UnifiedMemorySystem(config) # Store with metadata result = await memory_system.store_memory( content="Neural networks are effective for pattern recognition", namespace=("company", "ai", "research"), metadata={"source": "research_paper", "confidence": 0.95} ) # Advanced retrieval with filtering result = await memory_system.retrieve_memories( query="pattern recognition techniques", limit=10, memory_types=[MemoryType.SEMANTIC, MemoryType.PROCEDURAL], namespace=("company", "ai"), use_graph_rag=True )
Knowledge graph operations:
# Generate knowledge graph from stored memories kg_result = await memory_system.generate_knowledge_graph( namespace=("company", "projects") ) if kg_result.success: kg = kg_result.result["knowledge_graph"] print(f"Generated graph with {len(kg.nodes)} nodes") # Search for specific entities entity_result = await memory_system.search_entities("Alice") if entity_result.success: context = entity_result.result["entity_context"] print(f"Found entity: {context['entity'].name}") print(f"Connected to {context['total_connections']} other entities")
System monitoring and diagnostics:
# Get comprehensive statistics stats_result = await memory_system.get_memory_statistics() if stats_result.success: stats = stats_result.result print(f"Total memories: {stats['store_statistics']['total_count']}") print(f"Operations performed: {stats['system_statistics']['total_operations']}") # Run system health check diag_result = await memory_system.run_system_diagnostic() if diag_result.success: health = diag_result.result["system_health"] print(f"System health: {health}") for component, status in diag_result.result["component_diagnostics"].items(): print(f"{component}: {status['status']}")
Memory lifecycle management:
# Consolidate memories (remove duplicates, expired entries) consolidation_result = await memory_system.consolidate_memories( namespace=("user", "temp"), dry_run=True # Preview changes first ) if consolidation_result.success: result = consolidation_result.result["consolidation_result"] print(f"Would remove {result['duplicates_found']} duplicates") print(f"Would expire {result['expired_found']} old memories") # Actually perform consolidation if input("Proceed with consolidation? (y/n): ").lower() == 'y': final_result = await memory_system.consolidate_memories( namespace=("user", "temp"), dry_run=False )
Note
The UnifiedMemorySystem automatically selects the best agent for each operation based on the request type, available features, and performance considerations. Enable multi-agent coordination for the most intelligent behavior, or disable specific features for better performance in resource-constrained environments.
Initialize the unified memory system with comprehensive component setup.
Creates and configures all memory system components including stores, classifiers, knowledge graph generators, retrievers, and coordinators based on the provided configuration. All components are initialized and validated during construction.
- Parameters:
config (MemorySystemConfig) – MemorySystemConfig with all system settings and feature flags
- Raises:
ValueError – If required configuration parameters are missing or invalid
RuntimeError – If component initialization fails
Examples
Basic initialization:
config = MemorySystemConfig( store_type="memory", collection_name="my_memories" ) memory_system = UnifiedMemorySystem(config) print("System initialized successfully")
Production initialization with validation:
config = MemorySystemConfig( store_type="postgres", collection_name="prod_memories", default_namespace=("company", "prod"), enable_multi_agent_coordination=True, llm_config=AugLLMConfig(model="gpt-4") ) try: memory_system = UnifiedMemorySystem(config) # Validate initialization system_info = memory_system.get_system_info() assert system_info["initialized"] print(f"System ready with {len(system_info['components'])} components") except Exception as e: print(f"Initialization failed: {e}")
Note
Component initialization follows dependency order: store → classifier → KG generator → retrievers → coordinator. If any component fails to initialize, the entire system initialization will fail.
- async classify_memory(content, user_context=None)¶
Classify memory content.
- Parameters:
- Returns:
MemorySystemResult with classification result
- Return type:
- async consolidate_memories(namespace=None, dry_run=False)¶
Consolidate memories by removing duplicates and expired entries.
- Parameters:
- Returns:
MemorySystemResult with consolidation results
- Return type:
- async generate_knowledge_graph(namespace=None, force_regeneration=False)¶
Generate knowledge graph from memories.
- Parameters:
- Returns:
MemorySystemResult with knowledge graph
- Return type:
- async get_memory_statistics(namespace=None)¶
Get comprehensive memory statistics.
- async retrieve_memories(query, limit=10, namespace=None, memory_types=None, use_graph_rag=True, use_multi_agent=True)¶
Retrieve memories from the system.
- Parameters:
query (str) – Search query
limit (int) – Maximum number of memories to retrieve
namespace (tuple[str, Ellipsis] | None) – Memory namespace to search
memory_types (list[haive.agents.memory.core.types.MemoryType] | None) – Specific memory types to search
use_graph_rag (bool) – Whether to use graph RAG
use_multi_agent (bool) – Whether to use multi-agent coordination
- Returns:
MemorySystemResult with retrieved memories
- Return type:
- async run_system_diagnostic()¶
Run comprehensive system diagnostic.
- Returns:
MemorySystemResult with diagnostic results
- Return type:
- async search_entities(entity_name, namespace=None)¶
Search for entities in the knowledge graph.
- Parameters:
- Returns:
MemorySystemResult with entity information
- Return type:
- async store_memory(content, namespace=None, memory_type=None, importance=None, metadata=None)¶
Store a memory in the system.
- Parameters:
content (str) – Memory content to store
namespace (tuple[str, Ellipsis] | None) – Memory namespace (defaults to configured default)
memory_type (haive.agents.memory.core.types.MemoryType | None) – Force specific memory type (otherwise auto-classified)
importance (float | None) – Override importance score
- Returns:
MemorySystemResult with operation result
- Return type:
- async agents.memory.unified_memory_api.create_memory_system(store_type='memory', collection_name='haive_memories', enable_all_features=True)¶
Create a unified memory system with sensible default configuration.
This convenience function creates a UnifiedMemorySystem with commonly used settings, making it easy to get started without complex configuration.
- Parameters:
- Returns:
Fully configured and ready-to-use memory system
- Return type:
Examples
Quick start with in-memory storage:
# Create system with all features enabled memory_system = await create_memory_system() # System is ready to use immediately result = await memory_system.store_memory("Hello, world!") print(f"Stored memory: {result.success}")
Production setup with PostgreSQL:
memory_system = await create_memory_system( store_type="postgres", collection_name="company_memories", enable_all_features=True ) # Verify system health diag = await memory_system.run_system_diagnostic() print(f"System health: {diag.result['system_health']}")
Performance-focused setup:
# Disable resource-intensive features for speed memory_system = await create_memory_system( store_type="memory", collection_name="fast_cache", enable_all_features=False ) # System will use basic storage and retrieval only result = await memory_system.store_memory("Fast storage test")
Note
When enable_all_features=True, the system includes: - Automatic memory classification - Enhanced multi-strategy retrieval - Graph RAG with knowledge graph traversal - Multi-agent coordination for optimal routing
When enable_all_features=False, only basic storage and retrieval are enabled for maximum performance.
- async agents.memory.unified_memory_api.quick_memory_demo()¶
Comprehensive demonstration of the unified memory system capabilities.
This demo showcases the main features of the UnifiedMemorySystem including: - Memory storage with automatic classification - Intelligent retrieval with multiple strategies - Knowledge graph generation and analysis - System diagnostics and health monitoring - Performance metrics and statistics
Examples
Run the complete demo:
await quick_memory_demo()
Use as a template for your own integration:
# Copy relevant sections from this demo memory_system = await create_memory_system() # Store your data for item in your_data: await memory_system.store_memory(item) # Query your data result = await memory_system.retrieve_memories("your query")