# type: ignore """ modules/local_llm.py ================================================================================ FALLBACK LOCAL LLM - ÚLTIMA HIPÓTASE ================================================================================ Este módulo é usado SOMENTE quando TODAS as APIs externas falharem. Implementa um modelo local leve (TinyLlama ou equivalente) para respostas básicas em modo de emergência. Features: - Fallback final do sistema - Modelo pequeno (~1.5B parâmetros) - Respostas básicas em português/angolano - Não requer GPU ================================================================================ """ import os import re import time from typing import Optional, List, Dict, Any from datetime import datetime from .config import get_system_prompt # Imports opcionais com fallbacks try: import torch # type: ignore TORCH_AVAILABLE = True except Exception: TORCH_AVAILABLE = False torch = None # type: ignore import requests # type: ignore try: from huggingface_hub import hf_hub_download, InferenceClient # type: ignore HUGGINGFACE_HUB_AVAILABLE = True except Exception: HUGGINGFACE_HUB_AVAILABLE = False hf_hub_download = None InferenceClient = None try: from llama_cpp import Llama # type: ignore LLAMA_CPP_AVAILABLE = True except Exception: LLAMA_CPP_AVAILABLE = False Llama = None # type: ignore try: from loguru import logger # type: ignore LOGURU_AVAILABLE = True except Exception: LOGURU_AVAILABLE = False # Criar logger dummy class DummyLogger: def info(self, *args, **kwargs): pass def success(self, *args, **kwargs): pass def warning(self, *args, **kwargs): pass def error(self, *args, **kwargs): pass def debug(self, *args, **kwargs): pass logger = DummyLogger() # type: ignore try: from cachetools import TTLCache # type: ignore CACHETOOLS_AVAILABLE = True except Exception: CACHETOOLS_AVAILABLE = False # Implementação simples de cache fallback class TTLCache(dict): def __init__(self, maxsize=10, ttl=300, **kwargs): super().__init__(**kwargs) self.maxsize = maxsize self.ttl = ttl self._timestamps = {} def __setitem__(self, key, value): super().__setitem__(key, value) self._timestamps[key] = time.time() # Limpa itens antigos se necessário if len(self) > self.maxsize: oldest_key = min(self._timestamps.keys(), key=lambda k: self._timestamps[k]) self.pop(oldest_key, None) self._timestamps.pop(oldest_key, None) def get(self, key, default=None): # Verifica se expirou if key in self._timestamps: if time.time() - self._timestamps[key] > self.ttl: self.pop(key, None) self._timestamps.pop(key, None) return default return super().get(key, default) # Cache de prompts _prompt_cache: Any = None if CACHETOOLS_AVAILABLE: try: _prompt_cache = TTLCache(maxsize=10, ttl=300) except Exception: _prompt_cache = {} # ============================================================ # 🎯 CONFIGURAÇÕES DO FALLBACK LOCAL (GGUF via llama.cpp) # ============================================================ # Modelos locais suportados (do mais leve ao mais pesado - versão GGUF) LOCAL_LLM_MODELS = [ { "repo": "TheBloke/TinyLlama-1.1B-Chat-v1.0-GGUF", "file": "tinyllama-1.1b-chat-v1.0.Q4_K_M.gguf" # ~680MB }, { "repo": "TheBloke/phi-2-GGUF", "file": "phi-2.Q4_K_M.gguf" # ~1.7GB } ] # O prompt agora é importado de .config (get_system_prompt) # ============================================================ # 🏗️ CLASSE PRINCIPAL - LOCAL LLM FALLBACK # ============================================================ class LocalLLMFallback: """ Fallback local puro usando llama.cpp para quando TODAS as APIs externas falharem. Este motor é ultraleve consumindo menos de 1GB de RAM. IMPORTANTE: Esta classe só deve ser usada como ÚLTIMA opção. """ _instance = None _model_lock = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False import threading as _threading cls._instance._model_lock = _threading.Lock() return cls._instance def __init__(self): if self._initialized: return self._initialized = True # Componentes do modelo self._model = None # type: ignore self._model_path: Optional[str] = None self._deepseek_model: Optional[str] = None self._mistral_model: Optional[str] = None self._lexi_model: Optional[str] = None self._luna_model: Optional[str] = None self._multilingual_beast: Optional[str] = None self._is_loaded = False self._tokenizer = None # type: ignore self._pipeline = None # type: ignore # Configurações do Llama CPP / API Inference (Otimizados contra Alucinações) self._max_tokens = 1024 self._temperature = 0.85 self._top_p = 0.9 self._repetition_penalty = 1.15 self._ctx_size = 4096 self._max_consecutive_failures = 3 self._consecutive_failures = 0 self._is_hf_inference_mode = False self._hf_client = None # Estatísticas self._stats: Dict[str, Any] = { "total_calls": 0, "successful_calls": 0, "failed_calls": 0, "last_used": None, "model_loaded": False } # Tenta detectar e carregar modelo self._detect_and_load_model() def _detect_and_load_model(self) -> bool: """Configura o fallback via Cloud API (Hugging Face Inference).""" logger.info("Local LLM: Configurando fallback exclusivo via HuggingFace Cloud API.") try: import importlib as _iloc _cfgloc = _iloc.import_module('modules.config') _hf_fallback = getattr(_cfgloc, 'HF_TOKEN', None) except Exception: _hf_fallback = None hf_token: Optional[str] = os.getenv("HF_TOKEN") or _hf_fallback if hf_token: self._is_hf_inference_mode = True self._is_loaded = True # Nova Hierarquia AKIRA V21 - Usando config se disponível try: self._deepseek_model = getattr(_cfgloc, 'DEEPSEEK_MODEL', "deepseek-ai/DeepSeek-V3") self._mistral_model = getattr(_cfgloc, 'MISTRAL_MODEL_HF', "mistralai/Mistral-7B-Instruct-v0.3") except: self._deepseek_model = "deepseek-ai/DeepSeek-V3" self._mistral_model = "mistralai/Mistral-7B-Instruct-v0.3" self._lexi_model = "Orenguteng/Llama-3.1-8B-Lexi-Uncensored-V2" self._luna_model = "rhaymison/Mistral-8x7b-Quantized-portuguese-luana" self._multilingual_beast = "Qwen/Qwen2.5-72B-Instruct" self._model_path = self._deepseek_model # Default principal self._stats["model_loaded"] = True # Inicializa o cliente se possível if InferenceClient: try: self._hf_client = InferenceClient(token=hf_token) logger.success("✅ Fallback Cloud HF Inference configurado com sucesso.") except Exception as e: logger.warning(f"Erro ao inicializar InferenceClient: {e}") return True logger.error("❌ Fallback Local/Cloud indisponível: HF_TOKEN não encontrado.") return False def is_available(self) -> bool: """Verifica se o fallback está disponível (requer token ou modelo local).""" return self._is_loaded def is_operational(self) -> bool: """Verifica se o motor está pronto para gerar (Cloud ou Local).""" if getattr(self, '_is_hf_inference_mode', False): return self._is_loaded return self._is_loaded and self._model is not None def generate( self, prompt: str, system_prompt: Optional[str] = None, context_history: List[dict] = [], max_tokens: Optional[int] = None, temperature: Optional[float] = None ) -> Optional[str]: """Gera resposta usando modelo local ou nuvem HF.""" self._stats["total_calls"] += 1 max_new = max_tokens or self._max_tokens # Verifica disponibilidade if not self.is_operational(): self._stats["failed_calls"] += 1 return None # Usa cache se disponível cache_key = f"{prompt[:50]}:{system_prompt or 'default'}" if _prompt_cache is not None: cached = _prompt_cache.get(cache_key) if cached: logger.debug("Resposta encontrada em cache local") return cached try: # Prepara prompts (Centralizado em config.py) sys_prompt = system_prompt or get_system_prompt() # Formatação base compatível com a flag ChatML do Llama / TinyLlama formatted = f"<|system|>\n{sys_prompt}\n<|user|>\n{prompt}\n<|assistant|>\n" if getattr(self, '_is_hf_inference_mode', False): try: import importlib as _il2 _cfg2 = _il2.import_module('modules.config') _hf2 = getattr(_cfg2, 'HF_TOKEN', None) except Exception: _hf2 = None hf_token = os.getenv("HF_TOKEN") or _hf2 if not hf_token: logger.error("❌ Token HF não encontrado para a requisição de inferência HF") # Tentar prosseguir sem token se for modelo free (geralmente Llama 3 precisa) # Define o modelo base e as mensagens para as chamadas HF base_model = self._model_path or "Orenguteng/Llama-3.1-8B-Lexi-Uncensored-V2" messages = [] if sys_prompt: messages.append({"role": "system", "content": sys_prompt}) for turn in context_history: messages.append({ "role": turn.get("role", "user"), "content": turn.get("content", "") }) messages.append({"role": "user", "content": prompt}) # 1. HF ROUTER (Múltiplos providers) # Tentamos o modelo Lexi-V2 primeiro em todos os providers (especialmente featherless-ai) # Se falhar, tentamos o Llama-3.3-70B como fallback de alta potência no Router headers = { "Authorization": f"Bearer {hf_token}", "Content-Type": "application/json" } router_url = "https://router.huggingface.co/v1/chat/completions" # Providers disponíveis (featherless-ai é o que oficialmente suporta Lexi-V2) providers = ["featherless-ai", "nebius"] # Modelos para testar no Router (Luana/70B primeiro se for modo pesado) candidate_models = [] long_prompt = prompt.count('\n') >= 4 or len(prompt) > 800 prompt_lower = prompt.lower() # Palavras-chave exclusivas para o modelo Pesado (Qwen Abliterated: debates, polêmicas, ciência, etc) heavy_keywords = [ "debate", "discuti", "ideia", "explica", "acadêmica", "religião", "científica", "ciência", "filosófico", "filosofia", "ética", "etica", "moral", "radical", "polêmica", "polemica", "país", "analise", "fale sobre", "codigo", "programar", "política", "politi", "guerra", "mundo" ] # Palavras-chave exclusivas para Luna (Girias, cultura PT/AO/BR, informalidade) luna_keywords = [ "kkk", "mn", "informal", "foda", "lá ele", "puto", "mano", "porra", "fxe", "cassule" ] needs_heavy = long_prompt or any(x in prompt_lower for x in heavy_keywords) needs_luna = any(x in prompt_lower for x in luna_keywords) # Verifica a flag global de preferência por modelos pesados try: import importlib as _il3 _cfg3 = _il3.import_module('modules.config') prefer_heavy: bool = bool(getattr(_cfg3, 'PREFER_HEAVY_MODEL', False)) except Exception: prefer_heavy = False # Regra estrita: se for curtíssimo (ex: oi, tudo bem, hmm), NUNCA gasta o pesado palavras = len(prompt.split()) is_very_short = palavras <= 5 and not needs_heavy # 0. DEFINIR HIERARQUIA ESTRETA # 1. DeepSeek (Pesado/Padrão) -> 2. Mistral (Humano) -> 3. Lexi (Sem Censura) -> 4. Luna (Cultura) if needs_heavy and not is_very_short: # MENSAGEM COMPLEXA/LÓGICA: DeepSeek -> Mistral -> Lexi candidate_models.extend([self._deepseek_model, self._mistral_model, self._lexi_model]) elif needs_luna and not is_very_short: # MENSAGEM CULTURAL: Luna -> Mistral -> Lexi candidate_models.extend([self._luna_model, self._mistral_model, self._lexi_model]) elif "humano" in prompt_lower or "conversa" in prompt_lower: # MENSAGEM HUMANA: Mistral -> DeepSeek -> Lexi candidate_models.extend([self._mistral_model, self._deepseek_model, self._lexi_model]) else: # PADRÃO: DeepSeek como base se não for curto if is_very_short: candidate_models.extend([self._lexi_model, self._mistral_model]) else: # Hierarquia padrão solicitada: DeepSeek > Mistral > Lexi > Luna # Adicionado Qwen2.5-72B como fallback pesado de alta qualidade candidate_models.extend([ self._deepseek_model, "Qwen/Qwen2.5-72B-Instruct", self._mistral_model, self._lexi_model, self._luna_model ]) # Garantir apenas modelos únicos mantendo a ordem seen = set() candidate_models = [x for x in candidate_models if not (x in seen or seen.add(x))] for current_model in candidate_models: for provider in providers: model_with_provider = f"{current_model}:{provider}" # Ajuste dinâmico de template conforme a família do modelo current_messages = messages.copy() # Se for modelo Luana ou Mistral, aplicamos o template [INST] conforme a documentação _cm = str(current_model) if current_model else "" if "mistral" in _cm.lower() or "luana" in _cm.lower(): # Para Mistral via Chat API, geralmente o provedor já cuida da conversão, # mas podemos reforçar na primeira mensagem se necessário. # No caso da Luana específica, ela gosta do formato "Abaixo está uma instrução..." if "luana" in _cm.lower(): instruction = f"Abaixo está uma instrução que descreve uma tarefa, juntamente com uma entrada que fornece mais contexto.\nEscreva uma resposta que complete adequadamente o pedido.\n### instrução: {sys_prompt}\n### entrada: {prompt}" current_messages = [{"role": "user", "content": instruction}] # Extrair parâmetros específicos do modelo injetando agressividade e coerência try: import importlib as _il _cfg = _il.import_module('modules.config') _all_params: dict = getattr(_cfg, 'MODEL_PARAMETERS', {}) except Exception: _all_params = {} model_params: Dict[str, Any] = dict(_all_params.get(current_model, {})) payload = { "model": model_with_provider, "messages": current_messages, "max_tokens": max_tokens or model_params.get("max_tokens", max_new), "temperature": temperature or model_params.get("temperature", self._temperature), "top_p": model_params.get("top_p", self._top_p) } # Adicionar parâmetros extras se existirem para o motor HuggingFace (TGI/vLLM) for opt_param in ["top_k", "repetition_penalty", "frequency_penalty", "presence_penalty"]: if opt_param in model_params: payload[opt_param] = model_params[opt_param] try: logger.debug(f"🔁 Tentando HF Router: {model_with_provider}") resp = requests.post(router_url, headers=headers, json=payload, timeout=45) if resp.status_code == 200: data = resp.json() content = data.get("choices", [{}])[0].get("message", {}).get("content", "") if content and content.strip(): logger.success(f"✅ Sucesso via HF Router ({model_with_provider})") self._stats["last_model_used"] = current_model return self._process_successful_response(content, prompt, cache_key) elif resp.status_code == 401: auth_val = str(headers.get("Authorization", "")) token = auth_val.replace("Bearer ", "").strip() token_len = len(token) token_hint = f"{token[:5]}...{token[-2:]}" if token_len > 8 else "CURTO" try: reason = resp.json() except: reason = resp.text[:100] logger.error(f"❌ HF Router 401 (Unauthorized). Token: {token_hint} (Tam: {token_len}). Motivo: {reason}") # Token inválido ou expirado: levanta erro para api.py capturar e colocar em blacklist raise RuntimeError(f"HF Router 401 Unauthorized: {reason}") # Se o erro for de modelo não suportado por este provider, ignoramos silenciosamente no loop interno elif resp.status_code == 400: try: err_json = resp.json() err_str = str(err_json).lower() if "not supported" in err_str or "model_not_supported" in err_str: logger.debug(f"ℹ️ Provider '{provider}' não suporta {current_model}") continue logger.error(f"⚠️ Router '{provider}' rejeitou {current_model} (HTTP 400): {err_json}") except: logger.error(f"⚠️ Router '{provider}' rejeitou {current_model} (HTTP 400): {resp.text[:200]}") except (requests.exceptions.RequestException, ValueError): continue logger.error(f"❌ Todos os métodos HF falharam") self._consecutive_failures += 1 self._stats["failed_calls"] += 1 return None else: # ---------------------------------------------------- # EXECUTAR OFFLINE (GGUF CPU LLAMA.CPP) # ---------------------------------------------------- if not self._model: return None start_time = time.time() outputs = self._model( prompt=formatted, max_tokens=max_new, temperature=temperature or self._temperature, top_p=0.9, repeat_penalty=1.1, echo=False # IMPORTANT: Evita devolver o prompt na string de resposta (Semelhante ao antigo return_full_text=False) ) exec_time = time.time() - start_time logger.debug(f"[LLAMA CPP] Inferência CPU local GGUF completada em {exec_time:.2f}s") # Extrai resposta baseada no wrapper do create_completion if outputs and "choices" in outputs and len(outputs["choices"]) > 0: generated = outputs["choices"][0].get("text", "") # Garantir limpeza de possíveis sujidades de XML Chat templates response_text = self._extract_response(generated, formatted) if response_text: # Cache se disponível if _prompt_cache is not None: try: _prompt_cache[cache_key] = response_text except Exception: pass self._stats["successful_calls"] += 1 self._stats["last_used"] = datetime.now().isoformat() self._stats["last_model_used"] = "llama_local_gguf" self._consecutive_failures = 0 return response_text # Falha silenciosa self._consecutive_failures += 1 self._stats["failed_calls"] += 1 return None except Exception as e: logger.error(f"❌ Erro em fallback de emergência: {e}") self._consecutive_failures += 1 self._stats["failed_calls"] += 1 return None def _process_successful_response(self, text: str, prompt: str, cache_key: str) -> str: """Processa uma resposta bem-sucedida.""" res_text = self._extract_response(text, prompt) if _prompt_cache is not None: try: _prompt_cache[cache_key] = res_text except Exception: pass self._stats["successful_calls"] += 1 self._stats["last_used"] = datetime.now().isoformat() self._consecutive_failures = 0 return res_text def _extract_response(self, generated: str, prompt: str) -> str: """Extrai a resposta do texto gerado, removendo alucinações e metadados.""" if not generated: return "" response = generated # 1. Limpeza de tags de chat leakadas if "<|assistant|>" in response: response = response.split("<|assistant|>")[-1] elif "[/INST]" in response: response = response.split("[/INST]")[-1] elif "assistant\n" in response.lower(): parts = re.split(r'(?i)assistant\n', response) response = parts[-1] # 2. Remoção de prefixos repetitivos (Alucinações comuns do modelo) prefixes_to_strip = [ r'^### Akira ### Resposta:?\s*', r'^### Akira ###:?\s*', r'^### Resposta:?\s*', r'^Akira:?\s*', r'^🤖 AKIRA:?\s*', r'^Resposta:?\s*', r'^Assistant:?\s*' ] for pattern in prefixes_to_strip: response = re.sub(pattern, '', response, flags=re.IGNORECASE | re.MULTILINE) # 3. Se o modelo repetir o prompt do usuário no início if prompt.strip() in response[:len(prompt)+20]: response = response.replace(prompt.strip(), '', 1) return response.strip() def get_status(self) -> Dict[str, Any]: """Retorna status do fallback local.""" return { "available": self.is_available(), "operational": self.is_operational(), "model_path": self._model_path, "model_loaded": self._is_loaded, "consecutive_failures": self._consecutive_failures, "max_failures_allowed": self._max_consecutive_failures, "stats": self._stats.copy() } def reset_failures(self): """Reseta contador de falhas.""" self._consecutive_failures = 0 def should_use_fallback(self, api_failures: int = 0) -> bool: """ Decide se deve usar o fallback local. Args: api_failures: Número de falhas consecutivas de APIs Returns: True se deve usar fallback """ # Só usa se: # 1. Modelo está operacional # 2. Houve pelo menos 1 falha de API OU está explicitamente habilitado return ( self.is_operational() and (api_failures > 0 or os.getenv("USE_LOCAL_FALLBACK", "").lower() == "true") ) # ============================================================ # 🎯 FUNÇÃO PRINCIPAL DE FALLBACK # ============================================================ def get_local_fallback() -> LocalLLMFallback: """Retorna instância singleton do fallback local.""" return LocalLLMFallback() def generate_fallback_response( prompt: str, system_prompt: Optional[str] = None, api_failures: int = 0 ) -> Optional[str]: """ Gera resposta de fallback se necessário. Args: prompt: Prompt do usuário system_prompt: Prompt do sistema opcional api_failures: Número de falhas de API Returns: Resposta gerada ou None """ fallback = get_local_fallback() if fallback.should_use_fallback(api_failures): logger.info(f"🔴 Usando fallback local (API failures: {api_failures})") return fallback.generate(prompt, system_prompt) return None # ============================================================ # 🧪 MOCK PARA TESTES # ============================================================ class MockLocalLLM: """Mock para testes quando modelo não está disponível.""" def is_available(self) -> bool: return False def is_operational(self) -> bool: return False def generate(self, prompt: str, **kwargs) -> str: return "🤖 Modo de emergência: Todas as APIs falharam. Tente novamente mais tarde." def get_status(self) -> Dict[str, Any]: return {"available": False, "mock": True} # ============================================================ # 📤 EXPORTS # ============================================================ __all__ = [ "LocalLLMFallback", "get_local_fallback", "generate_fallback_response", "MockLocalLLM", "FALLBACK_SYSTEM_PROMPT", ]