File size: 1,915 Bytes
824bf31 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 |
"""Episodic memory for specific events and investigations."""
from typing import Any, Dict, List, Optional
from datetime import datetime
from .base import BaseMemory
class EpisodicMemory(BaseMemory):
"""Memory for specific investigation episodes and events."""
def __init__(self):
super().__init__()
self._episodes: List[Dict] = []
async def store(self, key: str, value: Any, metadata: Optional[Dict] = None) -> bool:
"""Store an episodic memory."""
episode = {
"key": key,
"value": value,
"metadata": metadata or {},
"timestamp": datetime.now().isoformat(),
"episode_id": len(self._episodes)
}
self._episodes.append(episode)
self._storage[key] = episode
return True
async def retrieve(self, key: str) -> Optional[Any]:
"""Retrieve an episode by key."""
episode = self._storage.get(key)
return episode["value"] if episode else None
async def search(self, query: str, limit: int = 10) -> List[Dict]:
"""Search episodes by query (stub implementation)."""
# TODO: Implement semantic search when vector DB is integrated
matching_episodes = []
query_lower = query.lower()
for episode in self._episodes[-limit:]: # Return recent episodes for now
episode_text = str(episode.get("value", "")).lower()
if query_lower in episode_text:
matching_episodes.append(episode)
return matching_episodes
async def clear(self) -> bool:
"""Clear all episodic memories."""
self._episodes.clear()
self._storage.clear()
return True
def get_recent_episodes(self, limit: int = 5) -> List[Dict]:
"""Get recent episodes."""
return self._episodes[-limit:] if self._episodes else [] |