Initialize memory systems by discovering and registering them (similar to tool). Args: memory_names: List of memory system names to initialize. If None, initialize all discovered memory systems.
(self, memory_names: Optional[List[str]] = None)
| 31 | self._registered_memories: Dict[str, MemoryConfig] = {} # memory_name -> MemoryConfig |
| 32 | |
| 33 | async def initialize(self, memory_names: Optional[List[str]] = None): |
| 34 | """Initialize memory systems by discovering and registering them (similar to tool). |
| 35 | |
| 36 | Args: |
| 37 | memory_names: List of memory system names to initialize. If None, initialize all discovered memory systems. |
| 38 | """ |
| 39 | self.base_dir = assemble_project_path(os.path.join(config.workdir, "memory")) |
| 40 | os.makedirs(self.base_dir, exist_ok=True) |
| 41 | self.save_path = os.path.join(self.base_dir, "memory.json") |
| 42 | self.contract_path = os.path.join(self.base_dir, "contract.md") |
| 43 | logger.info(f"| 📁 Memory Manager base directory: {self.base_dir} with save path: {self.save_path} and contract path: {self.contract_path}") |
| 44 | |
| 45 | # Initialize memory context manager |
| 46 | self.memory_context_manager = MemoryContextManager( |
| 47 | base_dir=self.base_dir, |
| 48 | save_path=self.save_path, |
| 49 | contract_path=self.contract_path |
| 50 | ) |
| 51 | await self.memory_context_manager.initialize(memory_names=memory_names) |
| 52 | |
| 53 | logger.info("| ✅ Memory systems initialization completed") |
| 54 | |
| 55 | async def register(self, memory: Union[Memory, Type[Memory]], *, override: bool = False, **kwargs: Any) -> MemoryConfig: |
| 56 | """Register a memory system or memory class asynchronously. |
nothing calls this directly
no test coverage detected