# type: ignore """ ================================================================================ AKIRA V21 ULTIMATE - CONTEXT BUILDER MODULE ================================================================================ Constrói prompts otimizados para LLM combinando: - Memória de curto prazo (100 mensagens) - Contexto de reply (prioritário) - Memória vetorial (fatos aprendidos) - Contexto emocional - Sistema adaptativo baseado em tamanho da pergunta Features: - Hierarquia correta de contexto (reply > curto prazo > vetorial) - Token budgeting inteligente - Ajuste adaptativo para perguntas curtas - Suporte a múltiplos provedores LLM ================================================================================ """ import os import sys import time import json import logging from typing import Optional, Dict, Any, List, Tuple from dataclasses import dataclass # Imports robustos com fallback - CORRIGIDO para usar modules. try: from . import config from .context_isolation import ContextIsolationManager, ConversationContext from .short_term_memory import ShortTermMemory, MessageWithContext from .reply_context_handler import ReplyContextHandler, ProcessedReplyContext from .lstm_extension import get_lstm_extension CONTEXT_BUILDER_AVAILABLE = True except ImportError: try: import modules.config as config from modules.context_isolation import ContextIsolationManager, ConversationContext from modules.short_term_memory import ShortTermMemory, MessageWithContext from modules.reply_context_handler import ReplyContextHandler, ProcessedReplyContext from modules.lstm_extension import get_lstm_extension CONTEXT_BUILDER_AVAILABLE = True except ImportError: CONTEXT_BUILDER_AVAILABLE = False config = None get_lstm_extension = None # type: ignore logger = logging.getLogger(__name__) # ============================================================ # CONFIGURAÇÃO # ============================================================ # Token budgets para diferentes componentes TOKEN_BUDGET_SYSTEM: int = 1500 TOKEN_BUDGET_REPLY: int = 800 # Para contexto de reply TOKEN_BUDGET_SHORT_TERM: int = 4000 # Para memória de curto prazo TOKEN_BUDGET_VECTOR: int = 1000 # Para memória vetorial TOKEN_BUDGET_MAC: int = 300 # Para contexto de drives (MAC) TOKEN_BUDGET_TOTAL: int = 8000 # Total disponível para contexto # Limiares para perguntas curtas SHORT_QUESTION_THRESHOLD: int = 5 # palavras @dataclass class PromptBuildResult: """ Resultado da construção do prompt. Attributes: system_prompt: Prompt do sistema (sem modificação) full_prompt: Prompt completo com contexto context_sections: Seções de contexto incluídas token_counts: Contagem de tokens por seção warnings: Avisos sobre limitações should_use_vector_memory: Se deve usar memória vetorial should_prioritize_reply: Se reply deve ser priorizado """ system_prompt: str = "" full_prompt: str = "" context_sections: Dict[str, str] = None token_counts: Dict[str, int] = None warnings: List[str] = None should_use_vector_memory: bool = True should_prioritize_reply: bool = False def __post_init__(self): if self.context_sections is None: self.context_sections = {} if self.token_counts is None: self.token_counts = {} if self.warnings is None: self.warnings = [] # ============================================================ # FUNÇÕES AUXILIARES # ============================================================ def estimar_tokens(texto: str) -> int: """Estima tokens em um texto (aproximação para português).""" if not texto: return 0 # Média de 4 caracteres por token em português return max(1, len(texto) // 4) def truncar_para_tokens(texto: str, max_tokens: int) -> str: """Trunca texto para caber no limite de tokens.""" if not texto or max_tokens <= 0: return "" tokens = texto.split() if len(tokens) <= max_tokens: return texto return " ".join(tokens[:max_tokens]) def is_pergunta_curta(texto: str) -> bool: """Verifica se é uma pergunta curta.""" if not texto: return False return len(texto.split()) <= SHORT_QUESTION_THRESHOLD def calcular_peso_contexto( mensagem: str, reply_context: Optional[ProcessedReplyContext] = None ) -> float: """ Calcula peso do contexto baseado no tamanho da mensagem e reply. Args: mensagem: Mensagem do usuário reply_context: Contexto de reply (opcional) Returns: Float entre 0.5 e 1.0 representando peso do contexto geral """ word_count = len(mensagem.split()) # Pergunta muito curta = menos contexto geral necessário if word_count <= 2: return 0.5 # Pergunta curta = contexto moderado if word_count <= SHORT_QUESTION_THRESHOLD: return 0.7 # Pergunta normal = contexto completo return 1.0 def get_mac_context() -> Dict[str, Any]: """ Obtém o estado atual dos drives do sistema como contexto para o LLM. Coleta informações de disco via psutil (se disponível) incluindo: - Uso de disco em cada partição - Espaço livre / total - Mount points Returns: Dict com informações de disco, ou dict vazio se psutil não disponível. """ try: import psutil except ImportError: return {} drives = [] for partition in psutil.disk_partitions(all=False): try: usage = psutil.disk_usage(partition.mountpoint) drives.append({ "mount": partition.mountpoint, "fstype": partition.fstype, "total_gb": round(usage.total / (1024 ** 3), 1), "used_gb": round(usage.used / (1024 ** 3), 1), "free_gb": round(usage.free / (1024 ** 3), 1), "percent": round(usage.percent, 1), }) except PermissionError: continue return {"drives": drives} # ============================================================ # CLASSE PRINCIPAL # ============================================================ class ContextBuilder: """ Construtor de prompts otimizados para LLM. Hierarquia de contexto: 1. System prompt (fixo) 2. Reply context (prioritário se existir) 3. Short-term memory (100 msgs sliding window) 4. Vector memory (fatos aprendidos) 5. User message (última) Adaptação para perguntas curtas: - Pergunta curta + reply: reply tem 100%, contexto geral 50% - Pergunta curta sem reply: contexto geral 70% - Pergunta normal: contexto geral 100% """ def __init__(self, config_module=None): """ Inicializa o builder. Args: config_module: Módulo de configuração (usa config se None) """ self.config = config_module or config self.isolation_manager = None self.lstm_extension = None # ← LSTM extension (optional) self._initialized = False if CONTEXT_BUILDER_AVAILABLE: try: self.isolation_manager = ContextIsolationManager() self._initialized = True except Exception as e: logger.warning(f"ContextBuilder: falha ao init isolation: {e}") def _ensure_initialized(self): """Garante inicialização.""" if not self._initialized and CONTEXT_BUILDER_AVAILABLE: try: self.isolation_manager = ContextIsolationManager() self._initialized = True except: pass def enable_lstm(self, db: Any) -> None: """ Habilita LSTM extension com database. Args: db: Instância de Database """ if get_lstm_extension: try: self.lstm_extension = get_lstm_extension(db) logger.info(" LSTM Extension habilitado em ContextBuilder") except Exception as e: logger.debug(f"LSTM initialization: {e}") def build_prompt( self, user_message: str, conversation_id: str, system_prompt: str = None, reply_context: Optional[ProcessedReplyContext] = None, short_term_memory: Optional[ShortTermMemory] = None, vector_memory_info: Optional[List[Dict[str, Any]]] = None, emocao_atual: str = "neutral", incluir_memoria_vetorial: bool = True, max_tokens_contexto: int = TOKEN_BUDGET_TOTAL, numero_usuario: Optional[str] = None ) -> PromptBuildResult: """ Constrói prompt completo para LLM. Args: user_message: Mensagem do usuário conversation_id: ID da conversa isolada system_prompt: Prompt do sistema (usa config se None) reply_context: Contexto de reply (opcional) short_term_memory: Memória de curto prazo (opcional) vector_memory_info: Fatos da memória vetorial (opcional) emocao_atual: Emoção atual do usuário incluir_memoria_vetorial: Se deve incluir memória vetorial max_tokens_contexto: Máximo de tokens para contexto Returns: PromptBuildResult com prompt completo """ result = PromptBuildResult() # Get system prompt system_prompt = system_prompt or getattr(self.config, 'get_system_prompt', lambda: getattr(self.config, 'SYSTEM_PROMPT', ''))() result.system_prompt = system_prompt # Inicializa seções sections = { "system": system_prompt, "reply_context": "", "short_term_context": "", "vector_memory": "", "mac_context": "", "emotional_context": "", "user_message": user_message } # Contadores de tokens tokens = { "system": estimar_tokens(system_prompt), "reply": 0, "short_term": 0, "vector": 0, "mac": 0, "emotional": 0, "user": estimar_tokens(user_message) } # Remaining budget after system and user remaining_budget = max_tokens_contexto - tokens["system"] - tokens["user"] # ===== 1. REPLY CONTEXT (PRIORITÁRIO!) ===== if reply_context and reply_context.is_reply: result.should_prioritize_reply = True # Para perguntas curtas com reply, mais tokens para reply if is_pergunta_curta(user_message): reply_budget = min(TOKEN_BUDGET_REPLY * 1.5, int(remaining_budget * 0.35)) remaining_budget -= reply_budget else: reply_budget = min(TOKEN_BUDGET_REPLY, int(remaining_budget * 0.25)) remaining_budget -= reply_budget # Constrói section do reply reply_section = self._build_reply_section(reply_context, user_message) reply_section = truncar_para_tokens(reply_section, reply_budget) sections["reply_context"] = reply_section tokens["reply"] = estimar_tokens(reply_section) # ===== 2. SHORT-TERM MEMORY ===== if short_term_memory: # Calcula peso baseado em tamanho da pergunta peso_contexto = calcular_peso_contexto(user_message, reply_context) stm_budget = min( int(TOKEN_BUDGET_SHORT_TERM * peso_contexto), int(remaining_budget * 0.7) ) stm_section = self._build_short_term_section( short_term_memory, reply_context, stm_budget ) sections["short_term_context"] = stm_section tokens["short_term"] = estimar_tokens(stm_section) remaining_budget -= tokens["short_term"] # ===== 2.5 LSTM CONTEXT (Long-Term Memory) ===== # Se STM não tem suficiente contexto sobre tema, LSTM ajuda if self.lstm_extension and conversation_id and numero_usuario: try: lstm_context = self.lstm_extension.get_context_for_prompt( conversation_id, numero_usuario ) # Condição: injetar LSTM somente se o tópico LSTM for relevante ao tópico atual # (ou se o short‑term não tem tópico). Fazemos overlap simples de palavras. _topic_lstm = (lstm_context or {}).get("topic_principal", "").lower() _topic_stm = short_term_memory.get("topic_principal", "").lower() if isinstance(short_term_memory, dict) else "" _inject = False if _topic_lstm and _topic_lstm != "unknown": if not _topic_stm or _topic_stm == "unknown": _inject = True else: # Overlap > 0.4 de palavras entre tópicos _words_lstm = set([w for w in _topic_lstm.split() if len(w) > 2]) _words_stm = set([w for w in _topic_stm.split() if len(w) > 2]) _ov = len(_words_lstm & _words_stm) / max(1, len(_words_lstm)) if _ov > 0.4: _inject = True if _inject and lstm_context and lstm_context.get("topic_principal"): lstm_section = self._build_lstm_section(lstm_context) if lstm_section: sections["lstm_context"] = lstm_section except Exception as e: logger.debug(f"LSTM context error: {e}") # ===== 3. VECTOR MEMORY ===== if incluir_memoria_vetorial and vector_memory_info: vector_budget = min(TOKEN_BUDGET_VECTOR, int(remaining_budget * 0.3)) vector_section = self._build_vector_section(vector_memory_info, vector_budget) sections["vector_memory"] = vector_section tokens["vector"] = estimar_tokens(vector_section) remaining_budget -= tokens["vector"] # ===== 3.5 MAC CONTEXT (Estado dos Drives) ===== mac_data = get_mac_context() if mac_data and mac_data.get("drives"): mac_section = self._build_mac_section(mac_data) mac_budget = min(TOKEN_BUDGET_MAC, int(remaining_budget * 0.15)) mac_section = truncar_para_tokens(mac_section, mac_budget) sections["mac_context"] = mac_section tokens["mac"] = estimar_tokens(mac_section) remaining_budget -= tokens["mac"] # ===== 4. EMOTIONAL CONTEXT ===== emotional_section = self._build_emotional_section(emocao_atual) sections["emotional_context"] = emotional_section tokens["emotional"] = estimar_tokens(emotional_section) # ===== 5. MONTA PROMPT COMPLETO ===== prompt_parts = [] # System if sections["system"]: prompt_parts.append(f"[SYSTEM]\n{sections['system']}\n[/SYSTEM]\n") # Emotional context (apenas se não neutral) if sections["emotional_context"]: prompt_parts.append(f"[EMOÇÃO ATUAL]\n{sections['emotional_context']}\n") # Reply context (prioritário!) if sections["reply_context"]: prompt_parts.append(f"[REPLY PRIORITÁRIO]\n{sections['reply_context']}\n") # Short-term context if sections["short_term_context"]: prompt_parts.append(f"[CONTEXTO RECENTE]\n{sections['short_term_context']}\n") # Vector memory if sections["vector_memory"]: prompt_parts.append(f"[MEMÓRIA APRENDIDA]\n{sections['vector_memory']}\n") # MAC context (drive state) if sections["mac_context"]: prompt_parts.append(f"[SISTEMA - DISCOS]\n{sections['mac_context']}\n") # User message prompt_parts.append(f"[MENSAGEM]\n{user_message}\n") result.full_prompt = "\n".join(prompt_parts) result.context_sections = sections result.token_counts = tokens # Warnings se orçamento estourado total_tokens = sum(tokens.values()) if total_tokens > max_tokens_contexto: result.warnings.append(f"Contexto grande: {total_tokens} tokens (limite: {max_tokens_contexto})") return result def _build_reply_section( self, reply_context: ProcessedReplyContext, user_message: str ) -> str: """Constrói seção de reply priorizado.""" parts = [] # Cabeçalho de prioridade if reply_context.priority_level >= 4: # CRÍTICO parts.append(" REPLY CRÍTICO - PERGUNTA CURTA ") elif reply_context.priority_level == 3: # REPLY TO BOT parts.append(" REPLY DIRETO AO BOT") else: parts.append(" REPLY") # Autor if reply_context.reply_to_bot: parts.append("Você está sendo diretamente mencionado!") else: parts.append(f"Respondendo a: {reply_context.quoted_author_name}") # Mensagem citada if reply_context.mensagem_citada: cited = reply_context.mensagem_citada[:300] parts.append(f"\nMsg citada:\n{cited}") # Contexto hint if reply_context.context_hint and reply_context.context_hint != "contexto_geral": parts.append(f"\nContexto: {reply_context.context_hint}") return "\n".join(parts) def _build_short_term_section( self, short_term_memory: ShortTermMemory, reply_context: Optional[ProcessedReplyContext] = None, max_tokens: int = TOKEN_BUDGET_SHORT_TERM ) -> str: """Constrói seção de memória de curto prazo.""" # Obtém mensagens do contexto messages = short_term_memory.get_context_window( include_replies=True, prioritize_replies=True, max_tokens=max_tokens ) if not messages: return "" parts = [] parts.append("(últimas mensagens - replies priorizados)") # Limita a quantidade para caber no orçamento included_count = 0 current_tokens = 0 for msg in messages: msg_tokens = estimar_tokens(msg.content) if current_tokens + msg_tokens > max_tokens: break # Formata mensagem if msg.role == "assistant": role = "[Akira]" else: _author = getattr(msg, 'author_name', None) or "Usuário" role = f"[{_author}]" content_preview = msg.content[:100] + ("..." if len(msg.content) > 100 else "") if msg.is_reply: parts.append(f"{role} [REPLY] {content_preview}") else: parts.append(f"{role} {content_preview}") current_tokens += msg_tokens included_count += 1 if not parts: return "" return "\n".join(parts) def _build_vector_section( self, vector_info: List[Dict[str, Any]], max_tokens: int = TOKEN_BUDGET_VECTOR ) -> str: """Constrói seção de memória vetorial.""" if not vector_info: return "" parts = [] parts.append("(fatos aprendidos nesta conversa)") current_tokens = 0 for item in vector_info[:10]: # Limita a 10 itens text = item.get("text", "") or item.get("mensagem", "") if not text: continue text_preview = text[:80] + ("..." if len(text) > 80 else "") current_tokens += estimar_tokens(text) if current_tokens > max_tokens: break parts.append(f"• {text_preview}") if len(parts) == 1: return "" return "\n".join(parts) def _build_emotional_section(self, emocao: str) -> str: """Constrói seção de contexto emocional.""" if emocao in ["neutral", "neutro"]: return "" emocoes_descritas = { "joy": "usuário parece feliz/contento", "felicidade": "usuário parece feliz/contento", "tristeza": "usuário parece triste", "triste": "usuário parece triste", "raiva": "usuário parece irritado/raivoso", "raivoso": "usuário parece irritado/raivoso", "amor": "usuário demonstra afeto", "medo": "usuário parece preocupado/assustado", "surpresa": "usuário parece surpreso", "surpreso": "usuário parece surpreso" } descricao = emocoes_descritas.get(emocao.lower(), f"usuário parece {emocao}") return f"Tom emocional: {descricao}" def _build_lstm_section(self, lstm_context: Dict[str, Any]) -> str: """ Constrói seção de contexto LSTM (longo prazo) — maximizada. Inclui todos os campos disponíveis para dar ao LLM o máximo de contexto histórico. """ if not lstm_context or not lstm_context.get("topic_principal"): return "" parts = [] topic = lstm_context.get("topic_principal") parts.append(f"(contexto histórico verificado - tema: {topic})") parts.append(f" Tema: {topic}") # Subtópicos subtopicas = lstm_context.get("subtopicas", []) if subtopicas: parts.append(f" Subtópicos: {', '.join(str(s)[:50] for s in subtopicas[:5])}") # Caminho da conversa conv_path = lstm_context.get("conversation_path", []) if conv_path: try: parts.append(f" Evolução: {' → '.join(str(p)[:30] for p in conv_path[:5])}") except Exception: pass # Perguntas pendentes unanswered = lstm_context.get("unanswered_questions", []) if unanswered: parts.append(f" Perguntas pendentes: {', '.join(str(u)[:60] for u in unanswered[:3])}") # Conhecimento demonstrado knowledge = lstm_context.get("assumed_knowledge", []) if knowledge: try: parts.append(f" Conhecimento do utilizador: {', '.join(str(k)[:40] for k in knowledge[:4])}") except Exception: pass # Padrão pattern = lstm_context.get("interaction_pattern") if pattern: parts.append(f" Padrão: {pattern}") # Histórico emocional emotion = lstm_context.get("emotion_history") if emotion: parts.append(f" Emoção histórica: {str(emotion)[:100]}") # Resumo summary = lstm_context.get("summary") if summary: parts.append(f" Resumo: {str(summary)[:200]}") # Preferências prefs = lstm_context.get("preferences") if prefs: parts.append(f" Preferências: {str(prefs)[:150]}") return "\n".join(parts) def _build_mac_section(self, mac_data: Dict[str, Any]) -> str: """Constrói seção de contexto de discos do sistema.""" drives = mac_data.get("drives", []) if not drives: return "" parts = [] parts.append("(estado atual dos discos)") for drive in drives: mount = drive.get("mount", "?") total = drive.get("total_gb", 0) free = drive.get("free_gb", 0) pct = drive.get("percent", 0) status = " OK" if pct < 90 else (" QUASE CHEIO" if pct < 95 else " CHEIO") parts.append(f"• {mount} — {free}GB livres de {total}GB ({pct}%{status})") return "\n".join(parts) # ============================================================ # HELPERS PARA API # ============================================================ def build_history_for_llm( self, short_term_memory: ShortTermMemory, reply_context: Optional[ProcessedReplyContext] = None, max_tokens: int = TOKEN_BUDGET_SHORT_TERM ) -> List[Dict[str, str]]: """ Constrói histórico formatado para LLM. Args: short_term_memory: Memória de curto prazo reply_context: Contexto de reply (opcional) max_tokens: Máximo de tokens Returns: Lista de dicts com role e content """ # Garante que reply_context está priorizado if reply_context and reply_context.is_reply: # Cria mensagem artificial para o reply reply_entry = { "role": "user", "content": f"[REPLY] {reply_context.get_reply_summary_for_llm(reply_context)}" } # Obtém resto do histórico history = short_term_memory.get_messages_for_llm( reply_context=None, # Já adicionado max_tokens=max_tokens - estimar_tokens(reply_entry["content"]) ) # Insere reply no início return [reply_entry] + history return short_term_memory.get_messages_for_llm(max_tokens=max_tokens) def estimate_prompt_tokens( self, user_message: str, reply_context: Optional[ProcessedReplyContext] = None, historico_size: int = 0 ) -> int: """ Estima tokens totais do prompt. Args: user_message: Mensagem do usuário reply_context: Contexto de reply historico_size: Tamanho do histórico em mensagens Returns: Estimativa de tokens """ system_tokens = TOKEN_BUDGET_SYSTEM reply_tokens = 0 if reply_context and reply_context.is_reply: reply_tokens = TOKEN_BUDGET_REPLY history_tokens = historico_size * 50 # Aproximação return system_tokens + reply_tokens + history_tokens + estimar_tokens(user_message) def get_conversation_context( self, numero_usuario: str, tipo_conversa: str, grupo_id: Optional[str] = None ) -> Tuple[Optional[ConversationContext], ShortTermMemory]: """ Obtém contexto isolado e memória de curto prazo. Args: numero_usuario: Número do usuário tipo_conversa: "pv" ou "grupo" grupo_id: ID do grupo Returns: Tupla (ConversationContext, ShortTermMemory) """ self._ensure_initialized() if not self.isolation_manager: return None, ShortTermMemory() context = self.isolation_manager.get_or_create_context( numero_usuario, tipo_conversa, grupo_id ) # Carrega short-term memory do contexto stm_data = context.short_memory if context else None stm = ShortTermMemory( conversation_id=context.context_id if context else "", context_data={"messages": stm_data} if stm_data else None ) return context, stm def __repr__(self) -> str: """Representação textual.""" return f"ContextBuilder(initialized={self._initialized})" # ============================================================ # FUNÇÕES DE FÁBRICA # ============================================================ def criar_context_builder(config_module=None) -> ContextBuilder: """ Factory function para criar ContextBuilder. Args: config_module: Módulo de configuração (opcional) Returns: ContextBuilder instance """ return ContextBuilder(config_module) # type: ignore