# type: ignore # treinamento.py # ================================================================ # TREINAMENTO AVANÇADO 3-NÍVEIS - AKIRA IA V21 ULTIMATE # ================================================================ # Arquitetura: Multi-nível (Emocional + NLP + API Adapter) # NLP Levels: Basic → Intermediate → Advanced (BART + Transformers) # Emoções: Análise avançada com BART + heurísticas # APIs: Mistral, Gemini, Groq, Cohere, Together, HuggingFace # ================================================================ import threading import time import json import hashlib from dataclasses import dataclass, field from typing import Optional, List, Dict, Any, Tuple, Callable from pathlib import Path from datetime import datetime import re import random # Imports opcionais com fallback (type: ignore para evitar erros de ambiente) try: import numpy as np # type: ignore NUMPY_AVAILABLE = True except Exception: NUMPY_AVAILABLE = False np = None # type: ignore try: from loguru import logger # type: ignore LOGURU_AVAILABLE = True except Exception: LOGURU_AVAILABLE = False # Criar logger dummy para evitar erros de tipo class DummyLogger: def info(self, *args, **kwargs): pass def success(self, *args, **kwargs): pass def warning(self, *args, **kwargs): pass def error(self, *args, **kwargs): pass def debug(self, *args, **kwargs): pass def exception(self, *args, **kwargs): pass logger = DummyLogger() # type: ignore try: from sentence_transformers import SentenceTransformer # type: ignore SENTENCE_TRANSFORMERS_AVAILABLE = True except Exception as e: SENTENCE_TRANSFORMERS_AVAILABLE = False SentenceTransformer = None # type: ignore try: import torch # type: ignore TORCH_AVAILABLE = True except Exception: TORCH_AVAILABLE = False torch = None # type: ignore try: from transformers import AutoTokenizer, AutoModelForSequenceClassification # type: ignore TRANSFORMERS_AVAILABLE = True except Exception: TRANSFORMERS_AVAILABLE = False AutoTokenizer = None # type: ignore AutoModelForSequenceClassification = None # type: ignore # Imports locais from . import config from .database import Database from .treinamento_modelo import get_model_trainer # ============================================================ # 🎯 CONFIGURAÇÕES DE TREINAMENTO # ============================================================ @dataclass class TrainingConfig: """Configuração do sistema de treinamento 3-níveis""" # Nível 1: Emoções enable_emotion_training: bool = True emotion_model: str = config.BART_EMOTION_MODEL emotion_confidence_threshold: float = 0.7 # Nível 2: NLP & Embeddings enable_nlp_training: bool = True embedding_model: str = config.EMBEDDING_MODEL embedding_dim: int = config.EMBEDDING_DIM # Nível 3: API Adapter enable_api_training: bool = True track_api_performance: bool = True # Gerais batch_size: int = 32 learning_rate: float = 0.001 max_samples_per_user: int = 100 training_interval_hours: int = 6 min_samples_for_training: int = 5 # Configuração ativa TRAINING_CONFIG = TrainingConfig() # ============================================================ # 🔧 EMBEDDINGS & MODELOS # ============================================================ class EmbeddingManager: """Gerenciador de embeddings com suporte a múltiplos modelos""" _instance = None _model_lock = threading.Lock() def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self._initialized = True self._model = None self._embedding_dim = None def load_model(self, model_name: Optional[str] = None) -> bool: """Carrega modelo de embeddings sob demanda""" if self._model is not None: return True with self._model_lock: if self._model is not None: return True if not SENTENCE_TRANSFORMERS_AVAILABLE: logger.warning("SentenceTransformers não disponível") return False model_to_load = model_name or TRAINING_CONFIG.embedding_model try: self._model = SentenceTransformer(model_to_load) self._embedding_dim = self._model.get_sentence_embedding_dimension() logger.success(f"✅ Embedding model carregado: {model_to_load} (dim={self._embedding_dim})") return True except Exception as e: logger.error(f"❌ Erro ao carregar embedding model: {e}") return False def generate_embedding(self, text: str) -> Optional[Any]: """Gera embedding para texto""" if not self.load_model(): return None try: emb = self._model.encode(text, convert_to_numpy=True) return emb except Exception as e: logger.warning(f"Erro ao gerar embedding: {e}") return None def generate_batch_embeddings(self, texts: List[str]) -> Optional[Any]: """Gera embeddings para batch de textos""" if not self.load_model(): return None try: embeddings = self._model.encode(texts, convert_to_numpy=True, batch_size=len(texts)) return embeddings except Exception as e: logger.warning(f"Erro ao gerar batch embeddings: {e}") return None def cosine_similarity(self, emb1: np.ndarray, emb2: np.ndarray) -> float: """Calcula similaridade de cossenos""" try: dot = np.dot(emb1, emb2) norm1 = np.linalg.norm(emb1) norm2 = np.linalg.norm(emb2) if norm1 == 0 or norm2 == 0: return 0.0 return float(dot / (norm1 * norm2)) except Exception: return 0.0 @property def embedding_dim(self) -> int: return self._embedding_dim or TRAINING_CONFIG.embedding_dim # Singleton embedding_manager = EmbeddingManager() # ============================================================ # 🎭 ANALISADOR DE EMOÇÕES (Via Singleton Central) # ============================================================ # Singleton importado para não duplicar o modelo BART em memória emotion_trainer = config.get_emotion_analyzer() # ============================================================ # 🧠 API ADAPTER TRAINER # ============================================================ class APIAdapterTrainer: """Treinador de adaptação para diferentes APIs (Mistral, Gemini, Groq, etc.)""" def __init__(self, db: Database): self.db = db self.api_stats: Dict[str, Dict[str, Any]] = {} self._init_api_tracking() def _init_api_tracking(self): """Inicializa tracking de APIs""" self.api_stats = { "mistral": {"success": 0, "failure": 0, "avg_response_time": 0, "total_tokens": 0}, "gemini": {"success": 0, "failure": 0, "avg_response_time": 0, "total_tokens": 0}, "groq": {"success": 0, "failure": 0, "avg_response_time": 0, "total_tokens": 0}, "cohere": {"success": 0, "failure": 0, "avg_response_time": 0, "total_tokens": 0}, "together": {"success": 0, "failure": 0, "avg_response_time": 0, "total_tokens": 0}, "huggingface": {"success": 0, "failure": 0, "avg_response_time": 0, "total_tokens": 0} } def record_api_call( self, provider: str, success: bool, response_time: float, tokens_used: int = 0, error: Optional[str] = None ): """Registra chamada de API para treinamento""" if provider not in self.api_stats: return stats = self.api_stats[provider] if success: stats["success"] += 1 # Média móvel do tempo de resposta n = stats["success"] stats["avg_response_time"] = ((n - 1) * stats["avg_response_time"] + response_time) / n stats["total_tokens"] += tokens_used else: stats["failure"] += 1 # Salva no banco self._save_api_stats(provider, stats) def _save_api_stats(self, provider: str, stats: Dict[str, Any]): """Salva estatísticas da API no banco""" try: self.db.salvar_aprendizado_detalhado( f"api_{provider}", "stats", json.dumps(stats) ) except Exception as e: logger.warning(f"Erro ao salvar stats da API {provider}: {e}") def get_best_provider(self) -> str: """Retorna o melhor provider baseado em成功率 e tempo""" best_score = -1 best_provider = "mistral" for provider, stats in self.api_stats.items(): if stats["success"] + stats["failure"] < 5: continue success_rate = stats["success"] / (stats["success"] + stats["failure"]) if (stats["success"] + stats["failure"]) > 0 else 0 avg_time = stats["avg_response_time"] # Score: sucesso alto + tempo baixo score = success_rate * 0.7 + (1 / (1 + avg_time)) * 0.3 if score > best_score: best_score = score best_provider = provider return best_provider def get_provider_stats(self, provider: str) -> Dict[str, Any]: """Retorna estatísticas de um provider""" return self.api_stats.get(provider, {}) # ============================================================ # 📊 HEURÍSTICAS E DICIONÁRIOS # ============================================================ # Palavras para análise heurística PALAVRAS_POSITIVAS = ['bom', 'ótimo', 'incrível', 'feliz', 'adorei', 'top', 'fixe', 'bué', 'show', 'legal', 'bacana', 'wah'] PALAVRAS_NEGATIVAS = ['ruim', 'péssimo', 'triste', 'ódio', 'raiva', 'chateado', 'merda', 'porra', 'odeio', 'caralho'] PALAVRAS_RUDES = ['caralho', 'puta', 'merda', 'fdp', 'vsf', 'krl', 'porra', 'desgraça'] # Gírias angolanas para treinamento GIRIAS_ANGOLANAS = { "puto": ("rapaz/rapariga", "casual"), "mano": ("amigo", "casual"), "kota": ("rapaz da cidade", "urbano"), "mwangolé": ("rapaz do subúrbio", "subúrbio"), "cota": ("dinheiro", "casual"), "fixe": ("bom/ótimo", "positivo"), "bué": ("muito", "intensificador"), "oroh": ("pessoa chata", "negativo"), "baza": ("terminar", "casual"), "kuduro": ("dança urbana", "cultural"), "sassa": ("sofisticado", "urbano"), "kalembe": ("ridículo", "negativo"), } # Intenções para treinamento INTENCOES_TREINAMENTO = { "saudacao": ["ola", "oi", "bom dia", "boa tarde", "boa noite", "como vai", "e aí"], "pergunta": ["?", "porquê", "porque", "como", "o que", "qual", "onde", "quando", "quanto"], "afirmacao": ["acho", "creio", "penso", "sei que", "tenho certeza"], "despedida": ["tchau", "até mais", "adeus", "fim", "parar"], "agradecimento": ["obrigado", "thanks", "grato", "agradecido"], "elogio": ["fixe", "bom trabalho", "parabéns", "incrível", "show"], "reclamacao": ["ruim", "péssimo", "odeio", "não gostei", "decepcionado"] } # ============================================================ # 🎯 ESTRUTURAS DE DADOS # ============================================================ @dataclass class Interacao: """Estrutura de uma interação para treinamento""" usuario: str mensagem: str resposta: str numero: str is_reply: bool = False mensagem_original: str = "" timestamp: float = field(default_factory=time.time) emocao: str = "neutral" confianca_emocao: float = 0.5 intencao: str = "pergunta" api_usada: str = "" tokens_usados: int = 0 response_time: float = 0.0 @dataclass class TrainingResult: """Resultado de um ciclo de treinamento""" nivel: str amostras_processadas: int embeddings_atualizados: int emocoes_aprendidas: int gírias_aprendidas: int api_adaptations: int duracao_segundos: float sucesso: bool erro: Optional[str] = None # ============================================================ # 🏗️ CLASSE PRINCIPAL DE TREINAMENTO # ============================================================ class Treinamento: """ Sistema de treinamento avançado 3-níveis: - Nível 1: Emoções (BART + Heurísticas) - Nível 2: NLP & Embeddings (SentenceTransformers) - Nível 3: API Adapter (Mistral, Gemini, Groq, etc.) """ def __init__( self, db: Database, contexto: Optional[Any] = None, interval_hours: int = 6 ): self.db = db self.contexto = contexto self.interval_hours = interval_hours # Threading self._thread = None self._running = False self._stop_event = threading.Event() # Componentes self.api_trainer = APIAdapterTrainer(db) self.model_trainer = get_model_trainer(db) # Usuários privilegiados self.privileged_users = getattr(config, 'PRIVILEGED_USERS', ('244937035662', 'isaac', 'isaac quarenta')) # Cache de treinamento self._training_cache: Dict[str, Any] = {} logger.info("🟢 Treinamento 3-níveis inicializado") # ============================================================ # 📝 REGISTRO DE INTERAÇÕES # ============================================================ def registrar_interacao( self, usuario: str, mensagem: str, resposta: str, numero: str = '', is_reply: bool = False, mensagem_original: str = '', api_usada: str = '', tokens_usados: int = 0, response_time: float = 0.0 ) -> Interacao: """ Registra interação e executa aprendizado em tempo real """ # Cria estrutura de interação interacao = Interacao( usuario=usuario, mensagem=mensagem, resposta=resposta, numero=numero, is_reply=is_reply, mensagem_original=mensagem_original, api_usada=api_usada, tokens_usados=tokens_usados, response_time=response_time ) try: # Salva no banco (com o modelo que gerou a resposta) self.db.salvar_mensagem( usuario, mensagem, resposta, numero, is_reply, mensagem_original, modelo_usado=api_usada or "desconhecido" ) # Aprendizado em tempo real self._aprender_em_tempo_real(interacao) # Registra API call se aplicável if api_usada: self.api_trainer.record_api_call( provider=api_usada, success=True, response_time=response_time, tokens_used=tokens_usados ) except Exception as e: logger.error(f"Erro ao registrar interação: {e}") if api_usada: self.api_trainer.record_api_call( provider=api_usada, success=False, response_time=response_time, error=str(e) ) return interacao def _aprender_em_tempo_real(self, interacao: Interacao): """Aprendizado em tempo real (Nível 1 + 2)""" if not interacao.numero: return # Combine mensagem + resposta para análise texto_completo = f"{interacao.mensagem} {interacao.resposta}" texto_lower = texto_completo.lower() # === NÍVEL 1: Análise de Emoções === # Correção Pylance: verifica se emotion_trainer está disponível if emotion_trainer is not None: analise_emocao = emotion_trainer.analisar(interacao.mensagem) interacao.emocao = analise_emocao.get('emocao', 'neutral') interacao.confianca_emocao = analise_emocao.get('confianca', 0.5) else: interacao.emocao = 'neutral' interacao.confianca_emocao = 0.5 # Salva emoção self.db.salvar_aprendizado_detalhado( interacao.numero, "emocao_atual", json.dumps({"emocao": interacao.emocao, "confianca": interacao.confianca_emocao}) ) # === NÍVEL 2: Embeddings === # Correção Pylance: verifica se embedding_manager e seu modelo estão disponíveis if embedding_manager is not None and embedding_manager.load_model(): embedding = embedding_manager.generate_embedding(texto_completo) if embedding is not None: self.db.salvar_embedding( interacao.numero, interacao.mensagem, interacao.resposta, embedding ) # === Análise de Intenção === intencao = self._detectar_intencao(texto_lower) interacao.intencao = intencao # === Heurística de Tom === tom = self._detectar_tom(texto_lower) self.db.registrar_tom_usuario( interacao.numero, tom, analise_emocao.get('confianca', 0.5), texto_lower[:200] ) # === Aprendizado de Gírias === self._aprender_girias(interacao.numero, texto_lower) def _detectar_intencao(self, texto: str) -> str: """Detecta intenção do texto""" for intencao, palavras in INTENCOES_TREINAMENTO.items(): if any(p in texto for p in palavras): return intencao return "pergunta" # Default def _detectar_tom(self, texto: str) -> str: """Detecta tom do texto""" rude_count = sum(1 for p in PALAVRAS_RUDES if p in texto) formal_count = sum(1 for p in ["senhor", "doutor", "por favor", "agradecido"] if p in texto) if rude_count > 0: return "rude" elif formal_count > 1: return "formal" elif any(p in texto for p in ["puto", "mano", "fixe", "kkk", "bué"]): return "informal" return "casual" def _aprender_girias(self, numero: str, texto: str): """Aprende gírias do texto""" for giria, (significado, _) in GIRIAS_ANGOLANAS.items(): if giria in texto: try: self.db.salvar_giria_aprendida( numero, giria, significado, texto[:100] ) except Exception as e: logger.warning(f"Erro ao salvar gíria {giria}: {e}") # ============================================================ # 🎓 TREINAMENTO EM 3 NÍVEIS # ============================================================ def train_all_levels(self) -> List[TrainingResult]: """ Executa treinamento completo em todos os níveis Returns: Lista de resultados para cada nível """ resultados = [] start_time = time.time() try: # Nível 1: Emoções logger.info("🎭 Treinando Nível 1: Emoções...") resultado_n1 = self._train_nivel_emocoes() resultados.append(resultado_n1) # Nível 2: NLP & Embeddings logger.info("🧠 Treinando Nível 2: NLP & Embeddings...") resultado_n2 = self._train_nivel_nlp() resultados.append(resultado_n2) # Nível 3: API Adapter logger.info("🔗 Treinando Nível 3: API Adapter...") resultado_n3 = self._train_nivel_api() resultados.append(resultado_n3) # Nível 4: MoE Experts logger.info("🤖 Treinando Nível 4: MoE Experts...") resultado_n4 = self._train_nivel_moe() resultados.append(resultado_n4) # Purificação e Segmentação Autónoma (Opcional, gera os JSONLs) try: self._purificar_e_segmentar_dataset() except: pass duracao_total = time.time() - start_time logger.success(f"✅ Treinamento completo: {duracao_total:.2f}s") except Exception as e: logger.error(f"❌ Erro no treinamento: {e}") resultados.append(TrainingResult( nivel="complete", amostras_processadas=0, embeddings_atualizados=0, emocoes_aprendidas=0, gírias_aprendidas=0, api_adaptations=0, duracao_segundos=time.time() - start_time, sucesso=False, erro=str(e) )) return resultados def _train_nivel_emocoes(self) -> TrainingResult: """Nível 1: Treinamento de emoções""" start_time = time.time() emocoes_aprendidas = 0 try: # Recupera usuários com interações usuarios = self._get_usuarios_para_treinamento() for usuario in usuarios: try: # Recupera mensagens recentes mensagens = self.db.recuperar_mensagens(usuario, limite=20) for msg, resp in mensagens: if msg and resp: analise = emotion_trainer.analisar(msg) # Salva aprendizado self.db.salvar_aprendizado_detalhado( usuario, f"emocao_{int(time.time())}", json.dumps(analise) ) emocoes_aprendidas += 1 except Exception as e: logger.warning(f"Erro ao treinar emoções para {usuario}: {e}") return TrainingResult( nivel="emocoes", amostras_processadas=len(usuarios), embeddings_atualizados=0, emocoes_aprendidas=emocoes_aprendidas, gírias_aprendidas=0, api_adaptations=0, duracao_segundos=time.time() - start_time, sucesso=True ) except Exception as e: return TrainingResult( nivel="emocoes", amostras_processadas=0, embeddings_atualizados=0, emocoes_aprendidas=0, gírias_aprendidas=0, api_adaptations=0, duracao_segundos=time.time() - start_time, sucesso=False, erro=str(e) ) def _train_nivel_nlp(self) -> TrainingResult: """Nível 2: Treinamento de NLP & Embeddings""" start_time = time.time() embeddings_atualizados = 0 try: if not embedding_manager.load_model(): raise Exception("Embedding model não disponível") usuarios = self._get_usuarios_para_treinamento() # Carrega modelo SentenceTransformers model = embedding_manager._model for usuario in usuarios: try: # Recupera mensagens mensagens = self.db.recuperar_mensagens(usuario, limite=20) # Prepara batch textos = [] for msg, resp in mensagens: if msg and resp: textos.append(f"{msg} {resp}") if textos: # Gera batch embeddings embeddings = embedding_manager.generate_batch_embeddings(textos) if embeddings is not None: # Salva embeddings no banco for i, (msg, resp) in enumerate(mensagens[:len(textos)]): if i < len(embeddings): self.db.salvar_embedding( usuario, msg, resp, embeddings[i] ) embeddings_atualizados += 1 except Exception as e: logger.warning(f"Erro ao treinar NLP para {usuario}: {e}") return TrainingResult( nivel="nlp", amostras_processadas=len(usuarios), embeddings_atualizados=embeddings_atualizados, emocoes_aprendidas=0, gírias_aprendidas=0, api_adaptations=0, duracao_segundos=time.time() - start_time, sucesso=True ) except Exception as e: return TrainingResult( nivel="api", amostras_processadas=0, embeddings_atualizados=0, emocoes_aprendidas=0, gírias_aprendidas=0, api_adaptations=0, duracao_segundos=time.time() - start_time, sucesso=False, erro=str(e) ) def _train_nivel_moe(self) -> TrainingResult: """Nivel 4: Treinamento Especialista MoE (Lexi, Qwen, Luana)""" start_time = time.time() examples_count = 0 try: # Especialistas suportados especialistas = ["roleplay", "debate", "cultural"] for esp in especialistas: # Dispara destilacao ou fine-tuning autonomo res = self.model_trainer.start_finetuning(especialidade=esp) if res.get("success"): examples_count += res.get("examples", res.get("count", 0)) return TrainingResult( nivel="moe_experts", amostras_processadas=examples_count, embeddings_atualizados=0, emocoes_aprendidas=0, gírias_aprendidas=0, api_adaptations=0, duracao_segundos=time.time() - start_time, sucesso=True ) except Exception as e: logger.error(f"Erro no nivel MoE: {e}") return TrainingResult( nivel="moe_experts", amostras_processadas=0, embeddings_atualizados=0, emocoes_aprendidas=0, gírias_aprendidas=0, api_adaptations=0, duracao_segundos=time.time() - start_time, sucesso=False, erro=str(e) ) def _train_nivel_api(self) -> TrainingResult: """Nível 3: Treinamento de API Adapter""" start_time = time.time() api_adaptations = 0 try: # Analisa performance das APIs for provider in self.api_trainer.api_stats.keys(): stats = self.api_trainer.api_stats[provider] total = stats["success"] + stats["failure"] if total > 0: success_rate = stats["success"] / total # Se success rate < 80%, ajusta estratégia if success_rate < 0.8: # Salva adaptação necessária self.db.salvar_aprendizado_detalhado( f"api_strategy_{provider}", "needs_adjustment", json.dumps({ "success_rate": success_rate, "avg_response_time": stats["avg_response_time"], "timestamp": time.time() }) ) api_adaptations += 1 return TrainingResult( nivel="api", amostras_processadas=0, embeddings_atualizados=0, emocoes_aprendidas=0, gírias_aprendidas=0, api_adaptations=api_adaptations, duracao_segundos=time.time() - start_time, sucesso=True ) except Exception as e: return TrainingResult( nivel="api", amostras_processadas=0, embeddings_atualizados=0, emocoes_aprendidas=0, gírias_aprendidas=0, api_adaptations=0, duracao_segundos=time.time() - start_time, sucesso=False, erro=str(e) ) def _get_usuarios_para_treinamento(self) -> List[str]: """Retorna lista de usuários para treinamento""" try: # Consulta usuários com mensagens result = self.db._execute_with_retry( "SELECT DISTINCT usuario FROM mensagens ORDER BY id DESC LIMIT 50" ) return [r[0] for r in result] if result else [] except Exception: return [] # ============================================================ # 🔄 LOOP PERIÓDICO # ============================================================ def _run_loop(self): """Loop de treinamento periódico""" interval = max(1, self.interval_hours) * 3600 while not self._stop_event.is_set(): try: if self._running: self.train_all_levels() except Exception as e: logger.exception(f"Erro no loop de treinamento: {e}") # Espera com suporte a parada for _ in range(int(interval)): if self._stop_event.is_set(): break time.sleep(1) def start_periodic_training(self): """Inicia treinamento periódico""" if self._running: return self._running = True self._stop_event.clear() self._thread = threading.Thread(target=self._run_loop, daemon=True) self._thread.start() logger.info(f"🚀 Treinamento periódico iniciado (intervalo: {self.interval_hours}h)") def stop(self): """Para treinamento periódico""" self._running = False self._stop_event.set() if self._thread: self._thread.join(timeout=5) logger.info("⏹️ Treinamento periódico parado") # ============================================================ # 📊 UTILITÁRIOS # ============================================================ def get_treinamento_status(self) -> Dict[str, Any]: """Retorna status do treinamento""" return { "running": self._running, "interval_hours": self.interval_hours, "embedding_available": embedding_manager.load_model(), "emotion_model_available": emotion_trainer.load_model(), "api_stats": self.api_trainer.api_stats, "privileged_users": len(self.privileged_users) } def obter_estatisticas(self) -> Dict[str, Any]: """ Retorna estatísticas do treinamento. Método para compatibilidade com testar_correcoes.py """ return { "status": self.get_treinamento_status(), "api_stats": self.api_trainer.api_stats, "usuarios_privilegiados": len(self.privileged_users), "embedding_disponivel": embedding_manager.load_model(), "emotion_model_disponivel": emotion_trainer.load_model() } def limpar_dataset(self) -> bool: """ Limpa o cache/dataset de treinamento. Método para compatibilidade com testar_correcoes.py """ try: self._training_cache.clear() logger.info("Dataset de treinamento limpo") return True except Exception as e: logger.error(f"Erro ao limpar dataset: {e}") return False def force_train(self) -> List[TrainingResult]: """Força treinamento imediato""" return self.train_all_levels() # ============================================================ # 🧹 SEGMENTAÇÃO AUTÓNOMA DE DATASET POR MODELO # ============================================================ def _purificar_e_segmentar_dataset(self, output_dir: str = "/akira/data/treino") -> Dict[str, int]: """ Extrai mensagens da BD, filtra as de baixa qualidade e exporta JSONL separados por especialista: - treino_roleplay_lexi.jsonl → Lexi / llama8b / genérico - treino_debate_qwen.jsonl → Qwen 72B / debates / teses - treino_cultural_luana.jsonl → Mistral-Luana / memes / gírias Retorna: dict com contagem de exemplos por ficheiro. """ import os # Padrões de classificação por modelo_usado MAPA_MODELOS: Dict[str, str] = { "lexi": "roleplay_lexi", "llama8b": "roleplay_lexi", "llama_local_gguf": "roleplay_lexi", "fallback_offline": "roleplay_lexi", "qwen": "debate_qwen", "qwen72b": "debate_qwen", "huihui": "debate_qwen", "featherless": "debate_qwen", "luana": "cultural_luana", "mistral": "cultural_luana", } # Palavras-chave de mensagens de erro (descartadas) PADROES_ERRO = ["eita!", "desculpa, estou off", "todos os provedores falharam", "erro", "exception", "system tá com problemas"] try: os.makedirs(output_dir, exist_ok=True) # Busca todas as mensagens com modelo registado rows = self.db._execute_with_retry( """SELECT usuario, mensagem, resposta, modelo_usado FROM mensagens WHERE resposta IS NOT NULL AND LENGTH(resposta) > 5 ORDER BY id DESC LIMIT 5000""" ) if not rows: logger.warning("⚠️ Nenhuma mensagem encontrada para segmentação") return {} # Agrupa por categoria de modelo buckets: Dict[str, List[Dict]] = { "roleplay_lexi": [], "debate_qwen": [], "cultural_luana": [], "outros": [] } for row in rows: usuario = row[0] or "" mensagem = row[1] or "" resposta = row[2] or "" modelo = (row[3] or "desconhecido").lower() # Filtra respostas de erro / muito curtas resposta_lower = resposta.lower() if any(p in resposta_lower for p in PADROES_ERRO): continue if len(resposta.strip()) < 5: continue # Detecta categoria categoria = "outros" for chave, cat in MAPA_MODELOS.items(): if chave in modelo: categoria = cat break buckets[categoria].append({ "instruction": mensagem, "output": resposta, "usuario": usuario, "modelo": modelo }) # Exporta ficheiros JSONL contagens: Dict[str, int] = {} for categoria, exemplos in buckets.items(): if not exemplos: continue nome_ficheiro = f"treino_{categoria}.jsonl" caminho = os.path.join(output_dir, nome_ficheiro) with open(caminho, "w", encoding="utf-8") as f: for ex in exemplos: f.write(json.dumps(ex, ensure_ascii=False) + "\n") contagens[nome_ficheiro] = len(exemplos) logger.info(f"📦 [{categoria}] → {len(exemplos)} exemplos → {caminho}") logger.success(f"✅ Segmentação concluída: {sum(contagens.values())} exemplos totais") return contagens except Exception as e: logger.error(f"❌ Erro na segmentação de dataset: {e}") return {}