Spaces:
Paused
Paused
| """Episodic memory implementation.""" | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| from datetime import UTC, datetime | |
| from pathlib import Path | |
| from typing import Any | |
| from hermes.config.settings import get_settings | |
| from hermes.core.types import EpisodicMemory | |
| logger = logging.getLogger(__name__) | |
| class EpisodicMemoryStore: | |
| """Episodic memory for storing task experiences.""" | |
| def __init__(self, storage_path: str | None = None) -> None: | |
| self.settings = get_settings() | |
| self._storage_path = Path(storage_path or "data/episodic_memory.json") | |
| self._episodes: list[EpisodicMemory] = [] | |
| self._load() | |
| def _load(self) -> None: | |
| """Load episodes from disk.""" | |
| try: | |
| if self._storage_path.exists(): | |
| data = json.loads(self._storage_path.read_text(encoding="utf-8")) | |
| self._episodes = [EpisodicMemory(**ep) for ep in data] | |
| except Exception as e: | |
| logger.warning(f"Could not load episodic memory: {e}") | |
| self._episodes = [] | |
| def _save(self) -> None: | |
| """Save episodes to disk.""" | |
| try: | |
| self._storage_path.parent.mkdir(parents=True, exist_ok=True) | |
| data = [ep.model_dump() for ep in self._episodes] | |
| self._storage_path.write_text( | |
| json.dumps(data, indent=2, default=str), encoding="utf-8" | |
| ) | |
| except Exception as e: | |
| logger.error(f"Could not save episodic memory: {e}") | |
| async def record_episode( | |
| self, | |
| task_id: str, | |
| task_description: str, | |
| actions: list[str], | |
| results: dict[str, Any], | |
| outcome: str = "", | |
| duration_seconds: float = 0.0, | |
| ) -> EpisodicMemory: | |
| """Record a new episode.""" | |
| episode = EpisodicMemory( | |
| task_id=task_id, | |
| task_description=task_description, | |
| actions=actions, | |
| results=results, | |
| outcome=outcome, | |
| duration_seconds=duration_seconds, | |
| timestamp=datetime.now(UTC), | |
| ) | |
| self._episodes.append(episode) | |
| self._save() | |
| return episode | |
| async def retrieve( | |
| self, | |
| query: str | None = None, | |
| task_id: str | None = None, | |
| limit: int = 10, | |
| ) -> list[EpisodicMemory]: | |
| """Retrieve episodes matching criteria.""" | |
| results = self._episodes | |
| if task_id: | |
| results = [ep for ep in results if ep.task_id == task_id] | |
| if query: | |
| query_lower = query.lower() | |
| results = [ | |
| ep for ep in results | |
| if query_lower in ep.task_description.lower() | |
| or any(query_lower in action.lower() for action in ep.actions) | |
| ] | |
| return sorted(results, key=lambda e: e.timestamp, reverse=True)[:limit] | |
| async def get_similar_episodes( | |
| self, task_description: str, limit: int = 5 | |
| ) -> list[EpisodicMemory]: | |
| """Get episodes similar to a task description.""" | |
| words = set(task_description.lower().split()) | |
| scored: list[tuple[EpisodicMemory, float]] = [] | |
| for ep in self._episodes: | |
| ep_words = set(ep.task_description.lower().split()) | |
| overlap = len(words & ep_words) | |
| total = len(words | ep_words) | |
| score = overlap / total if total > 0 else 0.0 | |
| scored.append((ep, score)) | |
| scored.sort(key=lambda x: x[1], reverse=True) | |
| return [ep for ep, _ in scored[:limit] if _ > 0] | |
| async def get_statistics(self) -> dict[str, Any]: | |
| """Get episodic memory statistics.""" | |
| if not self._episodes: | |
| return {"total_episodes": 0} | |
| durations = [ep.duration_seconds for ep in self._episodes] | |
| outcomes: dict[str, int] = {} | |
| for ep in self._episodes: | |
| outcome = ep.outcome or "unknown" | |
| outcomes[outcome] = outcomes.get(outcome, 0) + 1 | |
| return { | |
| "total_episodes": len(self._episodes), | |
| "avg_duration_seconds": sum(durations) / len(durations) if durations else 0, | |
| "outcomes": outcomes, | |
| "date_range": { | |
| "earliest": min(ep.timestamp for ep in self._episodes).isoformat(), | |
| "latest": max(ep.timestamp for ep in self._episodes).isoformat(), | |
| }, | |
| } | |
| async def clear(self) -> None: | |
| """Clear all episodes.""" | |
| self._episodes.clear() | |
| self._save() | |