Spaces:
Runtime error
Runtime error
| import gradio as gr | |
| import time | |
| import os | |
| import glob | |
| import torch | |
| import gc | |
| import re | |
| import json | |
| from pathlib import Path | |
| from omegaconf import OmegaConf | |
| from rag.constants.default_prompt import PROMT_USER_INTENT_TREE | |
| from rag.reader.directory_reader import DirectoryReader | |
| from rag.pipeline.milvus_bm25_retriever import HybridSearchRetrieverPipeline | |
| from rag.pipeline.contextual_generator_qa import ContextualQuestionGeneratorPipeline | |
| from rag.synthesizer.types import StreamingResponse | |
| from rag.config.configuration import ConfigurationManager | |
| torch.cuda.empty_cache() | |
| gc.collect() | |
| # === Cargar PDFs | |
| pdf_files = glob.glob(os.path.abspath(os.curdir) + "/data/*.pdf") | |
| reader = DirectoryReader(input_files=pdf_files) | |
| documents = reader.load_data() | |
| # === Configuración | |
| config = OmegaConf.load(Path("configs/config.yaml")) | |
| manager = ConfigurationManager(config) | |
| # === Inicializar Pipelines | |
| pipeline = HybridSearchRetrieverPipeline( | |
| documents=documents, | |
| splitter_config=manager.get_splitter_config(), | |
| index_retriver_config=manager.get_index_retriever_config(), | |
| embed_config=manager.get_embed_config(), | |
| milvus_config=manager.get_db_config(), | |
| rerank_config=manager.get_rerank_config(), | |
| response_config=manager.get_response_config(), | |
| llm_config=manager.get_llm_config() | |
| ) | |
| qgen_pipeline = ContextualQuestionGeneratorPipeline( | |
| llm_config=manager.get_llm_config(), | |
| embed_config=manager.get_embed_config(), | |
| splitter_config=manager.get_splitter_config(), | |
| qgen_config=manager.get_question_gen_config(), | |
| ) | |
| # === Intención básica | |
| def handle_intent(etiqueta: str, motivo: str) -> str: | |
| if etiqueta.lower() == "neutro": | |
| if "saludo" in motivo.lower(): | |
| return "👋 ¡Hola! ¿En qué puedo ayudarte hoy?" | |
| elif "despedida" in motivo.lower(): | |
| return "👋 ¡Hasta luego!" | |
| elif "agradecimiento" in motivo.lower(): | |
| return "😊 ¡Con gusto!" | |
| else: | |
| return "🙂 Gracias por tu mensaje." | |
| elif etiqueta.lower() == "negativo": | |
| return "⚠️ No puedo responder a eso. Por favor, intenta con algo relacionado al estatuto." | |
| return "" | |
| DEV_MODE = False # Cambia a True si quieres ver rutas de archivo en las fuentes | |
| def extract_sources(nodes): | |
| fuentes = [] | |
| for i, node in enumerate(nodes): | |
| meta = node.metadata or {} | |
| fuente = { | |
| "documento": meta.get("file_name", "Desconocido"), | |
| "página": meta.get("num_page", "¿?"), | |
| "autor": meta.get("Author", "Autor desconocido"), | |
| "fecha": meta.get("creation_date", "Sin fecha"), | |
| "ruta": meta.get("file_path", "No disponible") | |
| } | |
| fuente_md = ( | |
| f"**[{i + 1}] {fuente['documento']}**\n" | |
| f"> 📄 Página: `{fuente['página']}`\n" | |
| f"> ✍️ Autor: *{fuente['autor']}*\n" | |
| f"> 🗓️ Fecha de creación: {fuente['fecha']}" | |
| ) | |
| if DEV_MODE: | |
| fuente_md += f"\n> 🧪 Ruta (debug): `{fuente['ruta']}`" | |
| fuentes.append(fuente_md) | |
| return "\n\n".join(fuentes) if fuentes else "⚠️ No se encontraron fuentes relevantes." | |
| def detectar_intencion(query: str) -> str: | |
| try: | |
| result = pipeline.service_context.llm.predict( | |
| prompt=PROMT_USER_INTENT_TREE, | |
| query_str=query | |
| ) | |
| cleaned = re.sub(r"```(?:json)?|```", "", result).strip() | |
| parsed = json.loads(cleaned) | |
| etiqueta = parsed.get("etiqueta", "").lower() | |
| motivo = parsed.get("motivo", "").lower() | |
| return handle_intent(etiqueta, motivo) | |
| except Exception as e: | |
| print("⚠️ Error al detectar intención:", e) | |
| return "" | |
| def bot(history, use_intent, use_subq): | |
| user_msg = next( | |
| (m["content"] for m in reversed(history) if m["role"] == "user" and isinstance(m["content"], str)), | |
| None | |
| ) | |
| if not user_msg: | |
| return history, gr.update(visible=False), *[gr.update(visible=False)] * 10 | |
| try: | |
| if use_intent: | |
| intent_reply = detectar_intencion(user_msg) | |
| if intent_reply: | |
| history.append({"role": "assistant", "content": ""}) | |
| for char in intent_reply: | |
| history[-1]["content"] += char | |
| time.sleep(0.05) | |
| yield history, gr.update(visible=False), *[gr.update(visible=False)] * 10 | |
| return | |
| except Exception as e: | |
| print("⚠️ Error interpretando intención:", e) | |
| response, nodes = pipeline.main(user_msg) | |
| history.append({"role": "assistant", "content": ""}) | |
| if isinstance(response, StreamingResponse): | |
| for char in response.response_gen: | |
| history[-1]["content"] += char | |
| time.sleep(0.02) | |
| yield history, gr.update(visible=False), *[gr.update(visible=False)] * 10 | |
| subquestions = qgen_pipeline.run(user_msg, nodes) if use_subq else [] | |
| fuentes_md = extract_sources(nodes) | |
| history[-1]["sources"] = fuentes_md | |
| subq_updates = [ | |
| gr.update(value=q, visible=True) if i < len(subquestions) else gr.update(visible=False) | |
| for i, q in enumerate(subquestions + [""] * (10 - len(subquestions))) | |
| ] | |
| yield history, gr.update(value=f"📚 **Fuentes:**\n{fuentes_md}", visible=True), *subq_updates | |
| def add_message(history, msg_text): | |
| history.append({"role": "user", "content": msg_text}) | |
| return history, "" | |
| # === Toggle panel visibility | |
| def toggle_config(is_visible): | |
| return not is_visible, gr.update(visible=not is_visible) | |
| # === Estilo CSS | |
| css = """ | |
| body { | |
| background-color: #f8fafc; | |
| font-family: 'Segoe UI', sans-serif; | |
| } | |
| .gr-chatbot { | |
| background-color: #ffffff; | |
| border: 1px solid #d1fae5; | |
| border-radius: 12px; | |
| padding: 10px; | |
| box-shadow: 0 4px 12px rgba(0,0,0,0.03); | |
| } | |
| .gr-textbox { | |
| border-radius: 10px !important; | |
| border: 1px solid #a7f3d0; | |
| padding: 10px; | |
| flex-grow: 1; | |
| } | |
| .gr-button.enviar { | |
| background-color: #2ecc71; | |
| color: white; | |
| font-weight: 600; | |
| border-radius: 10px !important; | |
| padding: 10px 18px; | |
| margin-left: 8px; | |
| transition: background-color 0.3s ease; | |
| } | |
| .gr-button.enviar:hover { | |
| background-color: #27ae60; | |
| } | |
| .gr-markdown { | |
| background-color: #ecfdf5; | |
| border-left: 4px solid #10b981; | |
| padding: 10px; | |
| margin-top: 10px; | |
| border-radius: 8px; | |
| color: #064e3b; | |
| font-size: 15px; | |
| } | |
| .gr-button.subq { | |
| background-color: #d1fae5; | |
| border: none; | |
| padding: 8px 14px; | |
| border-radius: 9999px; | |
| cursor: pointer; | |
| font-size: 14px; | |
| margin: 4px 6px 0 0; | |
| transition: background-color 0.3s, transform 0.1s; | |
| box-shadow: 0 1px 2px rgba(0,0,0,0.05); | |
| } | |
| .gr-button.subq:hover { | |
| background-color: #a7f3d0; | |
| transform: scale(1.05); | |
| } | |
| #floating-config-wrapper { | |
| position: fixed; | |
| bottom: 20px; | |
| right: 20px; | |
| z-index: 9999; | |
| display: inline-flex; | |
| flex-direction: column; | |
| align-items: flex-end; | |
| width: auto; /* 👈 no ocupar ancho completo */ | |
| max-width: 250px; /* 👈 ancho controlado */ | |
| # pointer-events: none; /* 👈 deja pasar clics fuera del botón y panel */ | |
| } | |
| #config-btn { | |
| all: unset; /* 🚫 Quita todos los estilos por defecto */ | |
| display: flex; | |
| justify-content: center; | |
| align-items: center; | |
| width: 40px; | |
| height: 40px; | |
| font-size: 22px; | |
| background-color: transparent; /* 🟢 Sin fondo */ | |
| color: #2ecc71; /* 🟢 Verde icono */ | |
| border-radius: 50%; | |
| cursor: pointer; | |
| transition: background-color 0.2s ease; | |
| } | |
| #config-btn:hover { | |
| background-color: #d1fae5; /* 🟢 Suave fondo al pasar el mouse */ | |
| } | |
| #config-panel { | |
| background-color: #ecfdf5; | |
| border: 1px solid #a7f3d0; | |
| border-radius: 10px; | |
| padding: 12px; | |
| min-width: 220px; | |
| box-shadow: 0 6px 16px rgba(0,0,0,0.1); | |
| font-size: 14px; | |
| color: #064e3b; | |
| } | |
| #zonia-header h1 { | |
| font-family: 'Segoe UI', sans-serif; | |
| font-weight: bold; | |
| } | |
| #zonia-header p { | |
| font-family: 'Segoe UI', sans-serif; | |
| } | |
| .gr-chatbot { | |
| background: white; | |
| border-radius: 16px; | |
| border: 1px solid #d1fae5; | |
| box-shadow: 0 4px 14px rgba(0,0,0,0.05); | |
| padding: 12px; | |
| margin-bottom: 12px; | |
| } | |
| .gr-chatbot .message.user { | |
| background-color: #e0f2f1; | |
| border-radius: 12px; | |
| padding: 10px; | |
| margin: 6px 0; | |
| font-weight: 500; | |
| color: #065f46; | |
| } | |
| .gr-chatbot .message.assistant { | |
| background-color: #ecfdf5; | |
| border-radius: 12px; | |
| padding: 10px; | |
| margin: 6px 0; | |
| color: #065f46; | |
| border-left: 4px solid #10b981; | |
| } | |
| .gr-chatbot .message { | |
| font-family: 'Segoe UI', sans-serif; | |
| font-size: 15px; | |
| line-height: 1.6; | |
| } | |
| .gr-chatbot .avatar { | |
| display: none; /* Oculta los avatares si los hay */ | |
| } | |
| """ | |
| # === Interfaz | |
| with gr.Blocks(css=css) as app: | |
| gr.Markdown( | |
| """ | |
| <div style='padding: 1rem 1.5rem; border-radius: 12px; background: #ecfdf5; border-left: 5px solid #10b981; margin-bottom: 1.5rem'> | |
| <h1 style='margin: 0; font-size: 26px; color: #065f46;'>🤖 ZONIA</h1> | |
| <p style='margin: 0.3rem 0 0; font-size: 15px; color: #065f46;'>Un asistente inteligente para consultar estatutos de posgrado y reglamentos estudiantiles. Esta interfaz es una prueba para evaluación de usuarios.</p> | |
| </div> | |
| """, | |
| elem_id="zonia-header" | |
| ) | |
| config_visible = gr.State(False) | |
| toggle_intent = gr.Checkbox(visible=False, value=True, label="intent_flag") | |
| toggle_subq = gr.Checkbox(visible=False, value=True, label="subq_flag") | |
| chatbot = gr.Chatbot(label="Conversación", type="messages", show_copy_button=True) | |
| state = gr.State([]) | |
| sources_md = gr.Markdown(visible=False) | |
| subq_buttons = [gr.Button(visible=False, elem_classes="subq") for _ in range(10)] | |
| # Input + enviar | |
| with gr.Row(): | |
| msg = gr.Textbox(placeholder="Haz tu pregunta...") | |
| enviar = gr.Button("Enviar", elem_classes="enviar") | |
| # Botón de configuración y panel flotante | |
| with gr.Group(elem_id="floating-config-wrapper"): | |
| show_config_btn = gr.Button("⚙️", elem_id="config-btn") | |
| with gr.Column(visible=False, elem_id="config-panel") as config_panel: | |
| gr.Markdown("### ⚙️ Configuración") | |
| intent_checkbox = gr.Checkbox(label="Activar intención", value=True) | |
| subq_checkbox = gr.Checkbox(label="Activar subpreguntas", value=True) | |
| # Función toggle y sync | |
| def toggle_panel(is_visible): return not is_visible, gr.update(visible=not is_visible) | |
| def sync_flags(i, s): return i, s | |
| show_config_btn.click(toggle_panel, inputs=[config_visible], outputs=[config_visible, config_panel]) | |
| intent_checkbox.change(sync_flags, [intent_checkbox, subq_checkbox], [toggle_intent, toggle_subq]) | |
| subq_checkbox.change(sync_flags, [intent_checkbox, subq_checkbox], [toggle_intent, toggle_subq]) | |
| # Submit por enter | |
| msg.submit(add_message, [state, msg], [state, msg]).then( | |
| bot, [state, toggle_intent, toggle_subq], [chatbot, sources_md] + subq_buttons | |
| ) | |
| # Submit por botón | |
| enviar.click(add_message, [state, msg], [state, msg]).then( | |
| bot, [state, toggle_intent, toggle_subq], [chatbot, sources_md] + subq_buttons | |
| ) | |
| # Subpreguntas | |
| for btn in subq_buttons: | |
| btn.click(lambda txt, h: add_message(h, txt), [btn, state], [state, msg]).then( | |
| bot, [state, toggle_intent, toggle_subq], [chatbot, sources_md] + subq_buttons | |
| ) | |
| app.launch(True) |