MCPcopy Create free account
hub / github.com/SkyworkAI/DeepResearchAgent / restore

Method restore

src/memory/context.py:957–1002  ·  view source on GitHub ↗

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)

Source from the content-addressed store, hash-verified

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.

Callers

nothing calls this directly

Calls 9

buildMethod · 0.95
save_to_jsonMethod · 0.95
MemoryConfigClass · 0.90
warningMethod · 0.80
get_version_historyMethod · 0.80
register_versionMethod · 0.80
getMethod · 0.45
model_dumpMethod · 0.45
infoMethod · 0.45

Tested by

no test coverage detected