""" ================================================================================ THINKING ENGINE - Sistema de Pensamento Profundo Pré-Processamento ================================================================================ Similar a modelos com "thinking tokens" - analisa o que foi perguntado ANTES de gerar resposta, resultando em respostas mais acertivas. Features: - Análise multi-camada da pergunta/contexto - Embeddings especializados para pensamento - Detecção de intent implícito - Complexidade da pergunta - Relacionamentos com LSTM context - Cache de pensamentos ================================================================================ """ import json from typing import Dict, Any, Optional, List from loguru import logger from sentence_transformers import SentenceTransformer, util import numpy as np class ThinkingEngine: """Processa pensamento profundo antes de responder.""" def __init__(self, db=None): """Inicializa com modelo de embedding para análise profunda.""" self.db = db self.thinking_cache = {} self.model_thinking = None self._load_thinking_model() def _load_thinking_model(self): """Carrega modelo especializado para pensamento.""" try: # Usa o modelo centralizado do config (com fallback embutido) from . import config self.model_thinking = config.get_embedding_model("all-MiniLM-L6-v2") if self.model_thinking: logger.success("✅ ThinkingEngine: Modelo de pensamento carregado via config") else: logger.warning("⚠️ ThinkingEngine: Config retornou None para o modelo") except Exception as e: logger.warning(f"⚠️ ThinkingEngine: Erro ao carregar modelo: {e}") self.model_thinking = None def think( self, mensagem: str, contexto_lstm: Optional[Dict[str, Any]] = None, historico_recente: Optional[List[str]] = None, is_group: bool = False, usuario: str = None, llm_manager: Any = None ) -> Dict[str, Any]: """ Processa pensamento profundo sobre a pergunta/contexto. Args: mensagem: Mensagem do usuário contexto_lstm: Contexto LSTM (longo prazo) historico_recente: Últimas mensagens is_group: Se é em grupo usuario: Nome do usuário llm_manager: Instância de LLMManager para CoT Dinâmico (OpenRouter) Returns: Dict com análise profunda """ if not self.model_thinking: return self._thinking_fallback(mensagem) cache_key = f"{usuario}:{mensagem[:50]}" if cache_key in self.thinking_cache: logger.debug(f"🧠 ThinkingEngine: Pensamento recuperado do cache") return self.thinking_cache[cache_key] try: thinking_result = { "depth": self._analyze_question_complexity(mensagem), "intent": self._detect_intent(mensagem), "entities": self._extract_entities(mensagem), "context_relevance": self._analyze_context_relevance(mensagem, contexto_lstm), "related_topics": self._find_related_topics(mensagem, contexto_lstm), "assumptions": self._detect_assumptions(mensagem), "required_sources": self._identify_sources(mensagem), "response_strategy": self._plan_response_strategy(mensagem, is_group), "quality_markers": self._identify_quality_markers(mensagem), } # 🧠 CoT Dinâmico: Chama o OpenRouter para raciocínio estruturado dynamic_thought = self._generate_dynamic_thought( mensagem, contexto_lstm, historico_recente, is_group, llm_manager, usuario ) if dynamic_thought: thinking_result["dynamic_thought_trace"] = dynamic_thought # Cache por 30 minutos (300 chamadas) if len(self.thinking_cache) > 1000: self.thinking_cache.clear() self.thinking_cache[cache_key] = thinking_result logger.debug(f"🧠 ThinkingEngine: Pensamento realizado (depth={thinking_result['depth']})") return thinking_result except Exception as e: logger.warning(f"⚠️ ThinkingEngine erro: {e}") return self._thinking_fallback(mensagem) def _generate_dynamic_thought( self, mensagem: str, contexto_lstm: Optional[Dict[str, Any]], historico_recente: Optional[List[str]], is_group: bool, llm_manager: Any, usuario: str = "desconhecido" ) -> Optional[str]: """Usa o OpenRouter para gerar um plano de raciocínio passo a passo.""" if not llm_manager: logger.warning("⚠️ CoT Dinâmico abortado: llm_manager é None") return None if not hasattr(llm_manager, '_call_openrouter'): logger.warning(f"⚠️ CoT Dinâmico abortado: llm_manager ({type(llm_manager)}) não tem o método '_call_openrouter'") return None try: # Constrói um contexto enxuto para não gastar muitos tokens sys_prompt = ( "Atuas como o Motor Analítico Interno da Akira V21.\n" "A tua ÚNICA tarefa é gerar um rascunho de raciocínio (plano lógico) sobre como a Akira deve responder a esta mensagem, deixa sempre claro akira que sua resposta deve ser curta e direta e séria.\n" "Reflete sobre:\n" f"1. A emoção e intenção oculta de {usuario}.\n" "2. Que factos devem ser procurados no histórico.\n" "3. Qual o tom (direto, empático, sério) a usar.\n" f"NOTA: A pessoa a falar contigo chama-se '{usuario}'. Usa o nome real na tua análise em vez de 'o utilizador'.\n" "NÃO dês a resposta final. Apenas planeia a estratégia de resposta em menos de 80 palavras ed deia sugestões de resposta pra akira usar, lembrando ela não mandar emojis. GERA O TEU PENSAMENTO EXCLUSIVAMENTE EM PORTUGUÊS." ) if is_group: sys_prompt += "\nNOTA: Isto é um ambiente de GRUPO. Sê muito conciso e evita intervir desnecessariamente." if contexto_lstm: sys_prompt += "\n\n[MEMÓRIA LONGO PRAZO (LSTM)]" if 'topic_principal' in contexto_lstm: sys_prompt += f"\n- Tópico Principal: {contexto_lstm['topic_principal']}" if 'unanswered_questions' in contexto_lstm and contexto_lstm['unanswered_questions']: sys_prompt += f"\n- Perguntas Pendentes: {', '.join(contexto_lstm['unanswered_questions'][:2])}" if 'interaction_pattern' in contexto_lstm: sys_prompt += f"\n- Padrão do Utilizador: {contexto_lstm['interaction_pattern']}" if historico_recente: sys_prompt += "\n\n[MEMÓRIA CURTO PRAZO (LISTEN)]\nÚltimas mensagens da conversa:\n" # Pega mais mensagens para entender conversas paralelas for msg in historico_recente[-15:]: if isinstance(msg, dict) and "content" in msg: sys_prompt += f"{msg['content']}\n" else: sys_prompt += f"{msg}\n" logger.info("🧠 Gerando CoT Dinâmico via OpenRouter...") # Chamada ultrarrápida usando o modelo setado no config thought = llm_manager._call_openrouter( system_prompt=sys_prompt, context_history=[], # não passamos o histórico todo para ser super rápido user_prompt=mensagem, max_tokens=150 ) return thought except Exception as e: logger.warning(f"⚠️ Erro no CoT Dinâmico (Fallback ativado): {e}") return None def _analyze_question_complexity(self, mensagem: str) -> str: """Analisa complexidade da pergunta.""" msg_lower = mensagem.lower() # Sinais de complexidade complex_markers = { "muito": 0.3, "profundo": 0.4, "explique": 0.35, "detalhe": 0.35, "por quê": 0.4, "como": 0.3, "quando": 0.25, "onde": 0.2, "comparação": 0.5, "diferença": 0.4, "relação": 0.4, "múltiplo": 0.45, "vários": 0.4, "tanto": 0.35, } score = 0.1 # Base for marker, weight in complex_markers.items(): if marker in msg_lower: score += weight # Pontuação if "?" in mensagem: score += 0.1 if "!" in mensagem: score -= 0.1 score = min(1.0, score) if score < 0.2: return "simples" elif score < 0.5: return "moderada" elif score < 0.75: return "complexa" else: return "muito_complexa" def _detect_intent(self, mensagem: str) -> List[str]: """Detecta intent(s) implícito(s).""" intents = [] msg_lower = mensagem.lower() intent_markers = { "informação": ["o que", "como", "por quê", "sabe sobre", "fala sobre", "explica"], "ação": ["faz", "cria", "envia", "modifica", "deleta", "inicia"], "opinião": ["acha", "gosta", "prefere", "ache", "pense", "achei"], "confirmação": ["certo", "verdade", "é mesmo", "sério", "confirma"], "contexto": ["em relação", "sobre isso", "quanto a", "nisso"], "humor": ["kkk", "haha", "ué", "lol", ":)", "rsrs"], } for intent, markers in intent_markers.items(): if any(m in msg_lower for m in markers): intents.append(intent) return intents or ["indefinido"] def _extract_entities(self, mensagem: str) -> List[str]: """Extrai entidades mencionadas.""" # Simples: palavras maiúsculas ou nomes comuns palavras = mensagem.split() entities = [p.strip(".,!?;:") for p in palavras if len(p) > 3 and p[0].isupper()] return entities[:5] # Top 5 def _analyze_context_relevance( self, mensagem: str, contexto_lstm: Optional[Dict[str, Any]] ) -> float: """Quanto a mensagem se relaciona com contexto de longo prazo.""" if not contexto_lstm or not self.model_thinking: return 0.0 try: topic_lstm = contexto_lstm.get("topic_principal", "") if not topic_lstm: return 0.0 # Embedding similarity emb_msg = self.model_thinking.encode(mensagem, convert_to_tensor=False) emb_topic = self.model_thinking.encode(topic_lstm, convert_to_tensor=False) relevance = float(util.cos_sim(emb_msg, emb_topic)[0][0]) return max(0.0, min(1.0, relevance)) except: return 0.0 def _find_related_topics( self, mensagem: str, contexto_lstm: Optional[Dict[str, Any]] ) -> List[str]: """Encontra tópicos relacionados no LSTM.""" if not contexto_lstm: return [] topics = [] # Topics do LSTM (se houver) if contexto_lstm.get("subtopicas"): topics.extend(contexto_lstm["subtopicas"][:3]) if contexto_lstm.get("conversation_path"): topics.extend(contexto_lstm["conversation_path"][-3:]) return topics[:5] def _detect_assumptions(self, mensagem: str) -> List[str]: """Detecta assumptions que o usuário faz.""" assumptions = [] msg_lower = mensagem.lower() # Palavras que indicam assumption if "já" in msg_lower or "não sabe" in msg_lower: assumptions.append("assume_conhecimento_anterior") if "deve" in msg_lower or "deveria" in msg_lower: assumptions.append("expectativa_de_comportamento") if "sempre" in msg_lower or "nunca" in msg_lower: assumptions.append("generalização") return assumptions def _identify_sources(self, mensagem: str) -> List[str]: """Identifica que fontes seriam úteis.""" sources = [] msg_lower = mensagem.lower() if any(w in msg_lower for w in ["notícia", "última", "recente", "novo", "2024", "2025"]): sources.append("web_search") if any(w in msg_lower for w in ["wikipedia", "história", "quem foi", "quando"]): sources.append("wikipedia") if any(w in msg_lower for w in ["preço", "dólar", "bitcoin", "crypto", "cotação"]): sources.append("market_data") if any(w in msg_lower for w in ["clima", "tempo", "previsão", "chuva"]): sources.append("weather") return sources def _plan_response_strategy(self, mensagem: str, is_group: bool) -> str: """Define estratégia de resposta.""" msg_lower = mensagem.lower() # Contexto do grupo if is_group: if any(w in msg_lower for w in ["vocês", "vcs", "todos", "@all"]): return "grupo_completo" else: return "grupo_individual" else: return "privado" def _identify_quality_markers(self, mensagem: str) -> Dict[str, bool]: """Identifica marcadores de qualidade da resposta esperada.""" return { "needs_brevity": len(mensagem) < 20, "needs_detail": len(mensagem) > 100, "needs_humor": any(m in mensagem for m in ["kk", "kkk", ":)", "rsrs"]), "formal_tone": any(w in mensagem for w in ["sr.", "sra.", "prezado"]), "technical": any(w in mensagem.lower() for w in ["código", "api", "script", "função"]), } def _thinking_fallback(self, mensagem: str) -> Dict[str, Any]: """Fallback simples quando modelo não está disponível.""" return { "depth": "moderada", "intent": ["indefinido"], "entities": [], "context_relevance": 0.5, "related_topics": [], "assumptions": [], "required_sources": [], "response_strategy": "padrão", "quality_markers": { "needs_brevity": False, "needs_detail": False, "needs_humor": False, "formal_tone": False, "technical": False, }, } # Singleton global _thinking_engine_instance: Optional[ThinkingEngine] = None def get_thinking_engine(db=None) -> ThinkingEngine: """Retorna instância singleton do ThinkingEngine.""" global _thinking_engine_instance if _thinking_engine_instance is None: _thinking_engine_instance = ThinkingEngine(db=db) return _thinking_engine_instance