Spaces:
Sleeping
Sleeping
| 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 |