Spaces:
Running
Running
| """ | |
| Memory Engine - Persistent Conversation Memory System | |
| ====================================================== | |
| Provides ChatGPT-level conversation memory: | |
| 1. Short-term: Current session context | |
| 2. Long-term: Persistent user preferences and facts | |
| 3. Episodic: Past conversation summaries | |
| This enables the AI to remember: | |
| - User's name and preferences | |
| - Previous questions and answers | |
| - Context from earlier in the conversation | |
| """ | |
| import json | |
| import os | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Dict, List, Optional, Any | |
| from dataclasses import dataclass, field, asdict | |
| import logging | |
| from utils.paths import get_user_paths | |
| logger = logging.getLogger(__name__) | |
| class ConversationTurn: | |
| """A single turn in a conversation""" | |
| role: str # "user" or "assistant" | |
| content: str | |
| timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) | |
| metadata: Dict[str, Any] = field(default_factory=dict) | |
| class UserMemory: | |
| """Persistent memory about a user""" | |
| user_id: str | |
| name: Optional[str] = None | |
| company: Optional[str] = None | |
| role: Optional[str] = None | |
| preferences: Dict[str, Any] = field(default_factory=dict) | |
| facts: List[str] = field(default_factory=list) | |
| last_topics: List[str] = field(default_factory=list) | |
| created_at: str = field(default_factory=lambda: datetime.now().isoformat()) | |
| updated_at: str = field(default_factory=lambda: datetime.now().isoformat()) | |
| class MemoryEngine: | |
| """ | |
| Enterprise-grade memory system for AI conversations. | |
| Features: | |
| - Short-term memory: Last N turns of current conversation | |
| - Long-term memory: Persistent user facts and preferences | |
| - Episodic memory: Summaries of past conversations | |
| - Context window management: Smart truncation for LLM limits | |
| """ | |
| def __init__(self, user_id: str, max_short_term: int = 50): | |
| self.user_id = user_id | |
| self.max_short_term = max_short_term | |
| # Get storage paths | |
| self.paths = get_user_paths(user_id) | |
| self.memory_dir = self.paths.get("memory", Path(f"storage/users/{user_id}/memory")) | |
| self.memory_dir.mkdir(parents=True, exist_ok=True) | |
| # Memory files | |
| self.conversation_file = self.memory_dir / "conversation.json" | |
| self.user_memory_file = self.memory_dir / "user_memory.json" | |
| self.episodic_file = self.memory_dir / "episodic_memory.json" | |
| # Load memories | |
| self.short_term: List[ConversationTurn] = [] | |
| self.user_memory: UserMemory = self._load_user_memory() | |
| self._load_conversation() | |
| def _load_user_memory(self) -> UserMemory: | |
| """Load persistent user memory from disk""" | |
| try: | |
| if self.user_memory_file.exists(): | |
| with open(self.user_memory_file, 'r', encoding='utf-8') as f: | |
| data = json.load(f) | |
| return UserMemory(**data) | |
| except Exception as e: | |
| logger.warning(f"Error loading user memory: {e}") | |
| return UserMemory(user_id=self.user_id) | |
| def _save_user_memory(self): | |
| """Save user memory to disk""" | |
| try: | |
| self.user_memory.updated_at = datetime.now().isoformat() | |
| with open(self.user_memory_file, 'w', encoding='utf-8') as f: | |
| json.dump(asdict(self.user_memory), f, indent=2) | |
| except Exception as e: | |
| logger.error(f"Error saving user memory: {e}") | |
| def _load_conversation(self): | |
| """Load conversation history from disk""" | |
| try: | |
| if self.conversation_file.exists(): | |
| with open(self.conversation_file, 'r', encoding='utf-8') as f: | |
| data = json.load(f) | |
| self.short_term = [ | |
| ConversationTurn(**turn) for turn in data[-self.max_short_term:] | |
| ] | |
| except Exception as e: | |
| logger.warning(f"Error loading conversation: {e}") | |
| self.short_term = [] | |
| def _save_conversation(self): | |
| """Save conversation to disk""" | |
| try: | |
| with open(self.conversation_file, 'w', encoding='utf-8') as f: | |
| json.dump([asdict(turn) for turn in self.short_term], f, indent=2) | |
| except Exception as e: | |
| logger.error(f"Error saving conversation: {e}") | |
| def add_turn(self, role: str, content: str, metadata: Dict = None): | |
| """Add a conversation turn to memory""" | |
| turn = ConversationTurn( | |
| role=role, | |
| content=content, | |
| metadata=metadata or {} | |
| ) | |
| self.short_term.append(turn) | |
| # Trim to max size | |
| if len(self.short_term) > self.max_short_term: | |
| self.short_term = self.short_term[-self.max_short_term:] | |
| # Extract user info if user message | |
| if role == "user": | |
| self._extract_user_info(content) | |
| # Save to disk | |
| self._save_conversation() | |
| def _extract_user_info(self, content: str): | |
| """Extract and save user information from messages""" | |
| import re | |
| content_lower = content.lower() | |
| # Extract name | |
| name_patterns = [ | |
| r"(?:i'?m|i am|my name is|call me|this is)\s+([A-Z][a-z]+)", | |
| r"^([A-Z][a-z]+)\s+here", | |
| ] | |
| for pattern in name_patterns: | |
| match = re.search(pattern, content, re.IGNORECASE) | |
| if match: | |
| name = match.group(1).strip() | |
| if len(name) > 1 and name.lower() not in ['the', 'a', 'an', 'i']: | |
| self.user_memory.name = name | |
| self._save_user_memory() | |
| logger.info(f"📝 Remembered user name: {name}") | |
| break | |
| # Extract company | |
| company_patterns = [ | |
| r"(?:i work at|i'm from|from|at)\s+([A-Z][A-Za-z\s&]+(?:Inc|Corp|Ltd|LLC)?)", | |
| r"(?:company|organization|firm)\s+(?:is|called)\s+([A-Z][A-Za-z\s&]+)", | |
| ] | |
| for pattern in company_patterns: | |
| match = re.search(pattern, content, re.IGNORECASE) | |
| if match: | |
| company = match.group(1).strip() | |
| if len(company) > 2: | |
| self.user_memory.company = company | |
| self._save_user_memory() | |
| break | |
| # Track topics discussed | |
| topic_keywords = { | |
| 'revenue': ['revenue', 'sales', 'income'], | |
| 'customers': ['customer', 'client', 'buyer'], | |
| 'products': ['product', 'item', 'sku'], | |
| 'trends': ['trend', 'growth', 'forecast', 'predict'], | |
| 'comparison': ['compare', 'versus', 'vs', 'difference'], | |
| 'metrics': ['kpi', 'margin', 'profit', 'roi', 'cost'], | |
| 'segments': ['region', 'country', 'state', 'market', 'segment'], | |
| } | |
| for topic, keywords in topic_keywords.items(): | |
| if any(kw in content_lower for kw in keywords): | |
| if topic not in self.user_memory.last_topics: | |
| self.user_memory.last_topics.append(topic) | |
| if len(self.user_memory.last_topics) > 10: | |
| self.user_memory.last_topics = self.user_memory.last_topics[-10:] | |
| self._save_user_memory() | |
| def get_context_string(self, max_chars: int = 4000) -> str: | |
| """Get conversation context as a string for LLM""" | |
| parts = [] | |
| # Add user memory context | |
| if self.user_memory.name: | |
| parts.append(f"User's name: {self.user_memory.name}") | |
| if self.user_memory.company: | |
| parts.append(f"User's company: {self.user_memory.company}") | |
| if self.user_memory.last_topics: | |
| parts.append(f"Recent topics: {', '.join(self.user_memory.last_topics[-5:])}") | |
| if parts: | |
| parts.append("---") | |
| # Add recent conversation | |
| parts.append("Recent conversation:") | |
| for turn in self.short_term[-10:]: | |
| role = "User" if turn.role == "user" else "Assistant" | |
| content = turn.content[:500] # Truncate long messages | |
| parts.append(f"{role}: {content}") | |
| context = "\n".join(parts) | |
| # Truncate if too long | |
| if len(context) > max_chars: | |
| context = context[-max_chars:] | |
| return context | |
| def get_last_assistant_response(self) -> Optional[str]: | |
| """Get the last assistant response for follow-up context""" | |
| for turn in reversed(self.short_term): | |
| if turn.role == "assistant": | |
| return turn.content | |
| return None | |
| def get_last_user_query(self) -> Optional[str]: | |
| """Get the last user query""" | |
| for turn in reversed(self.short_term): | |
| if turn.role == "user": | |
| return turn.content | |
| return None | |
| def remember_fact(self, fact: str): | |
| """Store a fact about the user or their data""" | |
| if fact not in self.user_memory.facts: | |
| self.user_memory.facts.append(fact) | |
| if len(self.user_memory.facts) > 50: | |
| self.user_memory.facts = self.user_memory.facts[-50:] | |
| self._save_user_memory() | |
| def get_user_name(self) -> Optional[str]: | |
| """Get the user's name if known""" | |
| return self.user_memory.name | |
| def clear_conversation(self): | |
| """Clear current conversation but keep user memory""" | |
| self.short_term = [] | |
| self._save_conversation() | |
| # Global memory cache | |
| _memory_cache: Dict[str, MemoryEngine] = {} | |
| def get_memory_engine(user_id: str) -> MemoryEngine: | |
| """Get or create memory engine for a user""" | |
| if user_id not in _memory_cache: | |
| _memory_cache[user_id] = MemoryEngine(user_id) | |
| return _memory_cache[user_id] | |
| def add_to_memory(user_id: str, role: str, content: str, metadata: Dict = None): | |
| """Quick function to add a turn to memory""" | |
| engine = get_memory_engine(user_id) | |
| engine.add_turn(role, content, metadata) | |
| def get_conversation_context(user_id: str, max_chars: int = 4000) -> str: | |
| """Quick function to get conversation context""" | |
| engine = get_memory_engine(user_id) | |
| return engine.get_context_string(max_chars) | |
| def get_last_response(user_id: str) -> Optional[str]: | |
| """Get last assistant response for a user""" | |
| engine = get_memory_engine(user_id) | |
| return engine.get_last_assistant_response() | |
| def get_user_name(user_id: str) -> Optional[str]: | |
| """Get user's name if known""" | |
| engine = get_memory_engine(user_id) | |
| return engine.get_user_name() | |
| # Shared memory singleton for chart context and cross-mode memory | |
| _shared_memory: Dict[str, Any] = {} | |
| def get_shared_memory() -> Dict[str, Any]: | |
| """ | |
| Get the shared memory dictionary for storing cross-mode data. | |
| This is used for: | |
| - Chart context storage | |
| - Cross-mode communication | |
| - Temporary data caching | |
| """ | |
| return _shared_memory | |
| def set_shared_memory(key: str, value: Any): | |
| """Set a value in shared memory""" | |
| _shared_memory[key] = value | |
| def get_shared_memory_value(key: str, default: Any = None) -> Any: | |
| """Get a value from shared memory""" | |
| return _shared_memory.get(key, default) | |
| def clear_shared_memory(): | |
| """Clear all shared memory (use with caution)""" | |
| _shared_memory.clear() | |