from collections import deque import pandas as pd from config import Config class ConversationMemory: def __init__(self, max_history: int = Config.MAX_HISTORY): self.history = deque(maxlen=max_history) self.schema_cache = None def add_interaction(self, question: str, sql: str, result: str): self.history.append({ "question": question, "sql": sql, "result_summary": self._summarize_result(result) }) def _summarize_result(self, result) -> str: """Resumen ejecutivo para memoria de contexto""" if isinstance(result, pd.DataFrame): # Enfocado en datos CLAVE no en metadatos if len(result) == 1: return f"Único resultado: {result.iloc[0].to_dict()}" elif 'Cliente' in result.columns: top = result.nlargest(3, 'Neto') if 'Neto' in result.columns else result.head(3) return f"Top clientes: {top['Cliente'].tolist()}" else: return f"Filas: {len(result)}, Columnas: {list(result.columns)}" return str(result) def get_context(self, current_question: str) -> str: if not self.history: return "" last_relevant = [] for interaction in self.history: if "producto" in interaction['question'].lower() and "producto" in current_question.lower(): last_relevant.append(interaction) elif "cliente" in interaction['question'].lower() and "cliente" in current_question.lower(): last_relevant.append(interaction) context = "" for i, interaction in enumerate(last_relevant[-1:], 1): # Solo la última relevante context += ( f"Interacción #{i}: {interaction['question'][:50]}...\n" f"SQL: {interaction['sql'][:70]}...\n" f"Resultado: {interaction['result_summary']}\n\n" ) return context