import os
import ast
import gradio as gr
from llama_index.core import (
StorageContext,
load_index_from_storage,
)
import pandas as pd
import nest_asyncio
nest_asyncio.apply()
from logging_manager import LoggingManager
# OpenAI API Key:
OPENAI_API_KEY = os.environ.get('openai')
# HuggingFace Token:
HF_TOKEN = os.environ.get('hf')
# Cohere Rerank Token:
COHERE_TOKEN = os.environ.get('cohere')
# Context:
exec(os.environ.get('context'))
### LOG MANAGER ###
logger = LoggingManager(
repo_name="pharma-IA",
project_id="logs-dataintegrity",
hf_token=HF_TOKEN
)
# Context:
exec(os.environ.get('context2'))
# Interface
css = """
#component-2 * {
font-size: small !important;
}
#component-13 textarea {
background: transparent !important;
}
#component-16 * {
font-size: small !important;
}
#btn_select {
font-size: x-small;
width:140px;
margin:auto;
}
#component-21 {
padding: 0 !important;
}
#component-21 span{
padding: 4px 0 0 8px !important;
font-size: small !important;
}
#component-21 p {
padding: 0 0 0 14px !important;
}
#component-21 input {
caret-color: var(--neutral-400) !important;
pointer-events: auto !important;
}
#component-21 .secondary-wrap {
background: var(--neutral-200) !important;
border-radius: 4px !important;
}
#component-21 .icon-wrap {
padding: 0 !important;
width: 36px !important;
}
#select_list {
padding: 0 !important;
}
#select_list label.selected {
background: var(--secondary-200) !important;
border-color: var(--secondary-400) !important;
}
#select_list label.selected input {
background-color: var(--secondary-400) !important;
border-color: var(--secondary-400) !important;
}
.message-row.bubble.user-row .user{
background-color: var(--secondary-200) !important;
border-color: var(--secondary-400) !important;
}
#markdown table {
font-size: x-small !important;
}
table {
font-size: x-small !important;
padding: 2px 3px !important;
}
.html-container {
padding: 0 !important;
margin: 0 !important;
}
"""
# Lista de choices, excluyendo las últimas dos en la selección inicial
choices_with_tools = [
("EMA", retriever_1_tool),
("FDA", retriever_2_tool),
("Validación de Software", retriever_3_tool),
("Integridad de datos Internacional", retriever_4_tool),
("Estudios Clínicos", retriever_5_tool),
("Sistema MES", retriever_6_tool),
("Sistema LIMS", retriever_7_tool),
("Validación de Software e Integridad de Datos – Argentina", retriever_8_tool),
("Validación de Software e Integridad de Datos – Perú", retriever_9_tool),
("Validación de Software e Integridad de Datos – México", retriever_10_tool),
("Validación de Software e Integridad de Datos – Colombia", retriever_11_tool),
("Validación de Software e Integridad de Datos – Ecuador", retriever_12_tool),
("Validación de Software e Integridad de Datos – Paraguay", retriever_13_tool),
("Validación de Software e Integridad de Datos – Brasil", retriever_14_tool),
("Herramienta de comparación", retriever_summary_tool),
]
# Solo extraer los nombres para mostrarlos en la interfaz
selected_choices = ["EMA", "FDA", "Validación de Software", "Integridad de datos Internacional", "Estudios Clínicos", "Sistema MES", "Sistema LIMS"]
choice_labels = [label for label, _ in choices_with_tools]
# Inicializar el texto de evaluación vacío
result_evals = ""
result_metadata = ""
result_texts = ""
import asyncio
async def process_query(message):
"""Ejecuta el workflow para obtener la respuesta del modelo."""
global selected_choices, chat_history, query
try:
selected_retrievers = [(choice, tool) for choice, tool in choices_with_tools if choice in selected_choices]
retriever_tools = [tool for _, tool in selected_retrievers]
w = RAGWorkflow(timeout=200)
handler = w.run(
query=str(message),
llm_multimodal=gpt_4_1_multimodal,
llm_selector=gpt_4_1,
llm_history=gpt_4_1_mini,
llm_kg=gpt_4_1_mini,
retrieve_tools=retriever_tools,
bm25_top_k=2,
max_tools=3,
rerank_limit=5,
stream=True
)
response = ""
final_dict = None
async for event in handler.stream_events():
if isinstance(event, ProgressEvent):
print("ProgressEvent:", event.msg, flush=True)
response += event.msg
yield ("progress", response, None) # (type, response, final_dict)
await asyncio.sleep(0.01)
elif isinstance(event, StopEvent):
print("StopEvent:", event.result["response"], flush=True)
final_dict = event.result
yield ("final", event.result["response"], final_dict)
except Exception as e:
print(f"Error: {e}", flush=True)
yield ("error", f"Error: {e}", None)
async def llm_response(message, history, profile: gr.OAuthProfile | None):
global result_texts, result_metadata, selected_choices, chat_history
global kg_source_nodes, final_response, query
# Asignar "Usuario no ingresado" si profile es None
user_name = "Usuario no ingresado" if profile is None else profile.name
final_dict = None
chat_history = logger.get_user_history(user_name, 2)
async for chunk_type, chunk_response, chunk_dict in process_query(message):
if chunk_type == "progress":
yield chunk_response
elif chunk_type == "final":
final_dict = chunk_dict
yield chunk_response
elif chunk_type == "error":
yield chunk_response
# After streaming is complete, process the final data if we have it
if final_dict is not None:
try:
final_response = final_dict
query=str(message)
# Extraer la información de los metadatos y textos de la respuesta
result_metadata = "\n".join(extraer_informacion_metadata(final_dict, kg_source_nodes, max_results=20) or [])
result_texts = extraer_textos_metadata(final_dict, max_results=20) or []
# Guardar la conversación en el dataset
logger.save_interaction(message, final_dict["response"], user_name)
logger.save_node_references(message, final_dict["source_nodes"], kg_source_nodes)
except Exception as e:
print(f"Error processing final data: {e}", flush=True)
with gr.Blocks(theme=gr.themes.Base(), css=css) as demo:
# Función para actualizar las choices seleccionadas
def update_selected_choices(choices):
global selected_choices
print("Selección actualizada: " + str(choices))
selected_choices = list(set(choices))
# Alternar la selección
def toggle_all(selected):
if len(selected) == len(choice_labels):
update_selected_choices([])
return []
else:
update_selected_choices(choice_labels)
return choice_labels
# Referencias
def get_ref():
return {simple_ref: gr.Markdown(result_metadata), texts: gr.HTML(str(result_texts))}
# Logs
def get_logs(selected_month):
df = logger.get_audit_trail(selected_month) # Carga el DataFrame
# Eliminar las columnas que no queremos mostrar (incluyendo las nuevas)
columns_to_hide = ['Document Nodes', 'KG Nodes', 'response_node_ids', 'kg_node_ids']
df = df.drop(columns=[col for col in columns_to_hide if col in df.columns], errors='ignore')
# Transformar la columna "User Message" para mostrar solo el texto y los archivos
if "User Message" in df.columns:
def format_user_message(msg):
try:
if isinstance(msg, str):
data = ast.literal_eval(msg)
text = data.get("text", "")
files = data.get("files", [])
if files:
return f"{text} (Adjunto: {', '.join(files)})"
return text
return msg
except:
return msg # Si hay error al parsear, devolver el mensaje original
df["User Message"] = df["User Message"].apply(format_user_message)
# Manejar valores NaN en la columna "Feedback"
if "Feedback" in df.columns:
df["Feedback"] = df["Feedback"].apply(lambda x: "-" if pd.isna(x) else x)
# Resto del código permanece igual...
# Preprocesar el texto Markdown para manejar correctamente los saltos de línea
def preprocess_markdown(text):
if not isinstance(text, str):
return text
# Reemplazar \n por dos espacios seguidos de \n (requerido por Markdown)
text = text.replace('\n', ' \n')
return text
# Función para envolver imágenes en enlaces y controlar tamaño
def process_images(html_content):
from bs4 import BeautifulSoup
soup = BeautifulSoup(html_content, 'html.parser')
for img in soup.find_all('img'):
# Crear enlace padre que abre en nueva pestaña
parent_link = soup.new_tag('a', href=img['src'], target='_blank')
img.wrap(parent_link)
# Aplicar estilos de tamaño máximo
img['style'] = "max-width: 300px; max-height: 300px; width: auto; height: auto; display: block;"
# Añadir indicador de que es clickeable
img['title'] = "Click para ver imagen completa"
return str(soup)
# Convierte la columna "Response" a Markdown con preprocesamiento
if "Response" in df.columns:
df["Response"] = df["Response"].apply(
lambda x: f'{preprocess_markdown(x)}
' if isinstance(x, str) else x
)
# Generar estilos CSS dinámicos para las columnas
column_styles = []
for col in df.columns:
if col == "Response":
# Dar más espacio a la columna Response
column_styles.append(f'.col-{col} {{ min-width: 60%; max-width: 70%; }}')
else:
# Columnas normales con ancho automático pero con máximo
column_styles.append(f'.col-{col} {{ width: auto; max-width: 30%; }}')
# Convierte el DataFrame en HTML con clases específicas por columna
table_html = df.to_html(index=False, escape=False, classes="table",
formatters={col: lambda x: f'{x}
'
for col in df.columns},
na_rep="-")
# Procesar las imágenes en el HTML generado
table_html = process_images(table_html)
# Genera el HTML completo
html_content = f"""
{table_html}
"""
# Escapar comillas dobles para el iframe
html_content_escaped = html_content.replace('"', '"')
return f''
# Grafo
def get_graph():
iframe_grafo = draw_graph()
return {grafo: gr.HTML(iframe_grafo)}
# Evaluaciones
def get_evals():
global result_evals
global final_response
global query
# Verificar si 'final_response' está vacío
if not final_response:
if result_evals:
# Extraer solo el texto de la consulta
try:
# Si query es string, intentar convertir a dict
if isinstance(query, str):
import ast
query_dict = ast.literal_eval(query)
query_text = query_dict.get('text', 'Consulta sin texto')
elif isinstance(query, dict):
query_text = query.get('text', 'Consulta sin texto')
else:
query_text = str(query)
except:
query_text = str(query) # Fallback si hay error en la conversión
return {evals: gr.HTML(f"""
{result_evals}
Esta evaluación corresponde a la consulta: {query_text}
Se necesita una respuesta completa para iniciar la evaluación.
""")}
# Ejecuta la evaluación si final_response está disponible
result_evals = evaluate()
# Reiniciar 'final_response' después de la evaluación
final_response = ""
# Extraer solo el texto de la consulta para el resultado final
try:
# Si query es string, intentar convertir a dict
if isinstance(query, str):
import ast
query_dict = ast.literal_eval(query)
query_text = query_dict.get('text', 'Consulta sin texto')
elif isinstance(query, dict):
query_text = query.get('text', 'Consulta sin texto')
else:
query_text = str(query)
except:
query_text = str(query) # Fallback si hay error en la conversión
# Devolver el resultado de la evaluación
return {
evals: gr.HTML(f"""
{result_evals}
Esta evaluación corresponde a la consulta: {query_text}