Update app.py

app.py

@@ -1,84 +1,183 @@
-import re
-import os
 import gradio as gr
-import random
-from fuzzywuzzy import fuzz
-
-# ✅ Función para cargar el dataset desde el archivo local
-def cargar_dataset(filename="datos_alojamientos.txt"):
-    dataset = []
-    try:
-        if not os.path.exists(filename):
-            print(f"⚠️ Archivo no encontrado: {filename}")
-            return []
-        
-        with open(filename, "r", encoding="utf-8") as f:
-            contenido = f.read().strip()
-            alojamientos = [bloque.strip() for bloque in contenido.split("---") if bloque.strip()]
-        print(f"✅ Cargados {len(alojamientos)} alojamientos.")
-        return alojamientos
-    except Exception as e:
-        print(f"❌ Error al cargar el dataset: {e}")
-        return []
-
-# ✅ Función para limpiar y normalizar texto
-def limpiar_texto(texto):
-    return re.sub(r'\s+', ' ', texto.strip().lower())
-
-# ✅ Función para extraer información del alojamiento
-def obtener_valor(texto, patron, valor_predeterminado="No especificado"):
-    match = re.search(patron, texto, re.IGNORECASE | re.MULTILINE)
-    return match.group(1).strip() if match else valor_predeterminado
-
-# ✅ Función para buscar alojamientos según la cantidad de plazas solicitada
-def buscar_alojamientos(pregunta):
-    resultados = []
-    cantidad_solicitada = next((int(p) for p in pregunta.split() if p.isdigit()), None)
-    
-    for alojamiento in dataset:
-        texto_alojamiento = limpiar_texto(alojamiento)
-        plazas_totales = obtener_valor(alojamiento, r"(?:plazas totales\s*:\s*)(\d+)")
-        tipos = re.findall(r"(departamento|monoambiente|cabaña).*?plazas\s*:\s*(\d+).+?)\n", texto_alojamiento, re.IGNORECASE | re.DOTALL)
-        
-        for tipo, plazas in tipos:
-            if int(plazas) >= cantidad_solicitada:
-                resultados.append((alojamiento, tipo, plazas, plazas_totales))
-    
+from transformers import AutoModelForCausalLM, AutoTokenizer
+from collections import defaultdict, OrderedDict
+import re
+import torch
+from threading import Lock
+
+# Configuración inicial
+model_name = "microsoft/DialoGPT-small"
+device = "cuda" if torch.cuda.is_available() else "cpu"
+tokenizer = AutoTokenizer.from_pretrained(model_name)
+model = AutoModelForCausalLM.from_pretrained(model_name).to(device)
+model.eval()  # Modo evaluación para mayor rapidez
+
+# Cargar alojamientos en memoria con preprocesamiento
+def cargar_alojamientos():
+    with open("alojamientos.txt", "r", encoding="utf-8") as file:
+        alojamientos = file.read().split("\n\n")  # Separar por alojamientos
+    return {idx: alojamiento for idx, alojamiento in enumerate(alojamientos)}
+
+alojamientos_info = cargar_alojamientos()
+
+# Índice invertido optimizado
+indice_palabras = defaultdict(set)
+for idx, alojamiento in alojamientos_info.items():
+    for palabra in re.split(r'\W+', alojamiento.lower()):
+        if len(palabra) > 2:  # Ignorar palabras cortas
+            indice_palabras[palabra].add(idx)
+
+# Caché LRU con límite de tamaño
+class LRUCache:
+    def __init__(self, capacity=100):
+        self.cache = OrderedDict()
+        self.capacity = capacity
+        self.lock = Lock()
+
+    def get(self, key):
+        with self.lock:
+            if key not in self.cache:
+                return None
+            self.cache.move_to_end(key)
+            return self.cache[key]
+
+    def put(self, key, value):
+        with self.lock:
+            if key in self.cache:
+                self.cache.move_to_end(key)
+            self.cache[key] = value
+            if len(self.cache) > self.capacity:
+                self.cache.popitem(last=False)
+
+cache_respuestas = LRUCache()
+cache_paginas = LRUCache()
+
+# Función de búsqueda optimizada
+def buscar_alojamiento(consulta):
+    consulta = consulta.lower()
+    cached = cache_respuestas.get(consulta)
+    if cached is not None:
+        return cached
+
+    # Extraer número de plazas si existe
+    match_plazas = re.search(r"\b(\d+)\s*(personas|plazas)\b", consulta)
+    plazas = int(match_plazas.group(1)) if match_plazas else None
+
+    # Búsqueda por palabras clave
+    palabras = set(re.split(r'\W+', consulta))
+    indices = set()
+    for palabra in palabras:
+        if palabra in indice_palabras:
+            indices.update(indice_palabras[palabra])
+
+    # Filtrar por plazas si aplica
+    resultados = [
+        alojamientos_info[idx] for idx in indices
+        if not plazas or f"Plazas: {plazas}" in alojamientos_info[idx]
+    ]
+
+    cache_respuestas.put(consulta, resultados)
+    cache_paginas.put(consulta, 0)  # Reiniciar paginación
     return resultados
 
-# ✅ Función para formatear la respuesta
-def formatear_respuesta(resultados):
+# Formateo visual optimizado
+def formatear_alojamiento(texto):
+    lineas = texto.split("\n")
+    resultado = ["----------------------------------------------------------"]
+    iconos = {
+        "Alojamiento:": "🏠", "Dirección:": "📍", "Teléfono:": "📞", "Email:": "📧",
+        "Plazas:": "👤", "Mascotas": "���", "Wifi": "📶", "Directv": "📺",
+        "Ropa blanca": "🛏️", "Habilitación provincial": "📝", "Servicios:": "🛠"
+    }
+
+    for linea in lineas:
+        for clave, icono in iconos.items():
+            if clave in linea:
+                resultado.append(f"{icono} **{linea}**")
+                break
+        else:
+            if "Descripción:" in linea:
+                resultado.append(f"\n🔹 **{linea.replace('Descripción:', '').strip()}**")
+            elif linea.strip():
+                resultado.append(f"   - {linea.strip()}")
+
+    resultado.append("----------------------------------------------------------")
+    return "\n".join(resultado)
+
+# Paginación y resultados
+def mostrar_resultados(consulta):
+    resultados = buscar_alojamiento(consulta)
     if not resultados:
-        return "No encontré alojamientos que coincidan con tu búsqueda."
-    
-    respuesta = "Aquí tienes algunas opciones:\n\n"
-    for alojamiento, tipo, plazas, plazas_totales in resultados[:3]:
-        nombre = obtener_valor(alojamiento, r"(?:alojamiento\s*:\s*)(.+)")
-        direccion = obtener_valor(alojamiento, r"(?:dirección\s*:\s*)(.+)")
-        telefono = obtener_valor(alojamiento, r"(?:teléfono\s*:\s*)(.+)")
-        email = obtener_valor(alojamiento, r"(?:email\s*:\s*)(.+)")
-        servicios = obtener_valor(alojamiento, r"(?:servicios\s*:\s*)(.+)")
-        
-        respuesta += (f"🏠 {nombre}\n📍 {direccion}\n📞 {telefono}\n📧 {email}\n"
-                      f"Plazas Totales: {plazas_totales}\n🏡 {tipo} - Plazas: {plazas}\n🛠 Servicios: {servicios}\n\n---\n\n")
-    return respuesta
-
-# ✅ Cargar dataset
-dataset = cargar_dataset()
-
-# ✅ Interfaz con Gradio
-with gr.Blocks() as iface:
-    gr.Markdown("### Bienvenido a Valer-IA, tu informante turístico virtual 🏡")
-    chat_historial = gr.Textbox(label="Historial de chat", lines=15, interactive=False)
-    pregunta_input = gr.Textbox(label="Escribe tu consulta:")
-    
-    def actualizar_chat(historial, nueva_pregunta):
-        resultados = buscar_alojamientos(nueva_pregunta)
-        respuesta = formatear_respuesta(resultados)
-        return historial + "\n\n" + respuesta, ""
-
-    preguntar_btn = gr.Button("Preguntar")
-    preguntar_btn.click(actualizar_chat, inputs=[chat_historial, pregunta_input], outputs=[chat_historial, pregunta_input])
-
-if __name__ == "__main__":
-    iface.launch()
+        return "Lo siento, no encontré información exacta. Intenta preguntar de otra manera.", ""
+
+    pagina = cache_paginas.get(consulta) or 0
+    inicio, fin = pagina * 3, (pagina + 1) * 3
+    resultados_pagina = resultados[inicio:fin]
+
+    respuesta = "\n\n".join(formatear_alojamiento(r) for r in resultados_pagina)
+    pregunta_mas = ""
+    if fin < len(resultados):
+        cache_paginas.put(consulta, pagina + 1)
+        pregunta_mas = "¿Quieres ver más resultados? Escribe 'sí' para continuar."
+    elif pagina > 0:
+        pregunta_mas = "¿Quieres ver resultados anteriores? Escribe 'atrás' para volver."
+
+    return respuesta, pregunta_mas
+
+# Generación con DialoGPT
+def generar_respuesta_dialoGPT(message):
+    prompt = f"""
+    Eres un asistente virtual de turismo especializado en alojamientos.
+    Responde de manera clara y concisa a: "{message}"
+    Si no tiene relación con alojamientos, di: "Lo siento, no puedo ayudarte con esta consulta."
+    """
+    with torch.no_grad():
+        inputs = tokenizer.encode(prompt + tokenizer.eos_token, return_tensors="pt").to(device)
+        reply_ids = model.generate(
+            inputs, max_length=100, temperature=0.7, top_p=0.9,
+            pad_token_id=tokenizer.eos_token_id, do_sample=True
+        )
+    return tokenizer.decode(reply_ids[:, inputs.shape[-1]:][0], skip_special_tokens=True)
+
+# Lógica del chat
+historial_respuestas = []
+
+def chat(message):
+    global historial_respuestas
+    message = message.strip().lower()
+
+    if message == "sí" and historial_respuestas:
+        consulta = historial_respuestas[-1]["consulta"]
+        respuesta, pregunta_mas = mostrar_resultados(consulta)
+    elif message == "atrás" and historial_respuestas:
+        consulta = historial_respuestas[-1]["consulta"]
+        pagina = cache_paginas.get(consulta)
+        if pagina > 0:
+            cache_paginas.put(consulta, pagina - 1)
+        respuesta, pregunta_mas = mostrar_resultados(consulta)
+    else:
+        resultados = buscar_alojamiento(message)
+        if resultados:
+            respuesta, pregunta_mas = mostrar_resultados(message)
+        else:
+            respuesta = generar_respuesta_dialoGPT(message)
+            pregunta_mas = ""
+
+    historial_respuestas.append({"consulta": message, "respuesta": respuesta})
+    if len(historial_respuestas) > 10:
+        historial_respuestas.pop(0)
+
+    return "\n\n".join(f"**Pregunta:** {h['consulta']}\n\n{h['respuesta']}" for h in historial_respuestas), pregunta_mas
+
+# Interfaz con Gradio
+with gr.Blocks(title="Chat de Turismo") as iface:
+    gr.Markdown("### Asistente de Turismo - Alojamientos")
+    output_box = gr.Textbox(label="Historial", lines=15, interactive=False)
+    input_box = gr.Textbox(label="Consulta", placeholder="Escribe aquí y presiona Enter...")
+    extra_box = gr.Textbox(label="Opciones", interactive=False)
+    send_button = gr.Button("Enviar")
+
+    send_button.click(chat, inputs=input_box, outputs=[output_box, extra_box])
+    input_box.submit(chat, inputs=input_box, outputs=[output_box, extra_box])
+
+iface.launch(share=True, inbrowser=True)