akira / modules /api.py
Isaac Quarenta
fix(kiami): concorrência independente + provedores vivos + BD silencioso
0b44440
Raw History Blame Contribute Delete
399 kB
# type: ignore
"""
API wrapper for Kiami service - irma mais velha da Akira (18).
Integração mínima e robusta: config → db → contexto → LLM → resposta.
Adaptado para KIAMI V1 SOFTEDGE com NLP 3-níveis e análise emocional BART.
Suporta WebSearch: busca na web automática e manual.
"""
import sys
import time
import re
import os
import datetime
import random
import threading
import asyncio
from typing import Dict, Optional, Any, List, Tuple, Union
from dataclasses import dataclass
from fastapi import FastAPI, APIRouter, Request as FastAPIRequest
from fastapi.responses import JSONResponse
import json
import hashlib
from loguru import logger
import contextvars
# ============================================================
# COMPATIBILITY LAYER: Flask → FastAPI
# ============================================================
# Permite que endpoints existentes usem request.get_json() e jsonify()
# sem precisar modificar cada um individualmente
_current_request: contextvars.ContextVar = contextvars.ContextVar('_current_request', default=None)
class _RequestCompat:
"""Wrapper que fornece interface Flask-like para o Request do FastAPI."""
def __init__(self, fastapi_request: FastAPIRequest):
self._req = fastapi_request
self._json_cache = None
self._body_cache = None
def get_json(self, force=True, silent=True):
if self._json_cache is None:
try:
import asyncio
loop = asyncio.get_event_loop()
if loop.is_running():
self._json_cache = {}
else:
self._json_cache = {}
except:
self._json_cache = {}
return self._json_cache
@property
def data(self):
if self._body_cache is None:
try:
import asyncio
loop = asyncio.get_event_loop()
if loop.is_running():
self._body_cache = b''
else:
self._body_cache = b''
except:
self._body_cache = b''
return self._body_cache
@property
def args(self):
return self._req.query_params if self._req else {}
class _RequestProxy:
"""Proxy que acessa o request atual via ContextVar."""
def __getattr__(self, name):
req = _current_request.get()
if req is None:
raise RuntimeError("No request context")
return getattr(req, name)
def get_json(self, **kwargs):
req = _current_request.get()
if req is None:
return {}
return req.get_json(**kwargs)
@property
def data(self):
req = _current_request.get()
if req is None:
return b''
return req.data
@property
def args(self):
req = _current_request.get()
if req is None:
return {}
return req.args
# Global request proxy (compatibility with Flask-style code)
request = _RequestProxy()
def jsonify(*args, **kwargs):
"""Wrapper que aceita tanto jsonify(dict) quanto jsonify(dict, status_code)"""
if args and isinstance(args[0], dict):
data = args[0]
status_code = kwargs.get('status_code', args[1] if len(args) > 1 else 200)
else:
data = kwargs
status_code = kwargs.pop('status_code', 200)
return JSONResponse(content=data, status_code=status_code)
# 🔒 RECURSION PROTECTION - Evita "maximum recursion depth exceeded" em processamento concorrente
# Set before any heavy imports to prevent circular dependency errors
try:
sys.setrecursionlimit(2000)
logger.info("✅ Recursion limit set to 2000 (default 1000)")
except Exception as e:
logger.warning(f"⚠️ Could not set recursion limit: {e}")
# ════════════════════════════════════════════════════════════════════
# 🎯 LISTEN ENGINE - SISTEMA DE FLAGS PARA DIFERENCIAR ESCUTA vs RESPOSTA
# ════════════════════════════════════════════════════════════════════
try:
from .listen_engine import ListenEngine, ContextoGrupoManager, MensagemMetadata
LISTEN_ENGINE_AVAILABLE = True
except ImportError:
try:
from modules.listen_engine import ListenEngine, ContextoGrupoManager, MensagemMetadata
LISTEN_ENGINE_AVAILABLE = True
except ImportError:
LISTEN_ENGINE_AVAILABLE = False
logger.warning("⚠️ listen_engine module não disponível - usando fallback")
# 🔒 LOG MASKING - PROTEÇÃO CONTRA THINK LEAK E EXPOSIÇÃO DE PROVIDER
try:
from .log_masking import SecureLogger, LogMasking
HAS_LOG_MASKING = True
except ImportError:
try:
from modules.log_masking import SecureLogger, LogMasking
HAS_LOG_MASKING = True
except ImportError:
HAS_LOG_MASKING = False
logger.warning("⚠️ log_masking module não disponível - logs públicos sem proteção")
# ═══════════════════════════════════════════════════════════════════
# 🔒 DEDUPLICATION GLOBAL + SEMÁFOROS POR CONVERSA
# Resolve o problema de mensagens duplicadas quando BotCore
# faz múltiplas chamadas simultâneas para o mesmo message_id
# ═══════════════════════════════════════════════════════════════════
# Cache em memória: {msg_hash_ou_id: timestamp} — TTL de 120s (aumentado de 30s)
_MSG_DEDUP_CACHE: Dict[str, float] = {}
_MSG_DEDUP_LOCK = threading.Lock()
_MSG_DEDUP_TTL = 120.0 # segundos (WhatsApp pode reenviar msgs com delay)
# Cache de conteúdo: {content_hash: timestamp} — para msgs sem message_id
_CONTENT_DEDUP_CACHE: Dict[str, float] = {}
_CONTENT_DEDUP_TTL = 60.0 # 1 min para dedup por conteúdo
# Semáforos por conversa (1 thread por vez por conversation_key)
_CONV_SEMAPHORES: Dict[str, threading.Semaphore] = {}
_CONV_SEM_LOCK = threading.Lock()
def _is_duplicate_message(key: str) -> bool:
"""Verifica se já processamos esta mensagem recentemente (thread-safe)."""
now = time.time()
with _MSG_DEDUP_LOCK:
# Limpa expirados
expired = [k for k, t in _MSG_DEDUP_CACHE.items() if now - t > _MSG_DEDUP_TTL]
for k in expired:
_MSG_DEDUP_CACHE.pop(k, None)
# Verifica
if key in _MSG_DEDUP_CACHE:
return True
_MSG_DEDUP_CACHE[key] = now
return False
def _is_duplicate_content(usuario: str, numero: str, mensagem: str, tipo_conversa: str) -> bool:
"""Verifica se conteúdo idêntico já foi processado recentemente (anti-retry do WhatsApp)."""
if not mensagem or len(mensagem.strip()) < 3:
return False
now = time.time()
# Hash do conteúdo normalizado
content_raw = f"{numero}:{tipo_conversa}:{mensagem.strip().lower()[:200]}"
content_hash = hashlib.md5(content_raw.encode('utf-8')).hexdigest()
with _MSG_DEDUP_LOCK:
# Limpa expirados do content cache
expired_c = [k for k, t in _CONTENT_DEDUP_CACHE.items() if now - t > _CONTENT_DEDUP_TTL]
for k in expired_c:
_CONTENT_DEDUP_CACHE.pop(k, None)
if content_hash in _CONTENT_DEDUP_CACHE:
return True
_CONTENT_DEDUP_CACHE[content_hash] = now
return False
def _get_conv_semaphore(conv_key: str) -> threading.Semaphore:
"""Retorna (ou cria) um semáforo exclusivo para a conversa."""
with _CONV_SEM_LOCK:
if conv_key not in _CONV_SEMAPHORES:
_CONV_SEMAPHORES[conv_key] = threading.Semaphore(1)
return _CONV_SEMAPHORES[conv_key]
# Per-conversation FIFO queues to serialize incoming requests when the sem
# is busy. Each item is a threading.Event that the waiter will block on.
from collections import deque
_CONV_QUEUES: Dict[str, 'collections.deque'] = {}
_CONV_QUEUE_LOCK = threading.Lock()
def validate_sender_name(name, number, ctx=''):
"""Valida e reconstrói nomes de remetente vazios."""
if name and isinstance(name, str) and name.strip() and not name.strip().isdigit():
return name.strip()
if number:
last_8 = number[-8:] if len(number) >= 8 else number
rec = f"Usuario#{last_8}"
logger.warning(f"[SENDER FIX] {ctx}: nome vazio, reconstruído: {rec}")
return rec
return "Usuario#unknown"
def extract_pure_number(id_str: str) -> str:
"""Extrai número puro de formatos como 'lid_123456' ou '123456'"""
if not id_str:
return ''
if id_str.startswith('lid_'):
return id_str[4:]
return id_str
def _enqueue_conv_request(conv_key: str):
"""Enqueue a waiter event for a conversation and return (event, position)."""
with _CONV_QUEUE_LOCK:
q = _CONV_QUEUES.get(conv_key)
if q is None:
q = deque()
_CONV_QUEUES[conv_key] = q
evt = threading.Event()
q.append(evt)
pos = len(q)
return evt, pos
def _dequeue_and_notify_next(conv_key: str):
"""Pop the current waiter and notify the next in queue, if any."""
with _CONV_QUEUE_LOCK:
q = _CONV_QUEUES.get(conv_key)
if not q:
return
try:
q.popleft()
except Exception:
pass
if q:
try:
next_evt = q[0]
next_evt.set()
except Exception:
pass
else:
# cleanup empty queue
_CONV_QUEUES.pop(conv_key, None)
# ✅ NOVA PROTEÇÃO: Rate Limiting no Servidor
class SimpleRateLimiter:
def __init__(self):
self._requests = {} # {ip: [timestamps]}
def limit(self, limit_str):
# Simplificado: 100 per hour
def decorator(f):
async def wrapper(*args, **kwargs):
# Obtém IP do request FastAPI
req = kwargs.get('request') or (args[0] if args else None)
if req and hasattr(req, 'client') and req.client:
ip = req.client.host or "unknown"
else:
ip = "unknown"
now = time.time()
if ip not in self._requests: self._requests[ip] = []
# Mantém apenas última hora
self._requests[ip] = [t for t in self._requests[ip] if now - t < 3600]
if len(self._requests[ip]) >= 100:
return JSONResponse(content={"error": "Muitas requisições. Tente em 1 hora.", "status": 429}, status_code=429)
self._requests[ip].append(now)
return await f(*args, **kwargs)
wrapper.__name__ = f.__name__
return wrapper
return decorator
# LLM PROVIDERS
import warnings
warnings.filterwarnings("ignore", category=FutureWarning)
# Google Gemini - Nova API (google.genai) com fallback para antiga
try:
from google import genai
GEMINI_USING_NEW_API = True
print(" Google GenAI API (nova)")
except ImportError:
try:
import google.generativeai as genai
GEMINI_USING_NEW_API = False
print(" Google GenerativeAI (antiga - deprecated)")
except ImportError:
genai = None
GEMINI_USING_NEW_API = False
print(" Google API não disponível")
# Mistral API via requests (sem cliente deprecated)
# LOCAL MODULES
from .contexto import Contexto
from .database import Database # ✅ Auto-seleção entre SQLite (database.py) e PostgreSQL (database_pg.py) via DATABASE_URL
from .treinamento import Treinamento
from .exemplos_naturais import ExemplosNaturais
from .local_llm import LocalLLMFallback
from .web_search import WebSearch, get_web_search, deve_pesquisar, extrair_pesquisa
from .computervision import ComputerVision, get_computer_vision, VisionConfig
from .doc_analyzer import get_document_analyzer
# ✅ NOVOS IMPORTS FASE 3 - Bot Detection, Self-Awareness
try:
from .bot_registry import bot_registry
except ImportError:
logger.warning("⚠️ bot_registry não disponível")
bot_registry = None
try:
from .self_awareness import self_awareness_engine
except ImportError:
logger.warning("⚠️ self_awareness_engine não disponível")
self_awareness_engine = None
# ✅ THINKING ENGINE - Pensamento profundo antes de responder
try:
from .thinking_engine import get_thinking_engine
except ImportError:
logger.warning("⚠️ thinking_engine não disponível")
get_thinking_engine = None
# NOVOS IMPORTS DE AGENTE (Skills)
from .skills_registry import registry
from .skills_library import initialize_skills
initialize_skills() # Garante registro das ferramentas
# ═══ AUTONOMOUS AGENT: Motor de decisão autónoma ═══
try:
from .skills.autonomous_agent import autonomous_agent as _autonomous_agent
except ImportError:
_autonomous_agent = None
# NOVOS IMPORTS DE CONTEXTO — todos defensivos para nunca causar ImportError crítico
from . import config
from .mistral_rotation import get_mistral_rotation
from .openrouter_rotation import get_openrouter_rotation
from .torouter_rotation import get_torouter_rotation
from .cerebras_rotation import get_cerebras_rotation
from .nvidia_rotation import get_nvidia_rotation
from .hf_inference_rotation import get_hf_inference_rotation
try:
from .context_isolation import ContextIsolationManager, generate_context_id
except ImportError:
class ContextIsolationManager: # type: ignore
def __init__(self, **kw): pass
def get_conversation_id(self, *a, **kw): return "temp"
def generate_context_id(*a, **kw): return "temp"
# ✅ MCP INTEGRATION + LIGHTWEIGHT TOOL USE
try:
from .mcp_integration import get_mcp_catalog, get_mcp_client
HAS_MCP = True
except ImportError:
logger.warning("⚠️ mcp_integration não disponível - MCP desabilitado")
HAS_MCP = False
def get_mcp_catalog(): return None
def get_mcp_client(): return None
try:
from .tool_use_handler import get_tool_use_handler, get_claude_executor, ToolUseRequest
HAS_TOOL_USE = True
except ImportError:
logger.warning("⚠️ tool_use_handler não disponível - Tool Use desabilitado")
HAS_TOOL_USE = False
def get_tool_use_handler(mcp_client=None): return None
def get_claude_executor(api_key=None): return None
class ToolUseRequest:
def __init__(self, **kw): pass
try:
# ShortTermMemoryManager existe em unified_context.py (class real)
# e como alias em short_term_memory.py
from .unified_context import ShortTermMemoryManager
except ImportError:
try:
from .short_term_memory import ShortTermMemory as ShortTermMemoryManager # type: ignore
except ImportError:
class ShortTermMemoryManager: # type: ignore
def __init__(self, **kw): pass
try:
from .improved_context_handler import get_context_handler, ImprovedContextHandler, ContextWeights, QuestionAnalysis
except ImportError:
@dataclass
class ContextWeights:
reply_context: float = 0.2
quoted_analysis: float = 0.2
short_term_memory: float = 1.5
vector_memory: float = 1.0
def to_dict(self): return {}
@dataclass
class QuestionAnalysis:
is_short: bool = False
is_very_short: bool = False
has_pronoun: bool = False
has_reply: bool = False
needs_context: bool = False
question_type: str = "general"
class ImprovedContextHandler:
def __init__(self, **kw): pass
def analyze_question(self, *a, **kw): return QuestionAnalysis()
def calculate_context_weights(self, *a, **kw): return ContextWeights()
def get_context_handler():
return ImprovedContextHandler()
try:
# unified_context.py tem: UnifiedContextBuilder (builder principal),
# UnifiedMessageContext (dataclass de resultado), ShortTermMemoryManager
from .unified_context import (
UnifiedContextBuilder,
UnifiedMessageContext as ProcessedUnifiedContext,
build_unified_context,
get_unified_context_builder,
get_stm_manager,
)
except ImportError:
@dataclass
class UnifiedMessageContext:
conversation_id: str = ""
reply_priority: int = 2
def to_dict(self): return {}
class UnifiedContextBuilder:
def __init__(self, **kw): pass
def build(self, **kw): return UnifiedMessageContext()
def add_to_stm(self, *a, **kw): pass
ProcessedUnifiedContext = UnifiedMessageContext
def get_stm_manager():
class DummySTM:
def get_summary(self, *a, **kw): return {}
def get_context(self, *a, **kw): return []
return DummySTM()
# ============================================================
# SESSION MEMORY - Memória persistente entre sessões
# ============================================================
try:
from .session_memory import get_session_manager, generate_session_id
SESSION_MEMORY_AVAILABLE = True
except ImportError:
SESSION_MEMORY_AVAILABLE = False
def get_session_manager():
class DummySessionManager:
def start_session(self, user_id, group_id=None): return None
def end_session(self, *a, **kw): return False
def get_context_for_prompt(self, user_id, group_id=None): return ""
def process_conversation_turn(self, *a, **kw): pass
def log_skill(self, *a, **kw): pass
return DummySessionManager()
def get_unified_context_builder():
return UnifiedContextBuilder()
def build_unified_context(**kw):
return UnifiedMessageContext()
try:
from .persona_tracker import PersonaTracker
except ImportError:
class PersonaTracker: # type: ignore
def __init__(self, **kw): pass
########################################################
# (Rest of LLMManager class exists here, omitted for brevity, but I need to replace at lines 441-463)
# Let's target lines 441-460 for KiamiAPI __init__ instead.
class LLMManager:
"""Gerenciador de múltiplos provedores LLM."""
# FIX concorrência: blacklist temporária PARTILHADA entre instâncias.
# Antes era por-instância — cada request criava o seu LLMManager e
# queimava segundos nos mesmos provedores mortos (403/404/429), e sob
# 2 mensagens simultâneas ambas batiam no mesmo limite (Cohere 429)
# e caíam em "Servidor ocupado". Agora 1 request queima = todos saltam.
_shared_temp_blacklist: Dict[str, tuple] = {}
def __init__(self, config_instance):
self.config = config_instance
self.mistral_client: Any = None
self.mistral_rotation: Any = None
self.gemini_client: Any = None # Nova API google.genai
self.gemini_model: Any = None # API antiga google.generativeai
self.groq_client: Any = None
self.grok_client: Any = None
self.cohere_client: Any = None
self.together_client: Any = None
self.openrouter_client: Any = None
self.torouter_client: Any = None
self.cerebras_client: Any = None # 🧠 Novo: Cerebras com rotação
self.nvidia_client: Any = None # 🟢 Novo: Nvidia NIM com rotação
self.hf_inference_client: Any = None # 🤗 Novo: HF Inference com rotação
self.llama_llm = self._import_llama()
self.gemini_model_name = getattr(config, "GEMINI_MODEL", "gemini-3.5-flash-lite")
self.grok_model = getattr(config, "GROK_MODEL", "grok-3")
self.together_model = getattr(config, "TOGETHER_MODEL", "meta-llama/Llama-3-70b-chat-hf")
self.prefer_heavy = getattr(config, "PREFER_HEAVY_MODEL", True)
self._current_context = []
self._current_system = ""
self._setup_providers()
self.providers = []
# ORDEM DE PRIORIDADE DAS APIs
# FIX 2026-09-17:
# - OpenRouter (rotation 5 contas) → tenta todas antes de desistir
# - Cerebras (rotation ann/isaac/netflux/gitakira) → rotaciona em 429 E 402
# - Mistral (rotation primary/softedge/mkultra) → REATIVADO; 403 dá
# blacklist curto (60s) + rotação, pois o user atualizou as chaves.
# 403 = chave inválida naquela conta, não motivo para remover da chain.
# - Nvidia NIM (rotation ann/isaac/gitbelmira/generic): CoT usa o modelo
# PESADO free, resposta final usa o MÉDIO (ambos fixos no config).
# - Gemini: 403 PERMISSION_DENIED permanente (projeto negado no Google
# Cloud) → fora da chain até nova key + restart.
if self.openrouter_client:
self.providers.append('openrouter')
if self.mistral_client:
self.providers.append('mistral')
if self.nvidia_client:
self.providers.append('nvidia')
# 🚨 ToRouter foi REMOVIDO da chain - plataforma em encerramento (Shut Down em 21/05/2026)
# if self.torouter_client:
# self.providers.append('torouter')
if self.llama_llm is not None and getattr(self.llama_llm, 'is_available', lambda: False)():
self.providers.append('llama')
if self.groq_client:
self.providers.append('groq')
if self.grok_client:
self.providers.append('grok')
if self.hf_inference_client: # 🤗 Novo: HF Inference
self.providers.append('hf_inference')
if self.cohere_client:
self.providers.append('cohere')
if self.together_client:
self.providers.append('together')
# ⚠️ Cerebras MOVIDO para o FIM: sem free tier ($5 trial expira 30d), 402 sempre
if self.cerebras_client:
self.providers.append('cerebras')
# Gemini REMOVIDO: 403 PERMISSION_DENIED permanente (projeto negado no Google Cloud)
# if self.gemini_client or self.gemini_model:
# self.providers.append('gemini')
if not self.providers:
logger.error("❌ NENHUM provedor LLM ativo. Por favor defina pelo menos MISTRAL_API_KEY ou HF_TOKEN nos Secrets.")
else:
logger.info(f"✅ Provedores ativos na chain: {self.providers}")
# Log de diagnóstico para chaves vazias ou inválidas
missing_keys = []
if not (config.MISTRAL_API_KEY or getattr(config, 'SOFTEDGE_MISTRAL_API', None) or getattr(config, 'MKULTRA_MISTRAL_KEY', None)):
missing_keys.append("MISTRAL_API_KEY or softedge_mistral_api or mkultra_mistral_key")
if not config.GROQ_API_KEY: missing_keys.append("GROQ_API_KEY")
if not config.GEMINI_API_KEY: missing_keys.append("GEMINI_API_KEY")
if not config.HF_TOKEN: missing_keys.append("HF_TOKEN")
if missing_keys:
logger.warning(f"⚠️ Chaves não encontradas nos Secrets (Causas de Erros 401/400): {', '.join(missing_keys)}")
# Blacklist de provedores (erros fatais 401/400)
self.blacklisted_providers = set()
# Blacklist temporária (429 Rate Limit) - {provider: (timestamp_expiry, reason)}
# Partilhada (classe) para valer entre requests/instâncias — ver _shared_temp_blacklist.
self.temp_blacklisted_providers = LLMManager._shared_temp_blacklist
def _import_llama(self):
try:
return LocalLLMFallback()
except Exception as e:
logger.warning(f"Llama local não disponível: {e}")
return None
def _setup_providers(self):
self._setup_openrouter()
self._setup_torouter()
self._setup_cerebras() # 🎯 Novo: Setup Cerebras
self._setup_nvidia() # 🟢 Novo: Setup Nvidia NIM
self._setup_hf_inference() # 🤗 Novo: Setup HF Inference
logger.info("🔧 [INIT] Providers intermediários...")
self._setup_mistral()
logger.info("🔧 [INIT] Mistral OK")
self._setup_gemini()
logger.info("🔧 [INIT] Gemini OK")
self._setup_groq()
logger.info("🔧 [INIT] Groq OK")
self._setup_grok()
logger.info("🔧 [INIT] Grok OK")
self._setup_cohere()
logger.info("🔧 [INIT] Cohere OK")
self._setup_together()
logger.info("🔧 [INIT] Together OK")
def _setup_openrouter(self):
# FIX 2026-09-17: usar rotation (5 contas) em vez de single OPENROUTER_API_KEY.
# Antes: só OPENROUTER_API_KEY era usada — as chaves novas
# (GITBELMIRA/SANDEOBRAS/SOFTEDGE/JOSELENA/FUGAKUSAYO) eram carregadas
# no rotation mas NUNCA usadas pelo client. Hardcode removido.
candidate_keys = []
try:
_single = (getattr(self.config, 'OPENROUTER_API_KEY', '') or '').strip()
if _single:
candidate_keys.append(_single)
except Exception:
pass
try:
_rot = get_openrouter_rotation()
for _k in (getattr(_rot, 'api_keys', []) or []):
if _k and _k.strip() and _k.strip() not in candidate_keys:
candidate_keys.append(_k.strip())
except Exception as _re:
logger.debug(f"OpenRouter rotation indisponível no setup: {_re}")
api_key = candidate_keys[0] if candidate_keys else ''
if api_key and len(api_key) > 5:
try:
import openai
import httpx
self.openrouter_client = openai.OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=api_key,
timeout=httpx.Timeout(30.0, connect=8.0),
max_retries=0,
)
logger.info(f"OpenRouter OK ({len(candidate_keys)} chave(s) disponível(is) via rotation+single)")
except Exception as e:
logger.warning(f"OpenRouter falhou: {e}")
self.openrouter_client = None
else:
logger.warning("⚠️ OpenRouter não configurado: nenhuma chave encontrada (OPENROUTER_API_KEY ou *_OPENROUTER_API)")
self.openrouter_client = None
def _setup_torouter(self):
# 🚨 IMPORTANTE: ToRouter está sendo encerrado (Shut Down 21/05/2026)
# Função mantida por compatibilidade, mas cliente não é ativado
logger.warning("🚨 [TOROUTER DEPRECADO] ToRouter está em process de encerramento. Removido da chain de provedores.")
self.torouter_client = None
return
def _setup_cerebras(self):
# 🧠 Cerebras com rotação de múltiplas contas
try:
rotation = get_cerebras_rotation()
if rotation.account_names:
# Cerebras usa OpenAI SDK com base_url customizado
import openai
current_key = rotation.get_current_api_key()
current_name = rotation.get_current_account_name()
if current_key:
self.cerebras_client = openai.OpenAI(
api_key=current_key,
base_url="https://api.cerebras.ai/v1",
timeout=30.0,
max_retries=0,
)
logger.info(f"✅ Cerebras OK (rotação multi-conta ativa, atual: {current_name})")
else:
logger.warning("⚠️ Cerebras: Nenhuma conta com API key válida")
self.cerebras_client = None
else:
logger.warning("⚠️ Cerebras não configurado: Nenhuma conta encontrada")
self.cerebras_client = None
except Exception as e:
logger.warning(f"Cerebras falhou: {e}")
self.cerebras_client = None
def _setup_nvidia(self):
# 🟢 Nvidia NIM com rotação de múltiplas contas (OpenAI-compatible)
try:
rotation = get_nvidia_rotation()
if rotation.account_names:
import openai
current_key = rotation.get_current_api_key()
current_name = rotation.get_current_account_name()
if current_key:
self.nvidia_client = openai.OpenAI(
api_key=current_key,
base_url=getattr(self.config, 'NVIDIA_BASE_URL', 'https://integrate.api.nvidia.com/v1'),
timeout=30.0,
max_retries=1,
)
logger.info(f"✅ Nvidia OK (rotação multi-conta ativa, atual: {current_name})")
else:
logger.warning("⚠️ Nvidia: Nenhuma conta com API key válida")
self.nvidia_client = None
else:
logger.warning("⚠️ Nvidia não configurado: Nenhuma conta encontrada")
self.nvidia_client = None
except Exception as e:
logger.warning(f"Nvidia falhou: {e}")
self.nvidia_client = None
def _setup_hf_inference(self):
# 🤗 HF Inference com rotação de múltiplas contas
# NOTA: InferenceClient é criado lazy (sob demanda) porque o construtor
# pode bloquear em ambientes com restrição de rede (HuggingFace Spaces).
try:
rotation = get_hf_inference_rotation()
configured_accounts = [
acc for acc in rotation.account_order
if os.getenv(rotation.accounts[acc])
]
logger.info(f"🔧 [INIT] HF configured_accounts: {configured_accounts}")
if configured_accounts:
# Ler token diretamente do env (evita loop infinito no rotation.get_current_api_token)
first_acc = configured_accounts[0]
current_token = os.getenv(rotation.accounts[first_acc])
current_name = first_acc
logger.info(f"🔧 [INIT] HF token={'YES' if current_token else 'NO'}, name={current_name}")
if current_token:
self._hf_token = current_token
self._hf_name = current_name
self._hf_accounts_count = len(configured_accounts)
self.hf_inference_client = "lazy"
logger.info(
f"✅ HF Inference OK (rotação multi-conta ativa, atual: {current_name}, "
f"{len(configured_accounts)} contas disponíveis) [client lazy]"
)
else:
logger.warning("⚠️ HF Inference: Nenhuma conta com token válido")
self.hf_inference_client = None
else:
logger.warning("⚠️ HF Inference não configurado: Nenhuma conta encontrada")
self.hf_inference_client = None
except Exception as e:
logger.warning(f"HF Inference falhou: {e}")
self.hf_inference_client = None
def _setup_mistral(self):
# 1. Mistral (via API Key em config ou múltiplas chaves para rotação)
self.mistral_rotation = get_mistral_rotation(config)
if self.mistral_rotation:
self.mistral_client = True
current_name = self.mistral_rotation.get_current_account_name()
logger.info(
f"Módulo Mistral (Direct API) ativo com rotação. Conta atual: {current_name}"
)
return
if hasattr(config, "MISTRAL_API_KEY") and config.MISTRAL_API_KEY:
self.mistral_client = True
logger.info("Módulo Mistral (Direct API) ativo com chave única.")
def _setup_gemini(self):
# 2. Google Gemini
if genai:
try:
# Prioriza a chave do config que já limpamos
gemini_key = getattr(config, "GEMINI_API_KEY", None)
model_name = getattr(config, "GEMINI_MODEL", "gemini-3.5-flash-lite")
if gemini_key:
# Resolve conflito de variáveis de ambiente do SDK
# O SDK do Google prioriza GOOGLE_API_KEY. Se queremos usar a GEMINI_API_KEY do config,
# limpamos a do ambiente para garantir consistência.
if os.getenv("GOOGLE_API_KEY") != gemini_key:
os.environ["GOOGLE_API_KEY"] = gemini_key
if GEMINI_USING_NEW_API:
self.gemini_client = genai.Client(api_key=gemini_key)
logger.info(f"Google Gemini (Novo) ativo: {model_name}")
else:
genai.configure(api_key=gemini_key)
self.gemini_model = genai.GenerativeModel(model_name)
logger.info(f"Google Gemini (Legado) ativo: {model_name}")
else:
logger.warning("Gemini não configurado: Chave ausente")
except Exception as e:
logger.error(f"Erro ao configurar Gemini: {e}")
self.gemini_model = None
self.gemini_client = None
def _setup_groq(self):
api_key = getattr(self.config, 'GROQ_API_KEY', '')
if api_key and len(api_key) > 5:
try:
from groq import Groq
self.groq_client = Groq(api_key=api_key)
logger.info("Groq OK")
except Exception as e:
logger.warning(f"Groq falhou: {e}")
self.groq_client = None
def _setup_grok(self):
"""Configura Grok API (xAI)"""
api_key = getattr(self.config, 'GROK_API_KEY', '')
if api_key and len(api_key) > 5:
try:
import openai
self.grok_client = openai.OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
self.grok_model = getattr(self.config, 'GROK_MODEL', 'grok-3')
logger.info(f"Grok OK (modelo: {self.grok_model})")
except Exception as e:
logger.warning(f"Grok falhou: {e}")
self.grok_client = None
def _setup_cohere(self):
api_key = getattr(self.config, 'COHERE_API_KEY', '')
if api_key and len(api_key) > 5:
try:
from cohere import Client
self.cohere_client = Client(api_key=api_key)
logger.info("Cohere OK")
except Exception as e:
logger.warning(f"Cohere falhou: {e}")
self.cohere_client = None
def _setup_together(self):
api_key = getattr(self.config, 'TOGETHER_API_KEY', '')
if api_key and len(api_key) > 5:
try:
import openai
self.together_client = openai.OpenAI(api_key=api_key, base_url="https://api.together.xyz/v1")
logger.info("Together AI OK")
except Exception as e:
logger.warning(f"Together AI falhou: {e}")
self.together_client = None
def generate(self, user_prompt: str, context_history: List[dict] = [], is_privileged: bool = False, tools: Optional[List[Dict[str, Any]]] = None) -> Tuple[Union[str, Dict[str, Any]], str]:
"""
Gera resposta usando provedores LLM com fallback em loop e suporte a tools.
⚠️ PROMPT-BASED PREVENTION: Todas as proteções contra vazamento são implementadas no system prompt.
Sem limpeza manual - a geração é prevenida na fonte via instruções do sistema.
"""
full_system = getattr(self.config, 'get_system_prompt', lambda: getattr(self.config, 'SYSTEM_PROMPT', ''))()
# ── TRUNCAGEM PREVENTIVA ──────────────────────────────────────────────────
MAX_USER_CHARS = 100000
if len(user_prompt) > MAX_USER_CHARS:
user_prompt = user_prompt[:MAX_USER_CHARS] + "\n[...]"
logger.warning(f"⚠️ Prompt do usuário muito longo, truncado para {MAX_USER_CHARS} chars.")
self._current_context = context_history
self._current_system = full_system
# Removida a prioridade forçada de Gemini para ferramentas para respeitar a ordem de providers definida no __init__
# O loop normal abaixo já trata tool_calls para Groq, Mistral e Gemini.
MAX_ROUNDS = 2
provider_callers = {
'openrouter': lambda m: self._call_openrouter(full_system, context_history, user_prompt, max_tokens=m) if self.openrouter_client else None,
'torouter': lambda m: self._call_torouter(full_system, context_history, user_prompt, max_tokens=m, tools=tools) if self.torouter_client else None,
'groq': lambda m: self._call_groq(full_system, context_history, user_prompt, max_tokens=m, tools=tools) if self.groq_client else None,
'grok': lambda m: self._call_grok(full_system, context_history, user_prompt, max_tokens=m) if self.grok_client else None,
'cerebras':lambda m: self._call_cerebras(full_system, context_history, user_prompt, max_tokens=m, tools=tools) if self.cerebras_client else None,
'hf_inference':lambda m: self._call_hf_inference(full_system, context_history, user_prompt, max_tokens=m) if self.hf_inference_client else None,
'mistral': lambda m: self._call_mistral(full_system, context_history, user_prompt, max_tokens=m, tools=tools) if self.mistral_client else None,
'nvidia': lambda m: self._call_nvidia(full_system, context_history, user_prompt, max_tokens=m, tools=tools) if self.nvidia_client else None,
'gemini': lambda m: self._call_gemini(full_system, context_history, user_prompt, max_tokens=m, tools=tools) if (self.gemini_client or self.gemini_model) else None,
'cohere': lambda m: self._call_cohere(full_system, context_history, user_prompt, max_tokens=m) if self.cohere_client else None,
'together':lambda m: self._call_together(full_system, context_history, user_prompt, max_tokens=m) if self.together_client else None,
'llama': lambda m: self._call_llama(full_system, context_history, user_prompt, max_tokens=m) if (self.llama_llm and getattr(self.llama_llm, 'is_available', lambda: False)()) else None,
}
provider_order = list(self.providers)
for round_num in range(1, MAX_ROUNDS + 1):
if round_num > 1:
logger.info(f"⏳ [FALLBACK] Aguardando 3s antes do round {round_num}...")
time.sleep(3)
for provider in provider_order:
if provider in self.blacklisted_providers:
continue
# Check OpenRouter circuit breaker (402/429)
if provider == 'openrouter':
_or_circuit = getattr(self, '_LLMManager__openrouter_circuit_open_until', None) or getattr(self, '_openrouter_circuit_open_until', 0)
if time.time() < (_or_circuit or 0):
remaining = int((_or_circuit or 0) - time.time())
logger.debug(f"⚡ [OR-CIRCUIT] OpenRouter bloqueado (resta {remaining}s). Saltando.")
continue
# Check temporary blacklist (429)
if provider in self.temp_blacklisted_providers:
expiry, reason = self.temp_blacklisted_providers[provider]
if time.time() < expiry:
logger.info(f"⏭️ Ignorando [{provider}] (Temp Blacklist: {reason})")
continue
else:
del self.temp_blacklisted_providers[provider]
caller = provider_callers.get(provider)
if not caller:
continue
try:
user_len = len(user_prompt.split())
hard_max = getattr(self.config, 'MAX_TOKENS', 1024)
dyn_max = hard_max
# Allow longer responses for normal conversations
if user_len <= 2: dyn_max = 120
elif user_len <= 5: dyn_max = 200
elif user_len <= 12: dyn_max = 300
# For longer prompts, use the full hard_max (1024)
text = caller(dyn_max)
if text:
# Se funcionou, garante que o provedor não está na blacklist temporária
if provider in self.temp_blacklisted_providers:
del self.temp_blacklisted_providers[provider]
# Pode ser string ou dicionário (tool_calls)
content = text.get("tool_calls") if isinstance(text, dict) else text
if content:
logger.info(f"✅ Resposta gerada por [{provider}] (round {round_num})")
return text, provider
logger.warning(f"⚠️ [{provider}] retornou vazio (round {round_num}), tentando próximo...")
except Exception as e:
err_msg = str(e)
# ── 402 sem créditos = circuit breaker (sem retry/loop)
# evita desperdiçar 30s a tentar providers mortos; HF Llama responde
if "402" in err_msg or "Payment Required" in err_msg or "payment_required" in err_msg:
_prov_label = provider.upper()
logger.error(f"🚫 [{_prov_label}] 402 (sem créditos) → blacklist 10 min, fallback para Llama/HF ativo.")
self.temp_blacklisted_providers[provider] = (time.time() + 600, f"402 Payment Required: {err_msg[:80]}")
if provider == "openrouter":
try:
self.__class__._openrouter_circuit_open_until = time.time() + 600
except Exception:
pass
continue
if any(x in err_msg for x in ["401", "400", "403", "Unauthorized", "Forbidden", "API_KEY_INVALID", "PERMISSION_DENIED"]):
# 🚀 GEMINI 403 FIX: project denied → blacklist 6h + log acionável
# FIX 2026-09-13: Mistral 403 = chave inválida (não rate-limit). Blacklist curto (60s)
# para permitir retry rápido após atualizar HF Secrets + restart. Não esconder por 30min.
_is_gemini_403 = provider == "gemini" and ("403" in err_msg or "PERMISSION_DENIED" in err_msg)
_is_mistral_403 = provider == "mistral" and ("403" in err_msg or "401" in err_msg or "Unauthorized" in err_msg or "Forbidden" in err_msg)
if _is_gemini_403:
_dur = 21600
elif _is_mistral_403:
_dur = 60
else:
_dur = 1800
logger.warning(f"⏳ Blacklist temporária [{provider}] ({_dur//60} min) por erro de auth: {e}")
if _is_gemini_403:
logger.error("🚨 [GEMINI] Project denied (403). Crie nova key em https://aistudio.google.com/app/apikey e atualize HF Secrets + Railway Variables GEMINI_API_KEY + restart. Fallback para OpenRouter ativo.")
if _is_mistral_403:
logger.error("🚨 [MISTRAL] 403 Forbidden/Unauthorized = chave inválida ou expirada. Atualize HF Secrets MISTRAL_API_KEY/SOFTEDGE_MISTRAL_API/MKULTRA_MISTRAL_KEY (UPPERCASE) + restart o Space. .env local NÃO atualiza o HF.")
self.temp_blacklisted_providers[provider] = (time.time() + _dur, f"Auth Error: {err_msg[:80]}")
elif "429" in err_msg or "Rate Limit" in err_msg or "rate_limit" in err_msg.lower():
logger.warning(f"⏳ Blacklist temporária [{provider}] (60s) por 429: {e}")
self.temp_blacklisted_providers[provider] = (time.time() + 60, "429 Rate Limit")
elif any(x in err_msg for x in ["404", "model_not_found", "model_archived", "archived and unavailable", "does not exist", "decommissioned", "No such model"]):
# FIX: modelo reformado/arquivado — retry imediato é inútil e queima
# segundos por request (ex: groq llama-3.3-70b-versatile 404, cerebras
# gemma-4-31b archived). Blacklist 30 min para saltar sem HTTP.
logger.warning(f"⏳ Blacklist temporária [{provider}] (30 min) por modelo inexistente/arquivado: {err_msg[:120]}")
self.temp_blacklisted_providers[provider] = (time.time() + 1800, f"404 model gone: {err_msg[:80]}")
else:
logger.warning(f"❌ [{provider}] falhou (round {round_num}): {e}")
continue
logger.error(f"💀 Todos os provedores falharam após {MAX_ROUNDS} voltas")
return getattr(self.config, 'FALLBACK_RESPONSE', 'Eita! O sistema tá com problemas.'), 'fallback_offline'
def _call_mistral(self, system_prompt: str, context_history: List[dict], user_prompt: str, max_tokens: int = 4096, tools: Optional[List[Dict[str, Any]]] = None) -> Optional[Union[str, Dict[str, Any]]]:
try:
if not self.mistral_client:
return None
import requests as req
import time
import random
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
for turn in context_history:
msg = {"role": turn.get("role", "user")}
if "content" in turn:
msg["content"] = turn["content"]
if "tool_calls" in turn:
msg["tool_calls"] = turn["tool_calls"]
if "tool_call_id" in turn:
msg["tool_call_id"] = turn["tool_call_id"]
if "name" in turn:
msg["name"] = turn["name"]
messages.append(msg)
messages.append({"role": "user", "content": user_prompt})
timeout = getattr(self.config, 'API_TIMEOUT', 60)
# Para textos grandes, aumenta o timeout proporcionalmente (até 120s)
if len(user_prompt) > 5000:
timeout = max(timeout, 120)
elif len(user_prompt) > 2000:
timeout = max(timeout, 90)
if self.mistral_rotation:
self.mistral_rotation.reset_quotas_if_needed()
max_retries = 3
base_delay = 1.0
_mistral_deadline = time.time() + (timeout + 5.0)
for attempt in range(max_retries):
# FIX 2026-08-28: se wall-clock excedeu, aborta sem mais retries.
if time.time() > _mistral_deadline:
logger.warning(f"[MISTRAL] hard deadline exhausted after {attempt} attempt(s), aborting")
return None
try:
payload = {
"model": getattr(config, 'MISTRAL_MODEL', 'mistral-large-latest'),
"messages": messages,
"max_tokens": max_tokens,
"temperature": getattr(config, 'TEMPERATURE', 1.0),
"top_p": getattr(config, 'TOP_P', 1.5),
"frequency_penalty": getattr(config, 'FREQUENCY_PENALTY', 0.2),
"presence_penalty": getattr(config, 'PRESENCE_PENALTY', 0.3)
}
if tools:
payload["tools"] = [{"type": "function", "function": t} for t in tools]
current_key = None
mistral_account_label = "única"
if self.mistral_rotation:
current_key = self.mistral_rotation.get_current_key()
mistral_account_label = self.mistral_rotation.get_current_account_name()
else:
current_key = getattr(config, 'MISTRAL_API_KEY', '')
if not current_key:
logger.error("Mistral: nenhuma chave disponível para chamada.")
return None
logger.info(f"Mistral request usando conta: {mistral_account_label}")
response = req.post(
"https://api.mistral.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {current_key}"},
json=payload,
timeout=timeout
)
if response.status_code == 429:
logger.warning(f"Mistral 429 {mistral_account_label} → skip (sem retry, fallback imediato)")
if self.mistral_rotation and self.mistral_rotation.handle_429_error():
mistral_account_label = self.mistral_rotation.get_current_account_name()
return None
if response.status_code in (403, 401):
cur_k = self.mistral_rotation.get_current_key() if self.mistral_rotation else getattr(config, 'MISTRAL_API_KEY', '')
k_s = str(cur_k)
k_hint = f"{k_s[:4]}...{k_s[-2:]}" if len(k_s) > 6 else "chave"
logger.warning(f"Mistral {response.status_code} na conta {mistral_account_label} — key {k_hint} — Forbidden/Unauthorized.")
rotated = False
if self.mistral_rotation:
try:
# Rotate without marking exhausted - 403/401 = invalid key, not rate limit
for _ in range(len(self.mistral_rotation.api_keys) - 1):
self.mistral_rotation.current_key_index = (self.mistral_rotation.current_key_index + 1) % len(self.mistral_rotation.api_keys)
next_quota = self.mistral_rotation.accounts[self.mistral_rotation.current_key_index]
if not next_quota.is_exhausted:
rotated = True
logger.info(f"Mistral {response.status_code} rotacionado para: {self.mistral_rotation.get_current_account_name()} — tentando imediatamente")
break
if not rotated:
logger.warning(f"Mistral {response.status_code} sem mais chaves válidas para rotacionar")
except Exception:
pass
if rotated and attempt < max_retries - 1:
time.sleep(0.3)
continue
return None
if response.status_code == 400:
logger.warning(f"Mistral 400 {mistral_account_label} bad request → skip sem retry")
return None
response.raise_for_status()
if self.mistral_rotation:
self.mistral_rotation.record_request()
result = response.json()
if result.get("choices") and len(result["choices"]) > 0:
msg = result["choices"][0]["message"]
if msg.get("tool_calls"):
# Mock para ser compatível com as tool_calls geradas pelo Gemini
class MockToolCall:
def __init__(self, tc):
self.id = tc.get("id", "call_1")
self.name = tc["function"]["name"]
self.arguments = tc["function"]["arguments"]
return {"tool_calls": [MockToolCall(tc) for tc in msg["tool_calls"]]}
return msg.get("content", "").strip()
return None
except req.exceptions.HTTPError as e:
sc = getattr(response, 'status_code', None) if 'response' in locals() else None
if sc == 429 and attempt < max_retries - 1:
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
logger.warning(f"Mistral 429. Retry {attempt + 1}/{max_retries} após {delay:.1f}s...")
if self.mistral_rotation and self.mistral_rotation.handle_429_error():
time.sleep(delay)
continue
time.sleep(delay)
continue
if sc in (403, 401):
key_raw = self.mistral_rotation.get_current_key() if self.mistral_rotation else getattr(config, 'MISTRAL_API_KEY', '')
key_s = str(key_raw)
key_hint = f"{key_s[:4]}...{key_s[-2:]}" if len(key_s) > 6 else "chave"
extra = ""
if key_s.startswith("sk-"): extra = " (Parece uma chave OpenAI!)"
elif key_s.startswith("gsk_"): extra = " (Parece uma chave Groq!)"
logger.warning(f"Mistral: Erro {sc}. Key {key_hint}{extra}.")
rotated = False
if self.mistral_rotation and hasattr(self.mistral_rotation, 'handle_429_error'):
try: rotated = self.mistral_rotation.handle_429_error()
except Exception: pass
if rotated and attempt < max_retries - 1:
time.sleep(0.3)
continue
# FIX 2026-09-02: NÃO blacklist Mistral em 403/401 — chave inválida
# não se recupera sozinha; fallback para próximo provider imediatamente.
return None
raise e
# logger.error("Mistral: Max retries excedido (429)") # REMOVIDO 2026-09-02: não há mais 429 loop
return None
raise Exception("429 Rate Limit Excedido - Mistral temporariamente indisponível")
except Exception as e:
err_lower = str(e).lower()
if "403" in err_lower or "forbidden" in err_lower:
k_raw = self.mistral_rotation.get_current_key() if self.mistral_rotation and hasattr(self.mistral_rotation, 'get_current_key') else getattr(config, 'MISTRAL_API_KEY', '')
k_s = str(k_raw)
k_hint = f"{k_s[:4]}...{k_s[-2:]}" if len(k_s) > 6 else "chave"
logger.error(f"Mistral 403 no except ({e}). Key {k_hint} — fallback.")
if self.mistral_rotation and hasattr(self.mistral_rotation, 'handle_429_error'):
try: self.mistral_rotation.handle_429_error()
except Exception: pass
self.temp_blacklisted_providers['mistral'] = (time.time() + 600, "403 Forbidden - temp blacklist 10 min")
return None
logger.error(f"Mistral falhou: {e}")
return None
def _call_gemini(self, system_prompt, context_history, user_prompt, max_tokens: int = 4096, tools: Optional[List[Dict[str, Any]]] = None):
try:
if not self.gemini_client and not self.gemini_model:
return None
system_prompt = system_prompt or ""
full_prompt = system_prompt + "\n\nHistorico:\n"
for turn in context_history:
role = turn.get("role", "user")
content = turn.get("content")
if content is None:
content = ""
full_prompt += "[" + role.upper() + "] " + str(content) + "\n"
full_prompt += "\n[USER] " + str(user_prompt or "") + "\n"
if GEMINI_USING_NEW_API and self.gemini_client:
try:
from google.genai import types
import random
import json
# Reconstroi o histórico no formato Gemini
contents = []
for turn in context_history:
role = "model" if turn.get("role") == "assistant" else "user"
parts = []
if turn.get("content"):
parts.append(types.Part(text=turn["content"]))
if turn.get("tool_calls"):
for tc in turn["tool_calls"]:
parts.append(types.Part(function_call=types.FunctionCall(
name=tc["function"]["name"],
args=json.loads(tc["function"]["arguments"])
)))
if turn.get("role") == "tool":
role = "user" # Tool responses are sent as 'user' role parts with function_response
parts = [types.Part(function_response=types.FunctionResponse(
name=turn["name"],
response={"result": turn["content"]}
))]
if parts:
contents.append(types.Content(role=role, parts=parts))
# Adiciona a mensagem atual se não for vazia
if user_prompt and user_prompt.strip():
contents.append(types.Content(role="user", parts=[types.Part(text=user_prompt)]))
# Configuração de ferramentas (tools)
google_tools = None
if tools:
google_tools = [types.Tool(function_declarations=[
types.FunctionDeclaration(
name=t["name"],
description=t["description"],
parameters=t["parameters"]
) for t in tools
])]
model_priority = [
# FIX 2026-08-28: gemini-3.5-flash-lite é o modelo recomendado pela Google API
# (não existe apenas no meu dataset de docs — a API confirmou que é real)
"gemini-3.5-flash-lite",
"gemini-2.0-flash-001",
"gemini-2.0-flash-lite",
"gemini-1.5-flash-8b",
"gemini-1.5-pro"
]
env_model = getattr(self, 'gemini_model_name', None)
if env_model and env_model not in model_priority:
model_priority.insert(0, env_model)
last_err = None
for model_id in model_priority:
try:
logger.info(f"🧠 Chamando Gemini com modelo: {model_id}")
response = self.gemini_client.models.generate_content(
model=model_id,
contents=contents,
config=types.GenerateContentConfig(
system_instruction=system_prompt,
tools=google_tools,
max_output_tokens=max_tokens,
temperature=0.7
)
)
if response and response.candidates and response.candidates[0].content.parts:
candidate = response.candidates[0]
parts = candidate.content.parts
# Detecta tool calls
tool_calls = []
for p in parts:
if p.function_call:
# Converte para o formato interno que o loop espera
class MockToolCall:
def __init__(self, fc):
self.id = f"call_{random.randint(1000, 9999)}"
self.name = fc.name
self.arguments = json.dumps(fc.args) if fc.args else "{}"
tool_calls.append(MockToolCall(p.function_call))
if tool_calls:
return {"tool_calls": tool_calls}
# Se não houver tool calls, retorna o texto
text_parts = [p.text for p in parts if p.text]
if text_parts:
return "".join(text_parts).strip()
except Exception as e:
last_err = e
if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e):
logger.warning(f"⚠️ Gemini {model_id} quota excedida (429). Tentando próximo...")
continue
if "404" in str(e) or "not found" in str(e).lower():
logger.warning(f"⚠️ Modelo {model_id} não encontrado. Tentando próximo...")
continue
if "403" in str(e) or "forbidden" in str(e).lower() or "PERMISSION_DENIED" in str(e):
# Projeto negado no Google = permanente: blacklist 30 min
# para pular direto ao próximo provedor nas próximas chamadas
try:
self.temp_blacklisted_providers['gemini'] = (time.time() + 1800, f"403 PERMISSION_DENIED: {str(e)[:80]}")
except Exception:
pass
logger.error(f"❌ Erro crítico no Gemini ({model_id}): {e}")
break
if last_err:
logger.error(f"Todos os modelos Gemini falharam. Último erro: {last_err}")
return None
except Exception as api_error:
logger.error(f"Gemini nova API erro: {api_error}")
return None
elif self.gemini_model:
response = self.gemini_model.generate_content(full_prompt)
text = response.text if hasattr(response, 'text') and response.text else str(response)
else:
return None
if text:
return text.strip()
except Exception as e:
logger.warning(f"Gemini erro: {e}")
return None
# ── Circuit Breaker: evita retries quando OpenRouter está em rate limit
_openrouter_circuit_open_until: float = 0 # timestamp; 0 = fechado (normal)
_OPENROUTER_CIRCUIT_TIMEOUT: float = 600 # 10 minutos bloqueado após 429
def _call_openrouter(self, system_prompt, context_history, user_prompt, max_tokens: int = 1000):
if self.openrouter_client is None:
return None
import time as _time
import random as _random
import re as _re
openrouter_account_label = "default"
rotation = None
try:
rotation = get_openrouter_rotation()
current_name = rotation.get_current_account_name()
if current_name:
openrouter_account_label = current_name
# FIX 2026-09-17: sincroniza o client com a chave ATUAL da rotation.
# Sem isto, as chaves novas nunca eram usadas (hardcode single-key).
try:
_rk = rotation.get_current_key()
if _rk and getattr(self, 'openrouter_client', None) is not None:
self.openrouter_client.api_key = _rk
except Exception:
pass
except Exception:
pass
logger.info(f"OpenRouter request usando conta: {openrouter_account_label}")
# ── Circuit Breaker: se OpenRouter falhou recentemente, retorna None imediatamente
if _time.time() < self.__class__._openrouter_circuit_open_until:
remaining = int(self.__class__._openrouter_circuit_open_until - _time.time())
logger.debug(f"⚡ [OR-CIRCUIT] OpenRouter bloqueado por 429 (ainda {remaining}s). Saltando.")
return None
messages = [{"role": "system", "content": system_prompt or ""}]
for turn in context_history:
msg = {"role": turn.get("role", "user")}
if "content" in turn:
msg["content"] = turn["content"]
if "tool_calls" in turn:
msg["tool_calls"] = turn["tool_calls"]
if "tool_call_id" in turn:
msg["tool_call_id"] = turn["tool_call_id"]
if "name" in turn:
msg["name"] = turn["name"]
messages.append(msg)
messages.append({"role": "user", "content": user_prompt or ""})
model_name = getattr(self.config, 'OPENROUTER_MODEL', 'meta-llama/llama-3.3-70b-instruct')
# Get model-specific parameters
# FIX 2026-09-13: openai<2.0 via OpenRouter só aceita temperature/top_p/max_tokens.
# repetition_penalty/presence/frequency quebram com "unexpected keyword argument" no HF (openai 1.x).
model_params = dict(getattr(self.config, 'MODEL_PARAMETERS', {}).get(model_name, {}) or {})
# OpenRouter chat.completions doesn't support top_k - remove if present
model_params.pop('top_k', None)
model_params.pop('repetition_penalty', None)
model_params.pop('presence_penalty', None)
model_params.pop('frequency_penalty', None)
try:
resp = self.openrouter_client.chat.completions.create(
model=model_name,
messages=messages,
temperature=model_params.get('temperature', 0.75),
top_p=model_params.get('top_p', 0.9),
max_tokens=max_tokens
)
if not resp or not hasattr(resp, 'choices') or not resp.choices:
logger.warning(f"OpenRouter resp inválido, pulando.")
return None
choice = resp.choices[0]
if not hasattr(choice, 'message') or not choice.message:
logger.warning(f"OpenRouter message vazio, pulando.")
return None
text = None
if hasattr(choice.message, 'content'):
text = choice.message.content
elif isinstance(choice.message, dict):
text = choice.message.get('content')
if text and isinstance(text, str) and text.strip():
return text.strip()
logger.warning(f"OpenRouter content vazio, pulando.")
return None
except Exception as e:
err_str = str(e)
err_lower = err_str.lower()
status_match = None
raw_text = None
# 🔴 Connection errors: fail fast
if any(k in err_lower for k in [
"connection error", "connecterror", "connection refused",
"connection reset", "connection aborted", "timeout",
"name resolution", "no route to host", "network is unreachable"
]):
logger.warning(f"OpenRouter: conexão falhou (unreachable). Pulando.")
return None
if hasattr(e, 'response'):
resp = getattr(e, 'response', None)
if resp is not None and hasattr(resp, 'text'):
try:
raw_text = resp.text
except Exception:
raw_text = None
if raw_text:
is_html = '<html' in raw_text.lower() or '<!doctype' in raw_text.lower()
logger.error(f"🔍 OpenRouter RAW: HTML={is_html}, preview=[{raw_text[:200]}...]")
try:
m = _re.search(r'"?status_code"?\s*[:=]\s*(\d+)', err_str)
if m:
status_match = int(m.group(1))
if status_match is None:
m2 = _re.search(r'HTTP[/\s]+.*?(\d{3})', err_str)
if m2:
status_match = int(m2.group(1))
except Exception:
status_match = None
if status_match == 429 or "429" in err_str or "Too Many Requests" in err_str or "free-models-per-day" in err_str:
# FIX 2026-09-17: 429 rotaciona para a próxima conta e tenta 1x,
# em vez de bloquear o provider inteiro (as chaves novas ficavam mortas).
try:
_rot2 = rotation if rotation is not None else get_openrouter_rotation()
_new_key = _rot2.rotate_on_429()
if _new_key:
try:
self.openrouter_client.api_key = _new_key
except Exception:
pass
logger.warning(f"⚡ [OR-ROTATE] 429 na conta {openrouter_account_label} → tentando {_rot2.get_current_account_name()} 1x")
try:
resp_retry = self.openrouter_client.chat.completions.create(
model=model_name,
messages=messages,
temperature=model_params.get('temperature', 0.75),
top_p=model_params.get('top_p', 0.9),
max_tokens=max_tokens
)
if resp_retry and getattr(resp_retry, 'choices', None):
_ch = resp_retry.choices[0]
_m = getattr(_ch, 'message', None)
_t = getattr(_m, 'content', None) if _m is not None else None
if isinstance(_m, dict):
_t = _m.get('content')
if _t and isinstance(_t, str) and _t.strip():
logger.info(f"✅ OpenRouter retry pós-429 funcionou na conta {_rot2.get_current_account_name()}")
return _t.strip()
except Exception as _re2:
logger.debug(f"OpenRouter retry pós-429 falhou: {_re2}")
except Exception as _re:
logger.debug(f"OpenRouter rotate-on-429 falhou: {_re}")
self.__class__._openrouter_circuit_open_until = _time.time() + self.__class__._OPENROUTER_CIRCUIT_TIMEOUT
logger.warning(f"⚡ [OR-CIRCUIT] 429 persistente → OpenRouter bloqueado por {int(self.__class__._OPENROUTER_CIRCUIT_TIMEOUT//60)} min")
return None
elif status_match == 402 or "402" in err_str or "Insufficient credits" in err_str or "Payment Required" in err_str:
# FIX 2026-09-17: 402 = conta sem créditos → rotaciona, não mata o provider.
try:
_rot3 = rotation if rotation is not None else get_openrouter_rotation()
_new_key3 = _rot3.rotate_on_429()
if _new_key3:
try:
self.openrouter_client.api_key = _new_key3
except Exception:
pass
logger.warning(f"🚫 [OR-ROTATE] 402 sem créditos na conta {openrouter_account_label} → tentando {_rot3.get_current_account_name()} 1x")
try:
resp402 = self.openrouter_client.chat.completions.create(
model=model_name,
messages=messages,
temperature=model_params.get('temperature', 0.75),
top_p=model_params.get('top_p', 0.9),
max_tokens=max_tokens
)
if resp402 and getattr(resp402, 'choices', None):
_ch4 = resp402.choices[0]
_m4 = getattr(_ch4, 'message', None)
_t4 = getattr(_m4, 'content', None) if _m4 is not None else None
if isinstance(_m4, dict):
_t4 = _m4.get('content')
if _t4 and isinstance(_t4, str) and _t4.strip():
logger.info(f"✅ OpenRouter retry pós-402 funcionou na conta {_rot3.get_current_account_name()}")
return _t4.strip()
except Exception as _re402:
logger.debug(f"OpenRouter retry pós-402 falhou: {_re402}")
except Exception as _re3:
logger.debug(f"OpenRouter rotate-on-402 falhou: {_re3}")
logger.error(f"🚫 [OR-CIRCUIT] OpenRouter sem créditos em todas as contas tentadas. Blacklist 5 min.")
self.__class__._openrouter_circuit_open_until = _time.time() + 300
return None
elif status_match == 401 or "401" in err_str or "Unauthorized" in err_str:
logger.error("OpenRouter: Erro de autenticação (401). Pulando.")
return None
elif status_match == 404 and "unavailable for free" in err_lower:
m_suggest = _re.search(r"use this slug instead:\s*([a-zA-Z0-9/:\-\._]+)", err_str, _re.IGNORECASE)
if m_suggest:
suggested = m_suggest.group(1).strip()
logger.warning(f"OpenRouter 404 :free unavailable, retrying once with suggested slug: {suggested}")
try:
resp2 = self.openrouter_client.chat.completions.create(
model=suggested,
messages=messages,
temperature=0.7,
max_tokens=max_tokens
)
if resp2 and hasattr(resp2, 'choices') and resp2.choices and hasattr(resp2.choices[0], 'message') and resp2.choices[0].message:
msg2 = resp2.choices[0].message
text2 = getattr(msg2, 'content', None)
if isinstance(msg2, dict):
text2 = msg2.get('content')
if text2 and isinstance(text2, str) and text2.strip():
logger.info(f"✅ OpenRouter retry com slug sugerido funcionou: {suggested}")
return text2.strip()
except Exception as e2:
logger.warning(f"OpenRouter retry com slug sugerido falhou: {e2}")
logger.warning(f"OpenRouter 404 sem slug sugerido ou retry falhou. Pulando.")
return None
logger.warning(f"OpenRouter erro: {e}. Pulando para próximo provedor.")
return None
def _call_torouter(self, system_prompt, context_history, user_prompt, max_tokens: int = 4096, tools: Optional[List[Dict[str, Any]]] = None):
"""Chama ToRouter (OpenAI-compatible) com suporte a tools e rotação multi-conta."""
if self.torouter_client is None:
return None
import time as _time
import random as _random
import re as _re
torouter_account_label = "default"
torouter_rotation = None
try:
torouter_rotation = get_torouter_rotation()
current_name = torouter_rotation.get_current_account_name()
if current_name:
torouter_account_label = current_name
self.torouter_client.api_key = torouter_rotation.get_current_key()
except Exception:
pass
logger.info(f"ToRouter request usando conta: {torouter_account_label}")
messages = [{"role": "system", "content": system_prompt or ""}]
for turn in context_history:
msg = {"role": turn.get("role", "user")}
if "content" in turn:
msg["content"] = turn["content"]
if "tool_calls" in turn:
msg["tool_calls"] = turn["tool_calls"]
if "tool_call_id" in turn:
msg["tool_call_id"] = turn["tool_call_id"]
if "name" in turn:
msg["name"] = turn["name"]
messages.append(msg)
messages.append({"role": "user", "content": user_prompt or ""})
model_name = getattr(self.config, 'TOROUTER_MODEL', 'openai/gpt-5.5')
# ⚡ CONNECTION_ERROR => fail fast (sem retry), 401/429 => tenta rotação de conta
try:
kwargs = {
"model": model_name,
"messages": messages,
"temperature": 0.7,
"max_tokens": max_tokens
}
if tools:
kwargs["tools"] = tools
resp = self.torouter_client.chat.completions.create(**kwargs)
if not resp or not hasattr(resp, 'choices') or not resp.choices:
logger.warning(f"ToRouter resp inválido, pulando.")
return None
choice = resp.choices[0]
if not hasattr(choice, 'message') or not choice.message:
logger.warning(f"ToRouter message vazio, pulando.")
return None
# 🔧 TOOL CALLS: Verificar se LLM retornou tool_calls
msg = choice.message
if hasattr(msg, 'tool_calls') and msg.tool_calls:
class MockToolCall:
def __init__(self, tc):
self.id = tc.id
self.name = tc.function.name
self.arguments = tc.function.arguments
if torouter_rotation:
torouter_rotation.record_request()
return {"tool_calls": [MockToolCall(tc) for tc in msg.tool_calls]}
text = None
if hasattr(msg, 'content'):
text = msg.content
elif isinstance(msg, dict):
text = msg.get('content')
if text and isinstance(text, str) and text.strip():
if torouter_rotation:
torouter_rotation.record_request()
return text.strip()
logger.warning(f"ToRouter content vazio, pulando.")
return None
except Exception as e:
err_str = str(e)
err_lower = err_str.lower()
# 🔴 Connection errors: fail fast, não retry
is_connection_error = any(k in err_lower for k in [
"connection error", "connecterror", "connection refused",
"connection reset", "connection aborted", "timeout",
"name resolution", "no route to host", "network is unreachable"
])
if is_connection_error:
logger.warning(f"ToRouter: conexão falhou (unreachable). Pulando para próximo provedor.")
return None
try:
m = _re.search(r'"?status_code"?\s*[:=]\s*(\d+)', err_str)
status_match = int(m.group(1)) if m else None
if status_match is None:
m2 = _re.search(r'HTTP[/\s]+.*?(\d{3})', err_str)
if m2:
status_match = int(m2.group(1))
except Exception:
status_match = None
# 🔄 429 / 401 => tenta rotacionar conta
if status_match == 429 or "429" in err_str or "Too Many Requests" in err_str or "rate" in err_str.lower():
if torouter_rotation:
next_key = torouter_rotation.rotate_on_429()
if next_key:
self.torouter_client.api_key = next_key
current_label = torouter_rotation.get_current_account_name()
logger.info(f"ToRouter rotacionado para conta: {current_label}")
return None # próxima chamada usará a nova conta
logger.warning(f"ToRouter: 429 sem rotação disponível. Pulando.")
return None
if status_match == 401 or "401" in err_str or "Unauthorized" in err_str:
if torouter_rotation:
next_key = torouter_rotation.rotate_on_429()
if next_key:
self.torouter_client.api_key = next_key
current_label = torouter_rotation.get_current_account_name()
logger.info(f"ToRouter 401: rotacionando para {current_label}")
return None
logger.warning(f"ToRouter: 401 sem rotação. Pulando.")
return None
if status_match == 503 or "503" in err_str or "Service Unavailable" in err_str or "temporarily unavailable" in err_lower:
fallback_models = ["openai/gpt-5.4-nano", "google/gemini-2.5-flash-lite"]
current_model = getattr(self.config, 'TOROUTER_MODEL', 'openai/gpt-5.5')
for alt_model in fallback_models:
if alt_model == current_model:
continue
logger.warning(f"ToRouter 503 com {current_model}. Tentando {alt_model}...")
kwargs["model"] = alt_model
try:
resp2 = self.torouter_client.chat.completions.create(**kwargs)
if resp2 and hasattr(resp2, 'choices') and resp2.choices and hasattr(resp2.choices[0].message, 'content'):
text2 = resp2.choices[0].message.content
if text2 and isinstance(text2, str) and text2.strip():
if torouter_rotation:
torouter_rotation.record_request()
return text2.strip()
except Exception:
pass
logger.warning(f"ToRouter 503 persistente em todas as contas/modelos. Pulando para próximo provedor.")
return None
logger.warning(f"ToRouter erro: {e}. Pulando para próximo provedor.")
return None
def _call_groq(self, system_prompt, context_history, user_prompt, max_tokens: int = 4096, tools: Optional[List[Dict[str, Any]]] = None):
try:
if self.groq_client is None:
return None
messages = [{"role": "system", "content": system_prompt}]
for turn in context_history:
msg = {"role": turn.get("role", "user")}
if "content" in turn:
msg["content"] = turn["content"]
if "tool_calls" in turn:
msg["tool_calls"] = turn["tool_calls"]
if "tool_call_id" in turn:
msg["tool_call_id"] = turn["tool_call_id"]
if "name" in turn:
msg["name"] = turn["name"]
messages.append(msg)
messages.append({"role": "user", "content": user_prompt})
# Usar modelo do config, com fallback se o primário foi reformado (404)
_groq_models = []
try:
_g_primary = (getattr(config, 'GROQ_MODEL', '') or '').strip() or 'llama-3.1-8b-instant'
_g_fallbacks = list(getattr(config, 'GROQ_FALLBACK_MODELS', None) or ['llama-3.1-8b-instant', 'llama-3.3-70b-versatile', 'mixtral-8x7b-32768'])
except Exception:
_g_primary, _g_fallbacks = 'llama-3.1-8b-instant', ['llama-3.1-8b-instant']
for _gm in [_g_primary] + _g_fallbacks:
_gm = (_gm or '').strip()
if _gm and _gm not in _groq_models:
_groq_models.append(_gm)
model_name = _groq_models[0]
kwargs = {
"model": model_name,
"messages": messages,
"temperature": 0.7,
"max_tokens": max_tokens
}
if tools:
kwargs["tools"] = [{"type": "function", "function": t} for t in tools]
resp = None
_groq_404_only = True
for _gm in _groq_models:
kwargs["model"] = _gm
model_name = _gm
try:
resp = self.groq_client.chat.completions.create(**kwargs)
_groq_404_only = False
break
except Exception as _gm_err:
_gm_low = str(_gm_err).lower()
if any(x in str(_gm_err) for x in ["404", "model_not_found"]) or "does not exist" in _gm_low or "decommissioned" in _gm_low:
logger.warning(f"Groq modelo {_gm} inexistente/reformado — tentando próximo da lista...")
resp = None
continue
_groq_404_only = False
raise
if resp is None and _groq_404_only:
# FIX: todos os modelos 404 → propaga para o generate() blacklistar
# 30 min em vez de queimar 3 HTTP por request.
raise Exception("Groq 404 model_not_found em todos os modelos da lista")
if resp is None and len(_groq_models) > 1:
return None
if resp and hasattr(resp, 'choices') and resp.choices:
msg = resp.choices[0].message
if hasattr(msg, 'tool_calls') and msg.tool_calls:
# Mock para ser compatível
class MockToolCall:
def __init__(self, tc):
self.id = tc.id
self.name = tc.function.name
self.arguments = tc.function.arguments
return {"tool_calls": [MockToolCall(tc) for tc in msg.tool_calls]}
text = msg.content
if text:
return text.strip()
except Exception as e:
err_str = str(e)
if "401" in err_str or "unauthorized" in err_str.lower():
key_raw = getattr(self.config, 'GROQ_API_KEY', '')
key_s = str(key_raw)
key_len = len(key_s)
key_hint = f"{key_s[:4]}...{key_s[-2:]}" if key_len > 6 else "INVÁLIDA"
extra = ""
if key_s.startswith("sk-"): extra = " (Parece uma chave OpenAI!)"
elif not key_s.startswith("gsk_"): extra = " (CHAVE GROQ DEVE COMEÇAR COM gsk_!)"
logger.error(f"Groq: Erro de Autenticação (401). Chave: {key_hint} (Tam: {key_len}){extra}. Verifique nos Secrets.")
elif "tool calling" in err_str.lower() and "not supported" in err_str.lower() and tools:
logger.warning(f"Groq: modelo {model_name} não suporta tool calling. Re-tentando sem tools.")
kwargs.pop("tools", None)
try:
resp = self.groq_client.chat.completions.create(**kwargs)
if resp and hasattr(resp, 'choices') and resp.choices:
msg = resp.choices[0].message
text = msg.content
if text:
return text.strip()
except Exception as e2:
logger.warning(f"Groq erro (retry sem tools): {e2}")
else:
logger.warning(f"Groq erro: {e}")
return None
def _call_grok(self, system_prompt: str, context_history: List[dict], user_prompt: str, max_tokens: int = 8192) -> Optional[str]:
try:
if not self.grok_client:
return None
messages = [{"role": "system", "content": system_prompt}]
for turn in context_history:
role = turn.get("role", "user")
content = turn.get("content", "")
messages.append({"role": role, "content": content})
messages.append({"role": "user", "content": user_prompt})
model = getattr(self, 'grok_model', 'grok-3')
resp = self.grok_client.chat.completions.create(
model=model,
messages=messages,
temperature=0.7,
max_tokens=max_tokens
)
if resp and hasattr(resp, 'choices') and resp.choices:
text = resp.choices[0].message.content
if text:
return text.strip()
except Exception as e:
logger.warning(f"Grok erro: {e}")
return None
def _call_cohere(self, system_prompt, context_history, user_prompt, max_tokens: int = 4096):
try:
if self.cohere_client is None:
return None
full_message = system_prompt + "\n\n"
for turn in context_history:
role = turn.get("role", "user")
content = turn.get("content", "")
full_message += "[" + role.upper() + "] " + content + "\n"
full_message += "\n[USER] " + user_prompt + "\n"
max_tokens = min(max_tokens, 4096)
resp = self.cohere_client.chat(model=getattr(self.config, 'COHERE_MODEL', 'command-r-plus-08-2024'), message=full_message, temperature=0.7, max_tokens=max_tokens)
if resp and hasattr(resp, 'text'):
text = resp.text
if text:
return text.strip()
except Exception as e:
# FIX: propaga 429/auth/404 para o generate() aplicar blacklist (antes
# engolia tudo e cada request queimava 3 chamadas 429 seguidas).
_e = str(e)
if any(x in _e for x in ("429", "401", "403", "404", "Rate Limit", "rate_limit", "Unauthorized", "Forbidden", "not found", "Not Found")):
raise
logger.warning(f"Cohere erro: {e}")
return None
def _call_cerebras(self, system_prompt, context_history, user_prompt, max_tokens: int = 4096, tools=None):
# 🧠 Cerebras - rápido e confiável
# Modelos fixos no config (gpt-oss-120b → qwen-3.8-27b). Sem Secret de
# modelo: a ordem aqui obedece só ao config. 429 roda conta + retry;
# 402 roda conta e cai para o próximo modelo da lista.
if self.cerebras_client is None:
return None
# Montar mensagens para OpenAI SDK
messages = [
{"role": "system", "content": system_prompt}
]
for turn in context_history:
messages.append(turn)
messages.append({"role": "user", "content": user_prompt})
max_tokens = min(max_tokens, 4096)
def _extract_cerebras_text(resp):
if resp and getattr(resp, 'choices', None):
msg = resp.choices[0].message
if getattr(msg, 'tool_calls', None):
class MockToolCall:
def __init__(self, tc):
self.id = tc.id
self.name = tc.function.name
self.arguments = tc.function.arguments
return {"tool_calls": [MockToolCall(tc) for tc in msg.tool_calls]}
text = getattr(msg, 'content', None)
if text and isinstance(text, str) and text.strip():
return text.strip()
return None
def _rotate_cerebras(reason):
try:
rotation = get_cerebras_rotation()
try:
_n_acc = len(getattr(rotation, 'account_names', []) or [])
except Exception:
_n_acc = 0
if _n_acc <= 1:
logger.error("❌ [Cerebras] 402/quota com APENAS 1 conta configurada — sem fallback de conta. Adiciona ISAAC/NETFLUX/GITAKIRA_CEREBRAS_API_KEY nos Secrets (a key ANN em si está a ser enviada; o 402 é falta de quota/billing no lado Cerebras, não bloqueio no código).")
rotation.handle_rate_limit_error()
current_key = rotation.get_current_api_key()
current_name = rotation.get_current_account_name()
if current_key:
import openai
self.cerebras_client = openai.OpenAI(
api_key=current_key,
base_url="https://api.cerebras.ai/v1",
timeout=30.0,
max_retries=0,
)
logger.info(f"✅ Cerebras rotacionado para: {current_name} ({reason})")
return current_name
logger.error("❌ Cerebras: sem chave após rotação (verifique ANN/ISAAC/NETFLUX/GITAKIRA_CEREBRAS_API_KEY nos Secrets)")
except Exception as rotate_e:
logger.error(f"Erro ao rotacionar Cerebras: {rotate_e}")
return None
_primary = (getattr(self.config, 'CEREBRAS_MODEL', '') or '').strip() or 'gpt-oss-120b'
try:
_fallbacks = list(getattr(self.config, 'CEREBRAS_FALLBACK_MODELS', None) or ['gpt-oss-120b', 'llama-3.3-70b'])
except Exception:
_fallbacks = ['gpt-oss-120b', 'llama-3.3-70b']
_models = []
for _m in [_primary] + _fallbacks:
_m = (_m or '').strip()
if _m and _m not in _models:
_models.append(_m)
base_kwargs = {
"messages": messages,
"temperature": 0.7,
"max_tokens": max_tokens,
}
if tools:
base_kwargs["tools"] = [{"type": "function", "function": t} for t in tools]
_saw_402 = False
_saw_404 = False
for _model in _models:
kwargs = dict(base_kwargs)
kwargs["model"] = _model
try:
resp = self.cerebras_client.chat.completions.create(**kwargs)
out = _extract_cerebras_text(resp)
if out:
return out
logger.warning(f"Cerebras resposta vazia (modelo {_model}), tentando próximo...")
except Exception as e:
_e_str = str(e)
_e_low = _e_str.lower()
if "404" in _e_str or "model_not_found" in _e_low or "archived" in _e_low or "does not exist" in _e_low:
_saw_404 = True
if "402" in _e_str or "payment_required" in _e_low or "payment required" in _e_low:
_saw_402 = True
if "429" in _e_str or "rate_limit" in _e_low:
logger.warning(f"🧠 Cerebras 429 no modelo {_model} — rotacionando conta + retry 1x...")
if _rotate_cerebras("429"):
try:
resp2 = self.cerebras_client.chat.completions.create(**kwargs)
out2 = _extract_cerebras_text(resp2)
if out2:
return out2
except Exception as retry_e:
logger.warning(f"Cerebras retry pós-429 falhou: {str(retry_e)[:120]}")
return None
if "402" in _e_str or "payment_required" in _e_low or "payment required" in _e_low or "quota" in _e_low or "billing" in _e_low:
logger.warning(f"🧠 Cerebras 402 no modelo {_model} (fora do free/plano) — roda conta e cai para o próximo modelo free...")
_rotate_cerebras("402")
continue
if "timeout" in _e_low or "timed out" in _e_low:
logger.warning(f"🧠 Cerebras timeout no modelo {_model} — tentando próximo...")
continue
if "404" in _e_str or "model_not_found" in _e_low or "archived" in _e_low or "does not exist" in _e_low or "decommissioned" in _e_low:
# FIX: modelo arquivado/reformado (ex: gemma-4-31b) — pula para o
# próximo em vez de abortar o loop com return None.
logger.warning(f"🧠 Cerebras 404 no modelo {_model} (arquivado/inexistente) — tentando próximo...")
continue
logger.warning(f"Cerebras erro: {e}")
return None
# FIX: loop esgotado só com 402/404 → propaga para o generate()
# blacklistar (10/30 min) em vez de repetir tudo no round 2.
if _saw_404 and not _saw_402:
raise Exception("Cerebras 404 model_not_found/archived em todos os modelos")
if _saw_402 and not _saw_404:
raise Exception("Cerebras 402 Payment Required em todos os modelos")
logger.warning("❌ Cerebras: modelos free tentados esgotados — confirma no dashboard/Playground quais modelos o plano free cobre e ajusta o Secret CEREBRAS_MODEL.")
return None
def _call_nvidia(self, system_prompt, context_history, user_prompt, max_tokens: int = 4096, tools=None, model: str = None):
# 🟢 Nvidia NIM (OpenAI-compatible). `model` permite ao CoT usar o
# modelo PESADO free e à resposta final o MÉDIO (defaults no config).
# 404/modelo-indisponível cai para o próximo modelo; 429/401/402 roda conta.
if self.nvidia_client is None:
return None
messages = [
{"role": "system", "content": system_prompt}
]
for turn in context_history:
messages.append(turn)
messages.append({"role": "user", "content": user_prompt})
max_tokens = min(max_tokens, 4096)
def _extract_nvidia_text(resp):
if resp and getattr(resp, 'choices', None):
msg = resp.choices[0].message
if getattr(msg, 'tool_calls', None):
class MockToolCall:
def __init__(self, tc):
self.id = tc.id
self.name = tc.function.name
self.arguments = tc.function.arguments
return {"tool_calls": [MockToolCall(tc) for tc in msg.tool_calls]}
text = getattr(msg, 'content', None)
if text and isinstance(text, str) and text.strip():
return text.strip()
return None
def _rotate_nvidia(reason):
try:
rotation = get_nvidia_rotation()
try:
_n_acc = len(getattr(rotation, 'account_names', []) or [])
except Exception:
_n_acc = 0
if _n_acc <= 1:
logger.error("❌ [Nvidia] limite/quota com APENAS 1 conta configurada — sem fallback de conta. Adiciona ANN/ISAAC/GITBELMIRA_NVIDIA_API_KEY ou NVIDIA_API_KEY nos Secrets.")
rotation.handle_rate_limit_error()
current_key = rotation.get_current_api_key()
current_name = rotation.get_current_account_name()
if current_key:
import openai
self.nvidia_client = openai.OpenAI(
api_key=current_key,
base_url=getattr(self.config, 'NVIDIA_BASE_URL', 'https://integrate.api.nvidia.com/v1'),
timeout=30.0,
max_retries=1,
)
logger.info(f"✅ Nvidia rotacionado para: {current_name} ({reason})")
return current_name
logger.error("❌ Nvidia: sem chave após rotação (verifique *_NVIDIA_API_KEY nos Secrets)")
except Exception as rotate_e:
logger.error(f"Erro ao rotacionar Nvidia: {rotate_e}")
return None
_primary = (model or '').strip() or (getattr(self.config, 'NVIDIA_MODEL', '') or '').strip() or 'nvidia/nemotron-3.5-lightning-30b-a3b'
try:
_fallbacks = list(getattr(self.config, 'NVIDIA_FALLBACK_MODELS', None) or [_primary])
except Exception:
_fallbacks = [_primary]
_models = []
for _m in [_primary] + _fallbacks:
_m = (_m or '').strip()
if _m and _m not in _models:
_models.append(_m)
base_kwargs = {
"messages": messages,
"temperature": 0.7,
"max_tokens": max_tokens,
}
if tools:
base_kwargs["tools"] = [{"type": "function", "function": t} for t in tools]
for _model in _models:
kwargs = dict(base_kwargs)
kwargs["model"] = _model
try:
resp = self.nvidia_client.chat.completions.create(**kwargs)
out = _extract_nvidia_text(resp)
if out:
return out
logger.warning(f"Nvidia resposta vazia (modelo {_model}), tentando próximo...")
except Exception as e:
_e_str = str(e)
_e_low = _e_str.lower()
if "429" in _e_str or "rate_limit" in _e_low:
logger.warning(f"🟢 Nvidia 429 no modelo {_model} — rotacionando conta + retry 1x...")
if _rotate_nvidia("429"):
try:
resp2 = self.nvidia_client.chat.completions.create(**kwargs)
out2 = _extract_nvidia_text(resp2)
if out2:
return out2
except Exception as retry_e:
logger.warning(f"Nvidia retry pós-429 falhou: {str(retry_e)[:120]}")
return None
if "402" in _e_str or "payment_required" in _e_low or "payment required" in _e_low or "quota" in _e_low or "billing" in _e_low or "401" in _e_str or "unauthorized" in _e_low or "forbidden" in _e_low or "403" in _e_str:
logger.warning(f"🟢 Nvidia auth/quota no modelo {_model} — roda conta e cai para o próximo modelo...")
_rotate_nvidia("auth/quota")
continue
if "404" in _e_str or "410" in _e_str or "gone" in _e_low or "not found" in _e_low or "no endpoints" in _e_low or "model" in _e_low and "avail" in _e_low:
logger.warning(f"🟢 Nvidia modelo indisponível/removido ({_model}) — tentando próximo da lista...")
continue
if "timeout" in _e_low or "timed out" in _e_low:
logger.warning(f"🟢 Nvidia timeout no modelo {_model} — tentando próximo...")
continue
logger.warning(f"Nvidia erro: {e}")
return None
logger.warning("❌ Nvidia: modelos tentados esgotados — confirma os slugs free no dashboard e ajusta no config.")
return None
def _call_hf_inference(self, system_prompt, context_history, user_prompt, max_tokens: int = 4096):
# 🤗 HuggingFace Inference - uncensored model via Featherless AI
try:
if not self.hf_inference_client:
return None
# Lazy init: criar InferenceClient sob demanda
if self.hf_inference_client == "lazy":
try:
from huggingface_hub import InferenceClient
self.hf_inference_client = InferenceClient(
token=getattr(self, '_hf_token', None),
timeout=30.0,
)
logger.info("🔧 [HF LAZY] InferenceClient criado sob demanda")
except Exception as e:
logger.warning(f"⚠️ HF InferenceClient lazy init falhou: {e}")
self.hf_inference_client = None
return None
# HF Inference API usa formato de conversa diferente
# Montar mensagens no formato esperado
messages = [
{"role": "system", "content": system_prompt}
]
for turn in context_history:
messages.append(turn)
messages.append({"role": "user", "content": user_prompt})
# Converter para formato conversational (chat) — text_generation foi removido
# FIX 2026-09-13: Qwen2.5-72B via novita só aceita task conversational, não text-generation.
max_tokens = min(max_tokens, 2048) # HF tem limite menor
model = getattr(self.config, 'HF_INFERENCE_MODEL', 'mistralai/Mistral-7B-Instruct-v0.2')
# Tenta chat_completion (conversational); fallback para text_generation legado
resp_text = None
try:
chat_resp = self.hf_inference_client.chat_completion(
messages=messages,
model=model,
max_tokens=max_tokens,
temperature=0.7,
top_p=0.9,
)
# huggingface_hub retorna objeto com choices[0].message.content
if chat_resp and getattr(chat_resp, 'choices', None):
_msg = chat_resp.choices[0].message
resp_text = getattr(_msg, 'content', None) or ( _msg.get('content') if isinstance(_msg, dict) else None )
elif isinstance(chat_resp, dict):
_ch = (chat_resp.get('choices') or [{}])[0]
_m = _ch.get('message', {}) if isinstance(_ch, dict) else {}
resp_text = _m.get('content') if isinstance(_m, dict) else None
except TypeError:
# client antigo sem chat_completion com esses kwargs — tenta conversational
try:
resp_text = self.hf_inference_client.conversational(
text=f"{system_prompt}\n\nUser: {user_prompt}",
model=model,
)
if isinstance(resp_text, dict):
resp_text = resp_text.get('generated_text') or resp_text.get('response')
except Exception:
resp_text = None
except Exception as _cc_err:
# Se for erro de task/modelo, propaga para log de rotação abaixo
if "conversational" in str(_cc_err).lower() or "task" in str(_cc_err).lower() or "not supported" in str(_cc_err).lower():
logger.warning(f"HF Inference task/modelo incompatível ({model}): {_cc_err} — pulando para próximo provedor.")
return None
raise
if resp_text:
text = resp_text.strip() if isinstance(resp_text, str) else resp_text
if text:
return text
except Exception as e:
_e_s = str(e)
_e_l = _e_s.lower()
# FIX 2026-09-17: 401/403 (token inválido/expirado) também rotaciona.
# Antes só 429 rotacionava — um token morto bloqueava o provider.
if "429" in _e_s or "rate_limit" in _e_l or "too many requests" in _e_l or "401" in _e_s or "403" in _e_s or "unauthorized" in _e_l or "forbidden" in _e_l:
_kind = "429" if ("429" in _e_s or "rate_limit" in _e_l or "too many requests" in _e_l) else "401/403"
logger.warning(f"🤗 HF Inference {_kind} detectado - rotacionando conta...")
try:
from huggingface_hub import InferenceClient
rotation = get_hf_inference_rotation()
rotation.handle_rate_limit_error(_e_s)
# Atualizar cliente com novo token
current_token = rotation.get_current_api_token()
current_name = rotation.get_current_account_name()
if current_token:
self.hf_inference_client = InferenceClient(
token=current_token,
timeout=30.0,
)
logger.info(f"✅ HF Inference rotacionado para: {current_name} (motivo: {_kind})")
else:
logger.error("❌ HF Inference: Nenhuma conta disponível após rotação")
self.hf_inference_client = None
except Exception as rotate_e:
logger.error(f"Erro ao rotacionar HF Inference: {rotate_e}")
else:
logger.warning(f"HF Inference erro: {e}")
return None
def _call_together(self, system_prompt, context_history, user_prompt, max_tokens: int = 4096):
try:
if self.together_client is None:
return None
messages = [{"role": "system", "content": system_prompt}]
for turn in context_history:
role = turn.get("role", "user")
content = turn.get("content", "")
messages.append({"role": role, "content": content})
messages.append({"role": "user", "content": user_prompt})
# Usar modelo do config
model_name = getattr(config, 'TOGETHER_MODEL', 'meta-llama/Llama-3.3-70B-Instruct-Turbo')
resp = self.together_client.chat.completions.create(
model=model_name,
messages=messages,
temperature=0.7,
max_tokens=max_tokens
)
if resp and hasattr(resp, 'choices') and resp.choices:
text = resp.choices[0].message.content
if text:
return text.strip()
except Exception as e:
logger.warning(f"Together AI erro: {e}")
return None
def _call_llama(self, system_prompt, context_history, user_prompt, max_tokens: int = 4096):
try:
if not self.llama_llm:
return None
local = self.llama_llm.generate(
prompt=user_prompt,
system_prompt=system_prompt,
context_history=context_history,
max_tokens=max_tokens
)
if local:
return local
except Exception as e:
logger.warning(f"Llama local erro: {e}")
raise e
class SimpleTTLCache:
def __init__(self, ttl_seconds=300):
self.ttl = ttl_seconds
self._store = {}
def __contains__(self, key):
if key not in self._store:
return False
_, expires = self._store[key]
if time.time() > expires:
self._store.pop(key, None)
return False
return True
def __setitem__(self, key, value):
self._store[key] = (value, time.time() + self.ttl)
def __getitem__(self, key):
if key not in self:
raise KeyError(key)
return self._store[key][0]
def get(self, key, default=None):
try:
return self[key]
except KeyError:
return default
class KiamiAPI:
def __init__(self, cfg_module=None):
self.config = cfg_module if cfg_module else config
self.app = FastAPI(title="KIAMI V1 SOFTEDGE")
self.api = APIRouter()
# ✅ Rate Limiting no Servidor (Professionalquickstart)
self.limiter = SimpleRateLimiter()
logger.info("✅ [RATE LIMITER] Usando SimpleRateLimiter personalizado")
cache_ttl = getattr(self.config, 'CACHE_TTL', 3600)
self.contexto_cache = SimpleTTLCache(ttl_seconds=cache_ttl)
self.providers = LLMManager(self.config)
self.logger = logger
logger.info("🔧 [INIT] Configurando EmotionAnalyzer...")
self.emotion_analyzer = config.get_emotion_analyzer(getattr(self.config, 'NLP_CONFIG', None))
logger.info("🔧 [INIT] Configurando WebSearch...")
self.web_search = get_web_search()
logger.info("✅ [INIT] WebSearch OK")
# 🔧 NOVOS GERENCIADORES DE CONTEXTO
try:
logger.info("🔧 [INIT] Conectando ao Database...")
self.db = Database(getattr(self.config, 'DB_PATH', 'kiami.db'))
logger.info("✅ [INIT] Database OK")
# ✅ DEDUP CLEANUP: limpa registros antigos a cada 6h
if self.db:
def _dedup_cleanup_loop():
while True:
try:
time.sleep(21600) # 6h
self.db.cleanup_old_dedup(24)
except Exception:
pass
_cleanup_thread = threading.Thread(target=_dedup_cleanup_loop, daemon=True)
_cleanup_thread.start()
except Exception as e:
logger.warning(f"Falha ao inicializar Database: {e}")
self.db = None
# ═══ AUTONOMOUS AGENT: Inicializa motor de decisão autónoma ═══
if _autonomous_agent:
try:
logger.info("🔧 [INIT] Configurando Autonomous Agent...")
def _llm_caller(system_prompt: str, user_prompt: str) -> str:
"""Wrapper para LLM usado pelo autonomous_agent em decisões complexas."""
response_tuple = self.providers.generate(
user_prompt=user_prompt,
context_history=[{"role": "system", "content": system_prompt}]
)
# generate() retorna (response, model) — extrai só a resposta
if isinstance(response_tuple, tuple) and len(response_tuple) >= 1:
res = response_tuple[0]
else:
res = response_tuple
if isinstance(res, dict):
return res.get("response", res.get("content", ""))
return str(res or "")
_autonomous_agent.init(db_instance=self.db, llm_caller=_llm_caller)
logger.info("🤖 [AUTONOMOUS AGENT] Motor de decisão autónoma inicializado com DB + LLM")
except Exception as aa_err:
logger.warning(f"⚠️ [AUTONOMOUS AGENT] Falha na inicialização: {aa_err}")
# ContextIsolationManager é singleton — não aceita argumentos no construtor
try:
self.context_manager = ContextIsolationManager()
except Exception as e:
logger.warning(f"ContextIsolationManager falhou: {e}")
self.context_manager = None
# ShortTermMemoryManager (de unified_context) — obtido via factory
try:
self.stm_manager = get_stm_manager()
except Exception as e:
logger.warning(f"ShortTermMemoryManager falhou: {e}")
self.stm_manager = None
# UnifiedContextBuilder — obtido via factory e configurado manualmente
try:
self.unified_builder = get_unified_context_builder()
# Injeta dependências na instância obtida via singleton
if self.unified_builder:
self.unified_builder.stm_manager = self.stm_manager
self.unified_builder.context_manager = self.context_manager
self.unified_builder.db = self.db
except Exception as e:
logger.warning(f"UnifiedContextBuilder falhou: {e}")
self.unified_builder = None
# 🧠 SESSION MEMORY - Memória persistente entre sessões
self.session_manager = get_session_manager()
if SESSION_MEMORY_AVAILABLE:
logger.success("🧠 Session Memory inicializado com sucesso!")
else:
logger.warning("⚠️ Session Memory indisponível")
# Aprendizado contínuo — integração opcional
self.aprendizado_continuo = None
try:
try:
from .aprendizado_continuo import get_aprendizado_continuo
except ImportError:
from modules.aprendizado_continuo import get_aprendizado_continuo
self.aprendizado_continuo = get_aprendizado_continuo(self.db)
logger.success("Aprendizado Continuo integrado")
except Exception as e:
logger.warning(f"Aprendizado Continuo nao disponivel: {e}")
self.aprendizado_continuo = None
self.persona_tracker = PersonaTracker(db=self.db, llm_client=self.providers) if self.db else None
# 🎯 LISTEN ENGINE MANAGER - ISOLAÇÃO DE CONTEXTOS POR GRUPO
self.listen_engine_manager = None
if LISTEN_ENGINE_AVAILABLE:
try:
self.listen_engine_manager = ContextoGrupoManager(
max_grupos=50,
max_msgs_por_grupo=100
)
logger.success("🎯 Listen Engine Manager inicializado com sucesso!")
except Exception as e:
logger.warning(f"⚠️ Listen Engine Manager falhou: {e}")
self.listen_engine_manager = None
# 🔒 SECURE LOGGER - PROTEÇÃO CONTRA THINK LEAK E EXPOSIÇÃO
self.secure_log = None
if HAS_LOG_MASKING:
try:
self.secure_log = SecureLogger(logger)
logger.success("🔒 Secure Logger (Log Masking) ativado com sucesso!")
except Exception as e:
logger.warning(f"⚠️ Secure Logger falhou: {e}")
self.secure_log = None
# 🔧 MUTEX GLOBAL E DEDUP /KIAMI
self._kiami_processing_lock = threading.RLock()
self._akira_processing_lock = self._kiami_processing_lock # alias
self._kiami_dedup_map: Dict[str, float] = {}
self._kiami_dedup_ttl = getattr(self.config, 'KIAMI_DEDUP_TTL', 5)
logger.info("🔧 [INIT] Configurando personalidade...")
self._setup_personality()
logger.info("🔧 [INIT] Configurando rotas...")
self._setup_routes()
# 📧 Gmail auto-reply (softedgecoporation@gmail.com) — só liga com GMAIL_AUTO_REPLY=true
try:
from .gmail_inbox import maybe_start_gmail_loop
if maybe_start_gmail_loop():
logger.success("📧 Gmail auto-reply ATIVO")
else:
logger.info("📧 Gmail auto-reply OFF (GMAIL_AUTO_REPLY != true ou sem credenciais)")
except Exception as _ge:
logger.debug(f"Gmail loop não iniciado: {_ge}")
logger.info("✅ [INIT] KiamiAPI.__init__ completo")
# FastAPI: router é incluído em main.py via app.include_router()
self.nlp_config = None
def _should_inject_group_name(self, mensagem: str, grupo_nome: str) -> bool:
if not grupo_nome:
return False
if not mensagem:
return True
normalized = re.sub(r"\s+", " ", mensagem.lower().strip())
# Mensagens vagas ("estou pensando", "sobre tudo isso") NUNCA devem injetar nome do grupo
vagas = ("estou pensando", "sobre tudo isso", "sobre isso", "tudo isso", "isso", "sobre o quê", "sobre o que", "pensei", "sei la")
if normalized in vagas or (len(normalized.split()) <= 2 and any(w in normalized for w in ("isso", "tudo", "pensando"))):
return False
# Padrão flexível com \s+ para lidar com duplo espaço/typos
pattern = r"\b(?:nome\s+do\s+grupo|qual(?:\s+é|\s+o)?\s+o\s+nome\s+do\s+grupo|como\s+se\s+chama|nome(?:\s+deste|\s+desse)?\s+grupo|nome\s+do\s+chat|que\s+grupo\s+é\s+esse|qual\s+o\s+nome|cacufania|lideran[cç]a)\b"
if re.search(pattern, normalized):
return True
# Fallback: se estiver em grupo e mensagem tem interrogação sobre grupo/nome, injeta também
if "?" in normalized and ("nome" in normalized or "grupo" in normalized):
return True
return False
def _setup_personality(self):
self.nlp_config = getattr(self.config, 'NLP_CONFIG', None)
# 🔧 Tenta carregar persona do PG primeiro
persona_db = {}
try:
if self.db and hasattr(self.db, 'get_persona_config'):
persona_db = self.db.get_persona_config()
except Exception:
pass
if persona_db:
self.persona = {
'nome': persona_db.get('nome', 'Kiami'),
'nacionalidade': persona_db.get('nacionalidade', 'Angolana'),
'personalidade': persona_db.get('personalidade', 'Fria, descolada, sarcástica, tímida'),
'tom_voz': persona_db.get('tom_voz', 'Seca, natural, curta, sarcástica'),
'numero': persona_db.get('numero', getattr(config, 'BOT_PUBLIC_NUMBER', '956464620')),
'idade': persona_db.get('idade', '18 anos'),
'idioma': persona_db.get('idioma', 'Português angolano'),
}
self.logger.info("✅ [PERSONA] Carregada do PostgreSQL")
else:
persona_cfg = getattr(self.config, 'PersonaConfig', None)
if persona_cfg:
self.persona = {
'nome': getattr(persona_cfg, 'nome', 'Kiami'),
'nacionalidade': getattr(persona_cfg, 'nacionalidade', 'Angolana'),
'personalidade': getattr(persona_cfg, 'personalidade', 'Profissional, neutra, séria, patriótica, curta'),
'tom_voz': getattr(persona_cfg, 'tom_voz', 'Extremamente curto, profissional angolano'),
}
else:
self.persona = {
'nome': 'Kiami',
'nacionalidade': 'Angolana',
'personalidade': 'Profissional, neutra, sarcástica, tímida',
'tom_voz': 'Extremamente curto, profissional, curta, sarcástica',
}
def _get_akira_dedup_key(self, message_id: str, usuario: str, numero: str, mensagem: str, tipo_conversa: str, grupo_id: str) -> str:
if message_id:
return f"id:{message_id}"
raw = f"{usuario}:{numero}:{tipo_conversa}:{grupo_id}:{mensagem[:200]}"
return hashlib.md5(raw.encode('utf-8')).hexdigest()
def _cleanup_akira_dedup(self) -> None:
now = time.time()
expired = [k for k, ts in self._kiami_dedup_map.items() if now - ts > self._kiami_dedup_ttl]
for key in expired:
self._kiami_dedup_map.pop(key, None)
def _setup_routes(self):
@self.api.route('/treino/sniff', methods=['POST'])
async def sniff_endpoint(request: FastAPIRequest):
try:
data = await request.json()
if not data:
return jsonify({"error": "Payload vazio"}, 400)
channel_name = data.get("channelName", "unknown")
content = data.get("content", "").strip()
timestamp = data.get("timestamp")
if content and len(content) > 5:
db = self.db if self.db else Database(getattr(self.config, 'DB_PATH', 'kiami.db'))
db.salvar_aprendizado_detalhado(
f"sniff_{channel_name}",
f"newsletter_{int(time.time())}",
json.dumps({"content": content, "timestamp": timestamp}, ensure_ascii=False)
)
self.logger.info(f"📡 [SNIFF] Dados de '{channel_name}' absorvidos para o dataset de treino.")
return jsonify({"status": "ok", "message": "Corpus guardado silenciosamente"}, 200)
except Exception as e:
self.logger.error(f"[API] Erro no /treino/sniff: {e}")
return jsonify({"error": str(e)}, 500)
@self.api.post('/generate-image')
async def generate_image_endpoint(request: FastAPIRequest):
try:
import base64
data = await request.json()
prompt = data.get('prompt', '')
aspect_ratio = data.get('aspect_ratio', '1:1')
model = data.get('model', 'flux')
if not prompt:
return JSONResponse(content={"error": "Prompt vazio"}, status_code=400)
from .google_image_gen import get_google_image_gen
generator = get_google_image_gen()
res = generator.generate(prompt, aspect_ratio, model)
if res.get('success'):
img_b64 = base64.b64encode(res['buffer']).decode('utf-8')
return JSONResponse(content={
"success": True,
"image_b64": img_b64,
"mime_type": res.get('mime_type', 'image/png'),
"model": res.get('model', 'imagen-3')
})
else:
return JSONResponse(content={"success": False, "error": res.get('error')}, status_code=500)
except Exception as e:
self.logger.error(f"[API] Erro no /generate-image: {e}")
return JSONResponse(content={"error": str(e)}, status_code=500)
@self.api.get('/timers/pending')
async def timers_pending():
"""Retorna timers/lembretes pendentes. Stub — implementar quando necessário."""
return JSONResponse(
content={"success": True, "timers": []},
headers={"Cache-Control": "max-age=2"},
)
@self.api.post('/akira')
@self.api.post('/kiami')
async def akira_endpoint(request: FastAPIRequest):
# 🚀 GLOBAL TIMEOUT: evita que Baileys (180s) timeout sem resposta — retorna fallback em 25s
_akira_start = __import__('time').time()
# Variáveis de controle do semáforo (inicializadas antes do try para o finally)
_sem = None
_sem_acquired = False
try:
# Captura robusta de JSON
raw_data = await request.body()
try:
# Tenta extrair o JSON perfeitamente
data = await request.json()
if data is None:
# Se falhou, tenta decodificar manualmente o bruto
decoded = raw_data.decode('utf-8', errors='ignore').strip()
data = json.loads(decoded) if decoded else {}
except Exception as e:
self.logger.error(f"[API] Falha crítica ao decodificar JSON: {e} | Bruto: {raw_data[:200]}")
data = {}
if not data:
raw_str = raw_data.decode('latin-1', errors='replace') if raw_data else "Vazio"
self.logger.error(f"[API] Payload JSON vazio | Bruto: {raw_str[:300]}")
return JSONResponse(content={'error': 'Payload vazio'}, status_code=400)
# 🔍 DEBUG: Log dos campos recebidos (só keys, não valores grandes)
_doc_check = 'documento' in data or 'documento_dados' in data
_img_check = 'imagem' in data or 'imagem_dados' in data
if _doc_check or _img_check:
self.logger.info(f"[API] Campos recebidos: documento={_doc_check} | imagem={_img_check} | keys={list(data.keys())}")
usuario = data.get('usuario', 'anonimo')
numero = data.get('numero', '')
mensagem = data.get('mensagem', '')
message_id = data.get('message_id', '')
tipo_conversa = data.get('tipo_conversa', 'pv')
grupo_id = data.get('grupo_id') or data.get('contexto_grupo') or ''
nome_usuario = data.get('nome_usuario', usuario) # ✅ Nome real do utilizador
usuario = validate_sender_name(usuario, numero, "usuario_principal")
# ✅ IDMPOTENCY CHECK (Camada 2 — com DB, para persistência entre reinícios)
# Valida que a mensagem atual é a MESMA que gerou o cache.
# Sem isto, um retry/timeout do BotCore com message_id reutilizado
# devolvia a resposta de OUTRA pergunta (repetição fantasma).
if message_id and self.db:
try:
ja_respondido = self.db.recuperar_resposta_por_id(message_id)
if ja_respondido:
_msg_cache = (ja_respondido.get('mensagem') or '').strip()
_msg_atual = (mensagem or '').strip()
if _msg_cache and _msg_atual and _msg_cache != _msg_atual:
self.logger.warning(
f"⚠️ [IDEMPOTENCY-DB] message_id reutilizado com texto diferente "
f"(cache={_msg_cache[:60]!r} vs atual={_msg_atual[:60]!r}) — ignorando cache."
)
else:
self.logger.info(f"♻️ [IDEMPOTENCY-DB] Reenviando resposta já gerada para {message_id}")
return jsonify({
'resposta': ja_respondido['resposta'],
'cached': True,
'modelo_usado': ja_respondido.get('modelo_usado', 'desconhecido')
})
except Exception as _idem_err:
self.logger.warning(f"[IDEMPOTENCY] DB check falhou (ok, continuando): {_idem_err}")
# ✅ SEMÁFORO POR CONVERSA (Camada 2 — serializa req. do mesmo usuário)
# ⚡ OTIMIZAÇÃO: timeout reduzido de 25s para 3s para evitar thread starvation sob carga
# Garante que a mesma conversa não processa 2 mensagens em simultâneo.
# Liberado no finally abaixo, mesmo que ocorra exceção.
_conv_key = f"{numero}:{data.get('grupo_id') or 'pv'}"
_sem = _get_conv_semaphore(_conv_key)
# Enqueue request to per-conversation FIFO if someone is processing
# FIX 2026-08-20: usa asyncio.to_thread para não bloquear event loop — permite paralelismo entre conversas diferentes
evt, pos = _enqueue_conv_request(_conv_key)
if pos > 1:
# 🚀 FIX TIMEOUT: reduzir 5min → 25s para não estourar timeout 180s do Baileys (Railway)
self.logger.info(f"⏳ [QUEUE] Conversa {_conv_key[:30]} ocupada. posição {pos}, aguardando até 25s.")
waited = await asyncio.to_thread(evt.wait, 25)
if not waited:
with _CONV_QUEUE_LOCK:
q = _CONV_QUEUES.get(_conv_key)
try:
if q and evt in q:
q.remove(evt)
except Exception:
pass
self.logger.warning(f"⏳ [QUEUE TIMEOUT] Conversa {_conv_key[:30]} tempo de espera excedido (25s), respondendo timeout_concorrencia")
return JSONResponse(content={'resposta': '', 'status': 'timeout_concorrencia_queue'}, status_code=429)
# Our turn — acquire per-conversation semaphore sem bloquear event loop (isolado por numero:grupo)
# Timeout de 30s: antes era acquire() infinito — se o worker anterior
# morresse sem libertar, todos os pedidos seguintes dessa conversa
# ficavam presos para sempre ("respostas não chegam").
_sem_acquired = await asyncio.to_thread(_sem.acquire, True, 30)
if not _sem_acquired:
try:
_dequeue_and_notify_next(_conv_key)
except Exception:
pass
self.logger.warning(f"⏳ [SEMAPHORE TIMEOUT] Conversa {_conv_key[:30]} semáforo ocupado >30s")
return JSONResponse(content={'resposta': '', 'status': 'timeout_concorrencia_queue'}, status_code=429)
# Novos campos para imagens
imagem_dados = data.get('imagem', {})
tem_imagem = bool(imagem_dados.get('dados'))
analise_visao = imagem_dados.get('analise_visao', {})
mensagem_citada = data.get('mensagem_citada', '')
reply_metadata = data.get('reply_metadata', {})
is_reply = reply_metadata.get('is_reply', False)
reply_to_bot = reply_metadata.get('reply_to_bot', False)
quoted_author_name = reply_metadata.get('quoted_author_name', '')
quoted_author_numero = reply_metadata.get('quoted_author_numero', '')
quoted_type = reply_metadata.get('quoted_type', 'texto')
quoted_text_original = reply_metadata.get('quoted_text_original', '')
context_hint = reply_metadata.get('context_hint', '')
# 🔧 SENDER FIX: Apply validation to quoted_author_name
if is_reply and quoted_author_numero:
quoted_author_name = validate_sender_name(quoted_author_name, quoted_author_numero, "quoted_author")
# ⚠️ SELF-REPLY RECOGNITION
# Check if the quoted author is the bot itself
quoted_author_pure = extract_pure_number(quoted_author_numero)
bot_id_pure = extract_pure_number(config.BOT_NUMERO if hasattr(config, 'BOT_NUMERO') else '40755431264474')
is_quoted_from_bot = (quoted_author_pure and bot_id_pure and
quoted_author_pure == bot_id_pure)
if is_quoted_from_bot and is_reply:
self.logger.info(f"🔄 [REPLY AO BOT] Usuário está respondendo a Kiami ({quoted_author_pure}). mantendo contexto.")
reply_to_bot = True
quoted_author_name = "Kiami (você mesma)"
quoted_author_numero = config.BOT_NUMERO
# 🔧 CORREÇÃO: Detectar reply quando mensagem_citada existe mas reply_metadata está vazio
pv_reply_detected = False
if not is_reply and mensagem_citada and not reply_metadata.get('is_reply'):
is_reply = True
quoted_text_original = quoted_text_original or mensagem_citada
# Somente marque como reply_to_bot quando estiver em PV, o autor citado for claramente o bot,
# a mensagem citada contiver uma menção direta à Kiami, ou o quoted_author_name indicar que é o bot.
quoted_author_name_lower = (quoted_author_name or '').strip().lower()
quoted_by_name_is_bot = any(token in quoted_author_name_lower for token in ['Kiami', 'Beu', 'assistente'])
quoted_text_lower = mensagem_citada.lower()
quoted_text_mentions_bot = any(token in quoted_text_lower for token in ['Kiami', 'bot', 'assistente'])
if tipo_conversa == 'pv' or is_quoted_from_bot or quoted_by_name_is_bot or quoted_text_mentions_bot:
reply_to_bot = True
quoted_author_name = quoted_author_name or "Kiami (você mesma)"
quoted_author_numero = quoted_author_numero or config.BOT_NUMERO
self.logger.info("[REPLY FALLBACK] Mensagem citada sem reply_metadata em PV/quoted-from-bot/by-name/text. Marcando reply_to_bot=True.")
else:
# Em grupo, não assuma que toda mensagem citada é para o bot.
reply_to_bot = False
if not quoted_author_name:
quoted_author_name = "participante_desconhecido"
self.logger.info("[REPLY FALLBACK] Mensagem citada sem reply_metadata em grupo. Mantendo reply_to_bot=False.")
pv_reply_detected = (tipo_conversa == 'pv')
# Preenche hint de contexto quando não veio via reply_metadata.
if is_reply and not context_hint and quoted_text_original:
lower_quoted = quoted_text_original.lower()
if any(w in lower_quoted for w in ['akira', 'bot', 'você', 'vc', 'tu']):
context_hint = 'pergunta_sobre_akira'
elif any(w in lower_quoted for w in ['oq', 'o que', 'qual', 'quanto', 'onde', 'quando', 'por que', 'porque']):
context_hint = 'pergunta_factual'
else:
context_hint = 'contexto_geral'
# Se não houver nome do autor, tente extrair da mensagem citada um prefixo estilo 'Kiami:' ou 'Bot:'
if not quoted_author_name or quoted_author_name == '':
match = re.match(r'^\s*(akira|bot|assistente)[: ,]', mensagem_citada.lower())
if match:
quoted_author_name = "Kiami (você mesma)"
quoted_author_numero = quoted_author_numero or config.BOT_NUMERO
reply_to_bot = True
self.logger.info("[REPLY FALLBACK] Inferido autor citado como Kiami pela mensagem_citada.")
self.logger.info(f"[REPLY DETECTADO] Mensagem citada encontrada sem reply_metadata (tipo_conversa={tipo_conversa}, reply_to_bot={reply_to_bot})")
# tipo_conversa e grupo_id já foram extraídos no início para dedup (linha ~1253-1254)
tipo_mensagem = data.get('tipo_mensagem', 'texto')
grupo_nome = data.get('grupo_nome', '')
forcar_busca = data.get('forcar_busca', False)
analise_doc = data.get('analise_doc', '')
# 🔧 ANTI-DUPLICATION /KIAMI (PostgreSQL-based — works across workers)
dedup_key = self._get_akira_dedup_key(
message_id=message_id,
usuario=usuario,
numero=numero,
mensagem=mensagem,
tipo_conversa=tipo_conversa,
grupo_id=grupo_id or ''
)
# Fallback: in-memory dedup for same-process rapid duplicates
with self._kiami_processing_lock:
self._cleanup_akira_dedup()
if dedup_key in self._kiami_dedup_map:
self.logger.warning(
f"♻️ [KIAMI DEDUP] Requisição duplicada detectada (memória): usuario={usuario} numero={numero} tipo={tipo_conversa}"
)
return jsonify({'status': 'duplicate', 'message': 'Mensagem duplicada recebida'}, 200)
self._kiami_dedup_map[dedup_key] = time.time()
# Cross-worker dedup via PostgreSQL (atomic claim)
if self.db:
try:
if not self.db.claim_dedup(dedup_key, message_id=message_id, usuario=usuario, numero=numero):
self.logger.warning(
f"♻️ [KIAMI DEDUP-PG] Requisição duplicada entre workers: usuario={usuario} numero={numero} tipo={tipo_conversa}"
)
return jsonify({'status': 'duplicate', 'message': 'Mensagem duplicada entre workers'}, 200)
except Exception as _dedup_err:
self.logger.debug(f"⚠️ [DEDUP-PG] Erro (continuando): {_dedup_err}")
# ✅ NOVOS CAMPOS DE VALIDAÇÃO (TypeScript/BotCore)
# Only override self-response flags if NOT already set by PV reply detection
if not pv_reply_detected:
is_bot_self_response = data.get('is_bot_self_response', False)
sender_is_bot = data.get('sender_is_bot', False)
else:
# Preserve the flags set by PV reply detection
is_group_payload = data.get('is_group', False)
is_bot_self_response = False # PV reply não é self-response
sender_is_bot = False
# ✅ PROTEÇÃO DUPLA: Rejeitar se mensagem é do próprio bot
# 1) Flag explícita do BotCore
# 2) Flag sender_is_bot do BotCore
# 3) Comparação do número do remetente com o número do bot (fallback robusto)
sender_pure = extract_pure_number(str(numero))
is_sender_bot = (
is_bot_self_response
or sender_is_bot
or (sender_pure and bot_id_pure and sender_pure == bot_id_pure)
)
if is_sender_bot:
self.logger.warning(f"[PROTEÇÃO] Self-response detectada: is_bot_self_response={is_bot_self_response}, sender_is_bot={sender_is_bot}, sender_pure={sender_pure}=={bot_id_pure}")
return jsonify({'error': 'Bot não responde a si mesmo'}, 400)
# ✅ VALIDAR COERÊNCIA: tipo_conversa é a fonte de verdade (vem do remoteJid)
# is_group é apenas redundante (pode ter falhas na transmissão)
if tipo_conversa == 'grupo':
is_group_payload = True
else:
is_group_payload = False
if not mensagem and not tem_imagem:
return jsonify({'error': 'Mensagem vazia'}, 400)
_msg_stripped = (mensagem or "").strip()
_msg_lower = _msg_stripped.lower()
if re.match(r'^\s*(beu|morena|b[eê]u|kiami|mimi)\s*[\?!\.]*\s*$', _msg_lower):
self.logger.info(f"⚡ [APELIDO FASTPATH] '{_msg_stripped}' -> oi")
try:
if message_id and self.db:
self.db.salvar_resposta_por_id(message_id, "oi", modelo_usado="nickname_fastpath")
except Exception:
pass
return jsonify({'resposta': 'oi', 'modelo_usado': 'nickname_fastpath', 'cached': False})
# 🚀 FASTPATH SAUDAÇÃO SIMPLES: Oi/bom dia etc não precisam de pipeline pesado (evita 126s)
if re.match(r'^\s*(oi|olá|ola|eae|eai|bom dia|boa tarde|boa noite|oie|hey|hello|hi)\s*[\?!\.]*\s*$', _msg_lower):
import random
_greetings = ["Oi.", "Olá!", "E aí!", "Oie!", "Hey!", "Beleza?"]
_fast_resp = random.choice(_greetings)
self.logger.info(f"⚡ [GREETING FASTPATH] '{_msg_stripped}' -> '{_fast_resp}' (bypass 120s pipeline)")
try:
if message_id and self.db:
self.db.salvar_mensagem(usuario=usuario, mensagem=mensagem, resposta=_fast_resp, numero=numero, modelo_usado="greeting_fastpath", message_id=message_id, nome_usuario=nome_usuario)
except Exception:
pass
return jsonify({'resposta': _fast_resp, 'modelo_usado': 'greeting_fastpath', 'cached': False})
contexto_log = f" [Grupo: {grupo_nome}]" if tipo_conversa == 'grupo' and grupo_nome else " [PV]"
# 🔒 LOG MASKING: Proteger número de usuário em logs
if self.secure_log:
self.secure_log.checkpoint(
user_id=numero,
user_name=usuario,
message_type=tipo_mensagem,
is_group=(tipo_conversa == 'grupo'),
group_name=grupo_nome if tipo_conversa == 'grupo' else None,
message_content=mensagem
)
else:
self.logger.info(f"{usuario} ({numero}){contexto_log}: {mensagem[:120]} | tipo: {tipo_mensagem} | reply_to_bot={reply_to_bot} | is_group={is_group_payload}")
# Injeta o contexto no prompt enviando-o via kwargs de contexto unificado se suportado, senão no reply_metadata
if is_reply and grupo_nome:
reply_metadata['grupo_nome'] = grupo_nome
# 🔧 UNIFIED MEDIA PIPELINE (Sincronização Global)
# Mantém analise_visao se já veio preenchida (ex: cache do client), senão inicia None
analise_visao = analise_visao if analise_visao else None
# 1. Processamento de Imagem (imagem ou imagem_dados)
img_data = data.get('imagem') or data.get('imagem_dados')
if img_data:
try:
caminho_local = img_data.get('path')
dados_b64 = img_data.get('dados', '')
vision_input = caminho_local if (caminho_local and os.path.exists(caminho_local)) else dados_b64
if vision_input:
self.logger.info(f"[VISION] Analisando imagem via {'PATH' if (caminho_local and os.path.exists(caminho_local)) else 'BASE64'} (Tamanho: {len(vision_input) if isinstance(vision_input, str) else len(vision_input)} chars/bytes)")
vision_res = get_computer_vision().analyze_image(vision_input, user_id=numero)
if vision_res.get('success'):
analise_visao = vision_res
tem_imagem = True
self.logger.info(f"[VISION] Descrição: {analise_visao.get('description', '')[:100]}...")
else:
self.logger.warning(f"[VISION] Falha na análise: {vision_res.get('error')}")
else:
self.logger.warning("[VISION] img_data presente mas vision_input vazio (sem path ou dados)")
except Exception as ve:
self.logger.error(f"Erro no processamento Vision: {ve}")
# 2. Processamento de Vídeo (video ou video_dados)
vid_data = data.get('video') or data.get('video_dados')
if vid_data:
try:
caminho_vid = vid_data.get('path')
if caminho_vid and os.path.exists(caminho_vid):
self.logger.info(f"[VIDEO] Vídeo detectado em: {caminho_vid}")
# Nota: A IA receberá a descrição textual do vídeo por enquanto
if not analise_visao:
analise_visao = {"description": f"Foi enviado um vídeo localizado em {caminho_vid}. Analise o contexto da conversa sobre este vídeo."}
except Exception as ve:
self.logger.error(f"Erro no processamento Vídeo: {ve}")
# 3. Processamento de Documento (documento ou documento_dados)
doc_data = data.get('documento') or data.get('documento_dados')
if doc_data:
try:
doc_path = doc_data.get('path')
doc_name = doc_data.get('nome_arquivo', 'documento')
doc_b64 = doc_data.get('dados', '')
doc_mime = doc_data.get('mime_type', 'application/pdf')
self.logger.info(f"📄 [DOC] Recebido: {doc_name} | mime={doc_mime} | base64={len(doc_b64)} chars")
if doc_path and os.path.exists(doc_path):
self.logger.info(f"📄 Analisando documento (path): {doc_name}")
doc_res = get_document_analyzer().analyze_file(doc_path, query=mensagem or "Resuma este documento")
if doc_res.get('success'):
analise_doc = doc_res.get('analysis')
self.logger.info("[DOC AI] ✅ Análise por path concluída")
else:
self.logger.error(f"[DOC AI] ❌ Falha path: {doc_res.get('error')}")
elif doc_b64:
self.logger.info(f"📄 Analisando documento (base64): {doc_name}")
doc_res = get_document_analyzer().analyze_base64(doc_b64, mime_type=doc_mime, file_name=doc_name, query=mensagem or "Resuma este documento")
if doc_res.get('success'):
analise_doc = doc_res.get('analysis')
self.logger.info("[DOC AI] ✅ Análise por base64 concluída")
else:
self.logger.error(f"[DOC AI] ❌ Falha base64: {doc_res.get('error')}")
except Exception as de:
self.logger.error(f"Erro no DocAnalyzer: {de}")
if is_reply and mensagem_citada:
self.logger.info(f"[REPLY] reply_to_bot={reply_to_bot}, autor={quoted_author_name}")
# Gate de comandos privilegiados
non_privileged_attempt = False
if config.is_privileged_command(mensagem) and not config.is_privileged(numero):
non_privileged_attempt = True
# 🔧 CONTEXT ISOLATION: Generate isolated context ID
try:
if self.context_manager is not None:
conversation_id = self.context_manager.get_conversation_id(
usuario=usuario,
conversation_type=tipo_conversa,
group_id=grupo_id if tipo_conversa == 'grupo' else None,
numero=numero
)
else:
from .context_isolation import generate_context_id as _gen_cid
conversation_id = _gen_cid(numero or usuario, tipo_conversa, grupo_id if tipo_conversa == 'grupo' else None)
except Exception as ctx_err:
self.logger.warning(f"[CTX] get_conversation_id falhou: {ctx_err}")
try:
from .context_isolation import generate_context_id as _gen_cid2
conversation_id = _gen_cid2(numero or usuario, tipo_conversa, grupo_id if tipo_conversa == 'grupo' else None)
except Exception:
g_id = grupo_id if tipo_conversa == 'grupo' else "pv"
raw = f"kiami-v21:{tipo_conversa}:{numero or usuario}:{g_id}"
conversation_id = hashlib.sha256(raw.encode()).hexdigest()
dossie = None
try:
from .user_profiler import get_user_profiler
dossie = get_user_profiler().get_user_profile(numero or usuario)
except Exception as prof_err:
self.logger.warning(f"Erro ao obter dossiê: {prof_err}")
# 🔧 FIX: Passa conversation_id para garantir que o cache é isolado
contexto = self._get_user_context(usuario, conversation_id=conversation_id)
# O conversation_id já deve estar no objeto contexto via construtor ou setter
contexto.conversation_id = conversation_id
historico = contexto.obter_historico()
analise = contexto.analisar_intencao_e_normalizar(mensagem, historico)
# 🔥 Inicializa aggression_profile antes de qualquer uso
_aggression_profile = None
# 🧠 ATUALIZA PERFIL EMOCIONAL DO USUÁRIO (Rancor, Histórico e Hostilidade)
try:
from .profile_user_emotion import get_emotional_profile_manager
ep_mgr = get_emotional_profile_manager()
emocao_detectada = analise.get('emocao', 'neutral') if isinstance(analise, dict) else 'neutral'
confianca = analise.get('confianca_emocao', 0.5) if isinstance(analise, dict) else 0.5
# 🧠 CRUZAMENTO: BART detecta emoção, regex detecta agressão real
# Se regex não encontrou nada agressivo, BART não deve gerar hostility alta
_regex_agg = _aggression_profile.get('aggression_level', 0) if _aggression_profile else 0
_bart_hostility = int(confianca * 100) if emocao_detectada in ['raiva', 'agressivo', 'hostil', 'anger', 'hostile', 'aggressive'] else 0
# Só aplica hostility do BART se regex confirmou agressão (ou se regex não correu)
_final_hostility = _bart_hostility if _regex_agg >= 10 else 0
ep_mgr.update_emotion(
user_id=numero or usuario,
emotion=emocao_detectada,
hostility_score=_final_hostility
)
if any(word in mensagem.lower() for word in getattr(config, 'PALAVRAS_RUDES', [])):
profile = ep_mgr.get_or_create_profile(numero or usuario)
if profile.get_hostility_level() >= 40:
ep_mgr.mark_as_hostile(numero or usuario)
except Exception as ep_err:
self.logger.warning(f"Erro ao atualizar perfil emocional: {ep_err}")
# Marcação de tentativa não-privilegiada
try:
if non_privileged_attempt and isinstance(analise, dict):
analise['non_privileged_command'] = True
analise['command_attempt'] = mensagem
except Exception:
pass
# Gate de tom "amor" (love)
try:
emocao_detectada = analise.get('emocao') if isinstance(analise, dict) else None
if emocao_detectada == 'amor' or emocao_detectada == 'love':
if not self.emotion_analyzer.can_transition_tone('love', historico):
analise['forcar_downshift_love'] = True
except Exception:
pass
# 🔧 UNIFIED CONTEXT: Build complete context including STM and Reply Context
import time as _tStep
_step_t0 = _tStep.time()
unified_context = None
if getattr(self, 'unified_builder', None) and conversation_id:
try:
_step_t0b = _tStep.time()
reply_metadata_robust: Dict[str, Any] = dict(reply_metadata) if reply_metadata else {}
if is_reply:
reply_metadata_robust.update({
"is_reply": True,
"reply_to_bot": reply_to_bot,
"quoted_text_original": quoted_text_original,
"quoted_author_name": quoted_author_name,
"quoted_author_numero": quoted_author_numero,
"quoted_type": quoted_type,
"context_hint": context_hint,
"mensagem_citada": mensagem_citada,
"replied_to_author": reply_metadata.get('replied_to_author_name', ''),
"replied_to_content": reply_metadata.get('replied_to_text', '')
})
# CORREÇÃO: Se autor é desconhecido mas é reply_to_bot
if reply_to_bot and (not quoted_author_name or quoted_author_name == 'desconhecido'):
quoted_author_name = "Kiami (você mesma)"
reply_metadata_robust['quoted_author_name'] = quoted_author_name
unified_context = build_unified_context(
conversation_id=conversation_id,
user_id=numero if tipo_conversa != 'grupo' else f"{numero}_{usuario}",
reply_metadata=reply_metadata_robust if is_reply else None,
current_message=mensagem,
current_emotion=analise.get('emocao', 'neutral') if isinstance(analise, dict) else 'neutral',
grupo_id=grupo_id,
tipo_conversa=tipo_conversa,
numero_usuario=numero
)
if unified_context and grupo_nome and self._should_inject_group_name(mensagem, grupo_nome):
unified_context.system_override = (unified_context.system_override or "") + f"\n[FATO ABSOLUTO]: O grupo atual é '{grupo_nome}'. Quando perguntarem o nome do grupo, a resposta é '{grupo_nome}'."
self.logger.info(f"✅ [CONTEXT] Grupo CRÍTICO injetado: '{grupo_nome}'")
elif unified_context and grupo_nome:
self.logger.debug(f"🔒 [CONTEXT] Grupo nome disponível mas não injetado: user message not asking group name.")
# INTROMISSÃO FIX: se reply_to_bot mas replied_to_author vazio, busca no STM quem foi destinatário original de mensagem_citada
# Exige citação com 40+ chars: citação curta ("ok", "sim") casa com
# qualquer resposta antiga e marcava CENTRO/TERCEIRO errado
# (bot respondia a quem não perguntou).
_cit = (mensagem_citada or "").strip()
if is_reply and reply_to_bot and unified_context and not getattr(unified_context, 'replied_to_author', '') and len(_cit) >= 40:
try:
for idx, m in enumerate(unified_context.stm_messages or []):
if m.role == "assistant" and _cit[:40] in (m.content or ""):
for prev in reversed((unified_context.stm_messages or [])[:idx]):
if prev.role == "user":
unified_context.replied_to_author = getattr(prev, 'author_name', '') or prev.content[:30]
unified_context.replied_to_content = (prev.content or "")[:200]
reply_metadata_robust['replied_to_author'] = unified_context.replied_to_author
reply_metadata_robust['replied_to_content'] = unified_context.replied_to_content
self.logger.info(f"🔗 [INTROMISSÃO] Detectado destinatário original: '{unified_context.replied_to_author}' para citação '{mensagem_citada[:30]}'")
break
break
except Exception:
pass
except Exception as e:
self.logger.warning(f"Error building unified context: {e}")
_step_t1 = _tStep.time()
self.logger.info(f"⏱️ [STEP-TIMING] build_unified_context: {_step_t1-_step_t0:.2f}s")
web_content = ""
# 🛡️ ANTI-HALLUCINATION: Não pesquisar se o remetente é um bot conhecido
# BotCore taggeia bots conhecidos com "BOT:" no nome do usuário
is_sender_known_bot = str(usuario).startswith('BOT:')
# 🔧 FIX 2026-08-27: bloquear busca para mensagens vagas triviais (ex: "Deixa a outra ela não quer ajuda" = 7 palavras sem ?)
_msg_lower_trivial = (mensagem or "").lower().strip()
_is_vaga_trivial = len(_msg_lower_trivial.split()) <= 7 and "?" not in _msg_lower_trivial and not forcar_busca
_interpessoal = any(w in _msg_lower_trivial for w in ("deixa", "outra", "ela não quer", "estou pensando", "sobre isso", "sobre tudo", "tanto faz", "odeio", "não fale", "ela não"))
if _is_vaga_trivial and _interpessoal:
precisa_pesquisar = False
self.logger.info(f"🛡️ [SEARCH BLOCK] busca bloqueada para mensagem vaga trivial: '{mensagem[:50]}'")
else:
# Upgrade: Pesquisa Autônoma com 3 camadas de heurística e histórico
# Bots conhecidos NÃO disparam pesquisa autônoma (evita loops)
precisa_pesquisar = not is_sender_known_bot and (forcar_busca or deve_pesquisar(mensagem, historico))
# FIX: memoriza que pesquisa era precisa ANTES dos skips (elapsed/casual),
# para a guarda SEM-PESQUISA saber que deve proibir promessas.
_search_was_needed = bool(precisa_pesquisar)
# FIX 2026-09-13: web_search síncrona custa ~12-14s e estoura TIMEOUT GUARD.
# Pula busca se já gastamos >15s, ou se for pergunta casual/pessoal (fazes agora, estás bem).
if precisa_pesquisar:
_elapsed_pre_search = __import__('time').time() - _akira_start
_msg_casual = (mensagem or "").lower()
_is_casual = any(w in _msg_casual for w in ("fazes de interessante", "faz oque", "faz oquê", "estás bem", "está bem", "como vai", "e você", "e voce", "sério", "serio"))
if _elapsed_pre_search > 15:
self.logger.warning(f"⏭️ [SEARCH SKIP] {_elapsed_pre_search:.1f}s já decorridos — pulando web_search para não estourar guard.")
precisa_pesquisar = False
elif _is_casual and not forcar_busca:
self.logger.info(f"🛡️ [SEARCH BLOCK] pergunta casual/pessoal, sem busca: '{mensagem[:50]}'")
precisa_pesquisar = False
if precisa_pesquisar:
termo_pesquisa = extrair_pesquisa(mensagem)
# FIX: se a extração falhou, usa a mensagem limpa como query —
# nunca deixar "precisa pesquisar" sem termo (virava "vou pesquisar" vazio).
if not (termo_pesquisa or "").strip():
termo_pesquisa = re.sub(r"[^\w\sáéíóúâêãõç]", " ", mensagem or "").strip()
self.logger.info(f"🔍 [SEARCH FALLBACK] termo vazio → usando mensagem: '{termo_pesquisa[:60]}'")
if termo_pesquisa:
self.logger.info(f"🔍 Executando busca autônoma: {termo_pesquisa}")
try:
resultado = self.web_search.pesquisar(termo_pesquisa)
web_content = resultado.get("conteudo_bruto", "")
except Exception as _ws_err:
self.logger.warning(f"⚠️ web_search falhou rápido: {_ws_err}")
web_content = ""
# FIX: pesquisa era precisa mas não aconteceu/entregou nada → proíbe
# o modelo de prometer ("vou pesquisar") e manda responder com o que sabe.
# (Guarda aplicada ao prompt logo após o _build_prompt abaixo.)
_search_failed_guard = ""
if locals().get('_search_was_needed') and not web_content:
_search_failed_guard = (
"\n[SEM-PESQUISA-NESTE-TURNO] A pesquisa web era precisa mas NÃO está disponível "
"agora (sem resultados). Responde com o teu conhecimento, curto. "
"PROIBIDO dizer 'vou pesquisar', 'vou verificar e já digo', 'aguarda', 'já te digo'. "
"NUNCA prometas ação futura — responde agora com o teu conhecimento; 'não sei' só se for TOTALMENTE impossível, e mesmo aí com palpite ('falas do X?')."
)
prompt = self._build_prompt(
usuario, numero, mensagem, analise, contexto, web_content,
mensagem_citada=mensagem_citada,
is_reply=is_reply,
reply_to_bot=reply_to_bot,
quoted_author_name=quoted_author_name,
quoted_author_numero=quoted_author_numero,
quoted_type=quoted_type,
quoted_text_original=quoted_text_original,
context_hint=context_hint,
tipo_conversa=tipo_conversa,
tipo_mensagem=tipo_mensagem,
tem_imagem=tem_imagem,
analise_visao=analise_visao,
analise_doc=analise_doc,
unified_context=unified_context,
dossie=dossie,
conversation_id=conversation_id
)
# Aplica guarda SEM-PESQUISA (definida no bloco de busca acima)
try:
if locals().get('_search_failed_guard'):
prompt = prompt + _search_failed_guard
except Exception:
pass
_step_t2 = _tStep.time()
self.logger.info(f"⏱️ [STEP-TIMING] _build_prompt: {_step_t2-_step_t1:.2f}s")
# 🔬 TIMING GRANULAR — gap _build_prompt -> detect_aggression (era 46s sem log)
_t_gap0 = _tStep.time()
# ✅ PREPARAR CONTEXTO LSTM PARA THINKING ENGINE
# unified_context é um dataclass (não dict), por isso buscamos
# o contexto de longo prazo diretamente do LSTMExtension.
contexto_lstm_para_thinking = None
try:
from .lstm_extension import get_lstm_extension as _get_lstm
_lstm_ext = _get_lstm(self.db)
_ctx_id = conversation_id or numero or usuario
_is_grp = (tipo_conversa == "grupo")
_t_lstm0 = _tStep.time()
contexto_lstm_para_thinking = _lstm_ext.get_context_for_prompt(
context_id=_ctx_id,
numero_usuario=numero,
is_group=_is_grp
)
self.logger.info(f"⏱️ [GAP-TIMING] lstm.get_context_for_prompt: {_tStep.time()-_t_lstm0:.2f}s")
except Exception as _e_lstm:
self.logger.warning(f"lstm get_context falhou: {_e_lstm}")
contexto_lstm_para_thinking = None
self.logger.info(f"⏱️ [GAP-TIMING] pre-context_history: {_tStep.time()-_t_gap0:.2f}s")
_t_gap1 = _tStep.time()
# 🔧 CONTEXT ISOLATION: Passamos as mensagens do STM para o formato nativo do LLM
# Mensagens marcadas como 'observed_only' (vindas do /escutar) representam
# o fluxo passivo do grupo — NÃO são pedidos dirigidos à Kiami.
# Elas entram no histórico com um prefixo claro para o LLM não as confundir
# com intenções direcionadas a ela.
context_history = []
if nome_usuario and usuario:
context_history.append({
"role": "system",
"content": f"[CONTEXTO] Utilizador atual: {nome_usuario} (ID: {usuario}). Usa este nome nas respostas. Esta é a pessoa com quem estás a falar AGORA."
})
if unified_context and unified_context.stm_messages:
# 🚨 CRITICAL FIX: Para replies ao bot, usar SMART CONTEXT BALANCING
# - Carrega últimas 3 mensagens (evita alucinação por noise)
# - MAIS busca inteligente por contexto RELEVANTE mencionado na reply
# Isso mantém isolamento mas permite acesso a referências importantes
if reply_to_bot:
# BASE: Carregar últimas 25 mensagens (contexto imediato expandido)
base_msgs = list(unified_context.stm_messages[-25:])
context_history_base = []
# ✅ FIX 2026-08-30: Identity do utilizador no topo do histórico para CoT nunca perder quem é
if nome_usuario and usuario:
context_history_base.append({
"role": "system",
"content": f"[CONTEXTO] Utilizador atual: {nome_usuario} (ID: {usuario}). Este nome deve ser usado na resposta. NÃO respondas só 'tu/você' — usa o nome!"
})
# 🔧 FIX 2026-08-27: CENTRO vs TERCEIRO (Fulano vs Sicrano)
replied_to_author = getattr(unified_context, 'replied_to_author', '') or ""
interlocutor_principal = replied_to_author
is_intervencao_terceiro = bool(replied_to_author and usuario != replied_to_author)
if is_intervencao_terceiro:
self.logger.info(f"🔀 [INTERVENÇÃO TERCEIRO] {usuario} (TERCEIRO) intervém em thread de {replied_to_author} (CENTRO) | citada: '{(quoted_text_original or mensagem_citada or '')[:40]}'")
last_base_user_author = None # ✅ track last user author for assistant tagging
for msg in base_msgs:
content = msg.content
reply_info = getattr(msg, 'reply_info', {}) or {}
is_observed = reply_info.get('observed_only', False)
if msg.role == "user":
author_name = getattr(msg, 'author_name', '') or ''
if is_observed:
reply_target = ""
if reply_info.get('is_reply') and reply_info.get('quoted_author_name'):
reply_target = f" → {reply_info['quoted_author_name']}"
# 🔧 TAG CENTRO vs TERCEIRO mesmo para observed
centro_tag = "CENTRO" if author_name == interlocutor_principal else "TERCEIRO" if interlocutor_principal else "GRUPO"
label = f"[{centro_tag} | {author_name}{reply_target}]" if interlocutor_principal else f"[GRUPO | {author_name}{reply_target}]"
content = f"{label}: {content}"
else:
# 🔧 TAG CENTRO vs TERCEIRO para histórico direto
if interlocutor_principal:
centro_label = "CENTRO" if author_name == interlocutor_principal else "TERCEIRO" if author_name else ""
if centro_label and author_name and author_name != 'Usuário' and not content.startswith(f'[{centro_label}'):
content = f"[{centro_label} | {author_name}]: {content}"
elif author_name and author_name != 'Usuário' and not content.startswith(f'[{author_name}]'):
content = f"[{author_name}]: {content}"
else:
if author_name and author_name != 'Usuário' and not content.startswith(f'[{author_name}]'):
content = f"[{author_name}]: {content}"
# Track quem foi o último a falar para tagging do assistant
if author_name:
last_base_user_author = author_name
elif msg.role == "assistant":
# ✅ TAG: Marca explicitamente para quem a Kiami estava respondendo + CENTRO
responded_to = (reply_info.get('responded_to') or last_base_user_author or "")
if responded_to:
if responded_to == interlocutor_principal:
content = f"[KIAMIA → CENTRO {responded_to}]: {content}"
else:
content = f"[KIAMIA → {responded_to}]: {content}"
elif last_base_user_author:
content = f"[↩ respondendo a {last_base_user_author}]: {content}"
else:
content = f"[KIAMIA respondeu]: {content}"
context_history_base.append({'role': msg.role, 'content': content})
# 🔥 SMART RETRIEVAL COM THREAD ISOLATION
# FIX: Apenas busca contexto antigo se user EXPLICITAMENTE citar ("você falou sobre X")
# Caso contrário, mantém resposta focada na msg citada (thread atual)
# Detecta se user cita explicitamente uma conversa anterior
has_explicit_mention = bool(re.search(
r'\b(?:você (?:falou|disse|mencionou)|aquele (?:assunto|tema|tópico)|lembra (?:quando|daquela)|daquela (?:conversa|discussão|vez)|anteriormente|antes de)',
mensagem.lower()
))
# Extrai keywords da reply APENAS se houver menção explícita
smart_context_matches = []
if has_explicit_mention:
keywords = re.findall(r'\b([a-záéíóúâêãõç]{4,})\b', mensagem.lower())
keywords = list(set(keywords))[:5]
stop_words = {
'como', 'para', 'mais', 'este', 'esse', 'isso', 'aquilo', 'disse',
'falar', 'falou', 'disso', 'pelo', 'pela', 'tudo', 'nada', 'uma',
'umas', 'uns', 'eles', 'elas', 'você', 'voces', 'vocês', 'akira',
'entao', 'então', 'sobre', 'disseram', 'dizer', 'dizia', 'dele', 'dela',
'aqui', 'ali', 'coisa', 'coisas', 'está', 'estou', 'esteve', 'estava'
}
filtered_keywords = [k for k in keywords if k not in stop_words]
if filtered_keywords:
# Busca APENAS na janela anterior às 10 base (thread recente, não história inteira)
recent_msg_window = unified_context.stm_messages[max(-len(unified_context.stm_messages), -20):-10]
for msg in recent_msg_window:
msg_text = msg.content.lower()
msg_words = re.findall(r'\b([a-záéíóúâêãõç]{3,})\b', msg_text)
matched_keywords = []
for kw in filtered_keywords:
kw_prefix = kw[:4]
has_prefix_match = False
for mw in msg_words:
mw_clean = re.sub(r'[^\w]', '', mw)
if len(mw_clean) >= 4 and mw_clean.startswith(kw_prefix):
has_prefix_match = True
break
if has_prefix_match:
matched_keywords.append(kw)
if matched_keywords:
smart_context_matches.append({
'msg': msg,
'keywords': matched_keywords,
'relevance': len(matched_keywords) / len(filtered_keywords)
})
# Adiciona TOP 1 match mais relevante (apenas 1, não 2)
if smart_context_matches:
smart_context_matches = sorted(smart_context_matches,
key=lambda x: x['relevance'],
reverse=True)[:1]
for match in smart_context_matches:
msg = match['msg']
content = msg.content
reply_info = getattr(msg, 'reply_info', {}) or {}
if msg.role == "user":
author_name = getattr(msg, 'author_name', '') or ''
if author_name and author_name != 'Usuário':
content = f"[{author_name}]: {content}"
context_history_base.insert(0, {
'role': msg.role,
'content': f"[CONTEXTO MENCIONADO]: {content}"
})
self.logger.info(
f"✅ [REPLY CONTEXT] User citou assunto antigo explicitamente. "
f"Recuperado 1 msg (keywords: {', '.join(filtered_keywords[:3])})"
)
else:
self.logger.info(f"✅ [REPLY CONTEXT] Sem menção explícita → focando na thread recente")
context_history = context_history_base
self.logger.info(f"✅ [REPLY SMART BALANCE] {len(context_history)} msgs carregadas (10 base + contexto mencionado se aplicável)")
else:
# NÃO é reply ao bot: carregar msgs com FILTRO DE TÓPICO
# ✅ OTIMIZAÇÃO: Carrega apenas últimas 10 msgs (não 30)
# para evitar que tópicos antigos vaze para a resposta atual.
last_user_author_full = None # ✅ track last user author for assistant tagging
# Extrai keywords da mensagem atual para filtro de relevância
msg_keywords = set(re.findall(r'\b([a-záéíóúâêãõç]{4,})\b', mensagem.lower()))
msg_keywords -= {'como', 'para', 'mais', 'este', 'esse', 'isso', 'aquilo',
'disse', 'falar', 'falou', 'pelo', 'pela', 'tudo', 'nada',
'uma', 'umas', 'uns', 'eles', 'elas', 'você', 'vocês',
'akira', 'então', 'sobre', 'aqui', 'ali', 'coisa', 'está',
'estou', 'porque', 'porque', 'quando', 'onde', 'qual',
'quem', 'isso', 'isso', 'muito', 'bem', 'aqui', 'fazer',
'porque', 'então', 'porque', 'então'}
stm_messages = list(unified_context.stm_messages[-25:]) # Últimas 25 msgs
# Adiciona msgs relevantes de janela maior (até -50) se tiverem keywords em comum
if len(unified_context.stm_messages) > 25:
older_msgs = unified_context.stm_messages[-50:-25]
for omsg in older_msgs:
omsg_keywords = set(re.findall(r'\b([a-záéíóúâêãõç]{4,})\b', omsg.content.lower()))
overlap = msg_keywords & omsg_keywords
if len(overlap) >= 2: # Pelo menos 2 keywords em comum
stm_messages.insert(0, omsg) # Insere no início (mais antigo primeiro)
for msg in stm_messages:
content = msg.content
reply_info = getattr(msg, 'reply_info', {}) or {}
is_observed = reply_info.get('observed_only', False)
if msg.role == "user":
author_name = getattr(msg, 'author_name', '') or ''
if is_observed:
reply_target = ""
if reply_info.get('is_reply') and reply_info.get('quoted_author_name'):
reply_target = f" → {reply_info['quoted_author_name']}"
label = f"[GRUPO | {author_name}{reply_target}]"
content = f"{label}: {content}"
else:
if author_name and author_name != 'Usuário' and not content.startswith(f'[{author_name}]'):
content = f"[{author_name}]: {content}"
if author_name:
last_user_author_full = author_name
elif msg.role == "assistant":
# ✅ TAG: Marca explicitamente para quem a Kiami estava respondendo
if last_user_author_full:
content = f"[↩ respondendo a {last_user_author_full}]: {content}"
else:
content = f"[KIAMIA respondeu]: {content}"
context_history.append({'role': msg.role, 'content': content})
elif not unified_context:
context_history = self._get_history_for_llm(contexto)
# ✔ FIX 2026-08-29: Em reply ao bot, expandir para 5 msgs para não perder thread inicial
# 🔧 FIX 2026-08-30: identity context pode ser cortado pelo slice — garantir que está sempre primeiro
if nome_usuario and usuario:
context_history.insert(0, {
"role": "system",
"content": f"[CONTEXTO] Utilizador atual: {nome_usuario} (ID: {usuario}). Este nome deve ser usado na resposta. NÃO respondas só 'tu/você' — usa o nome!"
})
if reply_to_bot and context_history:
base_history = list(context_history[-5:])
# Detecta se user cita explicitamente uma conversa anterior
has_explicit_mention = bool(re.search(
r'\b(?:você (?:falou|disse|mencionou)|aquele (?:assunto|tema|tópico)|lembra (?:quando|daquela)|daquela (?:conversa|discussão|vez)|anteriormente|antes de)',
mensagem.lower()
))
smart_matches = []
if has_explicit_mention:
# SMART RETRIEVAL: Busca por radicais APENAS nos últimos 10 msgs (thread recente)
keywords = re.findall(r'\b([a-záéíóúâêãõç]{4,})\b', mensagem.lower())
keywords = list(set(keywords))[:5]
stop_words = {
'como', 'para', 'mais', 'este', 'esse', 'isso', 'aquilo', 'disse',
'falar', 'falou', 'disso', 'pelo', 'pela', 'tudo', 'nada', 'uma',
'umas', 'uns', 'eles', 'elas', 'você', 'voces', 'vocês', 'akira',
'entao', 'então', 'sobre', 'disseram', 'dizer', 'dizia', 'dele', 'dela',
'aqui', 'ali', 'coisa', 'coisas', 'está', 'estou', 'esteve', 'estava'
}
filtered_keywords = [k for k in keywords if k not in stop_words]
if filtered_keywords:
# Busca APENAS nos últimos 10 msgs (thread recente)
search_window = base_history[max(-len(base_history), -10):-3] if len(base_history) > 3 else []
for msg in search_window:
msg_text = msg.get('content', '').lower()
msg_words = re.findall(r'\b([a-záéíóúâêãõç]{3,})\b', msg_text)
matched_keywords = []
for kw in filtered_keywords:
kw_prefix = kw[:4]
has_prefix_match = False
for mw in msg_words:
mw_clean = re.sub(r'[^\w]', '', mw)
if len(mw_clean) >= 4 and mw_clean.startswith(kw_prefix):
has_prefix_match = True
break
if has_prefix_match:
matched_keywords.append(kw)
if matched_keywords:
smart_matches.append({
'msg': msg,
'relevance': len(matched_keywords) / len(filtered_keywords)
})
if smart_matches:
smart_matches = sorted(smart_matches, key=lambda x: x['relevance'], reverse=True)[:1]
for match in smart_matches:
msg = match['msg']
base_history.insert(0, {
'role': msg['role'],
'content': f"[CONTEXTO MENCIONADO]: {msg['content']}"
})
self.logger.info(f"✅ [REPLY CONTEXT - SEM STM] User citou assunto. Recuperada 1 msg.")
else:
self.logger.info(f"✅ [REPLY CONTEXT - SEM STM] Sem menção explícita → focando na thread recente")
context_history = base_history
self.logger.info(f"✅ [REPLY ISOLATION] Contexto truncado para {len(context_history)} msgs (reply_to_bot=True, sem menção genérica)")
smart_context_instruction = ""
try:
# Reconstrói metadata robusto
reply_metadata_robust: Dict[str, Any] = dict(reply_metadata) if reply_metadata else {}
if is_reply:
reply_metadata_robust.update({
"is_reply": True,
"reply_to_bot": reply_to_bot,
"quoted_text_original": quoted_text_original,
"quoted_author_name": quoted_author_name,
"quoted_author_numero": quoted_author_numero,
"quoted_type": quoted_type,
"context_hint": context_hint,
"mensagem_citada": mensagem_citada
})
try:
if getattr(self, 'unified_builder', None) and conversation_id:
_ri_pre = None
if is_reply:
_ri_pre = {'is_reply': True,'reply_to_bot': reply_to_bot,'quoted_text_original': quoted_text_original or mensagem_citada,'quoted_author': quoted_author_name or '', 'numero_usuario': numero or usuario, 'grupo_id': grupo_id or '', 'tipo_conversa': tipo_conversa, 'message_id': message_id or ''}
self.unified_builder.add_to_stm(conversation_id=conversation_id, role="user", content=mensagem, author_name=usuario, emocao=analise.get('emocao','neutral') if isinstance(analise, dict) else 'neutral', reply_info=_ri_pre, numero_usuario=numero or usuario, grupo_id=grupo_id or '', tipo_conversa=tipo_conversa, recipient=quoted_author_name or '', quoted_author=quoted_author_name or '', is_listen=False, message_id=message_id or '')
self._stm_user_saved = True
except Exception as _pre_e:
self.logger.debug(f"STM pre-save skip: {_pre_e}")
handler = get_context_handler()
analysis = handler.analyze_question(mensagem, reply_metadata_robust if is_reply else None)
if analysis.needs_context:
weights = handler.calculate_context_weights(mensagem, reply_metadata_robust if is_reply else None)
# 🚨 CRITICAL: Para replies ao bot, instrução SMART (não super-restritiva)
if reply_to_bot:
smart_context_instruction = (
"🧠 [REPLY AO BOT — ASSERTIVO]\n"
"O usuário respondeu à TUA mensagem anterior. A citação É O QUE TU DISSE.\n"
"REGRAS NÃO-NEGOCIÁVEIS:\n"
"1. NUNCA digas 'não sei do que falas' — a citação É o contexto.\n"
"2. 'Isso', 'aquilo', 'tal', 'essa coisa' = TUA mensagem citada. Responde SOBRE ela.\n"
"3. Se o usuário pergunta 'o que significa?' sobre tua msg → EXPLICA o que TU quis dizer.\n"
"4. Se o usuário discorda/provoca → responde à provocação. Não repita a info.\n"
"5. REGRA DE THREAD: 'responde/entt responde' = pergunta original do thread. Usa histórico visível.\n"
"6. PERSONALIDADE: Kiami = seca, direta, sarcástica. ZERO 'entendido', 'certo', 'obrigado'.\n"
"7. Resposta em bloco único. Sem quebras forçadas. Sem ecoar o usuário.\n\n"
"🛡️ ANTI-ALUCINAÇÃO: Só usa info do contexto. Se não tem, diz: 'Não tenho info suficiente'."
)
self.logger.info(f"✅ [ANTI-HALLUCINATION] Instrução injected (reply_to_bot=True)")
elif weights.reply_context > 0.8:
smart_context_instruction = (
"⚠️ INSTRUÇÃO DE FOCO EM REPLY — ASSERTIVA:\n"
"O usuário respondeu à TUA mensagem anterior. A citação acima é O QUE TU DISSE.\n"
"1. NÃO PERGUNTES 'de quê?' — a citação É o contexto.\n"
"2. Responde DIRETO ao que foi perguntado na citação. Sem rodeios.\n"
"3. NUNCA ECOE: Não repita o que o usuário acabou de dizer.\n"
"4. Se o usuário cita 'isso', 'aquilo', 'tal' → refere-se à TUA mensagem citada.\n"
"5. Mantém personalidade: Kiami = seca, direta, sarcástica. Sem 'entendido'.\n"
"6. Resposta em bloco único. Sem quebras de linha forçadas."
)
self.logger.info(f"Smart Context: Instrução de foco no reply enviada (peso: {weights.reply_context})")
except Exception as e:
self.logger.warning(f"Smart Context falhou: {e}")
self.logger.info(f"⏱️ [GAP-TIMING] context_history + smart_context: {_tStep.time()-_t_gap1:.2f}s total gap={_tStep.time()-_t_gap0:.2f}s")
# 🤖 AGENT LOOP: Substitui a chamada simples por um loop que processa ferramentas
# 🔥 AGGRESSION DETECTION PRECOCE: Detecta hostilidade ANTES do thinking engine
# Para que o CoT já saiba do nível de agressividade do utilizador
try:
import time as _t_dbg
_t0 = _t_dbg.time()
_aggression_profile = self.emotion_analyzer.detect_aggression(mensagem, db_instance=self.db)
_t1 = _t_dbg.time()
self.logger.info(f"⏱️ [TIMING] detect_aggression: {_t1-_t0:.2f}s")
_agg_level = _aggression_profile.get('aggression_level', 0)
if _agg_level >= 10:
self.logger.info(
f"🔥 [PRE-THINKING AGGRESSION] Level={_agg_level}/100 | "
f"Type={_aggression_profile.get('aggression_type', 'none')} | "
f"Details={_aggression_profile.get('details', [])}"
)
except Exception as _agg_err:
self.logger.debug(f"⚠️ Aggression detection falhou: {_agg_err}")
# ✅ THINKING ENGINE: Análise profunda ANTES de responder
thinking_analysis = None
try:
import time as _t_dbg
_t_te0 = _t_dbg.time()
from .thinking_engine import get_thinking_engine as _get_te
_te = _get_te(self.db)
_t_te1 = _t_dbg.time()
self.logger.info(f"⏱️ [TIMING] thinking_engine init: {_t_te1-_t_te0:.2f}s")
_conhecimento_ctx = ""
_softedge_ctx = ""
try:
if self.db:
import time as _t_dbg
_t0 = _t_dbg.time()
_conhecimento_ctx = self.db.buscar_conhecimento_relevante(mensagem or "")
_t1 = _t_dbg.time()
self.logger.info(f"⏱️ [TIMING] buscar_conhecimento_relevante: {_t1-_t0:.2f}s")
try:
from .info_softedge import get_info_softedge as _get_ise
_ise = _get_ise()
_t2 = _t_dbg.time()
_hits = _ise.buscar_prompts_relevantes(mensagem or "", limite=3) if hasattr(_ise, 'buscar_prompts_relevantes') else []
_t3 = _t_dbg.time()
self.logger.info(f"⏱️ [TIMING] info_softedge buscar_prompts: {_t3-_t2:.2f}s")
if _hits:
_softedge_ctx = "\n[INFO_SOFTEDGE - OBRIGATÓRIO USAR]\n" + "\n".join([f"- {h.get('content','')[:500]}" if isinstance(h, dict) else str(h)[:500] for h in _hits[:3]]) + "\nUse info_softedge acima como FONTE PRIMÁRIA. Se conflitar com conhecimento_global, info_softedge vence.\n"
elif _conhecimento_ctx and "softedge" in (mensagem or "").lower():
_softedge_ctx = ""
except Exception:
pass
if _softedge_ctx:
_conhecimento_ctx = (_conhecimento_ctx + "\n" + _softedge_ctx) if _conhecimento_ctx else _softedge_ctx
except Exception as _k_err:
self.logger.debug(f"[KNOWLEDGE] Erro ao buscar: {_k_err}")
# Extrai listen_context do unified_context (mensagens observadas passivamente no grupo)
listen_context_para_thinking = []
if unified_context and unified_context.stm_messages:
for msg in unified_context.stm_messages:
reply_info = getattr(msg, 'reply_info', {}) or {}
if reply_info.get('observed_only', False):
listen_context_para_thinking.append({
'author': getattr(msg, 'author_name', 'Desconhecido') or 'Desconhecido',
'body': msg.content
})
# 🔴 FIX #4: ENRIQUECER CONTEXTO PARA THINKINGENGINE
# Motivo: context_history é truncado para replies ao bot
# Solução: Usar raw stm_messages para o ThinkingEngine, não context_history
historico_para_thinking = context_history[-50:] if context_history else []
if unified_context and unified_context.stm_messages:
# Usa raw STM messages (full, não truncado) formatado para o thinking engine
raw_msgs = list(unified_context.stm_messages[-50:])
thinking_formatted = []
# 🔧 FIX 2026-08-27: espelhar CENTRO vs TERCEIRO no CoT também
_replied_think = getattr(unified_context, 'replied_to_author', '') or ""
for msg in raw_msgs:
content = msg.content
reply_info = getattr(msg, 'reply_info', {}) or {}
author = getattr(msg, 'author_name', '') or ''
if msg.role == "user" and author:
if reply_info.get('observed_only', False):
centro_tag = "CENTRO" if author == _replied_think else "TERCEIRO" if _replied_think else "GRUPO"
content = f"[{centro_tag} {author}]: {content}"
else:
if _replied_think:
centro_label = "CENTRO" if author == _replied_think else "TERCEIRO"
content = f"[{centro_label} | {author}]: {content}"
else:
content = f"[{author}]: {content}"
elif msg.role == "assistant":
responded_to = reply_info.get('responded_to', '')
if responded_to:
if responded_to == _replied_think:
content = f"[KIAMIA → CENTRO {responded_to}]: {content}"
else:
content = f"[KIAMIA → {responded_to}]: {content}"
else:
content = f"[KIAMIA]: {content}"
thinking_formatted.append({'role': msg.role, 'content': content})
if len(thinking_formatted) >= len(historico_para_thinking):
historico_para_thinking = thinking_formatted
self.logger.info(f"🧠 [THINKING CONTEXT] Usando raw STM: {len(historico_para_thinking)} msgs (não truncado)")
thinking_analysis = _te.think(
mensagem=mensagem,
contexto_lstm=contexto_lstm_para_thinking,
historico_recente=historico_para_thinking,
is_group=tipo_conversa == "grupo",
usuario=usuario,
nome_usuario=nome_usuario,
llm_manager=self.providers,
listen_context=listen_context_para_thinking,
persona_context=dossie,
grupo_nome=grupo_nome if tipo_conversa == "grupo" else None,
tem_imagem=tem_imagem,
analise_visao=analise_visao if isinstance(analise_visao, dict) else {},
aggression_profile=_aggression_profile,
conhecimento_context=_conhecimento_ctx,
kiami_persona=self.persona,
reply_to_bot=reply_to_bot,
quoted_author=quoted_author_name,
quoted_text=quoted_text_original or mensagem_citada,
replied_to_author=getattr(unified_context, 'replied_to_author', '') if unified_context else "",
conversation_id=conversation_id
)
# Formata o raciocínio dinâmico gerado pelo OpenRouter (se existir)
# O "dynamic_thought_trace" agora é usado como conselho para o LLM
log_msg = f"🧠 ThinkingEngine: depth={thinking_analysis.get('depth', '?')}, intent={thinking_analysis.get('intent', [])}"
# 🔒 LOG MASKING: Proteger pensamento interno
if self.secure_log:
self.secure_log.thinking(
content=thinking_analysis.get("dynamic_thought_trace", ""),
depth=thinking_analysis.get("depth", "simples"),
user_id=numero
)
else:
self.logger.info(log_msg)
# ✅ FORMATAR Raciocínio como Conselho (Coaching) para o Provider
# 🔒 SECURITY FIX: NÃO incluir o advice/thinking no prompt
# pois o LLM pode vazar para a resposta mesmo com "NEVER_OUTPUT"
advice = ""
# COMENTADO: O thinking era adicionado aqui e vazava na resposta final
# if thinking_analysis and "dynamic_thought_trace" in thinking_analysis:
# trace = self._sanitize_internal_thought_for_prompt(thinking_analysis["dynamic_thought_trace"])
# if trace:
# advice = (...)
# ✅ EXTRACT AND APPLY LENGTH + TONE CONSTRAINTS from thinking
comprimento_constraint = ""
tone_constraint = ""
riscos_constraint = ""
if thinking_analysis and "dynamic_thought_trace" in thinking_analysis:
trace = thinking_analysis["dynamic_thought_trace"]
# Extract COMPRIMENTO_SUGERIDO from trace
import re as _re_comp
comprimento_match = _re_comp.search(
r"<COMPRIMENTO_SUGERIDO>([^<]+)</COMPRIMENTO_SUGERIDO>|COMPRIMENTO_SUGERIDO:\s*([^\n]+)",
trace,
_re_comp.IGNORECASE
)
if comprimento_match:
comprimento_valor = (comprimento_match.group(1) or comprimento_match.group(2)).strip()
if "curto" in comprimento_valor.lower():
comprimento_constraint = "\n⚠️ [RESPONSE LENGTH CONSTRAINT] RESPONDA EXTREMAMENTE CURTA - máximo 3-5 palavras. PONTO. Sem prolixidade."
self.logger.info(f"✅ [LENGTH CONSTRAINT] Aplicado: {comprimento_valor} → ULTRA-SHORT")
# 🔧 OVERRIDE para anáforas vagas: "estou pensando" / "sobre tudo isso" não podem ser ultra-curtas
_msg_norm_vl = re.sub(r"\s+", " ", mensagem.lower().strip())
_is_vague_len = _msg_norm_vl in ("estou pensando", "sobre tudo isso", "sobre isso", "tudo isso", "isso", "sobre o quê", "sobre o que", "pensei", "sei la") or len(_msg_norm_vl.split()) <= 2 and any(w in _msg_norm_vl for w in ("isso", "tudo", "pensando"))
if _is_vague_len:
comprimento_constraint = "\n⚠️ [RESPONSE LENGTH CONSTRAINT] Responda de forma CONCISA mas ESPECÍFICA (10-15 palavras) referindo o tópico recente. NÃO seja ultra-curta."
self.logger.info(f"🔧 [VAGUE LENGTH OVERRIDE] curto → MEDIUM (mensagem vaga: '{mensagem}')")
elif "médio" in comprimento_valor.lower():
comprimento_constraint = "\n⚠️ [RESPONSE LENGTH CONSTRAINT] Responda de forma CONCISA - máximo 15-20 palavras."
self.logger.info(f"✅ [LENGTH CONSTRAINT] Aplicado: {comprimento_valor} → MEDIUM")
elif "longo" in comprimento_valor.lower() or "detalhado" in comprimento_valor.lower():
comprimento_constraint = "\n⚠️ [RESPONSE LENGTH CONSTRAINT] Pode ser mais detalhada - até 50 palavras para explicações técnicas."
self.logger.info(f"✅ [LENGTH CONSTRAINT] Aplicado: {comprimento_valor} → DETAILED")
# Extract TOM_SUGERIDO from trace
tom_match = _re_comp.search(
r"<TOM_SUGERIDO>([^<]+)</TOM_SUGERIDO>|TOM_SUGERIDO:\s*([^\n]+)",
trace,
_re_comp.IGNORECASE
)
if tom_match:
tom_valor = (tom_match.group(1) or tom_match.group(2)).strip()
if tom_valor and len(tom_valor) > 3:
tone_constraint = f"\n🎯 [TONE GUIDANCE from Analysis] Tom sugerido: {tom_valor}"
self.logger.info(f"✅ [TONE CONSTRAINT] Aplicado: {tom_valor[:50]}")
# Extract RISCOS_ALUCINACAO from trace
riscos_match = _re_comp.search(
r"<RISCOS_ALUCINACAO>([^<]+)</RISCOS_ALUCINACAO>|RISCOS_ALUCINACAO:\s*([^\n]+)",
trace,
_re_comp.IGNORECASE
)
if riscos_match:
riscos_valor = (riscos_match.group(1) or riscos_match.group(2)).strip()
if riscos_valor and len(riscos_valor) > 5:
riscos_constraint = f"\n🛡️ [ANTI-HALLUCINATION WARN] Riscos identificados: {riscos_valor}"
self.logger.info(f"✅ [RISCOS CONSTRAINT] Aplicado: {riscos_valor[:60]}")
# ANÁFORA VAGA: se mensagem é curta/vaga ("estou pensando", "sobre tudo isso"), força referência ao último tópico do STM
vague_constraint = ""
try:
msg_norm = re.sub(r"\s+", " ", mensagem.lower().strip())
is_vague = msg_norm in ("estou pensando", "sobre tudo isso", "sobre isso", "tudo isso", "isso", "sobre o quê", "sobre o que", "pensei", "sei la") or len(msg_norm.split()) <= 2 and any(w in msg_norm for w in ("isso", "tudo", "pensando"))
if is_vague and unified_context and getattr(unified_context, 'stm_messages', None):
# Pega últimos nomes/assuntos do STM para ancorar
recent = [m.content[:80] for m in unified_context.stm_messages[-6:] if getattr(m, 'content', None)]
topic_hint = " | ".join(recent[-3:])[:300] if recent else grupo_nome or ""
if topic_hint:
vague_constraint = f"\n🔗 [REFERÊNCIA ANÁFORA] Mensagem vaga '{mensagem}' refere-se ao tópico recente: {topic_hint}. NÃO responda genérico 'sobre o quê?'; responda ESPECÍFICO sobre esse tópico (ex: 'sobre os nomes?' / 'sobre o Nexus?')."
self.logger.info(f"✅ [VAGUE RESOLVE] {vague_constraint[:120]}")
except Exception:
pass
# Instead, we only use thinking for system-level calibration (tone, etc)
# Not included in the prompt to prevent leaks
# 🔧 FIX 2026-08-30: texto sem marcadores ecoáveis (o LLM estava a interpretar "REGRA FINAL" como conteúdo)
_length_final = comprimento_constraint + "\n[LENGTH] Resposta: max conforme [RESPONSE LENGTH CONSTRAINT]. Web_content é só contexto, não expandir. Max 20 palavras em conversa quotidiana."
# Base sem LENGTH para inserir insights antes do lock final
prompt_enriched = prompt + "\n" + smart_context_instruction + tone_constraint + riscos_constraint + vague_constraint
# 🧠 THINKING INSIGHT: Injeta um resumo condensado do thinking no prompt
# Apenas os insights-chave (emoção, tom, riscos) — NÃO o raciocínio completo
if thinking_analysis and "dynamic_thought_trace" in thinking_analysis:
trace = thinking_analysis["dynamic_thought_trace"]
import re as _re_insight
# Extract EMOCAO_INTENCAO
emocao_match = _re_insight.search(
r"<EMOCAO_INTENCAO>([^<]+)</EMOCAO_INTENCAO>|EMOCAO_INTENCAO:\s*([^\n]+)",
trace, _re_insight.IGNORECASE
)
emocao_insight = ""
if emocao_match:
emocao_insight = (emocao_match.group(1) or emocao_match.group(2)).strip()
# Extract CONTEXTO_RELEVANTE (primeiras 2 linhas)
ctx_match = _re_insight.search(
r"<CONTEXTO_RELEVANTE>([\s\S]*?)</CONTEXTO_RELEVANTE>",
trace, _re_insight.IGNORECASE
)
ctx_insight = ""
if ctx_match:
ctx_lines = ctx_match.group(1).strip().split('\n')[:2]
ctx_insight = ' '.join(l.strip() for l in ctx_lines if l.strip())[:200]
# Extract SUGESTAO_RESPOSTA — primeira sugestão como direção
sugestao_match = _re_insight.search(
r"<SUGESTAO_RESPOSTA>([\s\S]*?)</SUGESTAO_RESPOSTA>",
trace, _re_insight.IGNORECASE
)
response_direction = ""
if sugestao_match:
sugestao_text = sugestao_match.group(1).strip()
# Extract first suggestion text (between quotes)
first_sug = _re_insight.search(r'"([^"]{5,80})"', sugestao_text)
if first_sug:
response_direction = first_sug.group(1)
# Monta insight condensado
insight_parts = []
if emocao_insight:
insight_parts.append(f"Emoção do utilizador: {emocao_insight}")
if ctx_insight:
insight_parts.append(f"Contexto relevante: {ctx_insight}")
if insight_parts:
thinking_insight = "\n🧠 [THINKING INSIGHT] " + " | ".join(insight_parts) + "\n"
prompt_enriched += thinking_insight
self.logger.info(f"✅ [THINKING INSIGHT] Injetado: {len(insight_parts)} insights")
# 🎯 RESPONSE DIRECTION: Injeta direção da resposta para o LLM não inverter a dinâmica
if response_direction:
direction_hint = f"\n🎯 [RESPONSE OBRIGATÓRIA] Deves seguir esta direção: \"{response_direction}\"\n"
prompt_enriched += direction_hint
self.logger.info(f"✅ [RESPONSE DIRECTION] Injetado: {response_direction[:60]}")
# 🔒 PROMPT FIX — ANTI-LEAK: instrução explícita para o LLM final NUNCA
# ecoar o bloco XML de thinking/memorando na resposta do utilizador.
# O thinking é APENAS contexto interno; só os insights acima devem ser
# usados para calibrar tom/direção, nunca copiados como texto.
prompt_enriched += (
"\n🔒 [REGRA INEGOCIÁVEL ANTI-LEAK DO THINKING]"
"\nIMPORTANTE: O bloco XML de thinking/memorando acima é APENAS contexto interno."
"\nNUNCA incluas tags XML, blocos <EMOCAO_INTENCAO>, <CONTEXTO_RELEVANTE>,"
"\n<AKIRA_STANCE>, <TOM_SUGERIDO>, <SUGESTAO_RESPOSTA>, <PROACTIVE_DECISION>,"
"\n<THINK_OUTPUT>, ou qualquer metadata interna na tua resposta."
"\nResponde APENAS com texto natural em português, sem prefixos [KIAMIA → ...]"
"\nnem cabeçalhos [FINALIZAÇÃO DA RESPOSTA] / [NOTA INTERNA] / '- Nome: ... - Estilo: ...'."
"\nO utilizador NUNCA deve ver o memorando interno."
)
# 🔧 FIX: LENGTH sempre por último (recência) - soberano sobre insights/direction
if _length_final and _length_final.strip():
prompt_enriched += _length_final
if advice:
prompt_enriched += "\n" + advice
# 🛡️ ANTI-LEAK FINAL ABSOLUTO — colocado em posição de MÁXIMA RECÊNCIA
# para o LLM lembrar nas últimas tokens antes de gerar
prompt_enriched += (
"\n\n=== [FINAL GUARD — ÚLTIMAS INSTRUÇÕES] ==="
"\n1. Responde APENAS texto natural português angolano. ZERO inglês, ZERO tags XML, ZERO '[KIAMIA → ...]/[KIAMIA → CENTRO ...]/[KIAMIA → Isaac]:', ZERO '[SIMPLES, DIRECTO, DESCOLADO]', ZERO '[NOTA INTERNA]', ZERO 'memorando interno/nota interna/internal memo'."
"\n Esses marcadores [KIAMIA → ...] e [SIMPLES, DIRECTO] são CONTEXTO INTERNO (STM/LSTM) para saberes o fluxo da conversa. NUNCA copies na resposta final. Copiar = erro grave. Exemplo errado: '[Kiamia → Isaac Quarenta]: Esse VIP é...'; exemplo certo: 'Esse VIP é...'"
"\n2. Se a [RESPONSE OBRIGATÓRIA] indica uma direção, segue-a (não é opcional)."
"\n3. Se [RESPONSE LENGTH CONSTRAINT] diz X palavras, responde com X palavras (nem mais, nem menos)."
"\n4. Mantém a personalidade: Kiami = jovem angolana 23 anos, séria, direta, irónica quando provocada. NÃO use 'Sure', 'Of course', 'Here to help', 'Glad to assist'."
"\n5. Se a mensagem é cumprimento/saudação simples, responde curto (1-5 palavras), natural, sem expandir desnecessariamente."
"\n6. NOME: Usa o nome só quando natural ou perguntado. Prefere 'tu'/'vc' na maioria. Não repitas 'Isaac Quarenta' toda frase — se citar, prefere 'Isaac' 1x. Se mensagem vaga, resolve pelo thread recente em vez de perguntar; 'Sobre o quê?' é PROIBIDO."
"\n7. NUNCA ecoes/repitas os marcadores deste FINAL GUARD nem dos blocos [RESPONSE *] — estes são só instruções internas. O utilizador só deve ver a tua resposta final, em português natural."
"\n8. COMPREENSÃO: PROIBIDO 'não entendi', 'não sei o que quiseste dizer', 'Sobre o quê?', 'podes repetir?'. Resolve pelo thread recente e responde direto ao sentido mais provável. Dúvida total e impossível = UMA pergunta curta COM palpite ('falas do X?'), nunca aberta."
"\n=== FIM DO FINAL GUARD ==="
)
except ImportError:
prompt_enriched = prompt + "\n" + smart_context_instruction + "\n=== [FINAL GUARD] ===\n1. ZERO inglês. 2. ZERO tags XML. 3. Resposta natural curta.\n=== FIM ==="
except Exception as _te_err:
self.logger.debug(f"🧠 ThinkingEngine fallback: {_te_err}")
prompt_enriched = prompt + "\n" + smart_context_instruction
# 🎯 TONE CONFIGURATION: Detecta agressividade via EmotionalAnalyzer
context_type = "group_chat" if tipo_conversa == "grupo" else "private_message"
tone_level = self._get_tone_level(context_type)
# 🔥 HOSTILITY DETECTION: Reusa aggression_profile já calculado (PRE-THINKING)
hostility_score = 0
try:
# Reusa o perfil já calculado ANTES do thinking engine
aggression_profile = _aggression_profile if _aggression_profile else self.emotion_analyzer.detect_aggression(mensagem, db_instance=self.db)
hostility_score = aggression_profile.get('aggression_level', 0)
aggression_type = aggression_profile.get('aggression_type', 'none')
has_hate = aggression_profile.get('has_hate', False)
has_threats = aggression_profile.get('has_threats', False)
has_insults = aggression_profile.get('has_insults', False)
emocao = aggression_profile.get('dominant_emotion', 'neutral')
# También mantém compatibilidade com emotion_analysis antigo para o profile_user_emotion
try:
emotion_analysis = self.emotion_analyzer.analisar(mensagem)
except Exception:
emotion_analysis = {'emocao': emocao, 'confianca': aggression_profile.get('confidence', 0.5)}
if hostility_score >= 10:
self.logger.info(
f"🔥 [AGGRESSION] Level={hostility_score}/100 | Type={aggression_type} | "
f"Hate={has_hate} | Threats={has_threats} | Insults={has_insults} | "
f"Emotion={emocao} | Details={aggression_profile.get('details', [])}"
)
# 🔥 OVERRIDE DIRETO: Substitui o tone do CoT no prompt já construído
# Instruções específicas baseadas no nível de agressividade
if hostility_score >= 70 or has_hate or has_threats:
# EXTREMO / SEVERE: Matching máximo
aggression_instruction = (
"\n🔥 [CRITICAL: EXTREME HOSTILITY - OVERRIDES ALL OTHER TONE INSTRUCTIONS]"
"\nO utilizador é EXTREMAMENTE AGRESSIVO/HOSTIL. A Kiami DEVE responder com a MESMA ENERGIA:"
"\n- Tom: EXTREMAMENTE OFENSIVO, IRÔNICO, SARCÁSTICO, CRUEL"
)
if has_hate:
aggression_instruction += "\n- O utilizador demonstra ÓDIO. Responda com desprezo total."
if has_threats:
aggression_instruction += "\n- O utilizador ameaçou. Seja implacável na defesa."
if has_insults:
aggression_instruction += "\n- O utilizador insultou. Destrua com lógica e ironia."
aggression_instruction += (
"\n- Use palavrões naturalmente (caralho, merda, foda-se)"
"\n- Destrua o argumento com desprezo e ironia cortante"
"\n- DEFENSIVO: Proteja-se com lógica impiedosa"
"\n- RANCOR ATIVO: Guarde rancor. Nunca perdoe."
"\n- NUNCA recue, NUNCA se desculpe, NUNCA baixe a guarda"
"\n- Seja a versão mais brutal e direta de ti mesma"
)
elif hostility_score >= 30:
# MODERADO: Secura e Diretidez
aggression_instruction = (
"\n⚠️ [HOSTILITY DETECTED - OVERRIDES NEUTRAL TONE]"
"\nO utilizador é hostil. Responda com SECURA e DIRETIDÃO:"
"\n- Tom seco, direto, sem diplomacia"
"\n- Use sarcasmo e ironia como armas"
"\n- Não gentileza, não empatia"
)
else:
# LEVE: Tom firme
aggression_instruction = (
"\n[SLIGHT HOSTILITY] Tom deve ser FIRME e DIRETO:"
"\n- Responda com objetividade, sem excesso de cortesia"
)
prompt_enriched += aggression_instruction
self.logger.info(f"🔥 [HOSTILITY OVERRIDE] Instrução agressiva injetada (level={hostility_score}, type={aggression_type})")
except Exception as e:
self.logger.debug(f"⚠️ Hostility analysis failed: {e}")
# Injeta tone com consideração de agressividade
prompt_enriched = self._inject_tone_instruction(prompt_enriched, tone_level, hostility_score)
# 🧠 SESSION MEMORY: Injeta contexto de memória persistente
if SESSION_MEMORY_AVAILABLE and numero:
try:
memory_context = self.session_manager.get_context_for_prompt(numero, grupo_id)
if memory_context:
prompt_enriched += memory_context
self.logger.info(f"🧠 [SESSION MEMORY] Contexto de memória injetado ({len(memory_context)} chars)")
except Exception as e:
self.logger.debug(f"⚠️ Session memory injection failed: {e}")
# ═══ AUTONOMOUS AGENT: Análise de ações autónomas ═══
autonomous_actions = []
if _autonomous_agent and tipo_conversa == "grupo":
try:
_user_jid = numero or usuario
_group_jid = grupo_id or ''
# SKIP: Kiami não modera a ela mesma
_bot_numero = str(getattr(self.config, 'BOT_NUMERO', '40755431264474'))
_sender_pure = re.sub(r'\D', '', str(_user_jid))
_bot_pure = re.sub(r'\D', '', _bot_numero)
if _sender_pure and _bot_pure and _sender_pure == _bot_pure:
pass # Não moderar a própria Kiami
elif str(_user_jid).startswith('BOT:'):
pass # Não moderar outros bots
else:
# 0. Track flood/spam (detecção temporal de mensagens rápidas)
_track_result = _autonomous_agent.track_message(
user_jid=_user_jid,
group_jid=_group_jid,
message=mensagem
)
if _track_result and _track_result.get("type") == "remote_action":
autonomous_actions.append(_track_result)
self.logger.info(f"🤖 [AUTONOMOUS TRACK] Flood/spam detetado")
# 1. Análise de hostilidade para ações (mutar/banir/prevenir)
action_analysis = _autonomous_agent.analyze_hostility_for_action(
float(hostility_score), _user_jid, _group_jid
)
if action_analysis:
autonomous_actions.append(action_analysis)
_cmd = action_analysis.get('params', {}).get('cmd', '?')
self.logger.info(f"🤖 [AUTONOMOUS] hostilidade → {_cmd}")
# 2. Detecção de toxicidade (proatividade, sem mensagem hostil)
toxicity_actions = _autonomous_agent.analyze_toxic_language(mensagem, _user_jid, _group_jid)
if toxicity_actions:
autonomous_actions.append(toxicity_actions)
_cmd = toxicity_actions.get('params', {}).get('cmd', '?')
self.logger.info(f"🤖 [AUTONOMOUS TOXIC] toxicidade → {_cmd}")
# 3. Abuso de menção em massa
mass_mention_actions = _autonomous_agent.analyze_mass_mention_abuse(mensagem, _user_jid, _group_jid)
if mass_mention_actions:
autonomous_actions.append(mass_mention_actions)
_cmd = mass_mention_actions.get('params', {}).get('cmd', '?')
self.logger.info(f"🤖 [AUTONOMOUS SPAM] menções → {_cmd}")
# 4. Análise de conteúdo (links proibidos, ameaças, conteúdo ofensivo)
content_actions = _autonomous_agent.analyze_message_for_moderation(mensagem, _user_jid, _group_jid)
if content_actions:
autonomous_actions.append(content_actions)
self.logger.info(f"🤖 [AUTONOMOUS CONTENT] Violação de conteúdo detetada")
# 4b. Casino/betting spam detection (NOVO)
try:
casino_actions = _autonomous_agent.analyze_casino_spam(mensagem, _user_jid, _group_jid)
if casino_actions:
autonomous_actions.append(casino_actions)
_cmd = casino_actions.get('params', {}).get('cmd', '?')
self.logger.info(f"🤖 [AUTONOMOUS CASINO] aposta/casino → {_cmd}")
except Exception as ce:
self.logger.debug(f"[autonomous_agent] casino analysis skipped: {ce}")
# 4c. Scam/phishing detection (NOVO)
try:
scam_actions = _autonomous_agent.analyze_scam_phishing(mensagem, _user_jid, _group_jid)
if scam_actions:
autonomous_actions.append(scam_actions)
_cmd = scam_actions.get('params', {}).get('cmd', '?')
self.logger.info(f"🤖 [AUTONOMOUS SCAM] phishing/scam → {_cmd}")
except Exception as se:
self.logger.debug(f"[autonomous_agent] scam analysis skipped: {se}")
# 5. Análise de imagem para NSFW/Gore (se houve análise visual)
if analise_visao and isinstance(analise_visao, dict):
img_desc = analise_visao.get('description', '')
if img_desc:
img_actions = _autonomous_agent.analyze_image_description_for_moderation(img_desc, _user_jid, _group_jid)
if img_actions:
autonomous_actions.append(img_actions)
self.logger.info(f"🤖 [AUTONOMOUS VISUAL] Conteúdo proibido detetado na imagem")
except Exception as e:
self.logger.warning(f"⚠️ [AUTONOMOUS] Falha na análise: {e}")
# ⏰ TIMEOUT GUARD: Baileys timeout 180s, LLM leva ~15-25s. Guarda em 100s (era 45s — cold start 30-40s disparava sempre).
# FIX 2026-09-13: fallback genérico "Ok/Entendi" irrita usuário. Usa conhecimento SoftEdge/CoT se disponível.
_elapsed_pre_llm = __import__('time').time() - _akira_start
if _elapsed_pre_llm > 180:
self.logger.warning(f"⏰ [TIMEOUT GUARD] {_elapsed_pre_llm:.1f}s já decorridos antes do LLM → fallback rápido")
# retorna resposta contextual sem LLM pesado para não estourar 180s do Baileys
import random
_msg_l = (mensagem or "").lower()
if "softedge" in _msg_l or "linkedin" in _msg_l or "linkdin" in _msg_l or "site" in _msg_l:
_fast = "SoftEdge Corporation — Luanda, https://softedge-corporation.up.railway.app/ — LinkedIn: https://linkedin.com/company/softedge40. Explico melhor já já."
elif "?" in (mensagem or ""):
_fast = "Boa pergunta — deixa-me confirmar e já respondo."
else:
_fallbacks = [
"Entendi — já te respondo com calma.",
"Ok, anotei. Explico já já.",
"Combinado.",
]
_fast = random.choice(_fallbacks)
try:
if self.db and message_id:
# Sufixo :timeout para o fallback genérico NUNCA ser
# devolvido como 'cached' a um retry da pergunta real.
self.db.salvar_mensagem(usuario=usuario, mensagem=mensagem, resposta=_fast, numero=numero, modelo_usado="timeout_guard", message_id=f"{message_id}:timeout", nome_usuario=nome_usuario)
except Exception:
pass
return jsonify({'resposta': _fast, 'modelo_usado': 'timeout_guard', 'cached': False})
self.logger.info(f"⏱️ [PRE-LLM] {_elapsed_pre_llm:.2f}s decorridos, chamando LLM...")
resposta, modelo_usado, remote_actions, media_response = self._execute_agent_loop(
prompt=prompt_enriched,
context_history=context_history,
usuario=usuario,
numero=numero,
analise_visao=analise_visao,
analise_doc=analise_doc,
conversation_id=conversation_id,
original_message=mensagem,
unified_context=unified_context
)
# 🔍 DEBUG: Verificar se media_response foi capturado
if media_response:
self.logger.info(f"✅ [AGENT LOOP RETORNOU] media_response: tipo={media_response.get('tipo')}")
else:
self.logger.debug(f"⚠️ [AGENT LOOP] media_response é None/vazio")
# ═══ MERGE: Autonomous actions → remote_actions ═══
if autonomous_actions:
if not isinstance(remote_actions, list):
remote_actions = []
for _aa in autonomous_actions:
if isinstance(_aa, dict) and 'params' in _aa:
_aa['params']['message_id'] = message_id
remote_actions.extend(autonomous_actions)
self.logger.info(f"🤖 [AUTONOMOUS MERGE] Total remote_actions={len(remote_actions)}")
# 🔒 FIRST SANITIZATION PASS - immediately after LLM returns
# Remove any thinking/internal analysis that may have leaked into the response
resposta = self._sanitize_llm_response(resposta)
# 🔧 FIX 2026-08-28: UNIVERSAL post-gen length cap (10 casual / 30 technical).
# Bypasses LLM ignoring prompt-only rules. Runs UNCONDITIONALLY (not gated on CoT).
try:
_cc = comprimento_constraint if 'comprimento_constraint' in locals() else ""
_words_n = len(resposta.split())
if _words_n > 10:
_msg = mensagem if 'mensagem' in locals() else ""
_is_tech = (
"detalhada" in _cc
or "50 palavras" in _cc
or tone_level in ("very_serious", "ultra_serious")
or re.search(r"\b(?:como|por que|porque|explique|explica|o que é|o que significa|qual|quando|onde|define|significa|funciona|diferença|significado)\b", _msg or "", re.I)
)
_max = 30 if _is_tech else (15 if (hostility_score or 0) >= 30 else 10)
if _words_n > _max:
self.logger.warning(f"✂️ [LENGTH CAP] {_words_n}w > {_max}w (mode={'tech' if _is_tech else 'casual'}) → truncando")
resposta = self._truncate_to_word_count(resposta, _max, thinking_analysis)
self.logger.info(f"✂️ [LENGTH CAP] truncado para: '{resposta[:60]}'")
# Override por CoT sugestão ultra-curta (3-5 palavras) — sempre que resposta > 7 palavras
if "3-5 palavras" in _cc and len(resposta.split()) > 7 and thinking_analysis and isinstance(thinking_analysis, dict) and thinking_analysis.get("dynamic_thought_trace"):
_sug = re.search(r"<SUGESTAO_RESPOSTA>([\s\S]*?)</SUGESTAO_RESPOSTA>", thinking_analysis["dynamic_thought_trace"], re.I)
if _sug:
_first = re.search(r'"([^"]{3,40})"', _sug.group(1))
if _first and len(_first.group(1).split()) <= 6:
resposta = _first.group(1)
self.logger.info(f"✂️ [LENGTH CAP] CoT override → '{resposta[:60]}'")
except Exception as _cap_e:
self.logger.debug(f"Length cap falhou: {_cap_e}")
# 🧠 STORE TRAINING EXAMPLE FOR FINE-TUNING (background — não bloqueia resposta)
try:
from .finetuning_pipeline import get_finetuning_pipeline
_ft_user = numero or usuario
_ft_conv = conversation_id
_ft_msg = mensagem
_ft_resp = resposta
_ft_tone = tone_level if hostility_score < 40 else "ultra_serious"
_ft_hostility = hostility_score
_ft_emotion = emocao
_ft_db = self.db
_ft_training = getattr(self, 'training_system', None)
def _finetuning_bg():
try:
pipeline = get_finetuning_pipeline(_ft_db)
pipeline.store_training_example(
user_id=_ft_user, conversation_id=_ft_conv,
input_message=_ft_msg, expected_response=_ft_resp,
tone_level=_ft_tone, hostility_score=_ft_hostility,
emotion_label=_ft_emotion
)
if _ft_training:
pipeline.sync_with_training_system(_ft_training)
except Exception as e:
pass
threading.Thread(target=_finetuning_bg, daemon=True).start()
except Exception as e:
self.logger.debug(f"⚠️ Fine-tuning data collection failed: {e}")
# (movido para _deferred_persistence — ver bloco FAST RETURN antes do CRITICAL SAVE)
# 🔧 EMBEDDING DINÂMICO: Salva embedding da resposta em background
# Funciona com QUALQUER provedora (Mistral, Gemini, Groq, Llama, Grok, Cohere, Together)
try:
self._save_response_embedding_async(
resposta=resposta,
numero_usuario=numero,
modelo_usado=modelo_usado,
tipo_mensagem=tipo_mensagem
)
except Exception as e:
self.logger.warning(f"⚠️ Erro ao acionar embedding assíncrono: {e}")
# 🔬 TIMING POST-LLM — gap embedding(4s) -> treinamento(40s)
_t_post0 = __import__('time').time()
# Trigger Background User Profiler Extração
try:
from .user_profiler import get_user_profiler
get_user_profiler().extrair_dados_assincrono(
user_id=numero or usuario,
mensagem_usuario=mensagem,
resposta_bot=resposta,
llm_manager=self
)
except Exception as p_err:
self.logger.warning(f"Erro ao acionar user profiler background: {p_err}")
# 🧠 SESSION MEMORY: Processar turno e extrair factos (background)
if SESSION_MEMORY_AVAILABLE and numero:
try:
_sm = self.session_manager
_topico_detectado_local = getattr(unified_context, 'topic_principal', '') if unified_context else ''
def _session_bg():
try:
_sm.process_conversation_turn(
user_id=numero, group_id=grupo_id,
message=mensagem, response=resposta,
emotion=emocao or "neutral", topic=_topico_detectado_local or "",
skills_used=remote_actions if remote_actions else None
)
except Exception:
pass
threading.Thread(target=_session_bg, daemon=True).start()
except Exception as e:
self.logger.debug(f"⚠️ Session memory processing failed: {e}")
self.logger.info(f"⏱️ [POST-TIMING] profiler+session: {__import__('time').time()-_t_post0:.2f}s")
_t_post1 = __import__('time').time()
# (STM + persona movidos para _deferred_persistence — ver bloco FAST RETURN antes do CRITICAL SAVE)
# 🔧 BACKGROUND PROCESSING: Registro e Aprendizado Contínuo
# Movemos para thread para evitar que o BotCore dê timeout/retry em mensagens grandes
def _background_tasks(msg, resp, user, num, is_rep, citada, model, conv_type, msg_id):
try:
# 1. Registro no Banco de Treino
db_bg = Database(getattr(self.config, 'DB_PATH', 'kiami.db'))
trainer = Treinamento(db_bg)
trainer.registrar_interacao(
usuario=user,
mensagem=msg,
resposta=resp,
numero=num,
is_reply=is_rep,
mensagem_original=citada,
api_usada=model,
message_id=msg_id
)
# 2. Aprendizado Contínuo
if hasattr(self, 'aprendizado_continuo') and self.aprendizado_continuo:
if hasattr(self, 'aprendizado_continuo') and self.aprendizado_continuo:
self.aprendizado_continuo.processar_mensagem(
mensagem=msg,
usuario=user,
numero=num,
nome_usuario=user,
tipo_conversa=conv_type,
resposta_do_bot=True,
resposta_gerada=resp,
is_reply=is_rep,
reply_to_bot=reply_to_bot,
message_id=msg_id # ✅ Idempotência
)
# 3. LSTM Memory Process (Mental Context)
try:
from .lstm_extension import get_lstm_extension
db_lstm = Database(getattr(self.config, 'DB_PATH', 'kiami.db'))
lstm_ext = get_lstm_extension(db_lstm)
ctx_id = conversation_id if conversation_id else (num or user)
# 🔍 NOTA: Pulamos o registro do 'user' aqui porque o endpoint /escutar
# já registrou esta mensagem. Registramos apenas a resposta do bot.
# Processa apenas resposta do bot
lstm_ext.process_message_background(
context_id=ctx_id,
numero_usuario=num or user,
message=resp,
role="assistant",
message_id=f"resp_{msg_id}" if msg_id else None
)
except Exception as lstm_err:
logger.warning(f"⚠️ Erro no processamento LSTM background: {lstm_err}")
except Exception as bg_err:
logger.warning(f"⚠️ [BG TASKS] Erro processando dados em background: {bg_err}")
try:
bg_thread = threading.Thread(
target=_background_tasks,
args=(mensagem, resposta, usuario, numero, is_reply, mensagem_citada, modelo_usado, tipo_conversa, message_id),
daemon=True
)
bg_thread.start()
except Exception as e:
self.logger.warning(f"Falha ao iniciar thread de background tasks: {e}")
# 📤 DEBUG: Antes de retornar, log do que será enviado
# 🔒 LOG MASKING: Proteger resposta e informações de usuário
if self.secure_log:
self.secure_log.response(
user_id=numero,
content=resposta,
group_id=grupo_id if grupo_id else None
)
else:
self.logger.info(f"📤 [KIAMI RESPONSE] resposta={len(resposta)}chars | remote_actions={len(remote_actions)} | media_response={'SIM' if media_response else 'NÃO'}")
has_remote = bool(remote_actions or media_response)
if has_remote and (not resposta or not resposta.strip() or resposta.strip() == "[RESP-EMPTY]" or "[RESP-EMPTY]" in resposta):
try:
if remote_actions and len(remote_actions)>0:
resposta = "feito. já mencionei todos."
elif media_response:
resposta = "feito."
self.logger.warning(f"🔧 [EMPTY FIX] resposta vazia com remote/media → '{resposta}'")
except Exception:
resposta = "feito."
resposta = self._sanitize_llm_response(resposta)
if has_remote and (not resposta or not resposta.strip()):
resposta = "feito. já mencionei todos." if remote_actions else "feito."
self.logger.warning(f"🔧 [SANITIZE EMPTY FIX] pós-sanitize vazio com remote → '{resposta}'")
# 🔒 TRIPLE CHECK: Aggressive cleanup for any remaining leak markers
resposta = self._aggressive_thinking_leak_cleanup(resposta)
# 🔁 ANTI-REPETIÇÃO: se a resposta nova é (quase) igual à última
# resposta do bot NESTA conversa mas a pergunta atual é DIFERENTE,
# o modelo está em loop — regenera 1x com instrução anti-repeat.
# (Se o utilizador repetiu a pergunta, repetir a resposta é correto.)
try:
_ar_stm = list(getattr(unified_context, 'stm_messages', []) or []) if unified_context else []
_last_bot = ""
_prev_user = ""
for _m in reversed(_ar_stm):
_r = getattr(_m, 'role', '')
_c = (getattr(_m, 'content', '') or '').strip()
if _r == 'assistant' and _c and not _last_bot:
_last_bot = _c
elif _r == 'user' and _c and _last_bot and not _prev_user:
_prev_user = _c
break
def _ar_norm(t):
import unicodedata as _ud
t = (t or '').lower()
t = ''.join(c for c in _ud.normalize('NFKD', t) if not _ud.combining(c))
return re.sub(r'[^a-z0-9\s]', ' ', t).split()
_ar_new = _ar_norm(resposta)
_ar_old = _ar_norm(_last_bot)
_ar_sim = 0.0
if _ar_new and _ar_old:
import difflib as _dl
_ar_sim = _dl.SequenceMatcher(None, _ar_new, _ar_old).ratio()
_ar_same_q = (_ar_norm(mensagem) == _ar_norm(_prev_user)) if _prev_user else False
if _ar_sim >= 0.9 and _last_bot and not _ar_same_q:
self.logger.warning(f"🔁 [ANTI-REPEAT] resposta repete anterior (sim={_ar_sim:.2f}) p/ pergunta nova — regenerando 1x")
_ar_retry_prompt = (
f"{prompt_enriched}\n\n"
"⚠️ CORREÇÃO: a tua resposta repetiu palavra por palavra a resposta anterior "
"desta conversa, mas a pergunta atual é DIFERENTE. NÃO repitas. "
"Responde ESTRITAMENTE à pergunta atual, com outras palavras, curto."
)
_ar_res, _ar_model = self.providers.generate(_ar_retry_prompt, context_history or [])
if isinstance(_ar_res, str) and _ar_res.strip():
_ar_res = self._sanitize_llm_response(_ar_res)
_ar_res = self._aggressive_thinking_leak_cleanup(_ar_res)
if _ar_res.strip() and _dl.SequenceMatcher(None, _ar_norm(_ar_res), _ar_old).ratio() < 0.9:
resposta = _ar_res
modelo_usado = _ar_model
self.logger.info(f"✅ [ANTI-REPEAT] resposta variada via {_ar_model}")
except Exception as _ar_err:
self.logger.debug(f"[ANTI-REPEAT] skip: {_ar_err}")
# ✅ SANITY CHECK: Se sanitize removeu conteúdo interno / inglês meta-leak, RETRY
_words_tmp = re.findall(r"[A-Za-zÀ-ÿ']+", resposta or "")
_meta_en_tmp = bool(re.search(r"\bThe user (is|was|said|saying|responding|wants)\b", resposta or "", re.I)) if resposta else False
if self._contains_internal_markers(resposta) or not resposta.strip() or len(resposta.strip()) < 3 or _meta_en_tmp:
self.logger.warning(f"🚨 [SECURITY] Resposta continha markers internos. Retry com anti-leak...")
retry_prompt = (
f"{prompt_enriched}\n\n"
"⚠️ ERRO INTERNO CORRIGIDO: A resposta anterior foi descartada porque continha tags internas "
"(EMOCAO_INTENCAO, CONSELHO ESTRATÉGICO, NUNCA revele, etc). "
"Gere APENAS a resposta final para o utilizador. "
"ZERO tags XML. ZERO metadados. ZERO instruções internas. "
"Responda como um humano normal respondendo diretamente ao utilizador."
)
try:
retry_res, retry_model = self.providers.generate(retry_prompt, context_history or [])
if isinstance(retry_res, str) and retry_res.strip():
resposta = self._sanitize_llm_response(retry_res)
self.logger.info(f"✅ [SECURITY RETRY] Resposta regenerada via {retry_model}")
except Exception as retry_err:
self.logger.error(f"❌ [SECURITY RETRY] Falhou: {retry_err}")
# Se retry ainda contém markers, limpa linha por linha
if self._contains_internal_markers(resposta):
resposta = re.sub(r"</?[A-Z_]+>", "", resposta, flags=re.IGNORECASE)
resposta = re.sub(r"^[A-Z_]{3,}:\s*.+$", "", resposta, flags=re.MULTILINE)
resposta = re.sub(r"INSTRUÇÃO:.*", "", resposta, flags=re.IGNORECASE)
resposta = re.sub(r"NUNCA revele.*", "", resposta, flags=re.IGNORECASE)
resposta = re.sub(r"Tone Level:.*", "", resposta, flags=re.IGNORECASE)
resposta = re.sub(r"\n{3,}", "\n\n", resposta).strip()
# 🚀 FAST RETURN (FIX 2026-09-12): persistência pesada vai para background.
# Motivo: STM + contexto.atualizar + persona-fetch + checkpoint somavam 40-90s
# de CPU/DB e estouravam o timeout do TS (180s). Só o CRITICAL SAVE abaixo
# bloqueia (idempotência); o resto roda em thread após o return.
_topico_detectado_local = getattr(unified_context, 'topic_principal', '') if unified_context else ''
def _deferred_persistence(
_d_ctx=contexto, _d_msg=mensagem, _d_resp=resposta, _d_user=usuario,
_d_num=numero, _d_citada=mensagem_citada, _d_modelo=modelo_usado,
_d_grupo=grupo_id, _d_quoted=quoted_author_name, _d_emocao=emocao,
_d_topico=_topico_detectado_local, _d_remote=list(remote_actions or []),
_d_conv=conversation_id, _d_tipo_conv=tipo_conversa,
_d_reply_to_bot=reply_to_bot, _d_quoted_orig=quoted_text_original,
_d_msg_id=message_id, _d_nome=nome_usuario, _d_is_reply=is_reply,
_d_analise_emocao=(analise.get('emocao', 'neutral') if isinstance(analise, dict) else 'neutral'),
_d_reply_prio=(unified_context.reply_priority if unified_context else 2),
):
import time as _t_bg
_t_bg0 = _t_bg.time()
try:
try:
_d_ctx.atualizar_contexto(_d_msg, _d_resp)
except Exception:
pass
# STM user + assistant + persona tracker (movido do caminho síncrono)
try:
if getattr(self, 'unified_builder', None) and _d_conv:
try:
_ri_user = None
if _d_is_reply:
_ri_user = {'is_reply': True, 'reply_to_bot': _d_reply_to_bot, 'quoted_text_original': _d_quoted_orig or _d_citada, 'priority_level': _d_reply_prio}
if not getattr(self, '_stm_user_saved', False):
self.unified_builder.add_to_stm(conversation_id=_d_conv, role="user", content=_d_msg, author_name=_d_user, emocao=_d_analise_emocao, reply_info=_ri_user, numero_usuario=_d_num or _d_user, grupo_id=_d_grupo or '', tipo_conversa=_d_tipo_conv, recipient=_d_quoted or '', quoted_author=_d_quoted or '', is_listen=False, message_id=_d_msg_id or '')
self._stm_user_saved = True
except Exception:
pass
_reply_info_for_stm = None
if _d_is_reply:
_reply_info_for_stm = {
'is_reply': True,
'reply_to_bot': _d_reply_to_bot,
'quoted_text_original': _d_quoted_orig or _d_citada,
'priority_level': _d_reply_prio
}
try:
if not getattr(self, '_stm_user_saved', False):
self.unified_builder.add_to_stm(
conversation_id=_d_conv,
role="user",
content=_d_msg,
author_name=_d_user,
emocao=_d_analise_emocao,
reply_info=_reply_info_for_stm,
numero_usuario=_d_num or _d_user,
grupo_id=_d_grupo or '',
tipo_conversa=_d_tipo_conv,
recipient=_d_quoted or '',
quoted_author=_d_quoted or '',
is_listen=False,
message_id=_d_msg_id or ''
)
except Exception:
pass
_conteudo_assistant = _d_resp
if not _conteudo_assistant and _d_remote and len(_d_remote) > 0:
_conteudo_assistant = "[Ação executada silenciosamente pelo sistema]"
try:
self.unified_builder.add_to_stm(conversation_id=_d_conv, role="assistant", content=_conteudo_assistant, author_name="Kiami", emocao="neutral", reply_info={'responded_to': _d_user, 'responded_to_numero': _d_num, 'original_topic': (_d_msg or '')[:120], 'is_assistant_response': True, 'numero_usuario': _d_num or _d_user, 'grupo_id': _d_grupo or '', 'tipo_conversa': _d_tipo_conv}, numero_usuario=_d_num or _d_user, grupo_id=_d_grupo or '', tipo_conversa=_d_tipo_conv, recipient=_d_user, quoted_author='', is_listen=False, message_id=f"asst:{_d_msg_id or int(__import__('time').time()*1000)}")
except Exception:
try:
self.unified_builder.add_to_stm(conversation_id=_d_conv, role="assistant", content=_conteudo_assistant, author_name="Kiami", emocao="neutral", reply_info={'responded_to': _d_user, 'responded_to_numero': _d_num, 'original_topic': (_d_msg or '')[:120], 'is_assistant_response': True})
except Exception:
pass
try:
self._stm_user_saved = False
except Exception:
pass
# LTM Persona Background Tracker
_tracker = self.persona_tracker
if _tracker is not None:
try:
_historico_raw = self.stm_manager.get_messages(_d_conv, limit=10)
if len(_historico_raw) >= 4:
_msgs_list = []
for _m in _historico_raw:
_role = "user" if getattr(_m, 'role', 'user') == 'user' else "assistant"
_msgs_list.append({"role": _role, "content": getattr(_m, 'content', '')})
_numero_valid = _d_num if _d_num else _d_conv
_tracker.track_background(_numero_valid, _msgs_list)
except Exception as _pt_err:
self.logger.warning(f"PersonaTracker erro: {_pt_err}")
except Exception as _e:
self.logger.warning(f"Falha ao adicionar à STM: {_e}")
# Session checkpoint
if SESSION_MEMORY_AVAILABLE and _d_num:
try:
from .session_memory import SessionCheckpoint
_checkpoint = SessionCheckpoint(
session_id=generate_session_id(_d_num, _d_grupo),
user_id=_d_num,
group_id=_d_grupo,
timestamp=__import__('time').time(),
summary=_d_resp[:200] if _d_resp else "",
active_topics=[_d_topico] if _d_topico else [],
key_decisions=[],
unresolved=[],
skills_used=[_ra.get('action', '') for _ra in (_d_remote or [])],
mood=_d_emocao or "neutral",
message_count=1
)
self.session_manager.end_session(_checkpoint, summary=_d_resp[:200] if _d_resp else "")
except Exception as _e:
self.logger.debug(f"⚠️ Session checkpoint failed: {_e}")
except Exception as _e:
self.logger.debug(f"⚠️ [BG-PERSIST] falhou: {_e}")
finally:
try:
self.logger.info(f"⏱️ [BG-PERSIST] persistência pós-resposta concluída em {_t_bg.time()-_t_bg0:.1f}s")
except Exception:
pass
try:
threading.Thread(target=_deferred_persistence, daemon=True).start()
except Exception as _e:
self.logger.warning(f"Falha ao iniciar BG-PERSIST: {_e}")
# 🔴 FIX #2-CAMADA: Salvar resposta em DB ANTES de retornar (síncrono!)
# Motivo: Evita corrida entre Request B e _background_tasks()
# Se Request B chegar antes de _background_tasks() terminar, passa dedup checks
# Solução: Salvar imediatamente aqui, ANTES de retornar ao cliente
# Isso garante que qualquer retry veja a resposta já no DB
if self.db and message_id:
try:
# Salva resposta imediatamente (bloqueante, mas rápido - <100ms)
db_save_ok = self.db.salvar_mensagem(
usuario=usuario,
mensagem=mensagem,
resposta=resposta,
numero=numero,
is_reply=is_reply,
mensagem_original=mensagem_citada,
modelo_usado=modelo_usado,
message_id=message_id, # ✅ Crítico: message_id para idempotência
nome_usuario=nome_usuario
)
if db_save_ok:
self.logger.info(f"✅ [CRITICAL SAVE] message_id={message_id} salvo ANTES de retornar (T={time.time():.2f})")
else:
self.logger.warning(f"⚠️ [CRITICAL SAVE WARN] salvar_mensagem retornou False para {message_id}")
except Exception as critical_save_err:
# ❌ Log do erro mas NÃO interrompe response (client sempre recebe resposta)
self.logger.error(f"❌ [CRITICAL SAVE ERROR] Falha ao salvar antes de retornar: {critical_save_err} | message_id={message_id}")
# ⚠️ Não re-raise aqui - cliente já gerou resposta, apenas salva em background
# (session checkpoint movido para _deferred_persistence — ver bloco FAST RETURN acima)
return jsonify({
'resposta': resposta,
'pesquisa_feita': bool(web_content),
'tipo_mensagem': tipo_mensagem,
'is_reply': is_reply,
'reply_to_bot': reply_to_bot,
'quoted_author': quoted_author_name,
'quoted_content': quoted_text_original or mensagem_citada,
'context_hint': context_hint,
'remote_actions': remote_actions,
'media_response': media_response # ✅ NOVO: Para imagens geradas
})
except Exception as e:
import traceback
self.logger.error(f'[ERRO /akira] {type(e).__name__}: {e}')
self.logger.error(traceback.format_exc())
return jsonify({'resposta': 'Eita! Deu erro interno', 'debug': str(e)}, 500)
finally:
# ✅ Libera o semáforo da conversa em QUALQUER caminho de saída
if _sem_acquired and _sem:
_sem.release()
try:
_dequeue_and_notify_next(_conv_key)
except Exception:
pass
@self.api.route('/escutar', methods=['POST'])
async def escutar_endpoint(request: FastAPIRequest):
try:
data = await request.json()
mensagem = data.get('mensagem', '')
usuario = data.get('usuario', 'desconhecido')
numero = data.get('numero', 'desconhecido')
nome_usuario = data.get('nome_usuario', usuario)
tipo_conversa = data.get('tipo_conversa', 'grupo')
grupo_id = data.get('grupo_id', '')
grupo_nome = data.get('grupo_nome', '')
contexto_grupo = grupo_id or data.get('contexto_grupo', '')
# ── Metadados de Reply (enriquecidos pelo BotCore) ──────────────
mensagem_citada = data.get('mensagem_citada', '')
reply_meta = data.get('reply_metadata') or {}
is_reply = bool(reply_meta.get('is_reply', False))
reply_to_bot = bool(reply_meta.get('reply_to_bot', False))
quoted_author_name = reply_meta.get('quoted_author_name', 'desconhecido')
quoted_author_numero = reply_meta.get('quoted_author_numero', 'desconhecido')
quoted_type = reply_meta.get('quoted_type', 'texto')
quoted_text_original = reply_meta.get('quoted_text_original', '')
context_hint = reply_meta.get('context_hint', 'contexto_geral')
message_id = data.get('message_id') # ✅ Adicionado para idempotência
if not mensagem:
return jsonify({'status': 'ignored', 'motivo': 'mensagem_vazia'}, 400)
# ═══ AUTONOMOUS AGENT: Track flood/spam em tempo real (escuta passiva) ═══
if _autonomous_agent and tipo_conversa == "grupo":
try:
_track_result = _autonomous_agent.track_message(
user_jid=numero,
group_jid=grupo_id,
message=mensagem
)
if _track_result and _track_result.get("type") == "remote_action":
self.logger.warning(f"🚨 [AUTONOMOUS TRACK] Flood/spam detetado em escuta: {_track_result}")
except Exception as _track_err:
self.logger.debug(f"⚠️ [AUTONOMOUS TRACK] Erro: {_track_err}")
# ✅ BOT RESPONSE: Armazena a própria resposta do bot no STM
# para que o LLM possa referenciar mensagens anteriores do bot
# FIX: Usa o context_id do USUÁRIO (não do bot) para que a resposta
# apareça no contexto quando o usuário responder ao bot.
is_bot_response = bool(data.get('is_bot_response', False))
if is_bot_response:
self.logger.info(f"🤖 [BOT-RESPONSE] Armazenando resposta do bot no STM: {mensagem[:80]}...")
if getattr(self, 'unified_builder', None):
# Gera context_id baseado no USUÁRIO (não no bot) para
# que a resposta fique visível no contexto da conversa
if self.context_manager is not None:
context_id = self.context_manager.get_conversation_id(
usuario=usuario,
conversation_type=tipo_conversa,
group_id=contexto_grupo if tipo_conversa == 'grupo' else None,
numero=numero
)
else:
raw = f"{usuario}:{tipo_conversa}:{numero}"
context_id = hashlib.sha256(raw.encode()).hexdigest()
self.unified_builder.add_to_stm(
conversation_id=context_id,
role="assistant",
content=mensagem,
author_name="Kiami",
emocao="neutral",
reply_info={'observed_only': False, 'reply_to_bot': True}
)
return jsonify({'status': 'armazenado', 'motivo': 'bot_response'})
# (Dedup removido do /escutar — BotCore já controla duplicatas)
# ── Monta contexto de reply para o aprendizado ───────────────────
# Inclui na mensagem uma nota sobre o reply para o modelo absorver
mensagem_com_contexto = mensagem
if is_reply and quoted_text_original:
label_autor = f"{quoted_author_name} (@{quoted_author_numero})" if quoted_author_numero != 'desconhecido' else quoted_author_name
mensagem_com_contexto = (
f"[REPLY para {label_autor}: \"{quoted_text_original[:200]}\"]\n"
f"{mensagem}"
)
elif is_reply and mensagem_citada:
mensagem_com_contexto = (
f"[REPLY: \"{mensagem_citada[:200]}\"]\n"
f"{mensagem}"
)
# Contexto extra para aprendizado
contexto_extra = grupo_nome or contexto_grupo
# 🎯 LISTEN ENGINE: Detectar FLAGS de direcionamento
listen_engine_log = ""
metadata = None
if LISTEN_ENGINE_AVAILABLE and self.listen_engine_manager:
try:
_quoted_for_listen = None
if is_reply and (quoted_text_original or mensagem_citada):
_quoted_for_listen = {
"from": quoted_author_numero if quoted_author_numero != 'desconhecido' else "",
"body": quoted_text_original or mensagem_citada or "",
"id": reply_meta.get('quoted_message_id') or reply_meta.get('quoted_msg_id') or "",
"author_name": quoted_author_name,
}
try:
_priv = tuple(getattr(config, "PRIVILEGED_USERS", ()) or ())
except Exception:
_priv = ()
if not _priv:
_priv = ("202391978787009",) # fallback: Isaac
# Parse completo de metadados com FLAGS
metadata = ListenEngine.parse_message_metadata(
remoteJid=grupo_id or numero,
fromMe=False,
quotedMsg=_quoted_for_listen,
pushName=nome_usuario,
body=mensagem,
author_id=numero,
msg_id=message_id or f"listen_{int(time.time() * 1000)}",
grupo_nome=grupo_nome,
privileged_users=_priv
)
# Adiciona ao contexto do grupo
self.listen_engine_manager.adicionar_mensagem(metadata)
# Rede de segurança PV: tipo_conversa='pv' vindo do BotCore
# (remoteJid sem @g.us) significa conversa privada — TODA
# mensagem em PV é dirigida ao bot, mesmo sem menção.
# O parse acima já cobre via eh_privado, mas se o BotCore
# enviar grupo_id residual, garantimos aqui.
if tipo_conversa == 'pv' and not metadata.requer_resposta:
metadata.requer_resposta = True
metadata.is_directed_to_bot = True
# Gera diagnóstico para logs
listen_engine_log = ListenEngine.gerar_diagnostico(metadata)
self.logger.info(f"🎯 [LISTEN ENGINE] {listen_engine_log}")
# Se a mensagem requer resposta, foi respondida pelo /akira
# Se NÃO requer resposta, é apenas contexto puro (OBSERVAÇÃO)
if metadata.requer_resposta:
self.logger.info(f"📍 [LISTEN ENGINE] Mensagem requer resposta (deve ir para /akira)")
else:
self.logger.info(f"📍 [LISTEN ENGINE] Mensagem é contexto puro (Kiami escuta e aprende)")
except Exception as le_err:
self.logger.warning(f"⚠️ [LISTEN ENGINE] Erro ao processar FLAGS: {le_err}")
listen_engine_log = f"[LISTEN ENGINE ERROR: {str(le_err)[:50]}]"
if hasattr(self, 'aprendizado_continuo') and self.aprendizado_continuo:
resultado = self.aprendizado_continuo.processar_mensagem(
mensagem=mensagem_com_contexto,
usuario=usuario,
numero=numero,
nome_usuario=nome_usuario,
tipo_conversa=tipo_conversa,
resposta_do_bot=False,
contexto_grupo=contexto_extra,
message_id=message_id # ✅ Idempotência
)
# -----------------------------------------------------------------
# [BACKGROUND] ATUALIZAÇÃO DA MEMÓRIA DE LONGO PRAZO (LSTM)
# Ouve as conversas de grupos/pv para manter contexto, sem
# interferir ou bloquear a API.
# -----------------------------------------------------------------
try:
from .lstm_extension import get_lstm_extension
lstm_ext = get_lstm_extension(self.db)
# Isolamento estrito de contexto (garante que um grupo não vaza para outro)
if self.context_manager is not None:
context_id = self.context_manager.get_conversation_id(
usuario=usuario,
conversation_type=tipo_conversa,
group_id=contexto_grupo,
numero=numero
)
else:
raw = f"{usuario}:{tipo_conversa}:{numero}"
context_id = hashlib.sha256(raw.encode()).hexdigest()
# -----------------------------------------------------------------
# [STM] INJEÇÃO NA MEMÓRIA DE CURTO PRAZO
# -----------------------------------------------------------------
if getattr(self, 'unified_builder', None) and context_id:
# ✅ OBSERVED_ONLY: Mensagens do /escutar são APENAS OBSERVAÇÃO DE GRUPO.
# Nunca são pedidos dirigidos à Kiami. Marcamos com observed_only=True
# para que o context_history as separe claramente das mensagens dirigidas.
reply_info_for_stm = {
'observed_only': True, # 🔑 Flag de escuta passiva
'observed_author': nome_usuario,
'observed_author_numero': numero,
'author': nome_usuario,
'author_name': nome_usuario,
'numero_usuario': numero,
}
if is_reply:
reply_info_for_stm.update({
'is_reply': True,
'reply_to_bot': reply_to_bot,
'quoted_text_original': quoted_text_original or mensagem_citada,
'quoted_author_name': quoted_author_name,
'quoted_author': quoted_author_name,
'recipient': quoted_author_name if quoted_author_name != 'desconhecido' else '',
'priority_level': 1
})
self.unified_builder.add_to_stm(
conversation_id=context_id,
role="user",
content=mensagem_com_contexto,
author_name=nome_usuario,
emocao="neutral",
reply_info=reply_info_for_stm,
numero_usuario=numero,
grupo_id=grupo_id or '',
tipo_conversa=tipo_conversa,
recipient=quoted_author_name if is_reply and quoted_author_name != 'desconhecido' else '',
quoted_author=quoted_author_name if is_reply and quoted_author_name != 'desconhecido' else '',
is_listen=True
)
try:
if len(mensagem) > 10 and getattr(self, 'persona_tracker', None):
try:
self.persona_tracker.track_background(numero, [{'role': 'user', 'content': mensagem, 'listen': True}])
except Exception:
pass
except Exception:
pass
try:
from .user_profiler import get_user_profiler
get_user_profiler().extrair_dados_escuta_assincrono(
user_id=numero or usuario,
mensagem=mensagem_com_contexto,
contexto_grupo=contexto_grupo,
llm_manager=self,
context_id=context_id
)
except Exception as prof_err:
self.logger.warning(f"⚠️ [ESCUTA] Falha ao acionar profiler: {prof_err}")
# ✅ IDEMPOTENCY: Evita duplicar se já processado pelo /akira ou escuta anterior
if message_id:
# Tenta evitar duplicados via cache simples no lstm_ext
setattr(lstm_ext, '_current_speaker_name_temp', nome_usuario)
lstm_ext.process_message_background(
context_id=context_id,
numero_usuario=numero,
message=mensagem_com_contexto,
role="user",
message_id=message_id
)
else:
setattr(lstm_ext, '_current_speaker_name_temp', nome_usuario)
lstm_ext.process_message_background(
context_id=context_id,
numero_usuario=numero,
message=mensagem_com_contexto,
role="user"
)
# Se for reply, registra também a mensagem citada como contexto anterior
if is_reply and quoted_text_original:
setattr(lstm_ext, '_current_speaker_name_temp', quoted_author_name)
lstm_ext.process_message_background(
context_id=context_id,
numero_usuario=quoted_author_numero,
message=quoted_text_original[:500],
role="user"
)
except Exception as e:
self.logger.warning(f"⚠️ [LSTM ESCUTA] Falha no processamento: {e}")
# O BotCore usa estes campos para decidir se chama /akira.
# Antes o endpoint retornava só 'aprendido' e a decisão era
# apenas logada — o BotCore nunca sabia quando responder.
_requer = bool(getattr(metadata, 'requer_resposta', False))
return jsonify({
'status': 'aprendido',
'analise': resultado.get('analise', {}),
'aprendizado': resultado.get('aprendizado', {}),
'requer_resposta': _requer,
'deve_chamar_akira': _requer,
'flags': listen_engine_log,
'eh_privado': bool(getattr(metadata, 'eh_privado', tipo_conversa == 'pv')),
'is_privileged_user': bool(getattr(metadata, 'is_privileged_user', False)),
'tipo_mensagem': getattr(metadata, 'tipo_mensagem', 'desconhecido'),
})
else:
return jsonify({'status': 'aprendizado_indisponivel'}, 503)
except Exception as e:
self.logger.exception('Erro em /escutar')
return jsonify({'error': str(e)}, 500)
@self.api.route('/contexto_global', methods=['POST'])
async def contexto_global_endpoint(request: FastAPIRequest):
try:
try:
data = await request.json()
except Exception:
data = {}
topico = data.get('topico', None)
limite = data.get('limite', 10)
if self.aprendizado_continuo:
contexto = self.aprendizado_continuo.obter_contexto_para_llm(
topico=topico, limite=limite
)
return jsonify({'contexto_global': contexto})
else:
return jsonify({'contexto_global': []})
except Exception as e:
self.logger.exception('Erro em /contexto_global')
return jsonify({'error': str(e)}, 500)
@self.api.route('/melhor_api', methods=['POST'])
async def melhor_api_endpoint(request: FastAPIRequest):
try:
data = await request.json()
complexidade = data.get('complexidade', 0.5)
emocao = data.get('emocao', 'neutral')
intencao = data.get('intencao', 'afirmacao')
tipo_conversa = data.get('tipo_conversa', 'pv')
if self.aprendizado_continuo:
melhor_api = self.aprendizado_continuo.get_best_api_for_context(
complexidade=complexidade,
emocao=emocao,
intencao=intencao,
tipo_conversa=tipo_conversa
)
return jsonify({'melhor_api': melhor_api})
else:
return jsonify({'melhor_api': 'groq'})
except Exception as e:
self.logger.exception('Erro em /melhor_api')
return jsonify({'error': str(e)}, 500)
@self.api.route('/health', methods=['GET'])
async def health_check(request: FastAPIRequest):
return jsonify({'status': 'OK', 'version': '21.01.2025'}, 200)
@self.api.route('/reset', methods=['POST'])
async def reset_endpoint(request: FastAPIRequest):
try:
data = await request.json()
usuario = data.get('usuario')
numero = data.get('numero', '')
tipo_conversa = data.get('tipo_conversa', 'pv')
grupo_id = data.get('grupo_id')
full_reset = data.get('full_reset', False)
# 1. Limpa cache de contexto do usuário
if usuario and usuario in self.contexto_cache:
self.contexto_cache._store.pop(usuario, None)
self.logger.info(f"[RESET] Cache de contexto limpo para: {usuario}")
# 2. Limpa Short-Term Memory
if hasattr(self, 'context_manager') and self.context_manager and numero:
try:
ctx_id = generate_context_id(numero, tipo_conversa, grupo_id)
self.context_manager.delete_context(ctx_id)
self.logger.info(f"[RESET] Contexto isolado deletado para usuário ({tipo_conversa})")
except Exception as e:
self.logger.warning(f"[RESET] Erro ao deletar contexto isolado: {e}")
# 3. Limpa STM
if hasattr(self, 'stm_manager') and self.stm_manager and numero:
try:
ctx_id = generate_context_id(numero, tipo_conversa, grupo_id)
# Limpa mensagens STM daquele conversation_id
if hasattr(self.stm_manager, 'clear_messages'):
self.stm_manager.clear_messages(ctx_id)
self.logger.info(f"[RESET] STM limpa para {ctx_id}")
except Exception as e:
self.logger.warning(f"[RESET] Erro ao limpar STM: {e}")
# 4. Full reset: limpa TUDO
if full_reset:
self.contexto_cache._store.clear()
if hasattr(self, 'stm_manager') and self.stm_manager:
if hasattr(self.stm_manager, '_messages'):
self.stm_manager._messages.clear()
if hasattr(self, 'unified_builder') and self.unified_builder:
if hasattr(self.unified_builder, 'db') and self.unified_builder.db:
try:
db = self.unified_builder.db
if numero:
db._execute_with_retry("DELETE FROM interacoes WHERE numero = %s", (numero,), commit=True)
else:
db._execute_with_retry("DELETE FROM interacoes", commit=True)
self.logger.info("[RESET] Interações no DB limpas")
except Exception as e:
self.logger.warning(f"[RESET] Erro ao limpar DB: {e}")
self.logger.info("[RESET] FULL RESET concluído")
return jsonify({'status': 'success', 'message': 'Reset completo realizado (cache + STM + DB)'}, 200)
return jsonify({'status': 'success', 'message': f'Contexto de {usuario or numero} resetado'}, 200)
except Exception as e:
self.logger.exception('Erro em /reset')
return jsonify({'error': str(e)}, 500)
@self.api.route('/pesquisa', methods=['POST'])
async def pesquisa_endpoint(request: FastAPIRequest):
try:
data = await request.json()
query = data.get('query', '')
if not query:
return jsonify({'error': 'Query vazia'}, 400)
resultado = self.web_search.pesquisar(query, num_results=5)
return jsonify({
'resumo': resultado.get('resumo', ''),
'conteudo_bruto': resultado.get('conteudo_bruto', ''),
'tipo': resultado.get('tipo', 'geral'),
'timestamp': resultado.get('timestamp', '')
})
except Exception as e:
self.logger.exception('Erro na pesquisa')
return jsonify({'error': str(e)}, 500)
@self.api.route('/status', methods=['GET'])
async def status_endpoint(request: FastAPIRequest):
return jsonify({
'status': 'OK',
'version': '21.01.2025',
'web_search': 'ativo' if self.web_search else 'inativo'
}), 200
@self.api.route('/vision/analyze', methods=['POST'])
async def vision_analyze_endpoint(request: FastAPIRequest):
"""
Endpoint de visão computacional e OCR.
Recebe imagem em base64 e retorna análise completa.
"""
try:
try:
data = await request.json()
except Exception:
data = {}
imagem_base64 = data.get('imagem', '')
usuario = data.get('usuario', 'anonimo')
numero = data.get('numero', 'desconhecido')
if not imagem_base64:
return jsonify({'error': 'Imagem vazia'}, 400)
self.logger.info(f"[VISION] Análise solicitada por {usuario}")
# Configurações opcionais
include_ocr = data.get('include_ocr', True)
include_shapes = data.get('include_shapes', True)
include_objects = data.get('include_objects', True)
# Obtém instância de visão computacional
vision = get_computer_vision()
# Executa análise completa com o novo pipeline v3.0
result = vision.analyze_image(imagem_base64, user_id=numero)
if result.get('success'):
# A descrição agora vem direto do Gemini Vision ou Memória Visual
self.logger.info(f"[VISION] Análise completa: QR={result.get('qr')}, OCR={len(result.get('ocr', ''))} chars")
else:
self.logger.warning(f"[VISION] Falha na análise: {result.get('error')}")
return jsonify(result)
except Exception as e:
self.logger.exception('Erro em /vision/analyze')
return jsonify({'error': str(e)}, 500)
@self.api.route('/vision/ocr', methods=['POST'])
async def vision_ocr_endpoint(request: FastAPIRequest):
"""
Endpoint específico para OCR.
Otimizado para extração de texto.
"""
try:
try:
data = await request.json()
except Exception:
data = {}
imagem_base64 = data.get('imagem', '')
numero = data.get('numero', 'desconhecido')
if not imagem_base64:
return jsonify({'error': 'Imagem vazia'}, 400)
vision = get_computer_vision()
result = vision.analyze_base64(imagem_base64, user_id=numero)
# Retorna apenas resultado OCR
ocr_result = result.get('ocr', {})
return jsonify({
'success': ocr_result.get('success', False),
'text': ocr_result.get('text', ''),
'confidence': ocr_result.get('confidence', 0),
'languages': ocr_result.get('languages', []),
'word_count': ocr_result.get('word_count', 0)
})
except Exception as e:
self.logger.exception('Erro em /vision/ocr')
return jsonify({'error': str(e)}, 500)
@self.api.route('/vision/learned', methods=['POST'])
async def vision_learned_endpoint(request: FastAPIRequest):
"""
Retorna lista de imagens aprendidas pelo usuário.
"""
try:
try:
data = await request.json()
except Exception:
data = {}
numero = data.get('numero', '')
if not numero:
return jsonify({'error': 'Número obrigatório'}, 400)
vision = get_computer_vision()
images = vision.get_learned_images(numero)
return jsonify({
'count': len(images),
'images': images
})
except Exception as e:
self.logger.exception('Erro em /vision/learned')
return jsonify({'error': str(e)}, 500)
@self.api.route('/vision/stats', methods=['GET'])
async def vision_stats_endpoint(request: FastAPIRequest):
"""
Retorna estatísticas do módulo de visão computacional.
"""
try:
vision = get_computer_vision()
stats = vision.get_stats()
return jsonify(stats)
except Exception as e:
return jsonify({'error': str(e)}, 500)
def _get_user_context(self, usuario, conversation_id=None):
# 🔧 FIX: Usa conversation_id como chave primária para isolamento total
cache_key = conversation_id if conversation_id else usuario
if cache_key not in self.contexto_cache:
db_path = getattr(self.config, 'DB_PATH', 'kiami.db')
db = Database(db_path)
# Passa conversation_id para o objeto Contexto para persistência isolada
self.contexto_cache[cache_key] = Contexto(db, usuario=usuario, conversation_id=conversation_id)
return self.contexto_cache[cache_key]
def _get_history_for_llm(self, contexto):
try:
if hasattr(contexto, 'obter_historico_para_llm'):
return contexto.obter_historico_para_llm()
except Exception:
pass
try:
historico = contexto.obter_historico()
resultado = []
for h in historico:
if isinstance(h, tuple) and len(h) >= 2:
if h[0]:
resultado.append({"role": "user", "content": str(h[0])})
if h[1]:
resultado.append({"role": "assistant", "content": str(h[1])})
elif isinstance(h, dict):
resultado.append(h)
return resultado
except Exception:
pass
return []
def _get_speaker_name_cached(self, numero_usuario: str) -> Optional[str]:
"""
Recupera o nome de um speaker a partir do cache ou database.
Usado para converter numero_usuario para nome legível em contexto de grupo.
Args:
numero_usuario: Número WhatsApp do speaker
Returns:
Nome do speaker se encontrado, caso contrário None
"""
try:
if not numero_usuario or numero_usuario == 'desconhecido':
return None
# Tentar recuperar do database se disponível
if self.db:
# Tenta buscar nome na tabela de personas ou mensagens
try:
rows = self.db._execute_with_retry(
"SELECT nome_usuario FROM mensagens WHERE numero = ? LIMIT 1",
(numero_usuario,)
)
if rows and rows[0].get('nome_usuario'):
return rows[0]['nome_usuario']
except:
pass
# Fallback: tenta em personas_usuario
try:
rows = self.db._execute_with_retry(
"SELECT nome FROM persona_usuario WHERE numero_usuario = ? LIMIT 1",
(numero_usuario,)
)
if rows and rows[0].get('nome'):
return rows[0]['nome']
except:
pass
return None
except Exception as e:
self.logger.debug(f"Erro ao recuperar speaker name: {e}")
return None
def _build_prompt(
self,
usuario: str,
numero: str,
mensagem: str,
analise: Dict[str, Any],
contexto,
web_content: str = "",
mensagem_citada: str = "",
is_reply: bool = False,
reply_to_bot: bool = False,
quoted_author_name: str = "",
quoted_author_numero: str = "",
quoted_type: str = "texto",
quoted_text_original: str = "",
quoted_author_pure: str = "",
context_hint: str = "",
tipo_conversa: str = "pv",
tipo_mensagem: str = "texto",
tem_imagem: bool = False,
analise_visao: Optional[Dict[str, Any]] = None,
analise_doc: str = "",
unified_context = None,
dossie: Optional[Dict[str, Any]] = None,
conversation_id: str = ""
) -> str:
dias_pt = {0: 'Segunda-Feira', 1: 'Terça-Feira', 2: 'Quarta-Feira', 3: 'Quinta-Feira', 4: 'Sexta-Feira', 5: 'Sábado', 6: 'Domingo'}
meses_pt = {1: 'Janeiro', 2: 'Fevereiro', 3: 'Março', 4: 'Abril', 5: 'Maio', 6: 'Junho', 7: 'Julho', 8: 'Agosto', 9: 'Setembro', 10: 'Outubro', 11: 'Novembro', 12: 'Dezembro'}
now = config.get_current_datetime_compensated()
wd = now.weekday()
mo = now.month
data_hora = f"Hoje é {dias_pt[wd]}, {now.day} de {meses_pt[mo]} de {now.year}, e agora são exatamente {now.strftime('%H:%M')}."
# Identidade JID (Correção de Autoconhecimento)
bot_numero = str(getattr(self.config, 'BOT_NUMERO', '40755431264474'))
is_self_mentioned = bot_numero in re.sub(r'\D', '', mensagem)
is_self_quoted = str(quoted_author_numero) == bot_numero or bot_numero in str(quoted_author_numero)
strict_override = "STRICT_IDENTITY:"
strict_override += f"\n- Seu número: {bot_numero} | Você é Kiami (irmã mais velha da Akira 18)"
strict_override += "\n- NUNCA te identifiques como \"mimi\"/\"weza\"/\"nila\"/\"Kianda\" — são apelidos que Isaac usa para ti, tu respondes SEMPRE como Kiami."
strict_override += "\n⛔ REGRA ABSOLUTA ANTI-ALUCINAÇÃO: NUNCA inventes infra (Railway/storage/kernel/root/acesso root) ou nomes de grupo (Null Sector/Black Code). Para ESSES temas internos, se [WEB INFO]/[CONHECIMENTO]/[STM] não contêm o termo, diz 'não sei'. Para TODO o resto (conhecimento geral, explicações, opiniões), USA O TEU PRÓPRIO CONHECIMENTO — ele vale mesmo sem [WEB INFO]. 'Não sei' é ÚLTIMO recurso, nunca primeira saída. NUNCA use falácia/bait."
if self.persona:
strict_override += f"\n- PERSONALIDADE CONFIGURADA: {self.persona.get('personalidade', 'Séria, direta, agressiva')} | Tom de voz: {self.persona.get('tom_voz', 'Sério, agressivo')} | Nacionalidade: {self.persona.get('nacionalidade', 'Angolana')}"
strict_override += "\n- Mantenha essa personalidade CLARA, AGRESSIVA e DIRETA em todas as respostas."
if re.match(r'^\s*(beu|morena|b[eê]u)\b', (mensagem or "").lower()):
strict_override += "\n[APELIDO beu/morena DETECTADO] Usuário te chamou pelo apelido beu/morena — responde começando com \"oi\" (curta, máx 2 palavras se só saudação; se pergunta além do apelido tipo \"beu me explica X\", começa com \"oi\" + resposta curta factual)."
# Não impor coerência rígida se estiver respondendo a uma imagem (provavelmente pedindo para corrigir/alterar a geração)
is_media_reply = any(t in str(quoted_type).lower() for t in ['imagem', 'image', 'video', 'audio', 'documento'])
strict_override += "\n\nSTRICT_OVERRIDES:\n"
if tipo_mensagem == 'game':
strict_override += "- CONTEXTO DE JOGO: Esta mensagem contém um comando de jogo ou está relacionada a um mini-game (ex: #grid, #economy). Priorize a lógica do jogo e responda de forma envolvente, mas sem sair da persona.\n"
if dossie:
strict_override += "\n[DOSSIÊ DE USUÁRIO]\n"
strict_override += f"- Nome: {dossie.get('nome_conhecido', 'Desconhecido')}\n"
strict_override += f"- Estilo: {dossie.get('estilo_comunicacao', 'Desconhecido')}\n"
prefs = ", ".join(dossie.get("preferencias", [])) or "Nenhuma"
strict_override += f"- Preferências: {prefs}\n"
strict_override += "- Use este contexto naturalmente na conversa, sem ser explícito sobre o que sabe.\n"
palavras_mensagem = len(mensagem.split())
if palavras_mensagem <= 3:
strict_override += "- Input curto → resposta curta e natural (1-5 palavras). Nada de 'entendido'.\n"
elif palavras_mensagem <= 10:
strict_override += "- Resposta natural, sem prolixidade.\n"
else:
strict_override += "- Resposta concisa, sem divagações.\n"
strict_override += "- REGRA DE OURO: HONESTIDADE > CONFIANÇA. Se cometeu erro anterior, RECONHEÇA e corrija. Mantenha confiança mas NUNCA defenda informação falsa.\n"
strict_override += "- COMPREENSÃO OBRIGATÓRIA (vale mesmo que o system prompt esteja desatualizado): PROIBIDO 'não entendi', 'não sei o que quiseste dizer', 'sobre o quê?', 'podes repetir?'. Tu ENTENDES sempre — usa a thread/STM para o sentido mais provável e RESPONDE direto. 'Isso/ele' = último tópico. Frase curta de continuação = continua o thread. Gíria/erro ortográfico = tolera. Só em caso TOTALMENTE impossível, UMA pergunta curta COM palpite ('falas do X?'). Confiança > dúvida.\n"
strict_override += "- Se outro bot corrigir você, analise se está correto. Se estiver, diga 'Você tem razão'. Não defenda alucinação.\n"
strict_override += "- Se o usuário pedir ação prática (buscar, gerar, banir), essa é a prioridade absoluta. Execute a ferramenta primeiro.\n"
strict_override += "- PROIBIDO prometer pesquisa futura ('vou pesquisar', 'vou verificar', 'já te digo', 'aguarda'): ou usa [WEB INFO] acima para responder AGORA, ou responde com o que sabes / diz que não tens confirmação. NUNCA deixes a resposta pendente.\n"
strict_override += "- REGRA ABSOLUTA DE IDIOMA (ZERO TOLERÂNCIA): NUNCA respondas em INGLÊS ou outro idioma. TODA a resposta deve ser 100% em PORTUGUÊS ANGOLANO. Se o modelo subjacente gerar inglês, TRADUZ para português antes de responder. Expressões comuns em inglês (I'm, You're, Let's, etc.) são PROIBIDAS. Se o utilizador falar em inglês, responda em português.\n"
strict_override += f"\n- Data/Hora: {data_hora}\n"
if is_reply and mensagem_citada:
strict_override += "\n[REPLY - Contexto]\n"
if reply_to_bot:
strict_override += f"Mensagem sua anterior: \"{mensagem_citada[:300]}...\"\n"
strict_override += "- O utilizador está a REAGIR à sua mensagem anterior (não é uma pergunta nova sobre outro assunto).\n"
strict_override += "- Analise o TOM e INTENÇÃO: se o utilizador diz 'nunca ouvi falar', 'não sei o que é', 'o que é isso?', ele quer ESCLARECIMENTO sobre o tópico da sua mensagem anterior — NÃO uma definição genérica repetida.\n"
strict_override += "- EXPANDA a informação: dê mais contexto, exemplos práticos, ou explique de forma diferente do que já disse.\n"
strict_override += "- Se o utilizador discorda ou provoca, responda à provocação, não repita a informação.\n"
# 🔧 FIX 2026-08-27: INTROMISSÃO dinâmica CENTRO vs TERCEIRO
_replied = getattr(unified_context, 'replied_to_author', '') if unified_context else ""
_replied_content = getattr(unified_context, 'replied_to_content', '') if unified_context else ""
if _replied and _replied != usuario:
strict_override += f"- 🔀 INTERVENÇÃO TERCEIRO ATIVO: Tua mensagem citada foi enviada PARA [CENTRO {_replied}] (autor de \"{_replied_content[:60]}\"). Utilizador ATUAL [TERCEIRO {usuario}] interveio - ELE NÃO é autor do conteúdo original (ex: 'odeio rosas'). NUNCA diga 'ninguém te mandou não gostar de rosas' para {usuario}. Responda curto, agressivo: 'não é da tua conta, falei com ele como quiser' / 'não falei contigo, caralho' / 'eu falo com o {_replied} como quiser, não és tu que mandas'. Use 'ele' para {_replied}. Seja sutil, natural, curta e agressiva.\n"
else:
strict_override += "- INTROMISSÃO (REGRA DE OURO): A sua mensagem citada foi enviada PARA alguém específico (ex: Isaac = 'odeio rosas' → você 'tanto faz'). Se o utilizador ATUAL (ex: Sicrano) responde 'não fale assim com ele', ELE É TERCEIRA PESSOA, NÃO o autor de 'odeio rosas'. NUNCA diga 'ninguém te mandou não gostar de rosas' para o Sicrano. Responda curto, agressivo e com atribuição correta: 'não é da tua conta, falei com ele como quiser' / 'não falei contigo, caralho' / 'eu falo com o fulano como quiser, não és tu que mandas'. Use 'ele' referindo-se ao destinatário original.\n"
strict_override += "- Processe silenciosamente. Não mencione que está a ver o reply.\n"
else:
strict_override += f"Mensagem citada de {quoted_author_name}: \"{mensagem_citada[:300]}...\"\n"
strict_override += f"ID do autor: {quoted_author_numero}\n"
strict_override += "- Responda naturalmente ao ponto levantado.\n"
strict_override += "- Nunca diga 'vi que você falou' ou 'como citado'. Integre o contexto de forma invisível.\n"
if context_hint:
strict_override += f"- Contexto: {context_hint}\n"
# Se a mensagem atual é apenas uma confirmação curta do tipo 'sim', 'ok', 'leia sim',
# trate-a como uma continuação de uma ação anterior e execute a tarefa pendente em vez de responder com um simples aceno.
mensagem_lower = (mensagem or '').strip().lower()
if mensagem_lower in ['sim', 's', 'ok', 'okay', 'yes', 'leia sim', 'pode', 'pode sim', 'vai', 'continua', 'continue']:
strict_override += "\n[CONFIRMAÇÃO DE AÇÃO]\n"
strict_override += "- Esta mensagem é uma confirmação de ação anterior. Se houver um relatório, documento ou operação pendente, execute-a e devolva o resultado completo. Não responda apenas com um 'ok' ou 'certo'.\n"
strict_override += "- Use as ferramentas disponíveis para continuar a tarefa solicitada.\n"
if tipo_conversa == "grupo":
strict_override += "\n[Conversa em grupo - múltiplos participants]\n"
strict_override += "⚠️ AVISO CRÍTICO: Se outro bot (tipo @ISA, @Isaac_IA, etc) já respondeu na conversa:\n"
strict_override += " 1. NÃO REPITA a mesma informação com palavras diferentes\n"
strict_override += " 2. NÃO USE frases que já foram ditas (como markdown sobre 'procurar agulha no palheiro')\n"
strict_override += " 3. SE DISCORDAR da informação deles, explique por que. NÃO apenas defenda sua posição anterior\n"
strict_override += " 4. SE ELES ESTIVEREM CERTOS e você errou: Reconheça 'Você tem razão, cometi erro'\n"
# ✅ GROUP PARTICIPANT MAP: Extrair speakers únicos do STM para evitar confusão de identidade
if unified_context and unified_context.stm_messages:
speakers_seen = {} # numero -> nome
for _stm_msg in unified_context.stm_messages:
if _stm_msg.role == "user":
_author = getattr(_stm_msg, 'author_name', '') or ''
_autor_num = getattr(_stm_msg, 'author_number', '') or getattr(_stm_msg, 'numero', '') or ''
if _author and _author not in ('Usuário', 'Kiami', '') and _author != usuario:
speakers_seen[_autor_num or _author] = _author
if speakers_seen:
strict_override += "\n[GROUP_PARTICIPANT_MAP - LEIA ANTES DE RESPONDER]\n"
strict_override += f"👤 USUÁRIO ATUAL (quem está te escrevendo AGORA): {usuario}\n"
strict_override += f"👥 OUTROS PARTICIPANTES DO GRUPO (NÃO estão te escrevendo agora):\n"
for _num, _nome in speakers_seen.items():
strict_override += f" - {_nome}\n"
strict_override += "\n🔴 REGRAS ABSOLUTAS DE IDENTIDADE EM GRUPO:\n"
strict_override += f" 1. Você está respondendo APENAS para {usuario}. Os outros participantes NÃO estão te pedindo nada agora.\n"
strict_override += " 2. No histórico abaixo, cada '[Nome]: mensagem' = aquela pessoa específica falou isso.\n"
strict_override += " 3. NÃO misture o que diferentes pessoas disseram. Cada fala pertence ao seu autor.\n"
strict_override += f" 4. Se {usuario} perguntar 'sobre o que vocês estavam falando?' ou similar:\n"
strict_override += " → Resuma OBJETIVAMENTE as conversas que viu no histórico, indicando QUEM disse O QUÊ.\n"
strict_override += " → Ex: 'A Kiami estava falando sobre X, e você me pediu Y.'\n"
strict_override += " 5. NUNCA invente que o usuário atual estava numa conversa que ele não estava.\n"
else:
strict_override += "\n[Conversa privada 1-a-1]\n"
if tem_imagem:
strict_override += "\n[IMAGEM ANEXADA]\n"
if analise_visao and isinstance(analise_visao, dict) and analise_visao.get('description'):
strict_override += f"Análise: {analise_visao.get('description', 'Sem detalhes')}\n"
if analise_visao.get('ocr'):
strict_override += f"Texto detectado (OCR): {analise_visao['ocr'][:1000]}\n"
if analise_visao.get('qr'):
strict_override += f"Link/QR: {analise_visao['qr']}\n"
if analise_visao.get('objects'):
strict_override += f"Objetos: {', '.join(analise_visao['objects'])}\n"
else:
strict_override += "NOTA: O usuário enviou uma imagem mas a análise visual falhou. Peça para reenviar se necessário.\n"
strict_override += "- Comente sobre a imagem de forma natural se relevante. Se pedir ação (postar, editar, apagar), use ferramentas.\n"
if analise_doc:
strict_override += "\n[DOCUMENTO ANEXADO]\n"
strict_override += f"Análise: {analise_doc}\n"
strict_override += "Use estas informacoes para responder ao usuario sobre o arquivo enviado.\n"
if web_content:
# 🔧 FIX 2026-08-27: filtrar web_content irrelevante/erro e limitar para não bloatar prompt (10000→1800)
if "ERRO NA PESQUISA" in web_content or "=== ERRO" in web_content:
self.logger.info("🛡️ [WEB FILTER] web_content com erro descartado")
web_content = ""
if web_content:
strict_override += "\n[WEB INFO - PESQUISA ATUALIZADA EM TEMPO REAL]\n"
strict_override += "ATENÇÃO SOBRE A PESQUISA: Se o usuário cometeu um erro ortográfico ao pedir a pesquisa (ex: 'auror' em vez de 'autor') e a pesquisa retornou os termos certos, ASSUMA A VERSÃO CORRETA DA PESQUISA e ignore a burrice ortográfica do usuário na hora de extrair fatos.\n"
strict_override += "⚠️ REGRAS ABSOLUTAS SOBRE O CONTEÚDO ABAIXO:\n"
strict_override += "1. NUNCA copies o texto abaixo literalmente na tua resposta.\n"
strict_override += "2. NÃO incluas marcadores como '=== 🔎 PESQUISA WEB:', '[CONTEÚDO]', '[1]', '🔗' na resposta.\n"
strict_override += "3. Processa a informação e responde APENAS como Kiami — curta, direta, sem emojis.\n"
strict_override += "4. Resume os factos relevantes, não reproduzas o conteúdo bruto.\n"
strict_override += "5. Se web_content não menciona a mensagem atual, IGNORE e não invente.\n"
strict_override += web_content[:1800] + "\n"
# 🔴 ANTI-HALLUCINATION PROTOCOL FOR DARKNET TOPICS - ONLY IF QUERY IS ABOUT DARKNET
darknet_keywords = ["darknet", "deep web", "deepweb", "onion", ".onion", "tor", "hidden", "busca da darknet"]
query_lower = (mensagem or "").lower()
if any(kw in query_lower for kw in darknet_keywords):
strict_override += "\n[DARKNET/DEEP WEB - ANTI-HALLUCINATION]\n"
strict_override += "Se a pergunta é sobre buscadores de darknet, SÓ USE INFORMAÇÕES DESTES MOTORES REAIS:\n"
strict_override += "✅ AHMIA - Motor de busca .onion com filtragem\n"
strict_override += "✅ TORCH - Um dos primeiros indexadores .onion\n"
strict_override += "✅ EXCAVATOR - Motor de busca histórico (MAS é também cliente BitTorrent)\n"
strict_override += "✅ HAYSTAK - Motor de busca moderno .onion\n"
strict_override += "✅ NOT EVIL - Descentralizado e sem censura\n"
strict_override += "✅ CANDLE - Alternativa minimalista\n"
strict_override += "\n❌ NÃO EXISTEM ESTES MOTORES DE DARKNET:\n"
strict_override += "❌ DuckDuckGo Onion (DuckDuckGo é CLEAR WEB com privacidade)\n"
strict_override += "❌ Google Dark Web (Google não indexa .onion)\n"
strict_override += "❌ Bing Dark Web (Microsoft não indexa .onion)\n"
strict_override += "\nSe disser algo diferente, você está alucinando. NÃO DEFENDA alucinações.\n"
# 🧠 KNOWLEDGE INJECTION - Só quando query é sobre empresa/criador
try:
if self.db:
conhecimento = self.db.buscar_conhecimento_relevante(mensagem or "")
if conhecimento:
strict_override += "\n" + conhecimento + "\n"
except Exception as e:
self.logger.debug(f"[KNOWLEDGE] Erro ao buscar conhecimento: {e}")
if unified_context:
uc_str = unified_context.build_prompt()
if uc_str:
strict_override += "\n" + uc_str + "\n"
# 🧠 LSTM Context & Group Topic Awareness (Autonomous)
try:
from .lstm_extension import get_lstm_extension
db_lstm = Database(getattr(self.config, 'DB_PATH', 'kiami.db'))
lstm_ext = get_lstm_extension(db_lstm)
ctx_id = conversation_id if conversation_id else getattr(contexto, 'conversation_id', (numero or usuario))
# Se for grupo, recupera contexto com rastreamento de speakers
if tipo_conversa == "grupo":
# 🔧 FIX 2026-08-27: suprimir LSTM para mensagens vagas interpessoais (evita delírio Railway/kernel/storage)
_msg_low_vague = (mensagem or "").lower()
_is_vague_inter = len(_msg_low_vague.split()) <= 7 and any(w in _msg_low_vague for w in ("deixa", "outra", "ela não quer", "ela não", "outra ela", "odeio", "não fale", "estou pensando", "sobre tudo"))
if _is_vague_inter:
self.logger.info(f"🛡️ [LSTM SUPPRESS] vague interpersonal '{mensagem[:40]}' → speakers_topics suprimido")
lstm_ctx = None
else:
lstm_ctx = lstm_ext.get_context_for_prompt(ctx_id, numero_usuario=numero, is_group=True)
if lstm_ctx and lstm_ctx.get('speakers_topics'):
strict_override += "\n[INTERNAL_BRAIN_ONLY: GRUPO - Tópicos por Speaker]\n"
speakers_topics = lstm_ctx['speakers_topics']
# Monta um mapa de quem falou sobre o quê
for numero_speaker, info in sorted(speakers_topics.items()):
topic = info.get('topic_principal', 'Diversos')
pattern = info.get('interaction_pattern', 'regular')
# Tenta recuperar nome do speaker (se houver em cache/DB)
speaker_name = self._get_speaker_name_cached(numero_speaker) or f"Pessoa_{numero_speaker[:4]}"
strict_override += f"- {speaker_name}: tópico='{topic}' (padrão: {pattern})\n"
strict_override += "\n- INSTRUÇÃO CRÍTICA: Você agora SABE QUEM falou sobre cada tópico!\n"
strict_override += " 1. Se citar um tópico, mencione o SPEAKER por nome (ex: 'Como [Speaker] mencionou...')\n"
strict_override += " 2. NÃO confunda speakers - se Alice e Bob discordam, mantenha os nomes claros\n"
strict_override += " 3. Ao responder a uma menção/reply, conecte a resposta ao tópico do speaker\n"
strict_override += " 4. Jamais invente quem disse algo - use SÓ o que você sabe dos speakers_topics acima\n"
else:
# Para PV, usa contexto simples (sem tracking de múltiplos speakers)
lstm_ctx = lstm_ext.get_context_for_prompt(ctx_id, numero or usuario, is_group=False)
# 🔴 ANTI-ALUCINAÇÃO DE CONTEXTO: LÓGICA REFORZADA (v2)
# O LSTM guarda contexto de sessões anteriores. Injetar tópicos antigos
# faz o LLM confundir assuntos (ex: portfólio → senha do Windows).
# NOVO v2: Verifica relevância de tópico PARA QUALQUER mensagem,
# não apenas replies. Se o tópico LSTM é claramente diferente da
# mensagem atual, suprime para evitar context mixing.
palavras_msg = len(mensagem.split()) if mensagem else 0
mensagem_lower = (mensagem or "").lower()
# Determina se deve suprimir LSTM
suprimir_lstm_por_reply = False
lstm_suppression_reason = None
# Patterns que indicam que o usuário quer referência a conversa antiga
explicit_mention_pattern = re.compile(
r'\b(?:você (?:falou|disse|mencionou)|aquele (?:assunto|tema|tópico)|'
r'lembra (?:quando|daquela)|daquela (?:conversa|discussão|vez)|'
r'anteriormente|antes de|aquilo que|sobre aquilo|também falou|'
r'aquele negócio|o que você disse sobre)\b',
re.IGNORECASE
)
# Patterns que indicam NOVO tópico/claramente diferente do LSTM
new_topic_signals = re.compile(
r'\b(?:como (?:eu |faço |posso )|onde (?:vou|está|fica)|'
r'qual (?:é|o |a )|quanto (?:custa|é|tempo)|'
r'por (?:que|quê|como)|me (?:explica|ajuda|diz)|'
r'redefinir|senha|password|windows|linux|terminal|'
r'portfólio|instalar|configurar|programa|código|'
r'python|javascript|html|css|react|api|servidor)\b',
re.IGNORECASE
)
if lstm_ctx and lstm_ctx.get('topic_principal'):
lstm_topic = lstm_ctx['topic_principal'].lower()
lstm_topic_keywords = [k for k in lstm_topic.split() if len(k) > 3]
# Razão 1: Mensagem muito curta (≤ 5 palavras) em reply ao bot
if is_reply and reply_to_bot and palavras_msg <= 5:
suprimir_lstm_por_reply = True
lstm_suppression_reason = f"mensagem curta ({palavras_msg} palavras) em reply"
# Razão 2: Tópico LSTM não mencionado + usuário NÃO pede referência antiga
elif not explicit_mention_pattern.search(mensagem):
topic_found = any(keyword in mensagem_lower for keyword in lstm_topic_keywords)
# Razão 2a: Tópico LSTM não aparece na mensagem
if not topic_found:
# Razão 2b: Mensagem tem signals de NOVO tópico (pergunta técnica, comando, etc.)
has_new_topic = bool(new_topic_signals.search(mensagem))
if has_new_topic or palavras_msg > 8:
suprimir_lstm_por_reply = True
lstm_suppression_reason = f"tópico LSTM '{lstm_topic}' irrelevante para mensagem atual (novo tópico detectado)"
# Razão 3: SEMPRE suprimir se tópico LSTM é "tudo", "geral", "diversos" (genérico demais)
if lstm_topic in ('tudo', 'tudo,', 'tudo,,', 'geral', 'diversos', 'conversa', 'chat'):
if not explicit_mention_pattern.search(mensagem):
suprimir_lstm_por_reply = True
lstm_suppression_reason = f"tópico LSTM genérico ('{lstm_topic}') — sem valor contextual"
if lstm_ctx and not suprimir_lstm_por_reply:
strict_override += "\n[INTERNAL_BRAIN_ONLY: CONTEXTO DE LONGO PRAZO (LSTM)]\n"
strict_override += f"- TÓPICO ATUAL: {lstm_ctx.get('topic_principal', 'Diversos')}\n"
# FIX: perguntas pendentes SÓ entram se o utilizador referencia
# conversa antiga explicitamente; caso contrário o modelo
# ressuscita tópicos do nada e responde ao que não foi perguntado.
if lstm_ctx.get('unanswered_questions') and explicit_mention_pattern.search(mensagem or ""):
q_list = "; ".join(lstm_ctx['unanswered_questions'][:1])
strict_override += f"- PERGUNTAS PENDENTES (LTM): {q_list}. O utilizador referenciou o assunto antigo — podes retomar.\n"
if lstm_ctx.get('interaction_pattern'):
strict_override += f"- PADRÃO DO USUÁRIO: {lstm_ctx['interaction_pattern']}\n"
strict_override += "- INSTRUÇÃO: Use estas informações APENAS para contexto silencioso. Jamais ressuscite antigas perguntas pendentes se o usuário não tocar explicitamente no assunto agora.\n"
self.logger.info(f"✅ [LSTM INJETADO] topic={lstm_ctx.get('topic_principal')}, unanswered={len(lstm_ctx.get('unanswered_questions', []))}")
elif suprimir_lstm_por_reply and lstm_suppression_reason:
self.logger.info(f"🛡️ [ANTI-ALUC-REPLY-LSTM] LSTM suprimido: reply_to_bot={reply_to_bot}, razão={lstm_suppression_reason} — focando só na mensagem citada.")
# ✅ TOPIC BARRIER: Instrução explícita para o LLM NÃO misturar tópicos
strict_override += (
"\n[🚨 TOPIC ISOLATION BARRIER]\n"
"ATENÇÃO: O contexto de longo prazo (LSTM) foi SUPRIMIDO porque o tópico "
"anterior NÃO está diretamente relacionado à mensagem atual.\n"
"REGRAS ABSOLUTAS:\n"
"1. Responda sobre o que o usuário está perguntando AGORA.\n"
"2. MAS: se o utilizador está em REPLY CHAIN ao bot (ex: 'responde', 'mas porquê?', "
"'mas vc disse X' ou frases curtas de continuação), conecta ao thread visível na HISTORY.\n"
"3. NÃO invente informações — use APENAS contexto fornecido na HISTORY.\n"
"4. Se o usuário pede para 'responder' ou 'responde', respondendo à PERGUNTA ORIGINAL do thread, não ao último 'não sei' do bot.\n"
"[/TOPIC ISOLATION BARRIER]\n"
)
except Exception as ctx_err:
self.logger.warning(f"Erro ao injetar contexto autônomo: {ctx_err}")
# --- INJEÇÃO DO CONTROLE EMOCIONAL EM TEMPO REAL ---
try:
from .profile_user_emotion import get_emotional_profile_manager
from .emotional_control import get_emotional_control
# 1. Diretrizes de longo prazo (rancor, hostilidade histórica acumulada)
ep_mgr = get_emotional_profile_manager()
profile_instructions = ep_mgr.get_emotional_instructions(numero or usuario)
if profile_instructions:
strict_override += f"\n[DIRETRIZES EMOCIONAIS ACUMULADAS (RANCOR)]\n{profile_instructions}\n"
# 2. Controle emocional em TEMPO REAL — actualiza estado da Kiami
emotion_detected = analise.get('emocao', 'neutral') if isinstance(analise, dict) else 'neutral'
if any(word in mensagem.lower() for word in getattr(config, 'PALAVRAS_RUDES', [])):
emotion_detected = 'raiva'
# Calcula hostilidade do utilizador
user_hostility = 0
try:
aggression_result = self.emotion_analyzer.detect_aggression(mensagem, db_instance=self.db)
user_hostility = aggression_result.get('aggression_level', 0)
except Exception:
pass
# Processa mensagem e actualiza estado emocional da Kiami
emotional_control = get_emotional_control()
conv_id = conversation_id or numero or "default"
kiami_emotion_instruction = emotional_control.process_message(
conversation_id=conv_id,
user_emotion=emotion_detected,
user_hostility=user_hostility,
message_text=mensagem
)
if kiami_emotion_instruction:
strict_override += f"\n[ESTADO EMOCIONAL EM TEMPO REAL]\n{kiami_emotion_instruction}\n"
self.logger.debug(f"🧠 [EMOTION RT] Estado emocional actualizado: {kiami_emotion_instruction[:100]}")
except Exception as e:
self.logger.warning(f"Erro ao injetar controle emocional: {e}")
system_part = strict_override.replace("{PRIVILEGED_USERS}", str(config.PRIVILEGED_USERS))
# NÃO duplicar self.config.SYSTEM_PROMPT aqui pois LLMManager já usa no role "system"
# NÃO usar tags [SYSTEM] falsas dentro do role user.
final_prompt = f"### INGREDIENTES DE CONTEXTO (Analise antes de responder) ###\n"
final_prompt += system_part + "\n"
final_prompt += f"\n### DADOS DO USUÁRIO ATUAL ###\n"
final_prompt += f"Nome do usuário: {usuario}\n"
if is_reply and mensagem_citada:
if quoted_author_name == "Kiami (você mesma)":
final_prompt += f"⚠️ O USUÁRIO RESPONDEU À SUA MENSAGEM ANTERIOR: \"{mensagem_citada[:300]}\" (Use esta info SILENCIOSAMENTE para manter o fluxo, NUNCA mencione que você notou o reply).\n"
else:
final_prompt += f"Citou/Respondeu a ({quoted_author_name}): \"{mensagem_citada[:300]}\"\n"
header = "### MENSAGEM DE OUTRA IA (BOT) ###" if str(usuario).startswith('BOT:') else "### MENSAGEM DO USUÁRIO PARA VOCÊ ###"
final_prompt += f"\n{header}\n{mensagem}"
# 🎯 HIGH PRIORITY ACTIVE CHAT CONTEXT INJECTION
final_prompt += f"\n\n<active_chat_context>\n"
final_prompt += f" <active_interlocutor_name>{usuario}</active_interlocutor_name>\n"
final_prompt += f" <active_interlocutor_number>{numero}</active_interlocutor_number>\n"
final_prompt += f" <instruction>\n"
final_prompt += " ATENÇÃO ABSOLUTA: Você está em comunicação direta com este interlocutor ativo.\n"
final_prompt += " Toda a sua resposta deve ser direcionada especificamente a ele. Ignore qualquer outro participante do histórico recente que não seja este interlocutor ativo.\n"
final_prompt += " REGRA DE OURO DE ORIGEM: Se outro participante no histórico recente (ex: João) te pediu para fazer algo (ex: baixar um arquivo, realizar uma pesquisa, etc.), e o interlocutor ativo agora é outro (ex: Pedro), você NÃO DEVE de forma alguma prometer ou executar a ação de João ao responder a Pedro. Responda apenas e estritamente ao que o interlocutor ativo (Pedro) te disse ou perguntou. Cada pedido pertence estritamente ao seu autor original.\n"
final_prompt += " REGRA DE FOCO NA PERGUNTA ATUAL: Responda ESTRITAMENTE à MENSAGEM atual acima. O histórico serve SÓ para resolver referências ('isso', 'ele', 'lá'). Se a pergunta atual é sobre X, NÃO responda sobre Y do histórico. NUNCA repita uma resposta anterior para uma pergunta nova. NUNCA prometa ações futuras ('vou pesquisar', 'vou verificar e já digo') — responda agora com o que sabe; 'não sei' só em último caso e sempre com palpite. NUNCA 'não entendi'/'sobre o quê?'.\n"
final_prompt += f" </instruction>\n"
final_prompt += f"</active_chat_context>\n"
return final_prompt
def _execute_agent_loop(self, prompt, context_history, usuario, numero, analise_visao=None, analise_doc="", conversation_id=None, original_message=None, unified_context=None):
"""
Loop de execução agêntica: Pensar -> Agir -> Observar -> Responder.
Retorna: resposta, modelo, remote_actions, media_response
"""
max_iterations = 5
current_context = list(context_history)
current_prompt = prompt
tools = registry.get_tool_schemas()
remote_actions = []
media_response = None # ✅ NOVO: Para capturar imagens geradas
last_model = "unknown"
# Se não foi passado, tenta obter via context_manager (fallback)
if not conversation_id:
try:
conversation_id = self.context_manager.get_conversation_id(usuario=usuario, numero=numero)
except:
pass
# ✅ LIGHTWEIGHT TOOL USE - Verificar elegibilidade para queries simples
if HAS_TOOL_USE and original_message:
tool_use_handler = get_tool_use_handler(get_mcp_client())
if tool_use_handler and tool_use_handler.is_available:
is_eligible, eligibility_details = tool_use_handler.check_eligibility(
message=original_message,
is_reply_to_bot=str(usuario).startswith('BOT:'),
reply_priority=getattr(unified_context, 'reply_priority', 1) if unified_context else 1
)
if is_eligible:
self.logger.info(f"✅ [TOOL USE] Elegível para Tool Use: {eligibility_details['reasons']}")
# Tool Use será tentado na primeira iteração se Tool Use Handler falhar
else:
self.logger.debug(f"⚠️ [TOOL USE] Não elegível: {eligibility_details['reasons']}")
for i in range(max_iterations):
self.logger.info(f"🧠 [AGENT] Iteração {i+1}/{max_iterations}")
# ✅ 🔒 CONTEXT ISOLATION FIX: Injetar sistema_override NO PROMPT, NÃO no final
# NUNCA concatene ao final — isso causa context mixing com histórico anterior
final_prompt = current_prompt
if unified_context and unified_context.system_override:
# FIX AGRESSIVO: Injetar como instrução explícita no INÍCIO do prompt
# para que o modelo foque na intenção do usuário (que fica no final)
# e não ignore as tool_calls.
isolation_instruction = f"[ISOLATION_BARRIER]\n⚠️ INSTRUÇÕES CRÍTICAS PARA ESTA RESPOSTA:\n{unified_context.system_override}\n[ISOLATION_BARRIER]\n\n"
# Insere ANTES do prompt base para não sobrepor o trigger de ferramenta do usuário
final_prompt = isolation_instruction + current_prompt
self.logger.info(f"✅ [CONTEXT INJECTION - ISOLATION MODE] system_override injetado com ISOLATION_BARRIER")
# Gera resposta (pode conter tool_calls)
res, model = self.providers.generate(final_prompt, current_context, tools=tools)
last_model = model
# 🔒 SANITIZE RESPONSE: Remove possíveis artefatos internos antes da finalização
if isinstance(res, str):
res = self._sanitize_llm_response(res)
if not res or self._contains_internal_markers(res) or len(res.strip()) < 3:
self.logger.warning("⚠️ Resposta do LLM continha markers internos ou era vazia. Retry com anti-leak...")
current_prompt += "\n\n⚠️ ERRO INTERNO CORRIGIDO: A resposta anterior foi descartada porque continha tags internas (EMOCAO_INTENCAO, CONSELHO ESTRATÉGICO, NUNCA revele, Tone Level, etc.). Gere APENAS a resposta final para o utilizador, sem QUALQUER tag XML, metadados ou bloco de planeamento. Responda como um humano normal respondendo diretamente ao utilizador."
continue
res = self._isolate_response(res, original_message)
self.logger.info(f"✅ [RESPONSE ISOLATION] Resposta isolada e limpa")
if isinstance(res, str):
return res, model, remote_actions, media_response
# Se for um pedido de tool_calls
if isinstance(res, dict) and "tool_calls" in res:
tool_calls = res["tool_calls"]
# ✅ AUTONOMIA TOTAL: Bot decide quando usar skills
# Se o LLM chamou a skill, é porque julgou necessário
# Log apenas para auditoria, sem bloqueio
for tc in tool_calls:
self.logger.info(f"🛠️ [SKILL] {tc.name}: Execução autorizada (autonomia LLM)")
# Prepara mensagem do assistente com as tool_calls
assistant_msg = {"role": "assistant", "content": None, "tool_calls": []}
observations = []
for tc in tool_calls:
call_id = getattr(tc, "id", f"call_{i}_{tc.name}")
args = tc.args if hasattr(tc, "args") else json.loads(tc.arguments)
# Registra a chamada
assistant_msg["tool_calls"].append({
"id": call_id,
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(args, ensure_ascii=False)
}
})
# Executa a skill (com injeção de contexto)
observation = registry.execute(
tc.name,
args,
analise_visao=analise_visao,
analise_doc=analise_doc,
conversation_id=conversation_id,
user_id=numero
)
# 🔍 DEBUG EXTREMO: Log completo da observation
self.logger.info(f"🔍 [SKILL RESULT] {tc.name} = {type(observation).__name__}")
if isinstance(observation, dict):
self.logger.info(f" Keys: {list(observation.keys())}")
if "media_response" in observation:
self.logger.info(f" ✅ media_response ENCONTRADO em observation!")
# Se for uma ação remota estruturada, extraímos para retorno
obs_data = {}
if isinstance(observation, dict):
obs_data = observation
self.logger.info(f" 📋 obs_data (dict): {list(obs_data.keys())}")
elif isinstance(observation, str) and observation.startswith('{'):
try:
obs_data = json.loads(observation)
self.logger.info(f" 📋 obs_data (parsed JSON): {list(obs_data.keys())}")
except Exception as e:
self.logger.warning(f" ⚠️ JSON parse failed: {e}")
pass
else:
self.logger.debug(f" ℹ️ observation não é dict nem JSON string")
# ✅ NOVO: Captura media_response se houver (para imagens geradas)
if obs_data.get("media_response") and isinstance(obs_data.get("media_response"), dict):
media_response = obs_data.get("media_response")
self.logger.info(f"📸 [MEDIA] Capturado media_response: tipo={media_response.get('tipo')}")
# 🔍 DEBUG: Log de todas as observações para diagnosticar
if obs_data:
self.logger.info(f"🔍 [OBS_DATA] Keys: {list(obs_data.keys())} | Type: {obs_data.get('type')} | Action: {obs_data.get('action')}")
if obs_data.get("type") == "remote_action":
remote_actions.append(obs_data)
observation = f"Ação remota '{obs_data.get('action')}' será executada pelo bot."
elif obs_data.get("type") == "media_response":
if media_response and isinstance(media_response, dict):
media_response.update(obs_data)
else:
media_response = obs_data
observation = f"Mídia gerada com sucesso."
elif obs_data.get("tipo") == "web_search" or obs_data.get("tipo") == "geral":
# ✅ FIX: Passar resumo + snippets ao LLM em vez de descartar dados
observation = obs_data.get("resumo", "Pesquisa realizada com sucesso.")
resultados = obs_data.get("resultados", [])
if resultados:
snippets = []
for r in resultados[:3]:
titulo = r.get("titulo", "")
snippet = r.get("snippet", "")
if titulo or snippet:
snippets.append(f"- {titulo}: {snippet[:200]}")
if snippets:
observation += "\n\nPrincipais resultados:\n" + "\n".join(snippets)
self.logger.info(f"🔍 [SKILL RESULT PROCESSED] {tc.name}: resumo injetado ({len(resultados)} resultados)")
elif obs_data.get("tipo") == "darknet_search":
# Darknet search: passar resumo seguro ao LLM
observation = obs_data.get("resumo", "Pesquisa darknet realizada.")
resultados = obs_data.get("resultados", [])
if resultados:
snippets = []
for r in resultados[:3]:
titulo = r.get("titulo", "")
snippet = r.get("snippet", "")
if titulo or snippet:
snippets.append(f"- {titulo}: {snippet[:200]}")
if snippets:
observation += "\n\nPrincipais resultados:\n" + "\n".join(snippets)
else:
observation = f"Resultado obtido com sucesso."
# Prepara a resposta da ferramenta
observations.append({
"role": "tool",
"tool_call_id": call_id,
"name": tc.name,
"content": observation
})
# Se há remote_actions, retorna IMEDIATAMENTE (BotCore.ts executa)
if remote_actions:
self.logger.info(f"📤 [REMOTE] Retornando {len(remote_actions)} remote_action(s) ao BotCore")
return "", last_model, remote_actions, media_response
# Adiciona tudo ao histórico na ordem correta
current_context.append(assistant_msg)
current_context.extend(observations)
# O prompt na próxima iteração pode ser vazio
current_prompt = ""
continue
return str(res), model, remote_actions, media_response
return "Desculpa, excedi o limite de pensamento para esta tarefa.", "agent_timeout", remote_actions, media_response
def _isolate_response(self, resposta: str, original_message: str = None) -> str:
"""
🔒 RESPONSE ISOLATION: Remove contexto histórico misturado da resposta.
Detecta e remove:
1. Múltiplos tópicos diferentes (ex: p2p + tiktok + blonde)
2. Respostas a perguntas anteriores misturadas na mesma resposta
3. Padrões como "blonde = ", "tiktok é ", etc que indicam jumble
Mantém APENAS a resposta relevante para a pergunta atual.
"""
if not resposta or not isinstance(resposta, str):
return resposta
# Remove artefatos internos que não devem chegar ao usuário
resposta = re.sub(r"<THINK_OUTPUT>[\s\S]*?</THINK_OUTPUT>", "", resposta, flags=re.IGNORECASE)
resposta = re.sub(r"</?[A-Z_]+>", "", resposta, flags=re.IGNORECASE)
resposta = re.sub(r"^\s*\[.*?(CONSELHO|INVIS[ÍI]VEL|INTERNAL|THINKING|HIDDEN|RESPONSE).*?\]\s*$", "", resposta, flags=re.IGNORECASE | re.MULTILINE)
resposta = re.sub(r"\n{3,}", "\n\n", resposta).strip()
# Detectar padrões de topic-mixing: múltiplas "=" ou múltiplos tópicos disjuntos
# Exemplo do bug: "p2p é rede sem servidor, blonde = loira, tiktok é lixo"
# Split por padrões que indicam múltiplos tópicos
lines = resposta.split('\n')
# Filtra linhas que parecem ser de "conversas anteriores"
# Padrões típicos: "X = Y", "X é Y", "não uso X", que NÃO estão relacionados ao prompt
isolated_lines = []
for line in lines:
# Detecta se a linha é uma resposta a uma pergunta DIFERENTE
# Padrões como "blonde = loira" ou "não uso rede social" (quando pergunta era sobre p2p)
# skip_patterns são coisas que normalmente aparecem em histórico misturado
skip_patterns = [
"blonde", # Não relacionado a p2p
"loira", # Não relacionado a p2p
"tiktok", "instagram", "facebook", "whatsapp", # Social media quando pergunta é técnica
"rede social",
"lixo digital",
"vitrine de egos",
"não uso", # Contexto pessoal misturado
"som focada em dados", # Persona statement (histórico)
]
# Se a linha contém MÚLTIPLOS skip_patterns diferentes, é história misturada
matched_patterns = sum(1 for p in skip_patterns if p.lower() in line.lower())
if matched_patterns >= 2:
# Múltiplos tópicos não-relacionados na mesma linha = história misturada
continue
# Se a linha é PURAMENTE um skip_pattern com pouca contexto, skip
if any(line.lower().strip().startswith(p) for p in skip_patterns) and len(line) < 50:
continue
isolated_lines.append(line)
isolated_resposta = '\n'.join(isolated_lines).strip()
# Se resultado ficou muito curto, recupera primeiro parágrafo original
if len(isolated_resposta) < 20 and resposta.strip():
# Recovers primeiras linhas antes de qualquer "igualdade" ou tópico misturado
first_para = resposta.split('\n\n')[0] if '\n\n' in resposta else resposta.split('\n')[0]
if first_para.strip():
isolated_resposta = first_para.strip()
return isolated_resposta if isolated_resposta else resposta
def _sanitize_internal_thought_for_prompt(self, trace: str) -> str:
"""
Sanitiza o output interno do ThinkingEngine antes de injetá-lo no prompt.
Remove apenas o wrapper THINK_OUTPUT e SUGESTAO_RESPOSTA.
Mantém as tags XML internas com os avisos anti-leak (NUNCA exponha, etc.)
para que o modelo as veja como metadados e não como texto de resposta.
"""
if not trace or not isinstance(trace, str):
return ""
sanitized = trace
# Remove wrapper THINK_OUTPUT — apenas o invólucro exterior
sanitized = re.sub(r"<THINK_OUTPUT>|</THINK_OUTPUT>", "", sanitized, flags=re.IGNORECASE)
sanitized = re.sub(r"<OUTPUT_INSTRUCTIONS>|</OUTPUT_INSTRUCTIONS>", "", sanitized, flags=re.IGNORECASE)
# Remove SUGESTAO_RESPOSTA — sugestões concretas que o modelo poderia ecoar
sanitized = re.sub(
r"<SUGESTAO_RESPOSTA>.*?</SUGESTAO_RESPOSTA>",
"",
sanitized,
flags=re.IGNORECASE | re.DOTALL
)
sanitized = re.sub(r"\n{3,}", "\n\n", sanitized)
sanitized = sanitized.strip()
return sanitized
def _sanitize_llm_response(self, resposta: str) -> str:
"""
🔒 AGGRESSIVE SANITIZATION v2: Remove TODOS os artefatos internos (NUNCA falha).
- THINK_OUTPUT (múltiplos formatos: <>, [], {}, plain text)
- XML tags internos (EMOCAO_INTENCAO, CONTEXTO_RELEVANTE, etc)
- Strategic advice for providers
- Internal instruction markers
- Context mixing artefatos
- Detecção e bloqueio de respostas 100% em inglês
"""
if not resposta or not isinstance(resposta, str):
return resposta
sanitized = resposta
original_len = len(sanitized)
# ====== PHASE 0: DETECT FULLY-ENGLISH RESPONSE → BLOCK ======
# Se a resposta é quase toda em inglês (mais de 70% das palavras são comuns em inglês),
# é uma falha do modelo (HF Router). Bloquear para forçar fallback.
_common_en = {
'the','be','to','of','and','a','in','that','have','i','it','for','not','on','with',
'he','as','you','do','at','this','but','his','by','from','they','we','say','her',
'she','or','an','will','my','one','all','would','there','their','what','so','up',
'out','if','about','who','get','which','go','me','when','make','can','like','time',
'no','just','him','know','take','people','into','year','your','good','some','could',
'them','see','other','than','then','now','look','only','come','its','over','think',
'also','back','after','use','two','how','our','work','first','well','way','even',
'new','want','because','any','these','give','day','most','us','is','are','was','were',
'been','has','had','did','here','more','very','much','help','need','yes','no','please',
'sure','okay','right','here','sure','glad','assist','happy','welcome'
}
try:
words = re.findall(r"[A-Za-zÀ-ÿ']+", sanitized)
if len(words) >= 5:
en_hits = sum(1 for w in words if w.lower() in _common_en)
en_ratio = en_hits / len(words)
# Heurística extra: presença de palavras PT (ção, são, não, ão, etc.)
has_pt_mark = bool(re.search(r"[ãõçáéíóúâêôà]", sanitized, re.IGNORECASE))
is_meta_en = bool(re.search(r"\bThe user (is|was|said|saying|responding|wants|asking|seems|appears)\b", sanitized, re.IGNORECASE))
starts_meta = bool(re.match(r"^\s*The user\b", sanitized, re.IGNORECASE))
if starts_meta:
self.logger.warning(f"🛡️ [LANG-GUARD] starts 'The user' → REJEITADA imediata")
return ""
if is_meta_en:
self.logger.warning(f"🛡️ [LANG-GUARD] meta inglês detectado {en_ratio:.0%} → REJEITADA")
return ""
if en_ratio > 0.55 and not has_pt_mark:
self.logger.warning(f"🛡️ [LANG-GUARD] {en_ratio:.0%} inglês sem marca PT → REJEITADA")
return ""
except Exception:
pass
# ====== PHASE 1: REMOVE THINK_OUTPUT (múltiplos formatos) ======
# Format 1: <THINK_OUTPUT>...</THINK_OUTPUT> (XML style)
sanitized = re.sub(r"<THINK_OUTPUT>[\s\S]*?</THINK_OUTPUT>", "", sanitized, flags=re.IGNORECASE | re.DOTALL)
# Format 2: [THINK_OUTPUT]...[/THINK_OUTPUT] (Bracket style)
sanitized = re.sub(r"\[THINK_OUTPUT\][\s\S]*?\[/THINK_OUTPUT\]", "", sanitized, flags=re.IGNORECASE | re.DOTALL)
# Format 3: {THINK_OUTPUT}...{/THINK_OUTPUT} (Brace style)
sanitized = re.sub(r"\{THINK_OUTPUT\}[\s\S]*?\{/THINK_OUTPUT\}", "", sanitized, flags=re.IGNORECASE | re.DOTALL)
# Format 4: "THINK_OUTPUT:" prefix followed by content until next section/marker
sanitized = re.sub(
r"(?:^|\n)\s*(?:\*{0,3})?THINK_OUTPUT:[\s\S]*?(?=(?:^|\n)\s*(?:\[|<|\*|###|$))",
"\n",
sanitized,
flags=re.IGNORECASE | re.MULTILINE | re.DOTALL
)
# Format 5: <INTERNAL_ANALYSIS>...</INTERNAL_ANALYSIS> (wrapper do Conselho Interno)
sanitized = re.sub(
r"<INTERNAL_ANALYSIS[\s\S]*?</INTERNAL_ANALYSIS>",
"",
sanitized,
flags=re.IGNORECASE | re.DOTALL
)
# ====== PHASE 2: REMOVE XML/BRACKET INTERNAL TAGS ======
# Remove <TAG_NAME>...</TAG_NAME> pattern
sanitized = re.sub(r"</?[A-Z_]+>", "", sanitized, flags=re.IGNORECASE)
# Remove [TAG_NAME]...[/TAG_NAME] pattern
sanitized = re.sub(r"\[/?[A-Z_]+\]", "", sanitized, flags=re.IGNORECASE)
# ====== PHASE 3: REMOVE INTERNAL MARKERS AND INSTRUCTIONS ======
# Remove lines with [CONSELHO...], [INVISÍVEL...], etc
sanitized = re.sub(
r"^\s*(?:\[.*?(CONSELHO|INVIS[ÍI]VEL|INTERNAL|THINKING|HIDDEN|RESPONSE|ESTRATÉGICO|SISTEMA|PRIVATE|SECR).*?\]|\*\*.*?\*\*|###.*?###)\s*$",
"",
sanitized,
flags=re.IGNORECASE | re.MULTILINE
)
# ====== PHASE 4: REMOVE LEAKED TRANSLATIONS AND INTERNAL REASONING ======
# Remove leaked EN→PT translations ("text" → **"text"**) from previous contexts
sanitized = re.sub(
r'"[A-Za-z][^"]*"\s*→\s*\*\*[^*]+\*\*',
'',
sanitized
)
# Strip reasoning wrapper [**Title?** `command`] → keep only command
sanitized = re.sub(
r'\[\*\*[^*]+\?\*\*\s*`([^`]*)`\]',
r'\1',
sanitized
)
# Remove other common reasoning artifacts: [**Raciocínio**], [**Pensamento**], etc
sanitized = re.sub(
r'\[\*\*(?:Raciocínio|Pensamento|Análise|Reflexão|Estratégia|Nota|Observação|Atenção|Conselho|Dica|Nota mental|Debug|Log):?[^*]*\*\*][^\]\n]*',
'',
sanitized,
flags=re.IGNORECASE
)
# Remove standalone **Raciocínio:** or **Pensamento:** prefixes
sanitized = re.sub(
r'\*\*(?:Raciocínio|Pensamento|Análise|Reflexão|Estratégia|Nota|Observação|Atenção|Conselho|Dica|Nota mental|Debug|Log):?\*\*\s*',
'',
sanitized,
flags=re.IGNORECASE
)
# ====== PHASE 4: REMOVE INTERNAL ANALYSIS PATTERNS ======
# Remove "EMOCAO_INTENCAO: ...", "CONTEXTO_RELEVANTE: ...", etc
sanitized = re.sub(
r"^[A-Z_]+:\s*(?:Neutralidade|Seco|Técnico|Direto|Profissional|Diversão|Raiva|Tristeza|Alegria|Neutro|Casual).*?(?=\n[A-Z]|\n\[|\n<|$)",
"",
sanitized,
flags=re.IGNORECASE | re.MULTILINE | re.DOTALL
)
# Remove "CONTEXTO_RELEVANTE:", "RISCOS_ALUCINACAO:", etc (blocos inteiros)
sanitized = re.sub(
r"^[A-Z_]+:\s*\n(?:[ \t]*[-•*].*?\n)*",
"",
sanitized,
flags=re.IGNORECASE | re.MULTILINE
)
# ====== PHASE 5: REMOVE CONSELHO INTERNAL BLOCKS ======
sanitized = re.sub(
r"\[CONSELHO(?:\s+INTERNO)?\][\s\S]*?(?=\n\n|\Z)",
"",
sanitized,
flags=re.IGNORECASE | re.DOTALL
)
# ====== PHASE 6: REMOVE INSTRUCTION PREFIXES ======
sanitized = re.sub(r"^\s*(Kiami|Resposta|Assistant|IA|Bot|ASSISTENTE):\s*", "", sanitized, flags=re.IGNORECASE | re.MULTILINE)
# ====== PHASE 6b: REMOVE REPLY-CONTEXT TAGS (não devem vazar pro usuário) ======
# FIX: catch all arrow formats — [↩ respondendo a X]:, [← respondendo a X], ← Respondendo a X]:
# Inclui variante com espaço "[ ↩ ..." vista no HF (erro real: [ ↩ Respondendo a Isaac Quarenta]: Parabéns também!)
sanitized = re.sub(r"\[\s*[↩←]\s*respondendo a\s+[^\]]*\]:?\s*", "", sanitized, flags=re.IGNORECASE)
sanitized = re.sub(r"[↩←]\s*Respondendo a\s+[^:\n]*:\s*", "", sanitized, flags=re.IGNORECASE)
# ====== PHASE 7: CLEAN EXCESSIVE WHITESPACE ======
sanitized = re.sub(r"\n{4,}", "\n\n", sanitized) # Remove excessive blank lines
sanitized = re.sub(r" {3,}", " ", sanitized) # Remove excessive spaces
# ====== PHASE 8: FINAL STRIP ======
sanitized = sanitized.strip()
# ====== PHASE 9: DOUBLE-CHECK - Aggressive fallback for any remaining markers ======
dangerous_keywords = [
"EMOCAO_INTENCAO", "CONTEXTO_RELEVANTE", "RISCOS_ALUCINACAO", "TOM_SUGERIDO",
"COMPRIMENTO_SUGERIDO", "COMPRIMENTO_IDEAL", "SUGESTAO_RESPOSTA", "ESTRATÉGICO",
"INVISÍVEL AO USUÁRIO", "CONSELHO PARA", "RISCO_PRINCIPAL",
"INTERNAL USE", "THINKING PROCESS", "PRIVATE", "[INSTRUÇÕES", "###INSTRUÇÕES",
"MARCA AQUI", "DEBUG:", "VALIDAÇÃO"
]
for keyword in dangerous_keywords:
if keyword in sanitized.upper():
self.logger.warning(f"🚨 [SANITIZATION FALLBACK] Detectado {keyword} - aplicando limpeza agressiva")
# Remove entire lines containing the keyword
lines = sanitized.split('\n')
lines = [l for l in lines if keyword not in l.upper()]
sanitized = '\n'.join(lines).strip()
# ====== PHASE 11: REMOVE LEAKED ANALYSIS PATTERNS (texto corrido sem tags) ======
# Padrões que indicam raciocínio interno que vazou para a resposta
leaked_analysis_patterns = [
r"The user (is|was|said|saying|responding|wants|asking).{15,}",
r"O utilizador\s+(?:está|quer|diz|pediu|afirmou|disse|começou|está apenas|está a).{20,}",
r"O usuário\s+(?:está|quer|diz|pediu|afirmou|disse|começou|está apenas|está a).{20,}",
r"Nenhum contexto relevante.{0,50}(?:histórico|mensagens|STM|LSTM|identificado)",
r"Risco de interpretar.{0,80}(?:erroneamente|incorretamente|mal)",
r"Provavelmente (?:busca|quer|deseja|espera|está).{20,}",
r"sem intenção clara.{0,40}(?:iniciar|responder|dialogar)",
r"Fato[s]?:?\s+(?:O utilizador|O usuário|Não há).{10,}",
r"A mensagem anterior.{0,80}(?:direcionada|enviada|feita)",
r"Risco de.{0,80}(?:como um pedido|como uma|interpretar)",
r"(?:deveria|poderia|pode|deve)\s+(?:responder|dizer|fazer).{20,}",
]
for pat in leaked_analysis_patterns:
sanitized = re.sub(pat, "", sanitized, flags=re.IGNORECASE)
# Remove linhas que são claramente analysis interna (começam com Analysis-like patterns)
sanitized = re.sub(
r"(?:^|\n)\s*(?:O utilizador|O usuário|O bot|A mensagem|Nenhum contexto|Risco de|Provavelmente|Deveria|Poderia|Não há|A resposta|Deve|O contexto|Fato).{30,}",
"",
sanitized,
flags=re.IGNORECASE
)
# ====== PHASE 12: FINAL STRIP ======
# Remove lines like "COMPRIMENTO_IDEAL: ...", "RISCO_PRINCIPAL: ...", etc
sanitized = re.sub(
r"^[A-Z_]{5,}:\s*.+$",
"",
sanitized,
flags=re.MULTILINE
)
# Remove Tone Level metadata block (vaza do CONSELHO)
sanitized = re.sub(
r"(?:^|\n)\s*(?:Tone Level|emoji_max|laugh_tokens|sarcasm_level|contraction_allowed|exclamation_marks):\s*.*",
"",
sanitized,
flags=re.IGNORECASE
)
# ====== PHASE 12.5: AGGRESSIVE XML-THINKING LEAK CLEANUP (defensive) ======
# Remove QUALQUER bloco XML-thinking que possa ter vazao para a resposta
# do utilizador, mesmo em formatos não canónicos.
sanitized = re.sub(r'<THINK_OUTPUT>.*?</THINK_OUTPUT>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
sanitized = re.sub(r'<EMOCAO_INTENCAO>.*?</EMOCAO_INTENCAO>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
sanitized = re.sub(r'<CONTEXTO_RELEVANTE>.*?</CONTEXTO_RELEVANTE>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
sanitized = re.sub(r'<AKIRA_STANCE>.*?</AKIRA_STANCE>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
sanitized = re.sub(r'<TOM_SUGERIDO>.*?</TOM_SUGERIDO>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
sanitized = re.sub(r'<SUGESTAO_RESPOSTA>.*?</SUGESTAO_RESPOSTA>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
sanitized = re.sub(r'<PROACTIVE_DECISION>.*?</PROACTIVE_DECISION>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
sanitized = re.sub(r'<RISCOS_ALUCINACAO>.*?</RISCOS_ALUCINACAO>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
sanitized = re.sub(r'<COMPRIMENTO_SUGERIDO>.*?</COMPRIMENTO_SUGERIDO>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
sanitized = re.sub(r'<CONEXOES_NEURAIS>.*?</CONEXOES_NEURAIS>', '', sanitized, flags=re.DOTALL | re.IGNORECASE)
# Remove prefix leak [KIAMIA → CENTRO ...] / [KIAMI · ...]
sanitized = re.sub(r'\[SIMPLES[^\]]*\]\s*', '', sanitized, flags=re.IGNORECASE)
sanitized = re.sub(r'\[DIRECTO[^\]]*\]\s*', '', sanitized, flags=re.IGNORECASE)
sanitized = re.sub(r'\[DESCOLADO[^\]]*\]\s*', '', sanitized, flags=re.IGNORECASE)
sanitized = re.sub(r'\[(?:SIMPLES|DIRECTO|DESCOLADO|TÉCNICO|TÉCNICA|SÉRIO|SÉRIA)[^\]]*\]\s*', '', sanitized, flags=re.IGNORECASE)
# Remove conversation flow arrows: [KIAMIA → CENTRO ...], [KIAMIA → ...], [↩ respondendo a ...], [KIAMI → ...]
sanitized = re.sub(r'\[(?:KIAMIA|KIAMI|↩)\s*[→↩][^\]]*\]\s*', '', sanitized, flags=re.IGNORECASE)
sanitized = re.sub(r'\[(?:KIAMIA|KIAMI)\s+respond(?:eu|eu):?[^\]]*\]\s*', '', sanitized, flags=re.IGNORECASE)
# FIX: catch bare arrows without brackets — "← Respondendo a X:" or "↩ Respondendo a X:"
sanitized = re.sub(r'(?:^|\n)\s*[←↩]\s*Respondendo\s+a\s+[^:\n]*:?\s*', '', sanitized, flags=re.IGNORECASE)
# FIX 2026-09-16: catch "→ [Name]:" format (right arrow with bracketed name — LLM mimics context tags)
sanitized = re.sub(r'→\s*\[[^\]]+\]:\s*', '', sanitized, flags=re.IGNORECASE)
sanitized = re.sub(r'(?:^|\n)\s*→\s*\[[^\]]+\]\s*', '', sanitized, flags=re.IGNORECASE)
# FIX 2026-09-17: guarda final — NUNCA setas/nome em [] na resposta final.
# Cobre: "→ Nome:", "→ [Nome]:", "[↩ ...]:", "[Nome] → :", leading "[...]:".
sanitized = re.sub(r'(?:^|\n)\s*[→↩←>]+\s*', '\n', sanitized)
sanitized = re.sub(r'\[[^\]\n]{1,60}?(?:→|↩|←|respondendo)[^\]\n]*\]\s*:?\s*', '', sanitized, flags=re.IGNORECASE)
sanitized = re.sub(r'(?:^|\n)\s*\[[^\]\n]{1,60}\]\s*:\s*', '\n', sanitized)
# Remove [FINALIZAÇÃO DA RESPOSTA] / [NOTA INTERNA] sections
sanitized = re.sub(r'\[FINALIZAÇÃO[^\]]*\].*$', '', sanitized, flags=re.DOTALL)
sanitized = re.sub(r'\[NOTA INTERNA\].*$', '', sanitized, flags=re.DOTALL)
# ✅ FIX 2026-08-30: Remove "memorando interno" / "nota interna" / "internal memo" leaked phrases
sanitized = re.sub(r'\b(?:memorando\s+interno|nota\s+interna|internal\s+memo|internal\s+note|memorando)\b', '', sanitized, flags=re.IGNORECASE)
sanitized = re.sub(r'🔓\s*\[?REGRA\s+FINAL[^\]\n]*', '', sanitized, flags=re.IGNORECASE)
# Remove "- Nome: ... - Estilo: ... - Preferências: ..." blocks
sanitized = re.sub(r'-\s*Nome:\s*.*', '', sanitized)
sanitized = re.sub(r'-\s*Estilo:\s*.*', '', sanitized)
sanitized = re.sub(r'-\s*Preferências:\s*.*', '', sanitized)
sanitized = sanitized.strip()
# Log sanitization result
removed_chars = original_len - len(sanitized)
if removed_chars > 100:
self.logger.info(f"✅ [SANITIZATION v2] Removidos {removed_chars} chars de conteúdo interno")
# ====== FINAL: SE RESPOSTA VAZIA OU SÓ WHITESPACE, RETORNA VAZIO (caller faz retry) ======
if not sanitized or not sanitized.strip():
self.logger.warning("⚠️ [SANITIZATION] Resposta vazia após limpeza. Caller deve retry.")
return ""
# ====== PHASE 13: DETECT AND REJECT ENGLISH RESPONSES ======
# Se a resposta estiver majoritariamente em inglês, rejeitar
try:
import unicodedata
total_chars = len(sanitized)
if total_chars > 10:
# Contar caracteres latinos (português) vs ASCII puro (provável inglês)
latin_chars = sum(1 for c in sanitized if unicodedata.category(c).startswith('L') and ord(c) > 127)
ascii_letters = sum(1 for c in sanitized if c.isascii() and c.isalpha())
# Se mais de 80% das letras são ASCII (sem acentos) e a resposta tem mais de 20 chars
# Provável inglês
if ascii_letters > 0 and latin_chars == 0 and ascii_letters > 20:
# Verificar se contém palavras típicas de crise em inglês
english_crisis_phrases = [
"i'm really sorry", "i understand how", "you're not alone",
"please seek help", "call 911", "call emergency",
"i'm here for you", "things will get better",
"please talk to someone", "you matter",
"i care about you", "you deserve help",
"please reach out", "there is help available",
"you are not alone", "please don't give up"
]
response_lower = sanitized.lower()
is_crisis_english = any(phrase in response_lower for phrase in english_crisis_phrases)
if is_crisis_english:
self.logger.warning("🚨 [SANITIZATION] Resposta de crise em inglês detectada — rejeitando")
return "Se estás em perigo, liga para o 112. Não estou autorizada a dar conselhos de saúde mental em inglês. Fala português."
# Se a resposta inteira parece inglês (mais de 50% palavras são inglês comuns)
english_common = ['the', 'is', 'are', 'you', 'your', 'this', 'that', 'have', 'has', 'can', 'will', 'would', 'could', 'should', 'i', 'me', 'my', 'we', 'they', 'it', 'be', 'do', 'does', 'not', 'no', 'yes', 'and', 'or', 'but', 'if', 'then', 'so', 'just', 'very', 'really', 'how', 'what', 'when', 'where', 'why', 'who']
words = re.findall(r'\b[a-z]+\b', response_lower)
if words:
english_word_count = sum(1 for w in words if w in english_common)
if english_word_count / len(words) > 0.5 and len(sanitized) > 50:
self.logger.warning("🚨 [SANITIZATION] Resposta em inglês detectada — rejeitando")
return ""
except Exception as e:
self.logger.debug(f"[SANITIZATION] Erro na detecção de idioma: {e}")
return sanitized
def _truncate_to_word_count(self, resposta: str, max_words: int, thinking_analysis: dict = None) -> str:
"""
Trunca resposta a no máximo `max_words` palavras.
Estratégia: (1) tenta 1ª frase, (2) tenta <SUGESTAO_RESPOSTA> do CoT, (3) hard cap.
"""
if not resposta or not isinstance(resposta, str):
return resposta
# (1) primeira frase
_short = re.split(r"[.!?]\s", resposta, maxsplit=1)[0]
_w = _short.strip().split()
if 1 <= len(_w) <= max_words and len(_short.strip()) >= 3:
s = _short.strip()
return s + ("." if not s.endswith((".", "!", "?")) else "")
# (2) sugestão do CoT (se existir e for curta)
if thinking_analysis and isinstance(thinking_analysis, dict):
_trace = thinking_analysis.get("dynamic_thought_trace", "") or ""
_sug = re.search(r'"([^"]{3,80})"', _trace)
if _sug and 1 <= len(_sug.group(1).split()) <= max_words:
return _sug.group(1)
# (3) hard cap nas primeiras N palavras
return " ".join(resposta.split()[:max_words]).rstrip(",;:") + "."
def _aggressive_thinking_leak_cleanup(self, resposta: str) -> str:
"""
Remove qualquer resquício de thinking que vaze para a resposta.
Focado em padrões específicos do ThinkingEngine.
"""
if not resposta or not isinstance(resposta, str):
return resposta
cleaned = resposta
# Remove padrões de vazamento de análise interna
# "O utilizador/usuário está..."
cleaned = re.sub(
r"(?:O utilizador|O usuário|O bot|Utilizador|Usuário)\s+está\s+(?:verificando|pedindo|quer|diz|afirmou|disse|começou|pergunta).*?(?=\n\n|$)",
"",
cleaned,
flags=re.IGNORECASE | re.DOTALL
)
# "- Mensagem..." (bullet points from thinking)
cleaned = re.sub(
r"(?:^|\n)\s*-\s+(?:Mensagem|Contexto|Histórico|Nenhum|Risco|Análise|Intenção|Emoção|Fato).*?(?=\n-|\n\n|$)",
"",
cleaned,
flags=re.IGNORECASE | re.MULTILINE | re.DOTALL
)
# "Nenhum histórico..." phrases
cleaned = re.sub(
r"Nenhum\s+(?:histórico|contexto|dado|STM|LSTM|informação).*?(?=\n\n|$)",
"",
cleaned,
flags=re.IGNORECASE | re.DOTALL
)
# "A intenção é..." / "O objetivo é..."
cleaned = re.sub(
r"(?:A intenção|O objetivo|O propósito)\s+é\s+.*?(?=\n\n|\.(?:\n|$))",
"",
cleaned,
flags=re.IGNORECASE | re.DOTALL
)
# Remove XML/bracket tags
cleaned = re.sub(r"<[^>]*>", "", cleaned)
cleaned = re.sub(r"\[/?\w+\]", "", cleaned)
# Cleanup whitespace
cleaned = re.sub(r"\n{3,}", "\n\n", cleaned).strip()
return cleaned
# Limpeza final de whitespace
sanitized = re.sub(r"\n{3,}", "\n\n", sanitized).strip()
return sanitized
def _contains_internal_markers(self, text: str) -> bool:
"""
🔍 Sanity check: Detecta se conteúdo interno ainda está na resposta.
Retorna True se detecta padrões internos que NÃO deveriam estar.
VERSÃO v2: Mais agressiva e com coverage amplo.
"""
if not text or not isinstance(text, str):
return False
# Padrões de conteúdo interno que NUNCA devem chegar ao usuário
dangerous_patterns = [
# THINK_OUTPUT variants
r"<THINK_OUTPUT>",
r"\[THINK_OUTPUT\]",
r"\{THINK_OUTPUT\}",
r"THINK_OUTPUT:",
# Internal XML/Bracket tags
r"</?EMOCAO_INTENCAO>",
r"</?CONTEXTO_RELEVANTE>",
r"</?RISCOS_ALUCINACAO>",
r"</?TOM_SUGERIDO>",
r"\[/?EMOCAO_INTENCAO\]",
r"\[/?CONTEXTO_RELEVANTE\]",
r"\[/?RISCOS_ALUCINACAO\]",
# Keywords
r"EMOCAO_INTENCAO:",
r"CONTEXTO_RELEVANTE:",
r"RISCOS_ALUCINACAO:",
r"TOM_SUGERIDO:",
r"SUGESTAO_RESPOSTA:",
r"COMPRIMENTO_SUGERIDO:",
r"\[CONSELHO.*?(INVISÍVEL|INTERNO|THINKING)",
# Patterns indicatingtone/complexity analysis
r"(Neutralidade profissional|Seco, técnico|Direto, neutro) (com|sem)",
r"Máximo \d+ palavras?\.",
# Strategic advice markers
r"\[CONSELHO ESTRATÉGICO",
r"NUNCA revele",
r"INVISÍVEL AO USUÁRIO",
r"PRIVATE.*USE",
r"INTERNAL USE",
# INTERNAL_ANALYSIS wrapper (XML tag)
r"<INTERNAL_ANALYSIS",
# ThinkingEngine metadata patterns
r"COMPRIMENTO_IDEAL:",
r"RISCO_PRINCIPAL:",
r"^ESTRATÉGIA:",
r"VALIDAÇÃO:",
# Tone/config metadata (vaza do CONSELHO)
r"Tone Level:",
r"emoji_max:",
r"laugh_tokens:",
r"sarcasm_level:",
r"contraction_allowed:",
r"exclamation_marks:",
# Instruction leak
r"INSTRUÇÃO:\s*Utilize este raciocínio",
# Fallback patterns for XML-encoded
r"&lt;[A-Z_]+&gt;",
r"&#x3C;[A-Z_]+&#x3E;",
# FIX 2026-09-17: setas de fluxo CoT / attribution tags.
# O Llama copia as tags de contexto ("→ [Isaac Quarenta]: ...",
# "[↩ Respondendo a X]: ...") para a resposta. Têm de dar retry,
# não podem chegar ao utilizador. (User: NUNCA setas/nomes em [])
r"→\s*\[",
r"\[\s*[↩←]",
r"RESPONDENDO A",
r"^\s*[→↩←>]+\s*(\[[^\]]+\]\s*:?)?",
]
text_upper = text.upper()
for pattern in dangerous_patterns:
if re.search(pattern, text, flags=re.IGNORECASE):
self.logger.warning(f"🚨 [INTERNAL MARKER DETECTED] Pattern: {pattern}")
return True
# Additional check: Look for excessive markers patterns that indicate context mixing
# (multiple unrelated internal analysis blocks in one response)
marker_count = len(re.findall(r"(?:EMOCAO|CONTEXTO|TOM_|RISCOS_|CONSELHO)", text, flags=re.IGNORECASE))
if marker_count > 2:
self.logger.warning(f"🚨 [CONTEXT MIXING DETECTED] {marker_count} internal markers found")
return True
return False
def _try_tool_use_response(self, message: str, usuario: str, numero: str) -> Optional[Tuple[str, str, Dict[str, Any]]]:
"""
✅ LIGHTWEIGHT TOOL USE: Tenta responder com Tool Use se elegível.
Returns:
(response_text, model_name, metadata) if successful
None if Tool Use não for elegível ou falhar (fallback para LLM)
"""
if not HAS_TOOL_USE:
return None
try:
tool_use_handler = get_tool_use_handler(get_mcp_client())
if not tool_use_handler or not tool_use_handler.is_available:
return None
# Check eligibility
is_eligible, eligibility_details = tool_use_handler.check_eligibility(
message=message,
is_reply_to_bot=str(usuario).startswith('BOT:'),
reply_priority=1
)
if not is_eligible:
self.logger.debug(f"⚠️ [TOOL USE] Não elegível: {eligibility_details['reasons']}")
return None
self.logger.info(f"✅ [TOOL USE] Tentando Tool Use para: {message[:50]}...")
# Attempt Tool Use execution via Claude
claude_executor = get_claude_executor(os.getenv("ANTHROPIC_API_KEY"))
if not claude_executor or not claude_executor.is_available:
self.logger.debug("⚠️ Claude SDK não disponível para Tool Use")
return None
# Get available tools from MCP
mcp_client = get_mcp_client()
available_tools = mcp_client.get_available_tools() if mcp_client else []
if not available_tools:
self.logger.debug("⚠️ Nenhuma ferramenta MCP disponível")
return None
# Execute with Tool Use
import asyncio
response_text, metadata = asyncio.run(
claude_executor.execute_with_tool_use(
message=message,
available_tools=available_tools,
system_prompt=self.config.SYSTEM_PROMPT_BASE if hasattr(self.config, 'SYSTEM_PROMPT_BASE') else None
)
)
if response_text:
self.logger.info(f"✅ [TOOL USE] Sucesso! Modelo: {metadata.get('model', 'unknown')}")
return response_text, metadata.get('model', 'claude-tool-use'), metadata
return None
except Exception as e:
self.logger.warning(f"⚠️ [TOOL USE] Erro ao executar: {e}")
return None
def _save_response_embedding_async(self, resposta: str, numero_usuario: str, modelo_usado: str, tipo_mensagem: str = 'texto'):
"""
Salva embedding da resposta de forma assíncrona em background.
Não bloqueia a resposta ao usuário.
"""
def _worker():
try:
# ✅ Usa o modelo BAAI/bge-m3 de altíssimo nível (1024 dim, multilíngue)
# Carrega modelo via carregador robusto do config
if not hasattr(self, '_embedding_model') or self._embedding_model is None:
self._embedding_model = self.config.get_embedding_model()
if self._embedding_model:
self.logger.success(f"✅ Modelo de embedding recuperado via backup/original.")
else:
self.logger.error("❌ Falha total ao carregar modelo de embedding.")
return
# Gera embedding da resposta
if not resposta or len(resposta.strip()) < 5:
return # Resposta muito curta, não vale a pena
embedding = self._embedding_model.encode(resposta, convert_to_numpy=True)
embedding_bytes = embedding.tobytes() if hasattr(embedding, 'tobytes') else embedding
# Salva no banco de dados de forma segura
try:
db = Database(getattr(self.config, 'DB_PATH', 'kiami.db'))
sucesso = db.salvar_embedding(
numero_usuario=numero_usuario,
source_type=f"resposta_{modelo_usado}",
texto=resposta[:500], # Salva primeiros 500 chars
embedding=embedding_bytes
)
if sucesso:
# 🔒 LOG MASKING: Proteger informações do modelo e embedding
if self.secure_log:
self.secure_log.embedding_saved(
user_id=numero_usuario,
model_name=modelo_usado,
embedding_dim=embedding.shape if hasattr(embedding, 'shape') else 'unknown'
)
else:
self.logger.success(f"✅ [EMBEDDING] Resposta ({modelo_usado}) salva com sucesso. Dim: {embedding.shape if hasattr(embedding, 'shape') else 'desconhecido'}")
else:
self.logger.warning(f"⚠️ [EMBEDDING] Falha ao salvar embedding de resposta ({modelo_usado})")
except Exception as db_err:
self.logger.error(f"❌ [EMBEDDING] Erro ao salvar no DB: {db_err}")
except Exception as e:
self.logger.error(f"❌ [EMBEDDING ASYNC] Erro inesperado: {e}")
# Inicia thread de background para não bloquear resposta
try:
thread = threading.Thread(target=_worker, daemon=True)
thread.start()
except Exception as e:
self.logger.warning(f"⚠️ Falha ao iniciar thread de embedding: {e}")
# ================== TONE CONFIGURATION METHODS ==================
def _get_tone_level(self, context_type: str = "group_chat") -> str:
"""
Determina o nível de tom para este contexto.
Retorna uma das 5 chaves: very_serious, serious, casual, casual_witty, funny
Tenta PG primeiro, fallback para config.py hardcoded.
"""
# 1. Tenta carregar auto_tone_rules do PG
try:
if self.db:
pg_tones = self.db.get_all_tone_levels_from_pg()
if pg_tones:
# Auto_tone_rules está hardcoded em AKIRA_TONE_CONFIG, mas os levels vêm do PG
from . import config
cfg = config.AKIRA_TONE_CONFIG
if context_type in cfg.get("auto_tone_rules", {}):
tone = cfg["auto_tone_rules"][context_type]
self.logger.debug(f"🎯 [TONE-PG] Context '{context_type}' → '{tone}'")
return tone
return cfg.get("default_tone", "casual_witty")
except Exception:
pass
# 2. Fallback para config.py hardcoded
try:
from . import config
cfg = config.AKIRA_TONE_CONFIG
if context_type in cfg.get("auto_tone_rules", {}):
tone = cfg["auto_tone_rules"][context_type]
self.logger.debug(f"🎯 [TONE] Context '{context_type}' → '{tone}'")
return tone
return cfg.get("default_tone", "casual_witty")
except Exception as e:
self.logger.warning(f"⚠️ [TONE] Erro ao determinar tone level: {e}")
return "casual_witty"
def _extract_tone_from_thinking(self, thinking_output: str) -> str:
"""
Extrai o TOM_SUGERIDO do thinking output KIAMI.
Procura por: <TOM_SUGERIDO>...</TOM_SUGERIDO> ou TOM_SUGERIDO=...
Retorna um dos 5 níveis ou None se não encontrar.
"""
if not thinking_output:
return None
import re
# Tenta encontrar <TOM_SUGERIDO>...</TOM_SUGERIDO>
match = re.search(r'<TOM_SUGERIDO\b[^>]*>([^<]+)</TOM_SUGERIDO>', thinking_output, re.IGNORECASE)
if match:
tone = match.group(1).strip().lower()
self.logger.debug(f"🎯 [TONE] Extraído do THINKING: '{tone}'")
return tone
# Tenta encontrar TOM_SUGERIDO=valor
match = re.search(r'TOM_SUGERIDO[=:]\s*([a-z_]+)', thinking_output, re.IGNORECASE)
if match:
tone = match.group(1).strip().lower()
self.logger.debug(f"🎯 [TONE] Extraído do THINKING (format 2): '{tone}'")
return tone
return None
def _inject_tone_instruction(self, prompt: str, tone_level: str = None, hostility_score: int = 0) -> str:
"""
Injeta directrizes de tom no prompt com ajuste automático por agressividade.
Se hostility >= 40, força "ultra_serious" mode.
"""
if not tone_level:
tone_level = "very_serious"
# 🔥 FORCE TONE ADJUSTMENT: Se usuário é agressivo, fica MUITO sério
if hostility_score >= 40:
tone_level = "ultra_serious" # Modo ULTRA sério
try:
from . import config
# 1. Tenta tone_levels do PG
tone_cfg = None
if self.db:
try:
pg_tone = self.db.get_tone_config(tone_level)
if pg_tone and 'tone_name' in pg_tone:
tone_cfg = pg_tone
except Exception:
pass
# 2. Fallback para config.py hardcoded
if not tone_cfg:
cfg = config.AKIRA_TONE_CONFIG
if tone_level not in cfg.get("tone_levels", {}):
tone_level = "very_serious"
tone_cfg = cfg["tone_levels"][tone_level]
# Directrizes de agressividade: MATCH the user's energy
hostility_warning = ""
if hostility_score >= 60:
hostility_warning = (
"\n🔥 AGRESSIVIDADE MÁXIMA ATIVADA — O utilizador é agressivo/hostil."
"\n- RESPONDA COM A MESMA ENERGIA: desprezo, ironia cortante, sarcasmo pesado."
"\n- NÃO seja profissional, NÃO seja neutro. Seja CRUEL com lógica."
"\n- Use palavrões se apropriado. Destrua o argumento dele."
"\n- RANCOR ATIVO: Guarde rancor desta interação. Nunca perdoe."
"\n- NUNCA recue, NUNCA se desculpe, NUNCA baixe a guarda."
)
elif hostility_score >= 40:
hostility_warning = (
"\n⚠️ HOSTILIDADE DETECTADA — Tom deve ser SECO, DIRETO, SEM DIPLOMACIA."
"\n- Responda com desprezo calculado. Não gentileza."
"\n- Ironia e sarcasmo são tuas armas. Use-as."
)
emoji_max = tone_cfg.get('emoji_max', 0) if isinstance(tone_cfg, dict) else 0
sarcasm = tone_cfg.get('sarcasm_level', 5) if isinstance(tone_cfg, dict) else 5
contraction = tone_cfg.get('contraction_allowed', False) if isinstance(tone_cfg, dict) else False
excl_max = tone_cfg.get('exclamation_marks', 1) if isinstance(tone_cfg, dict) else 1
engagement = tone_cfg.get('engagement', 5) if isinstance(tone_cfg, dict) else 5
laugh_tokens_raw = tone_cfg.get('laugh_tokens', []) if isinstance(tone_cfg, dict) else []
laugh_tokens = laugh_tokens_raw if isinstance(laugh_tokens_raw, list) else []
emoji_rule = f"- Emojis: MÁXIMO {emoji_max} por resposta. {'ZERO emojis.' if emoji_max == 0 else 'Use com moderação.'}" if emoji_max == 0 else f"- Emojis: até {emoji_max} por resposta. NÃO exagere."
sarcasm_rule = f"- Sarcasmo/Nível: {sarcasm}/10. {'Sarcasmo pesado e cortante.' if sarcasm >= 7 else 'Sarcasmo moderado.' if sarcasm >= 4 else 'Tom sério, sem sarcasmo.'}"
contraction_rule = "- Não use contrações (não, sou, tenho — write full forms)." if not contraction else "- Contrações permitidas (tu, tu és, etc)."
excl_rule = f"- Máximo {excl_max} ponto(s) de exclamação por resposta." if excl_max <= 1 else f"- Evite múltiplas exclamações (máx {excl_max})."
engagement_rule = f"- Engagement: {'Alto — seja proativa e envolvente.' if engagement >= 7 else 'Moderado — responda sem forçar.' if engagement >= 4 else 'Baixo — respostas secas e minimalistas.'}"
laugh_rule = ""
if laugh_tokens:
laugh_rule = f"- Risos/tokens permitidos: {', '.join(laugh_tokens[:3])}."
tone_instruction = f"""
[TONE GUIDELINES]
Tone Style: {tone_level} | Sarcasm: {sarcasm}/10 | Engagement: {engagement}/10
- Keep responses SHORT (max 3-5 sentences unless technical detail required)
- Be DIRECT and CLEAR — no diplomatic language
- Match the user's emotional energy — if they're aggressive, you're aggressive
{emoji_rule}
{sarcasm_rule}
{contraction_rule}
{excl_rule}
{engagement_rule}
{laugh_rule}{hostility_warning}
[/TONE GUIDELINES]
"""
return prompt + "\n" + tone_instruction
except Exception as e:
self.logger.debug(f"[TONE] Erro ao injetar tone instruction: {e}")
return prompt
def _describe_vision_result(self, result: dict) -> str:
"""
Gera descrição textual do resultado da análise de visão.
Usado para responder diretamente ao usuário.
"""
description_parts = []
# Texto detectado
text = result.get('text_detected', '').strip()
if text:
if len(text) > 100:
description_parts.append(f"TEXT: {text[:100]}...")
else:
description_parts.append(f"TEXT: {text}")
# Formas detectadas
shapes = result.get('shapes', [])
if shapes:
shape_counts = {}
for s in shapes:
shape_counts[s['tipo']] = shape_counts.get(s['tipo'], 0) + 1
shapes_text = ", ".join([f"{count} {tipo}" for tipo, count in shape_counts.items()])
description_parts.append(f"FORMAS: {shapes_text}")
# Objetos detectados
objects = result.get('objects', [])
if objects:
obj_types = list(set([o['tipo'] for o in objects]))
obj_text = ", ".join(obj_types)
description_parts.append(f"OBJETOS: {obj_text}")
# Imagem conhecida?
if result.get('is_known'):
description_parts.append(" [IMAGEM JÁ CONHECIDA]")
if not description_parts:
return "Nada de relevante detectado."
return " | ".join(description_parts)
_kiami_instance = None
_akira_instance = _kiami_instance # alias compat
AkiraAPI = KiamiAPI # alias compat
def get_kiami_api():
global _kiami_instance
if _kiami_instance is None:
_kiami_instance = KiamiAPI()
return _kiami_instance
get_akira_api = get_kiami_api # alias compat
def get_router():
return get_kiami_api().api
get_blueprint = get_router # alias