| from typing import Dict |
|
|
| from ..utils import create_object |
| from .base_memory import Memory |
|
|
|
|
| class MemoryManager: |
|
|
| def __init__(self, cfg) -> None: |
| self.cfg = cfg |
| self.memory_map: Dict[str, Memory] = {} |
|
|
| def create_instance(self, session_id): |
| self.memory_map[session_id] = create_object(self.cfg) |
|
|
| def get_memory(self, session_id=0, **kwargs) -> list: |
| return self.memory_map[session_id].get_memory(**kwargs) |
|
|
| def add(self, memory, session_id=0, **kwargs) -> None: |
| if session_id not in self.memory_map: |
| self.create_instance(session_id) |
| self.memory_map[session_id].add(memory, **kwargs) |
|
|
| def get(self, session_id=0) -> Memory: |
| return self.memory_map.get(session_id, None) |
|
|
| def reset(self, session_id=0) -> None: |
| if session_id in self.memory_map: |
| del self.memory_map[session_id] |
|
|
