""" 🧠 FINE-TUNING DATA PIPELINE - Gerencia dados de treinamento e pesos do modelo. Armazena conversas, calcula embeddings treináveis, gerencia ciclos de fine-tuning com PostgreSQL. Colabora com treinamento.py para aprendizado contínuo híbrido. Integra LoRA para eficiência em CPU + memória limitada (HF Spaces Free). """ import json import hashlib import numpy as np from datetime import datetime from typing import Dict, List, Optional, Tuple from loguru import logger import asyncio import os try: from sentence_transformers import SentenceTransformer, util from sentence_transformers.losses import CosineSimilarityLoss SENTENCE_TRANSFORMERS_AVAILABLE = True except ImportError: SENTENCE_TRANSFORMERS_AVAILABLE = False try: import torch import torch.nn as nn import torch.optim as optim TORCH_AVAILABLE = True except ImportError: TORCH_AVAILABLE = False try: from peft import LoraConfig, get_peft_model, PeftModel PEFT_AVAILABLE = True except ImportError: PEFT_AVAILABLE = False logger.warning("⚠️ PEFT (LoRA) não disponível - usar pip install peft") # ============================================================ # � LoRA ADAPTER - Eficiente para CPU + Memória Limitada # ============================================================ class LoRAAdapter: """ LoRA (Low-Rank Adaptation) - Reduz parâmetros treináveis em 99.9% Ideal para: CPU, HF Spaces Free, embeddings adaptativos """ def __init__(self, model_dim: int = 384, lora_rank: int = 8, db=None): self.logger = logger self.db = db self.model_dim = model_dim self.lora_rank = lora_rank self.lora_model = None self.base_model = None self.device = "cpu" # HF Spaces Free = CPU only self.lora_alpha = 32 self.lora_dropout = 0.1 self.logger.info(f"🦙 LoRA Adapter inicializado (r={lora_rank}, dim={model_dim}, device=CPU)") def create_lora_model(self, base_model: nn.Module) -> Optional[nn.Module]: """ Envolve modelo com LoRA. Reduz parâmetros: 100% → 0.1% """ if not PEFT_AVAILABLE: self.logger.warning("⚠️ PEFT não disponível, usando adapter manual") return base_model try: # LoRA config otimizado para CPU lora_config = LoraConfig( r=self.lora_rank, # Rank do adapter (8 = bom balanço) lora_alpha=self.lora_alpha, target_modules=["weight"], # Aplica em camadas lineares lora_dropout=self.lora_dropout, bias="none", task_type="CAUSAL_LM" ) # Envolve modelo com LoRA self.lora_model = get_peft_model(base_model, lora_config) # Estatísticas total_params = sum(p.numel() for p in self.lora_model.parameters()) trainable_params = sum(p.numel() for p in self.lora_model.parameters() if p.requires_grad) reduction = (1 - trainable_params / total_params) * 100 self.logger.info(f"✅ LoRA aplicado | Treináveis: {trainable_params:,} ({100-reduction:.2f}%) | Total: {total_params:,}") self.base_model = base_model return self.lora_model except Exception as e: self.logger.error(f"❌ Erro ao criar LoRA model: {e}") return base_model def train_step(self, input_tensor: torch.Tensor, target_tensor: torch.Tensor, learning_rate: float = 0.0001, accumulation_steps: int = 4) -> float: """ Treino com LoRA em CPU (gradient accumulation para memória limitada). """ if self.lora_model is None: return 0.0 try: optimizer = optim.AdamW( [p for p in self.lora_model.parameters() if p.requires_grad], lr=learning_rate, weight_decay=0.01 # Regularização ) self.lora_model.train() total_loss = 0.0 # Gradient accumulation (simula batch maior em CPU) for accum_step in range(accumulation_steps): # Forward pass output = self.lora_model(input_tensor) # Loss loss_fn = nn.MSELoss() loss = loss_fn(output, target_tensor) # Backprop (acumula) (loss / accumulation_steps).backward() total_loss += loss.item() # Update torch.nn.utils.clip_grad_norm_( [p for p in self.lora_model.parameters() if p.requires_grad], max_norm=1.0 # Evita exploding gradients em CPU ) optimizer.step() optimizer.zero_grad() avg_loss = total_loss / accumulation_steps self.logger.debug(f"🦙 LoRA step: loss={avg_loss:.4f}") return avg_loss except Exception as e: self.logger.error(f"❌ Erro em LoRA train step: {e}") return 0.0 def save_lora_weights(self, path: str) -> bool: """ Salva apenas LoRA weights (~1MB em vez de 4GB). Perfeito para HF Spaces. """ if self.lora_model is None: return False try: os.makedirs(os.path.dirname(path), exist_ok=True) # Salva apenas adapter (LoRA) self.lora_model.save_pretrained(path) # Estatística de espaço size_mb = sum(os.path.getsize(os.path.join(path, f)) for f in os.listdir(path)) / (1024 * 1024) self.logger.info(f"💾 LoRA weights salvos: {path} ({size_mb:.2f}MB)") return True except Exception as e: self.logger.error(f"❌ Erro ao salvar LoRA weights: {e}") return False def load_lora_weights(self, path: str) -> bool: """Carrega LoRA weights do checkpoint.""" if self.lora_model is None or self.base_model is None: return False try: self.lora_model = PeftModel.from_pretrained(self.base_model, path) self.logger.info(f"📂 LoRA weights carregados: {path}") return True except Exception as e: self.logger.error(f"❌ Erro ao carregar LoRA weights: {e}") return False def get_trainable_params_count(self) -> int: """Retorna quantidade de parâmetros treináveis.""" if self.lora_model is None: return 0 return sum(p.numel() for p in self.lora_model.parameters() if p.requires_grad) # ============================================================ # 🧠 EMBEDDING TRAINER - Com LoRA integrado # ============================================================ class EmbeddingTrainer: """ Gerencia embeddings com pesos treináveis (adapters). Integra LoRA para eficiência em CPU + memória limitada. """ def __init__(self, embedding_model: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2", db=None, use_lora: bool = True): self.logger = logger self.db = db self.embedding_model_name = embedding_model self.embedding_dim = 384 # MiniLM dimension self.model = None self.trainable_weights = None self.device = "cpu" # HF Spaces Free: CPU only self.use_lora = use_lora self.lora_adapter = None self._load_model() def _load_model(self): """Carrega modelo de embeddings com LoRA se disponível.""" try: if SENTENCE_TRANSFORMERS_AVAILABLE and TORCH_AVAILABLE: self.model = SentenceTransformer( self.embedding_model_name, device=self.device ) # Opção 1: LoRA (recomendado para CPU + HF Spaces) if self.use_lora and PEFT_AVAILABLE: self.lora_adapter = LoRAAdapter( model_dim=self.embedding_dim, lora_rank=8, # Otimizado para CPU db=self.db ) # Envolve transformer com LoRA if hasattr(self.model, 'model'): self.model.model = self.lora_adapter.create_lora_model(self.model.model) self.logger.info(f"✅ EmbeddingTrainer com LoRA carregado (CPU)") # Opção 2: Adapter linear simples (fallback) else: self.trainable_weights = nn.Linear( self.embedding_dim, self.embedding_dim, bias=True ).to(self.device) self.logger.info(f"✅ EmbeddingTrainer com adapter linear (CPU)") else: self.logger.warning("⚠️ Sentence-Transformers/Torch não disponível") except Exception as e: self.logger.error(f"❌ Erro ao carregar embedding model: {e}") def encode(self, texts: List[str]) -> np.ndarray: """Gera embeddings com aplicação de pesos treináveis.""" try: if self.model is None: return np.zeros((len(texts), self.embedding_dim)) # Embeddings base embeddings = self.model.encode(texts, convert_to_tensor=False) # Aplica pesos treináveis (se treinados) if TORCH_AVAILABLE and self.trainable_weights is not None: embeddings_tensor = torch.from_numpy(embeddings).float().to(self.device) embeddings_adapted = self.trainable_weights(embeddings_tensor).detach().cpu().numpy() return embeddings_adapted return embeddings except Exception as e: self.logger.error(f"❌ Erro ao gerar embeddings: {e}") return np.zeros((len(texts), self.embedding_dim)) def compute_similarity(self, text1: str, text2: str) -> float: """Calcula similaridade semântica entre dois textos.""" try: if self.model is None: return 0.5 emb1 = self.encode([text1])[0] emb2 = self.encode([text2])[0] # Similaridade cosseno similarity = np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2) + 1e-8) return float(similarity) except Exception as e: self.logger.debug(f"Erro ao calcular similaridade: {e}") return 0.5 def train_on_batch(self, input_texts: List[str], output_texts: List[str], learning_rate: float = 0.0001): """ Treina pesos adaptativos em um lote. Usa LoRA se disponível (99.9% menos parâmetros em CPU). Minimiza distância entre embeddings de entrada→saída esperada. """ if not TORCH_AVAILABLE: return 0.0 try: # 🦙 Opção 1: LoRA (CPU eficiente, ~1MB checkpoint) if self.lora_adapter and self.lora_adapter.lora_model: return self._train_lora(input_texts, output_texts, learning_rate) # 📈 Opção 2: Adapter linear simples (fallback) elif self.trainable_weights: return self._train_adapter(input_texts, output_texts, learning_rate) return 0.0 except Exception as e: self.logger.error(f"❌ Erro ao treinar: {e}") return 0.0 def _train_lora(self, input_texts: List[str], output_texts: List[str], learning_rate: float = 0.0001) -> float: """Treina com LoRA (99.9% menos parâmetros).""" try: # Gera embeddings with torch.no_grad(): input_emb = torch.from_numpy(self.model.encode(input_texts, convert_to_tensor=False)).float() output_emb = torch.from_numpy(self.model.encode(output_texts, convert_to_tensor=False)).float() # Treina LoRA com gradient accumulation (CPU-friendly) loss = self.lora_adapter.train_step( input_emb, output_emb, learning_rate=learning_rate, accumulation_steps=4 # Acumula 4 steps para CPU ) self.logger.debug(f"🦙 LoRA loss: {loss:.4f}") return loss except Exception as e: self.logger.error(f"❌ Erro em _train_lora: {e}") return 0.0 def _train_adapter(self, input_texts: List[str], output_texts: List[str], learning_rate: float = 0.0001) -> float: """Treina adapter linear simples (fallback).""" try: optimizer = optim.Adam(self.trainable_weights.parameters(), lr=learning_rate) # Embeddings input_emb = torch.from_numpy(self.model.encode(input_texts, convert_to_tensor=False)).float() output_emb = torch.from_numpy(self.model.encode(output_texts, convert_to_tensor=False)).float() # Passa através dos pesos treináveis input_adapted = self.trainable_weights(input_emb) # Loss: minimizar distância (CosineSimilarityLoss) loss_fn = nn.CosineEmbeddingLoss() loss = loss_fn( input_adapted, output_emb, torch.ones(len(input_texts)) ) # Backprop optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(self.trainable_weights.parameters(), max_norm=1.0) optimizer.step() loss_value = float(loss.detach().numpy()) self.logger.debug(f"📈 Adapter loss: {loss_value:.4f}") return loss_value except Exception as e: self.logger.error(f"❌ Erro em _train_adapter: {e}") return 0.0 def save_weights(self, path: str): """Salva pesos treináveis (LoRA ~1MB ou adapter ~10MB).""" try: os.makedirs(os.path.dirname(path), exist_ok=True) # 🦙 LoRA: salva apenas adapter (1MB) if self.lora_adapter and hasattr(self.lora_adapter, 'save_lora_weights'): self.lora_adapter.save_lora_weights(path) # 📈 Adapter linear: salva torch state dict elif TORCH_AVAILABLE and self.trainable_weights is not None: torch.save(self.trainable_weights.state_dict(), path) size_kb = os.path.getsize(path) / 1024 self.logger.info(f"💾 Adapter weights salvos: {path} ({size_kb:.2f}KB)") except Exception as e: self.logger.error(f"Erro ao salvar pesos: {e}") def load_weights(self, path: str): """Carrega pesos treináveis (LoRA ou adapter).""" try: if not os.path.exists(path): self.logger.warning(f"⚠️ Arquivo de pesos não encontrado: {path}") return # 🦙 LoRA if self.lora_adapter and hasattr(self.lora_adapter, 'load_lora_weights'): self.lora_adapter.load_lora_weights(path) # 📈 Adapter linear elif TORCH_AVAILABLE and self.trainable_weights is not None: self.trainable_weights.load_state_dict(torch.load(path, map_location='cpu')) self.logger.info(f"📂 Adapter weights carregados: {path}") except Exception as e: self.logger.error(f"Erro ao carregar pesos: {e}") def get_training_info(self) -> Dict[str, any]: """Retorna informações sobre modelo e treinamento.""" info = { 'device': self.device, 'embedding_dim': self.embedding_dim, 'using_lora': self.use_lora and self.lora_adapter is not None, 'model_type': 'LoRA' if (self.lora_adapter and self.lora_adapter.lora_model) else 'Adapter', } if self.lora_adapter and self.lora_adapter.lora_model: info['lora_rank'] = self.lora_adapter.lora_rank info['trainable_params'] = self.lora_adapter.get_trainable_params_count() info['model_size_kb'] = 1.0 # LoRA é ~1MB elif self.trainable_weights: info['trainable_params'] = sum(p.numel() for p in self.trainable_weights.parameters()) info['model_size_kb'] = 10.0 # Adapter linear ~10MB return info class FinetuningPipeline: """ Gerencia o ciclo completo de fine-tuning: 1. Coleta: Armazena conversas de usuários (entrada + resposta esperada) 2. Processamento: Calcula embeddings treináveis com pesos adaptativos 3. Armazenamento: Persiste em PostgreSQL 4. Recuperação: Fornece lotes para treinamento 5. Colaboração: Integra com treinamento.py para aprendizado híbrido 6. Repetição: Ciclos contínuos de melhoria com feedback """ def __init__(self, db, embedding_trainer: Optional[EmbeddingTrainer] = None): self.db = db self.logger = logger self.embedding_trainer = embedding_trainer or EmbeddingTrainer(db=db) self._initialize_tables() def _initialize_tables(self): """Cria tabelas PostgreSQL para fine-tuning com suporte a embeddings.""" try: with self.db._get_connection() as conn: cur = conn.cursor() # Tabela de exemplos de treinamento (expandida com embeddings) cur.execute(""" CREATE TABLE IF NOT EXISTS finetuning_examples ( id SERIAL PRIMARY KEY, user_id TEXT NOT NULL, conversation_id TEXT NOT NULL, input_message TEXT NOT NULL, expected_response TEXT NOT NULL, actual_response TEXT, quality_score INT DEFAULT 50, tone_level VARCHAR(50), hostility_score INT DEFAULT 0, embedding_vector BYTEA, embedding_input BYTEA, embedding_output BYTEA, similarity_score FLOAT DEFAULT 0.0, emotion_label VARCHAR(50), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, indexed BOOLEAN DEFAULT FALSE ); CREATE INDEX IF NOT EXISTS idx_finetuning_user ON finetuning_examples(user_id); CREATE INDEX IF NOT EXISTS idx_finetuning_quality ON finetuning_examples(quality_score); CREATE INDEX IF NOT EXISTS idx_finetuning_tone ON finetuning_examples(tone_level); CREATE INDEX IF NOT EXISTS idx_finetuning_emotion ON finetuning_examples(emotion_label); """) # Tabela de pesos e métricas de treinamento (expandida) cur.execute(""" CREATE TABLE IF NOT EXISTS training_metrics ( id SERIAL PRIMARY KEY, training_session_id TEXT UNIQUE NOT NULL, examples_used INT, avg_quality FLOAT, model_accuracy FLOAT, embedding_loss FLOAT, emotion_accuracy FLOAT, loss FLOAT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, weights_checkpoint BYTEA, embedding_weights BYTEA, status VARCHAR(50) ); """) # Tabela de histórico de ciclos de treinamento cur.execute(""" CREATE TABLE IF NOT EXISTS training_cycles ( id SERIAL PRIMARY KEY, cycle_number INT, cycle_type VARCHAR(50), started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, completed_at TIMESTAMP, examples_processed INT, improvement_pct FLOAT, embedding_improvement FLOAT, emotion_improvement FLOAT, status VARCHAR(50) ); """) # Tabela de feedback e colaboração (NOVO) cur.execute(""" CREATE TABLE IF NOT EXISTS training_feedback ( id SERIAL PRIMARY KEY, example_id INT, feedback_type VARCHAR(50), feedback_value FLOAT, source VARCHAR(50), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY(example_id) REFERENCES finetuning_examples(id) ); CREATE INDEX IF NOT EXISTS idx_feedback_example ON training_feedback(example_id); """) self.logger.info("✅ Fine-tuning tables initialized (with embeddings + collaboration)") except Exception as e: self.logger.warning(f"⚠️ Tables may already exist: {e}") def store_training_example(self, user_id: str, conversation_id: str, input_message: str, expected_response: str, tone_level: str = "very_serious", hostility_score: int = 0, emotion_label: str = "neutro") -> int: """ Armazena um exemplo de treinamento com embeddings treináveis. Colaboração: Integra dados com treinamento.py para aprendizado híbrido. """ try: # Gera embeddings com pesos adaptativos input_emb = self.embedding_trainer.encode([input_message])[0] output_emb = self.embedding_trainer.encode([expected_response])[0] similarity = self.embedding_trainer.compute_similarity(input_message, expected_response) # Serializa embeddings input_emb_bytes = np.frombuffer(input_emb.tobytes(), dtype=np.float32) output_emb_bytes = np.frombuffer(output_emb.tobytes(), dtype=np.float32) with self.db.get_connection_context() as conn: cur = conn.cursor() try: auto_quality = min(100, max(50, int(similarity * 100 + 30))) cur.execute(""" INSERT INTO finetuning_examples (user_id, conversation_id, input_message, expected_response, tone_level, hostility_score, emotion_label, quality_score, embedding_input, embedding_output, similarity_score) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id; """, (user_id, conversation_id, input_message, expected_response, tone_level, hostility_score, emotion_label, auto_quality, input_emb_bytes.tobytes(), output_emb_bytes.tobytes(), similarity)) result = cur.fetchone() if result is None: self.logger.error(f"❌ [FINETUNING] RETURNING não retornou ID: Verifique se a tabela está OK") return -1 # RealDictCursor retorna dict, não tupla - acessa por chave example_id = result['id'] if isinstance(result, dict) else result[0] self.logger.info(f"✅ [FINETUNING] Exemplo #{example_id} armazenado | Emotion={emotion_label} | Similarity={similarity:.3f}") return example_id except Exception as cur_err: self.logger.error(f"❌ [FINETUNING] Cursor error: {cur_err} | Type: {type(cur_err).__name__}") import traceback self.logger.debug(f"Traceback: {traceback.format_exc()}") raise except Exception as e: import traceback self.logger.error(f"❌ Erro ao armazenar exemplo: {e}") self.logger.debug(f"Traceback: {traceback.format_exc()}") return -1 def rate_example(self, example_id: int, quality_score: int, feedback_type: str = "manual"): """ Avalia a qualidade e registra feedback (colaboração com treinamento.py). """ try: quality_score = max(0, min(100, quality_score)) with self.db.get_connection_context() as conn: cur = conn.cursor() cur.execute(""" UPDATE finetuning_examples SET quality_score = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s; """, (quality_score, example_id)) # Registra feedback para colaboração cur.execute(""" INSERT INTO training_feedback (example_id, feedback_type, feedback_value, source) VALUES (%s, %s, %s, %s); """, (example_id, feedback_type, quality_score / 100.0, "finetuning_pipeline")) self.logger.debug(f"📊 Exemplo #{example_id} feedback={quality_score} | Type={feedback_type}") except Exception as e: self.logger.error(f"❌ Erro ao avaliar exemplo: {e}") def get_training_batch(self, batch_size: int = 32, min_quality: int = 60, include_emotions: bool = True) -> List[Dict]: """ Retorna um lote priorizado para treinamento híbrido. Colaboração: Inclui dados de emoção do treinamento.py. """ try: with self.db.get_connection_context() as conn: cur = conn.cursor() query = """ SELECT id, input_message, expected_response, tone_level, hostility_score, emotion_label, similarity_score FROM finetuning_examples WHERE quality_score >= %s ORDER BY quality_score DESC, similarity_score DESC, created_at DESC LIMIT %s; """ cur.execute(query, (min_quality, batch_size)) rows = cur.fetchall() batch = [] for row in rows: if isinstance(row, dict): # RealDictCursor retorna dict batch.append({ 'example_id': row['id'], 'input': row['input_message'], 'expected_output': row['expected_response'], 'tone_level': row['tone_level'], 'hostility_score': row['hostility_score'], 'emotion_label': row['emotion_label'], 'similarity_score': row['similarity_score'], }) else: # Tupla normal batch.append({ 'example_id': row[0], 'input': row[1], 'expected_output': row[2], 'tone_level': row[3], 'hostility_score': row[4], 'emotion_label': row[5], 'similarity_score': row[6], }) self.logger.info(f"📦 Training batch retrieved: {len(batch)} examples | min_quality={min_quality}") return batch except Exception as e: self.logger.error(f"❌ Erro ao recuperar batch: {e}") return [] def get_statistics(self) -> Dict: """Retorna estatísticas com análise de embeddings.""" try: with self.db.get_connection_context() as conn: cur = conn.cursor() # Totais cur.execute("SELECT COUNT(*) as count FROM finetuning_examples;") result = cur.fetchone() total = result['count'] if isinstance(result, dict) else (result[0] if result else 0) cur.execute("SELECT AVG(quality_score) as avg_quality, AVG(similarity_score) as avg_similarity FROM finetuning_examples;") result = cur.fetchone() if isinstance(result, dict): avg_quality = result['avg_quality'] if result else 0 avg_similarity = result['avg_similarity'] if result else 0 else: avg_quality = result[0] if result and result[0] else 0 avg_similarity = result[1] if result and result[1] else 0 # Por emotion cur.execute(""" SELECT emotion_label, COUNT(*) as count, AVG(quality_score) as avg_quality FROM finetuning_examples GROUP BY emotion_label; """) emotion_dist = {} for row in cur.fetchall(): if isinstance(row, dict): emotion_dist[row['emotion_label']] = {'count': row['count'], 'avg_quality': row['avg_quality']} else: emotion_dist[row[0]] = {'count': row[1], 'avg_quality': row[2]} # Por tone cur.execute(""" SELECT tone_level, COUNT(*) as count, AVG(quality_score) as avg_quality FROM finetuning_examples GROUP BY tone_level; """) tone_dist = {} for row in cur.fetchall(): if isinstance(row, dict): tone_dist[row['tone_level']] = {'count': row['count'], 'avg_quality': row['avg_quality']} else: tone_dist[row[0]] = {'count': row[1], 'avg_quality': row[2]} return { 'total_examples': total, 'avg_quality_score': round(avg_quality, 2), 'avg_embedding_similarity': round(avg_similarity, 3), 'emotion_distribution': emotion_dist, 'tone_distribution': tone_dist, } except Exception as e: self.logger.error(f"❌ Erro ao recuperar estatísticas: {e}") import traceback self.logger.debug(f"Traceback: {traceback.format_exc()}") return {} def start_training_cycle(self, cycle_type: str = "hybrid") -> str: """ Inicia ciclo de treinamento híbrido. cycle_type: "hybrid" (fine-tuning + emotions), "embedding", "emotion", "full" """ try: session_id = hashlib.md5(f"{datetime.now().isoformat()}".encode()).hexdigest() with self.db.get_connection_context() as conn: cur = conn.cursor() cur.execute("SELECT MAX(cycle_number) as max_cycle FROM training_cycles;") result = cur.fetchone() if isinstance(result, dict): current_cycle = (result['max_cycle'] if result['max_cycle'] else 0) + 1 else: current_cycle = (result[0] if result and result[0] else 0) + 1 cur.execute(""" INSERT INTO training_cycles (cycle_number, cycle_type, status) VALUES (%s, %s, 'started') RETURNING id; """, (current_cycle, cycle_type)) self.logger.info(f"🚀 [CYCLE {current_cycle}] Tipo={cycle_type} | Session={session_id}") return session_id except Exception as e: self.logger.error(f"❌ Erro ao iniciar ciclo: {e}") import traceback self.logger.debug(f"Traceback: {traceback.format_exc()}") return None def complete_training_cycle(self, session_id: str, improvement_pct: float = 0.0, embedding_improvement: float = 0.0, emotion_improvement: float = 0.0): """ Completa ciclo registrando melhorias em múltiplas dimensões. Colaboração: Registra progressos de fine-tuning e emotions. """ try: with self.db.get_connection_context() as conn: cur = conn.cursor() cur.execute(""" UPDATE training_cycles SET completed_at = CURRENT_TIMESTAMP, status = 'completed', improvement_pct = %s, embedding_improvement = %s, emotion_improvement = %s WHERE cycle_number = ( SELECT MAX(cycle_number) FROM training_cycles ); """, (improvement_pct, embedding_improvement, emotion_improvement)) self.logger.info(f"✅ [CYCLE COMPLETE] Fine-tuning={improvement_pct}% | Embedding={embedding_improvement}% | Emotion={emotion_improvement}%") except Exception as e: self.logger.error(f"❌ Erro ao completar ciclo: {e}") # ============================================================ # 🤝 COLLABORAÇÃO COM TREINAMENTO.PY # ============================================================ def sync_with_training_system(self, training_system): """ Colaboração: Sincroniza com treinamento.py. Permite que aprendizado_continuo do treinamento alimenta fine-tuning. """ try: # Obtém estatísticas de treinamento stats = self.get_statistics() if hasattr(training_system, 'registrar_interacao'): self.logger.info(f"🤝 Sincronizando com treinamento.py: {stats}") # Treinamento.py pode usar essas stats para ajustar sua estratégia return stats return stats except Exception as e: self.logger.error(f"❌ Erro ao sincronizar com treinamento.py: {e}") return {} def train_embedding_adapter(self, batch_size: int = 32, learning_rate: float = 0.0001) -> float: """ Treina o adapter de embeddings em um batch. Usa LoRA para eficiência em CPU + HF Spaces Free. Colaboração: Melhora representação semântica para qualidade de respostas. """ try: batch = self.get_training_batch(batch_size=batch_size, min_quality=70) if not batch: self.logger.warning("⚠️ Nenhum exemplo com quality >= 70 para treinar embeddings") return 0.0 inputs = [ex['input'] for ex in batch] outputs = [ex['expected_output'] for ex in batch] # Obtém info do modelo (LoRA vs adapter) model_info = self.embedding_trainer.get_training_info() model_type = model_info.get('model_type', 'unknown') # Treina pesos adaptativos (LoRA ou adapter linear) loss = self.embedding_trainer.train_on_batch(inputs, outputs, learning_rate) self.logger.info(f"🦙 Embedding adapter ({model_type}) treinado: {len(batch)} exemplos | Loss={loss:.4f} | LR={learning_rate}") return loss except Exception as e: self.logger.error(f"❌ Erro ao treinar embedding adapter: {e}") return 0.0 # Singleton # Singleton _finetuning_pipeline_instance = None def get_finetuning_pipeline(db=None): """Get or create singleton.""" _finetuning_pipeline_instance = None def get_finetuning_pipeline(db=None): """Get or create singleton.""" global _finetuning_pipeline_instance if _finetuning_pipeline_instance is None: if db is None: from .database_pg import get_database db = get_database() _finetuning_pipeline_instance = FinetuningPipeline(db) return _finetuning_pipeline_instance