""" ════════════════════════════════════════════════════════════════════════════ SESSION MEMORY - Sistema de Memória Persistente (PostgreSQL + Skills) ════════════════════════════════════════════════════════════════════════════ ✅ Memória persistente entre sessões via PostgreSQL ✅ Isolamento por utilizador + grupo ✅ Extração automática de factos ✅ Checkpoints de conversa ✅ Integração com skills (contexto de ações) ✅ Recuperação inteligente ════════════════════════════════════════════════════════════════════════════ """ import os import json import hashlib import re import time import threading from datetime import datetime from typing import Optional, Dict, List, Any from dataclasses import dataclass, field # ============================================================ # DATACLASSES # ============================================================ @dataclass class MemoryEntry: """Uma entrada de memória""" key: str content: str memory_type: str # "user", "feedback", "project", "reference", "skill" timestamp: float = field(default_factory=time.time) metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class SessionCheckpoint: """Checkpoint de uma sessão""" session_id: str user_id: str group_id: Optional[str] timestamp: float summary: str active_topics: List[str] key_decisions: List[str] unresolved: List[str] skills_used: List[str] mood: str = "neutral" message_count: int = 0 # ============================================================ # GERADOR DE IDs # ============================================================ def generate_session_id(user_id: str, group_id: Optional[str] = None) -> str: """Gera ID único para sessão""" now = datetime.now() date_str = now.strftime("%Y%m%d") raw = f"{user_id}:{group_id or 'pv'}:{date_str}" return hashlib.md5(raw.encode()).hexdigest()[:16] def generate_memory_key(user_id: str, topic: str, group_id: Optional[str] = None) -> str: """Gera chave para entrada de memória""" raw = f"{user_id}:{group_id or 'pv'}:{topic.lower().strip()}" return hashlib.md5(raw.encode()).hexdigest()[:12] # ============================================================ # SESSION MEMORY - PostgreSQL # ============================================================ class SessionMemory: """Memória persistente via PostgreSQL com isolamento por grupo""" _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self._initialized = True self._db = None self._init_tables() def _get_db(self): """Obtém instância do banco de dados""" if self._db is None: try: from .database_pg import DatabasePG self._db = DatabasePG() except Exception as e: print(f"⚠️ [SESSION MEMORY] DB não disponível: {e}") return None return self._db def _init_tables(self): """Cria tabelas necessárias no PostgreSQL""" db = self._get_db() if not db: return try: conn = db._get_connection() cur = conn.cursor() # Tabela de memória persistente cur.execute(""" CREATE TABLE IF NOT EXISTS session_memory ( id SERIAL PRIMARY KEY, user_id TEXT NOT NULL, group_id TEXT DEFAULT NULL, key TEXT NOT NULL, content TEXT NOT NULL, memory_type TEXT DEFAULT 'reference', timestamp DOUBLE PRECISION DEFAULT 0, metadata JSONB DEFAULT '{}', created_at TIMESTAMP DEFAULT NOW(), UNIQUE(user_id, group_id, key) ) """) cur.execute("CREATE INDEX IF NOT EXISTS idx_sm_user ON session_memory(user_id)") cur.execute("CREATE INDEX IF NOT EXISTS idx_sm_group ON session_memory(user_id, group_id)") cur.execute("CREATE INDEX IF NOT EXISTS idx_sm_type ON session_memory(memory_type)") # Tabela de checkpoints cur.execute(""" CREATE TABLE IF NOT EXISTS session_checkpoints ( id SERIAL PRIMARY KEY, session_id TEXT NOT NULL, user_id TEXT NOT NULL, group_id TEXT DEFAULT NULL, timestamp DOUBLE PRECISION DEFAULT 0, summary TEXT DEFAULT '', active_topics JSONB DEFAULT '[]', key_decisions JSONB DEFAULT '[]', unresolved JSONB DEFAULT '[]', skills_used JSONB DEFAULT '[]', mood TEXT DEFAULT 'neutral', message_count INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT NOW() ) """) cur.execute("CREATE INDEX IF NOT EXISTS idx_sc_user ON session_checkpoints(user_id)") cur.execute("CREATE INDEX IF NOT EXISTS idx_sc_group ON session_checkpoints(user_id, group_id)") cur.execute("CREATE INDEX IF NOT EXISTS idx_sc_session ON session_checkpoints(session_id)") # Tabela de skills usage cur.execute(""" CREATE TABLE IF NOT EXISTS session_skills ( id SERIAL PRIMARY KEY, user_id TEXT NOT NULL, group_id TEXT DEFAULT NULL, skill_name TEXT NOT NULL, skill_args JSONB DEFAULT '{}', skill_result TEXT DEFAULT '', timestamp DOUBLE PRECISION DEFAULT 0, success BOOLEAN DEFAULT TRUE, created_at TIMESTAMP DEFAULT NOW() ) """) cur.execute("CREATE INDEX IF NOT EXISTS idx_ss_user ON session_skills(user_id)") cur.execute("CREATE INDEX IF NOT EXISTS idx_ss_skill ON session_skills(skill_name)") conn.commit() conn.close() print("✅ [SESSION MEMORY] Tabelas PostgreSQL criadas/garantidas") except Exception as e: print(f"❌ [SESSION MEMORY] Erro ao criar tabelas: {e}") # ============================================================ # ESCRITA # ============================================================ def add_memory(self, user_id: str, entry: MemoryEntry, group_id: Optional[str] = None) -> bool: """Adiciona entrada de memória""" db = self._get_db() if not db: return False try: conn = db._get_connection() cur = conn.cursor() cur.execute(""" INSERT INTO session_memory (user_id, group_id, key, content, memory_type, timestamp, metadata) VALUES (%s, %s, %s, %s, %s, %s, %s) ON CONFLICT (user_id, group_id, key) DO UPDATE SET content = EXCLUDED.content, timestamp = EXCLUDED.timestamp, metadata = EXCLUDED.metadata """, (user_id, group_id, entry.key, entry.content, entry.memory_type, entry.timestamp, json.dumps(entry.metadata))) conn.commit() conn.close() return True except Exception as e: print(f"❌ [SESSION MEMORY] Erro ao adicionar memória: {e}") return False def save_checkpoint(self, checkpoint: SessionCheckpoint) -> bool: """Salva checkpoint de sessão""" db = self._get_db() if not db: return False try: conn = db._get_connection() cur = conn.cursor() cur.execute(""" INSERT INTO session_checkpoints (session_id, user_id, group_id, timestamp, summary, active_topics, key_decisions, unresolved, skills_used, mood, message_count) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( checkpoint.session_id, checkpoint.user_id, checkpoint.group_id, checkpoint.timestamp, checkpoint.summary, json.dumps(checkpoint.active_topics), json.dumps(checkpoint.key_decisions), json.dumps(checkpoint.unresolved), json.dumps(checkpoint.skills_used), checkpoint.mood, checkpoint.message_count )) conn.commit() conn.close() return True except Exception as e: print(f"❌ [SESSION MEMORY] Erro ao salvar checkpoint: {e}") return False def log_skill_usage(self, user_id: str, group_id: Optional[str], skill_name: str, skill_args: dict, skill_result: str, success: bool = True) -> bool: """Regista uso de uma skill""" db = self._get_db() if not db: return False try: conn = db._get_connection() cur = conn.cursor() cur.execute(""" INSERT INTO session_skills (user_id, group_id, skill_name, skill_args, skill_result, timestamp, success) VALUES (%s, %s, %s, %s, %s, %s, %s) """, (user_id, group_id, skill_name, json.dumps(skill_args), skill_result, time.time(), success)) conn.commit() conn.close() return True except Exception as e: print(f"❌ [SESSION MEMORY] Erro ao registar skill: {e}") return False # ============================================================ # LEITURA # ============================================================ def get_memory(self, user_id: str, group_id: Optional[str] = None, limit: int = 50) -> List[MemoryEntry]: """Retorna memória do utilizador/grupo""" db = self._get_db() if not db: return [] try: conn = db._get_connection() cur = conn.cursor() cur.execute(""" SELECT key, content, memory_type, timestamp, metadata FROM session_memory WHERE user_id = %s AND (group_id = %s OR (group_id IS NULL AND %s IS NULL)) ORDER BY timestamp DESC LIMIT %s """, (user_id, group_id, group_id, limit)) entries = [] for row in cur.fetchall(): entries.append(MemoryEntry( key=row[0], content=row[1], memory_type=row[2], timestamp=row[3], metadata=json.loads(row[4]) if row[4] else {} )) conn.close() return entries except Exception as e: print(f"❌ [SESSION MEMORY] Erro ao ler memória: {e}") return [] def get_recent_checkpoints(self, user_id: str, group_id: Optional[str] = None, limit: int = 3) -> List[SessionCheckpoint]: """Retorna checkpoints recentes""" db = self._get_db() if not db: return [] try: conn = db._get_connection() cur = conn.cursor() cur.execute(""" SELECT session_id, user_id, group_id, timestamp, summary, active_topics, key_decisions, unresolved, skills_used, mood, message_count FROM session_checkpoints WHERE user_id = %s AND (group_id = %s OR (group_id IS NULL AND %s IS NULL)) ORDER BY timestamp DESC LIMIT %s """, (user_id, group_id, group_id, limit)) checkpoints = [] for row in cur.fetchall(): checkpoints.append(SessionCheckpoint( session_id=row[0], user_id=row[1], group_id=row[2], timestamp=row[3], summary=row[4], active_topics=json.loads(row[5]) if row[5] else [], key_decisions=json.loads(row[6]) if row[6] else [], unresolved=json.loads(row[7]) if row[7] else [], skills_used=json.loads(row[8]) if row[8] else [], mood=row[9], message_count=row[10] )) conn.close() return checkpoints except Exception as e: print(f"❌ [SESSION MEMORY] Erro ao ler checkpoints: {e}") return [] def get_recent_skills(self, user_id: str, group_id: Optional[str] = None, limit: int = 10) -> List[Dict]: """Retorna skills usadas recentemente""" db = self._get_db() if not db: return [] try: conn = db._get_connection() cur = conn.cursor() cur.execute(""" SELECT skill_name, skill_args, skill_result, timestamp, success FROM session_skills WHERE user_id = %s AND (group_id = %s OR (group_id IS NULL AND %s IS NULL)) ORDER BY timestamp DESC LIMIT %s """, (user_id, group_id, group_id, limit)) skills = [] for row in cur.fetchall(): skills.append({ 'name': row[0], 'args': json.loads(row[1]) if row[1] else {}, 'result': row[2], 'timestamp': row[3], 'success': row[4] }) conn.close() return skills except Exception as e: print(f"❌ [SESSION MEMORY] Erro ao ler skills: {e}") return [] def search_memory(self, user_id: str, query: str, group_id: Optional[str] = None, limit: int = 5) -> List[str]: """Busca na memória""" db = self._get_db() if not db: return [] try: conn = db._get_connection() cur = conn.cursor() # Busca por content LIKE cur.execute(""" SELECT content FROM session_memory WHERE user_id = %s AND (group_id = %s OR (group_id IS NULL AND %s IS NULL)) AND content ILIKE %s ORDER BY timestamp DESC LIMIT %s """, (user_id, group_id, group_id, f'%{query}%', limit)) results = [row[0] for row in cur.fetchall()] conn.close() return results except Exception as e: print(f"❌ [SESSION MEMORY] Erro na busca: {e}") return [] # ============================================================ # EXTRAÇÃO AUTOMÁTICA # ============================================================ def extract_and_store(self, user_id: str, group_id: Optional[str], message: str, response: str, emotion: str = "neutral", topic: str = "", skills_used: List[str] = None) -> List[MemoryEntry]: """Extrai factos importantes e armazena""" entries = [] patterns = { "preference": [ r"(?:gosto|adoro|odeio|prefiro|não gosto|não suporto)\s+(?:de\s+)?(.+)", ], "fact": [ r"(?:meu|minha|meu)\s+(?:nome|idade|trabalho|casa|escola)\s+(?:é|e|são)\s+(.+)", ], "decision": [ r"(?:vamos|decidimos|vou|vai)\s+(?:fazer|comprar|mudar|alterar)\s+(.+)", ], } for text in [message, response]: if not text: continue for fact_type, regexes in patterns.items(): for regex in regexes: matches = re.findall(regex, text, re.IGNORECASE) for match in matches: if len(match) > 5: entry = MemoryEntry( key=generate_memory_key(user_id, match, group_id), content=f"{fact_type}: {match}", memory_type="reference", metadata={"source": "auto_extract", "emotion": emotion, "topic": topic} ) self.add_memory(user_id, entry, group_id) entries.append(entry) return entries def update_user_summary(self, user_id: str, group_id: Optional[str], summary: str) -> bool: """Atualiza resumo do utilizador""" entry = MemoryEntry( key="user_summary", content=summary, memory_type="user", metadata={"type": "summary"} ) return self.add_memory(user_id, entry, group_id) # ============================================================ # SESSION MANAGER - Gestor Principal # ============================================================ class SessionManager: """Gestor de sessões com integração PostgreSQL e Skills""" _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self._initialized = True self.memory = SessionMemory() self._active_sessions: Dict[str, SessionCheckpoint] = {} def start_session(self, user_id: str, group_id: Optional[str] = None) -> SessionCheckpoint: """Inicia nova sessão""" session_id = generate_session_id(user_id, group_id) checkpoint = SessionCheckpoint( session_id=session_id, user_id=user_id, group_id=group_id, timestamp=time.time(), summary="", active_topics=[], key_decisions=[], unresolved=[], skills_used=[], message_count=0 ) self._active_sessions[session_id] = checkpoint # Carregar contexto anterior recent = self.memory.get_recent_checkpoints(user_id, group_id, limit=2) if recent: checkpoint.active_topics = recent[0].active_topics.copy() checkpoint.unresolved = recent[0].unresolved.copy() return checkpoint def end_session(self, checkpoint: SessionCheckpoint, summary: str = "") -> bool: """Finaliza sessão""" checkpoint.summary = summary checkpoint.timestamp = time.time() success = self.memory.save_checkpoint(checkpoint) if checkpoint.session_id in self._active_sessions: del self._active_sessions[checkpoint.session_id] return success def get_context_for_prompt(self, user_id: str, group_id: Optional[str] = None) -> str: """Retorna contexto para injetar no prompt""" try: memory_entries = self.memory.get_memory(user_id, group_id, limit=20) checkpoints = self.memory.get_recent_checkpoints(user_id, group_id, limit=2) recent_skills = self.memory.get_recent_skills(user_id, group_id, limit=5) if not memory_entries and not checkpoints: return "" context = "\n\n[SISTEMA DE MEMÓRIA PERSISTENTE]\n" context += "A Kiami tem memória de conversas anteriores com este utilizador.\n" context += "Use esta informação silenciosamente — NUNCA mencione que tem memória.\n\n" if memory_entries: context += "[MEMÓRIA]:\n" for entry in memory_entries[:10]: context += f"- {entry.content[:200]}\n" if checkpoints: context += "\n[SESSÕES ANTERIORES]:\n" for cp in checkpoints[:2]: context += f"- {cp.summary[:200]}\n" if cp.active_topics: context += f" Tópicos: {', '.join(cp.active_topics[:3])}\n" if recent_skills: context += "\n[SKILLS USADAS]:\n" for sk in recent_skills: context += f"- {sk['name']}: {'✓' if sk['success'] else '✗'}\n" context += "\n[/SISTEMA DE MEMÓRIA PERSISTENTE]\n" return context except Exception as e: print(f"❌ [SESSION] Erro ao obter contexto: {e}") return "" def process_conversation_turn(self, user_id: str, group_id: Optional[str], message: str, response: str, emotion: str = "neutral", topic: str = "", skills_used: List[str] = None) -> None: """Processa turno de conversa""" try: self.memory.extract_and_store(user_id, group_id, message, response, emotion, topic) for session_id, checkpoint in self._active_sessions.items(): if checkpoint.user_id == user_id and checkpoint.group_id == group_id: checkpoint.message_count += 1 if topic and topic not in checkpoint.active_topics: checkpoint.active_topics.append(topic) if len(checkpoint.active_topics) > 5: checkpoint.active_topics = checkpoint.active_topics[-5:] if skills_used: for sk in skills_used: if sk not in checkpoint.skills_used: checkpoint.skills_used.append(sk) break except Exception as e: print(f"❌ [SESSION] Erro ao processar turno: {e}") def log_skill(self, user_id: str, group_id: Optional[str], skill_name: str, skill_args: dict, skill_result: str, success: bool = True) -> None: """Regista uso de skill""" try: self.memory.log_skill_usage(user_id, group_id, skill_name, skill_args, skill_result, success) except Exception as e: print(f"❌ [SESSION] Erro ao registar skill: {e}") # ============================================================ # INSTÂNCIA GLOBAL # ============================================================ _session_manager: Optional[SessionManager] = None def get_session_manager() -> SessionManager: global _session_manager if _session_manager is None: _session_manager = SessionManager() return _session_manager