from datetime import datetime, timedelta import locale import json from pathlib import Path from langchain.tools import tool from langchain_community.vectorstores import Chroma from langchain_community.document_loaders import TextLoader from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter vector_store = None CHROMA_PERSIST_DIR = Path("chroma") def init_knowledge_base(base_dir: Path, openai_key: str): global vector_store manifest_path = base_dir / "manifest.json" print(f"🧠 Inicializando base de conocimiento desde: {manifest_path}") if not manifest_path.exists(): print(f"⚠️ {manifest_path} no existe.") vector_store = None return try: with open(manifest_path, 'r', encoding='utf-8') as f: manifest = json.load(f) documents_metadata = manifest.get("documents", []) if not documents_metadata: print("⚠️ No se encontraron documentos en el manifest.") vector_store = None return all_docs = [] for doc_info in documents_metadata: rel_path = doc_info.get("knowledge_base_path") if not rel_path: continue # El path en manifest usa backslashes, lo convertimos a Path del sistema file_path = base_dir / rel_path.replace("\\", "/") if file_path.exists(): print(f"📖 Cargando: {file_path.name}") loader = TextLoader(str(file_path), encoding='utf-8') all_docs.extend(loader.load()) else: print(f"⚠️ Archivo no encontrado: {file_path}") if not all_docs: print("⚠️ No se cargó ningún documento.") vector_store = None return text_splitter = RecursiveCharacterTextSplitter(chunk_size=1200, chunk_overlap=200) texts = text_splitter.split_documents(all_docs) embeddings = OpenAIEmbeddings(openai_api_key=openai_key) # Si ya existe el directorio chroma, lo borramos para forzar recarga limpia if CHROMA_PERSIST_DIR.exists(): import shutil try: shutil.rmtree(CHROMA_PERSIST_DIR) print(f"🧹 Directorio de persistencia {CHROMA_PERSIST_DIR} limpiado.") except Exception as e: print(f"⚠️ No se pudo limpiar {CHROMA_PERSIST_DIR}: {e}") vector_store = Chroma.from_documents( texts, embeddings, persist_directory=str(CHROMA_PERSIST_DIR) ) print(f"✅ Base de conocimiento cargada: {len(texts)} fragmentos de {len(all_docs)} archivos.") except Exception as e: print(f"❌ Error cargando base de conocimiento: {e}") vector_store = None # ——————————————————————————————————————————— # Herramientas Auxiliares de Fecha # ——————————————————————————————————————————— def _formatear_fecha_es(fecha: datetime) -> str: """Formatea una fecha como YYYY-MM-DD (día de la semana en español).""" dias = ["lunes", "martes", "miércoles", "jueves", "viernes", "sábado", "domingo"] return f"{fecha.strftime('%Y-%m-%d')} ({dias[fecha.weekday()]})" @tool def obtener_fecha_actual() -> str: """Devuelve la fecha actual con el día de la semana. Útil para saber qué día es hoy. RECUERDA: Las herramientas de tiempo real (MCP) SOLO funcionan para esta fecha.""" return _formatear_fecha_es(datetime.now()) @tool def calcular_fecha(dias_offset: int) -> str: """ Calcula una fecha relativa a hoy sumando o restando días. Útil para fechas como 'mañana', 'ayer' o 'en 3 días'. Devuelve la fecha en formato YYYY-MM-DD (día de la semana). Args: dias_offset: Número de días a sumar (positivo) o restar (negativo). """ fecha = datetime.now() + timedelta(days=dias_offset) return _formatear_fecha_es(fecha) # ——————————————————————————————————————————— # Herramientas RAG / Históricas # ——————————————————————————————————————————— @tool def detectar_tipo_consulta(pregunta_usuario: str) -> str: """ Analiza la pregunta del usuario para determinar si busca información de TIEMPO_REAL o información HISTORICA/PLANIFICADA. """ pregunta_lower = pregunta_usuario.lower() historico_keywords = [ 'septiembre', 'octubre', 'noviembre', 'diciembre', 'enero', 'febrero', 'marzo', 'abril', 'mayo', 'junio', 'julio', 'agosto', 'mes pasado', 'meses pasados', 'meses anteriores', 'año pasado', 'este año', '2026', '2025', '2024', '2023', '2022', 'histórico', 'estadística', 'reporte', 'itinerario mensual', 'cuántos vuelos hubo', 'total de vuelos en', 'comparación', 'tendencia' ] tiempo_real_keywords = [ 'hoy', 'ahora', 'actual', 'actualmente', 'en este momento' ] # Si no es explícitamente "hoy", tratar como histórico/planificado para buscar en documentos for k in historico_keywords: if k in pregunta_lower: return "HISTORICO" for k in tiempo_real_keywords: if k in pregunta_lower: return "TIEMPO_REAL" # Fallback: si pregunta por un día específico (ej: "el jueves", "7 de febrero"), es HISTORICO (Documentos) return "HISTORICO" @tool def leer_itinerario_mensual(mes: str, anio: str = "2026") -> str: """ Lee el contenido completo del itinerario mensual de vuelos para un mes y año específicos. Útil para tener la visión completa de los vuelos programados. Args: mes: Nombre del mes (ej: 'enero', 'febrero', 'diciembre'). anio: Año (ej: '2025', '2026'). Por defecto '2026'. """ print(f"🔍 Herramienta leer_itinerario_mensual: mes='{mes}', anio='{anio}'") base_dir = Path(__file__).resolve().parent.parent.parent manifest_path = base_dir / "manifest.json" if not manifest_path.exists(): print(f"❌ No se encontró manifest.json en {manifest_path}") return "Error: No se encontró el archivo manifest.json en la raíz del cliente." try: with open(manifest_path, 'r', encoding='utf-8') as f: manifest = json.load(f) found_path = None mes_slug = mes.lower() print(f"📂 Buscando coincidencia para slug: '{mes_slug}' y año: '{anio}'") for doc in manifest.get("documents", []): orig_name = doc.get("original_name", "").lower() kb_path = doc.get("knowledge_base_path", "").lower() # Coincidencia más flexible: que el mes y el año estén en el nombre if mes_slug in kb_path or mes_slug in orig_name: if not anio or anio in kb_path or anio in orig_name: found_path = doc.get("knowledge_base_path") print(f"✨ Coincidencia encontrada: {found_path}") break if not found_path: print(f"⚠️ No se encontró itinerario para {mes} {anio} en el manifest.") return f"No se encontró un itinerario para {mes} {anio} en el manifest." file_path = base_dir / found_path.replace("\\", "/") if not file_path.exists(): return f"Error: El archivo {file_path} no existe físicamente." with open(file_path, 'r', encoding='utf-8') as f: content = f.read() return f"--- CONTENIDO DEL ITINERARIO ({mes.upper()} {anio}) ---\n\n{content}" except Exception as e: return f"Error al leer el itinerario: {str(e)}" @tool def buscar_en_documentos(consulta: str) -> str: """ Realiza una búsqueda semántica en la base de datos de itinerarios (Markdown/PDF). Útil para consultar vuelos de fechas específicas (ayer, mañana, próxima semana) o planes de vuelo futuros que no están en el sistema de tiempo real. """ global vector_store print(f"🔍 Buscando en documentos RAG: '{consulta}'") if vector_store is None: print("❌ El vector_store no está inicializado.") return "La base de conocimiento histórica no está disponible." try: docs = vector_store.similarity_search(consulta, k=5) if not docs: return "No se encontró información en los itinerarios históricos." resultados = [] for doc in docs: metadata_str = "" if hasattr(doc, 'metadata') and doc.metadata: source = doc.metadata.get('source', '') if source: filename = Path(source).stem metadata_str = f"[{filename}] " resultados.append(f"{metadata_str}{doc.page_content}") return "\n\n---\n\n".join(resultados) except Exception as e: return f"Error al buscar en documentos históricos: {str(e)}"