MCPcopy Index your code
hub / github.com/SkyworkAI/DeepResearchAgent / _load_from_code

Method _load_from_code

src/memory/context.py:235–338  ·  view source on GitHub ↗

Load memory systems from JSON file. JSON file content example: { "metadata": { "saved_at": str, # "YYYY-MM-DD HH:MM:SS" "num_memories": int, # total memory count "num_versions": int # total version count

(self)

Source from the content-addressed store, hash-verified

233 return memory_configs
234
235 async def _load_from_code(self):
236 """Load memory systems from JSON file.
237
238 JSON file content example:
239 {
240 "metadata": {
241 "saved_at": str, # "YYYY-MM-DD HH:MM:SS"
242 "num_memories": int, # total memory count
243 "num_versions": int # total version count
244 },
245 "memory_systems": {
246 "memory_name": {
247 "current_version": "1.0.0",
248 "versions": {
249 "1.0.0": {
250 "name": str,
251 "description": str,
252 "require_grad": bool,
253 "version": str,
254 "cls": Type[Memory],
255 "config": dict,
256 "metadata": dict,
257 "code": str
258 },
259 ...
260 }
261 }
262 }
263 }
264 """
265 memory_configs: Dict[str, MemoryConfig] = {}
266
267 # If save file does not exist yet, nothing to load
268 if not os.path.exists(self.save_path):
269 logger.info(f"| 📂 Memory config file not found at {self.save_path}, skipping code-based loading")
270 return memory_configs
271
272 # Load all memory configs from json file
273 try:
274 with open(self.save_path, "r", encoding="utf-8") as f:
275 load_data = json.load(f)
276 except json.JSONDecodeError as e:
277 logger.warning(f"| ⚠️ Failed to parse memory config JSON from {self.save_path}: {e}")
278 return memory_configs
279
280 metadata = load_data.get("metadata", {})
281 memories_data = load_data.get("memory_systems", {})
282
283 async def register_memory_class(memory_name: str, memory_data: Dict[str, Any]) -> Optional[Tuple[str, Dict[str, MemoryConfig], Optional[MemoryConfig]]]:
284 """Load all versions for a single memory from JSON."""
285 try:
286 current_version = memory_data.get("current_version", "1.0.0")
287 versions = memory_data.get("versions", {})
288
289 if not versions:
290 logger.warning(f"| ⚠️ Memory {memory_name} has no versions")
291 return None
292

Callers 1

initializeMethod · 0.95

Calls 8

gather_with_concurrencyFunction · 0.90
warningMethod · 0.80
register_versionMethod · 0.80
existsMethod · 0.45
infoMethod · 0.45
loadMethod · 0.45
getMethod · 0.45
valuesMethod · 0.45

Tested by

no test coverage detected