Restore a specific version of a memory system from history Args: memory_name: Name of the memory system version: Version string to restore auto_initialize: Whether to automatically initialize the restored memory Returns:
(self, memory_name: str, version: str, auto_initialize: bool = True)
| 955 | return False |
| 956 | |
| 957 | async def restore(self, memory_name: str, version: str, auto_initialize: bool = True) -> Optional[MemoryConfig]: |
| 958 | """Restore a specific version of a memory system from history |
| 959 | |
| 960 | Args: |
| 961 | memory_name: Name of the memory system |
| 962 | version: Version string to restore |
| 963 | auto_initialize: Whether to automatically initialize the restored memory |
| 964 | |
| 965 | Returns: |
| 966 | MemoryConfig of the restored version, or None if not found |
| 967 | """ |
| 968 | # Look up version from dict-based history (O(1) lookup) |
| 969 | version_config = None |
| 970 | if memory_name in self._memory_history_versions: |
| 971 | version_config = self._memory_history_versions[memory_name].get(version) |
| 972 | |
| 973 | if version_config is None: |
| 974 | logger.warning(f"| ⚠️ Version {version} not found for memory {memory_name}") |
| 975 | return None |
| 976 | |
| 977 | # Create a copy to avoid modifying the history |
| 978 | restored_config = MemoryConfig(**version_config.model_dump()) |
| 979 | |
| 980 | # Set as current active config |
| 981 | self._memory_configs[memory_name] = restored_config |
| 982 | |
| 983 | # Update version manager current version |
| 984 | version_history = await version_manager.get_version_history("memory", memory_name) |
| 985 | if version_history: |
| 986 | # Check if version exists in version history, if not register it |
| 987 | if version not in version_history.versions: |
| 988 | await version_manager.register_version("memory", memory_name, version) |
| 989 | version_history.current_version = version |
| 990 | else: |
| 991 | # If version history doesn't exist, register the version first |
| 992 | await version_manager.register_version("memory", memory_name, version) |
| 993 | |
| 994 | # Initialize if requested |
| 995 | if auto_initialize and restored_config.cls is not None: |
| 996 | await self.build(restored_config) |
| 997 | |
| 998 | # Persist to JSON (current_version changes) |
| 999 | await self.save_to_json() |
| 1000 | |
| 1001 | logger.info(f"| 🔄 Restored memory {memory_name} to version {version}") |
| 1002 | return restored_config |
| 1003 | |
| 1004 | async def get_variables(self, memory_name: Optional[str] = None) -> Dict[str, 'Variable']: |
| 1005 | """Get variables from memory systems, where each memory's code is used as the variable value. |
nothing calls this directly
no test coverage detected