import nest_asyncio
nest_asyncio.apply()
from llama_index.core import (
VectorStoreIndex,
ServiceContext,
SimpleDirectoryReader,
load_index_from_storage,
)
from llama_index.core.storage import StorageContext
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.prompts import PromptTemplate
from llama_index.core.response_synthesizers import TreeSummarize
from llama_index.core.query_pipeline import InputComponent
from llama_index.core.indices.knowledge_graph import KGTableRetriever
from llama_index.legacy.vector_stores.faiss import FaissVectorStore
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import Settings
import openai
import os
from github import Github
from datetime import datetime
import gradio as gr
import pandas as pd
# Context:
exec(os.environ.get('context'))
##### Graph start ##########
import networkx as nx
import matplotlib.pyplot as plt
from PIL import Image
from io import BytesIO
def draw_graph():
global kg_data
G = nx.DiGraph()
for source, relation, target in kg_data:
G.add_edge(source, target, label=relation)
# Utilizar spring_layout para mejorar la disposición de los nodos
pos = nx.spring_layout(G)
plt.figure(figsize=(12, 8))
# Ajustar el tamaño de los nodos
nx.draw(G, pos, with_labels=True, node_color='skyblue', node_size=400, edge_color='k', linewidths=1, font_size=8, font_weight='bold')
# Ajustar el tamaño de las flechas y el espaciado entre ellas
edge_labels = {}
for source, target, data in G.edges(data=True):
if 'label' in data:
edge_labels[(source, target)] = data['label']
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_size=7, font_weight='normal')
plt.title("Graph")
plt.axis('off')
buf = BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)
plt.close()
return Image.open(buf)
##### Graph end ##########
##### Refs start ##########
import re
def extraer_informacion_metadata(respuesta, max_results=10):
# Obtener source_nodes de la respuesta
source_nodes = respuesta.source_nodes
# Obtener page_labels, file_names y scores de source_nodes
page_file_info = [
f"Página {node.node.metadata.get('page_label', '')} del archivo {node.node.metadata.get('file_name', '')} (Relevance: {node.score:.6f} - Id: {node.node.id_})\n\n"
for node in source_nodes if node.score <= 1 # Excluir nodos con score > 1
]
# Limitar la cantidad de resultados
page_file_info = page_file_info[:max_results]
return page_file_info
import html
def extraer_textos_metadata(respuesta, max_results=10):
# Obtener source_nodes de la respuesta
source_nodes = respuesta.source_nodes
# Obtener información de página, archivo y texto de cada nodo
page_file_text_info = []
for node in source_nodes:
if node.score <= 1: # Excluir nodos con score > 1
page_label = node.node.metadata.get('page_label', '')
file_name = node.node.metadata.get('file_name', '')
text = node.node.text.strip()
# Escapar caracteres especiales en el texto
escaped_text = html.escape(text)
# Formatear con HTML
formatted_text = f"""
Página {page_label} del archivo {file_name}
{escaped_text}
"""
page_file_text_info.append(formatted_text.strip()) # Quitar espacios adicionales
# Limitar la cantidad de resultados
page_file_text_info = page_file_text_info[:max_results]
return ''.join(page_file_text_info) # Devolver como un string limpio
##### Refs end ##########
##### Logs start ##########
import pandas as pd
from datasets import load_dataset, Dataset, DatasetDict
from huggingface_hub import login, HfApi, file_exists, hf_hub_download, list_repo_files
# HuggingFace Token:
HF_TOKEN = os.environ.get('hf')
# Definiciones
repo_name = "pharma-IA"
project_id = "gmpcolombia"
def save_to_dataset(user_message, response_text, user):
current_month = datetime.now().strftime('%Y-%m')
filename = f"logs_{current_month}.csv"
repo_id = f"{repo_name}/logs-{project_id}"
if file_exists(repo_id=repo_id, filename=f"{filename}", repo_type="dataset", token=HF_TOKEN):
local_filepath = hf_hub_download(
repo_id=repo_id,
filename=f"{filename}",
repo_type="dataset",
token=HF_TOKEN
)
df = pd.read_csv(local_filepath)
else:
df = pd.DataFrame(columns=["timestamp", "user_message", "response_text", "flag", "user"])
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
new_data = pd.DataFrame([{
"timestamp": timestamp,
"user_message": user_message,
"response_text": response_text,
"flag": "",
"user": user
}])
df = pd.concat([df, new_data], ignore_index=True)
df.to_csv(filename, index=False)
api = HfApi()
api.upload_file(
path_or_fileobj=filename,
path_in_repo=f"{filename}",
repo_id=repo_id,
token=HF_TOKEN,
repo_type="dataset"
)
def normalize_text(text):
return text.strip().lower()
def print_like_dislike(x: gr.LikeData):
#print(f"Value: {x.value}")
#print(f"Liked: {x.liked}")
if x is not None:
text_value = x.value if isinstance(x.value, str) else x.value.get('value', '')
current_month = datetime.now().strftime('%Y-%m')
filename = f"logs_{current_month}.csv"
repo_id = f"{repo_name}/logs-{project_id}"
if file_exists(repo_id=repo_id, filename=f"{filename}", repo_type="dataset", token=HF_TOKEN):
local_filepath = hf_hub_download(
repo_id=repo_id,
filename=f"{filename}",
repo_type="dataset",
token=HF_TOKEN
)
df = pd.read_csv(local_filepath)
#print(df.head()) # Verifica el contenido del archivo CSV
normalized_value = normalize_text(text_value)
df['normalized_response_text'] = df['response_text'].apply(normalize_text)
response_indices = df.index[df['normalized_response_text'].str.contains(normalized_value, na=False, regex=False)].tolist()
print(f"Response Indices: {response_indices}")
if response_indices:
response_index = response_indices[-1]
print(f"Updating index: {response_index} with value: {x.liked}")
# Solo actualiza el valor de 'flag'
df['flag'] = df['flag'].astype(object)
df.at[response_index, 'flag'] = str(x.liked)
df = df.drop(columns=['normalized_response_text'])
df.to_csv(filename, index=False)
api = HfApi()
api.upload_file(
path_or_fileobj=filename,
path_in_repo=f"{filename}",
repo_id=repo_id,
token=HF_TOKEN,
repo_type="dataset"
)
else:
print("No matching response found to update.")
else:
print(f"File {filename} does not exist in the repository.")
else:
print("x is None.")
def save_evals_to_dataset(query, faithfulness_score, ans_relevancy_score, ctx_relevancy_score):
current_month = datetime.now().strftime('%Y-%m')
filename = f"logs_{current_month}.csv"
repo_id = f"{repo_name}/logs-{project_id}"
if file_exists(repo_id=repo_id, filename=f"{filename}", repo_type="dataset", token=HF_TOKEN):
local_filepath = hf_hub_download(
repo_id=repo_id,
filename=f"{filename}",
repo_type="dataset",
token=HF_TOKEN
)
df = pd.read_csv(local_filepath)
else:
print(f"File {filename} does not exist in the repository.")
return
# Normalizamos el query para la comparación
normalized_query = normalize_text(query).lower() # Convertimos a minúsculas
# Buscamos la última entrada que coincida con el query, convirtiendo a minúsculas en la comparación
matching_indices = df.index[df['user_message'].str.lower().str.contains(normalized_query, na=False, regex=False)].tolist()
if matching_indices:
last_index = matching_indices[-1] # Tomamos la última coincidencia
# Agregamos los puntajes a las columnas correspondientes
df.at[last_index, 'groundedness'] = faithfulness_score
df.at[last_index, 'answer_rel'] = ans_relevancy_score
df.at[last_index, 'context_rel'] = ctx_relevancy_score
df.to_csv(filename, index=False)
api = HfApi()
api.upload_file(
path_or_fileobj=filename,
path_in_repo=f"{filename}",
repo_id=repo_id,
token=HF_TOKEN,
repo_type="dataset"
)
else:
print("No matching query found in the dataset.")
# Función para verificar si un archivo existe en el repositorio
def file_exists(repo_id, filename, repo_type="dataset", token=None):
files = list_repo_files(repo_id=repo_id, repo_type=repo_type, token=token)
return filename in files
# Función para cargar las hojas de auditoría disponibles
def load_available_logs():
repo_id = f"{repo_name}/logs-{project_id}"
files = list_repo_files(repo_id=repo_id, repo_type="dataset", token=HF_TOKEN)
# Filtramos los archivos CSV con formato 'logs_YYYY-MM.csv'
available_months = [f.split('_')[1].replace('.csv', '') for f in files if f.startswith('logs_')]
return available_months
# Cargar los logs del mes seleccionado
def load_audit_trail(selected_month):
filename = f"logs_{selected_month}.csv"
repo_id = f"{repo_name}/logs-{project_id}"
if file_exists(repo_id=repo_id, filename=filename, repo_type="dataset", token=HF_TOKEN):
local_filepath = hf_hub_download(
repo_id=repo_id,
filename=filename,
repo_type="dataset",
token=HF_TOKEN
)
df = pd.read_csv(local_filepath)
# Convertir el campo 'timestamp' a una cadena con el formato UTC-0
df["timestamp"] = pd.to_datetime(df["timestamp"]).dt.strftime('%Y-%m-%d %H:%M:%S UTC-0')
# Ordenar por la columna timestamp de forma descendente
df = df.sort_values(by="timestamp", ascending=False)
# Renombrar las columnas para visualización
df = df.rename(columns={
"timestamp": "Marca de Tiempo",
"user_message": "Mensaje Usuario",
"response_text": "Respuesta",
"flag": "Etiqueta",
"user": "Usuario",
"groundedness": "Groundedness",
"answer_rel": "Answer Relev.",
"context_rel": "Context Relev."
})
return df
else:
return pd.DataFrame(columns=["Marca de Tiempo", "Mensaje Usuario", "Respuesta", "Etiqueta", "Usuario"])
##### Logs end ##########
##### Evaluate start ##########
from llama_index.core.evaluation import FaithfulnessEvaluator
from llama_index.core.evaluation import RelevancyEvaluator
from llama_index.core.evaluation import AnswerRelevancyEvaluator
from llama_index.core.evaluation import ContextRelevancyEvaluator
final_response = ""
query = ""
def ctx_relevancy_eval():
global final_response
global query
# Verificamos si 'final_response' tiene el atributo 'source_nodes'
if not hasattr(final_response, 'source_nodes'):
raise AttributeError("El objeto 'final_response' no tiene un atributo 'source_nodes'.")
# Obtener los source_nodes de la respuesta
source_nodes = final_response.source_nodes
# Extraer los textos de los source nodes
contexts = []
for node in source_nodes:
if node.score <= 1: # Excluir nodos con score > 1
text = node.node.text.strip()
contexts.append(text)
if not contexts:
raise ValueError("No se encontraron textos en los source nodes.")
evaluator = ContextRelevancyEvaluator(llm=gpt4omini)
evaluation_result = evaluator.evaluate(query=query, contexts=contexts)
# Extraer el puntaje de la evaluación
relevancy_score = evaluation_result.score
return relevancy_score
def faithfulness_eval():
global final_response
global query
# Verificamos si 'final_response' tiene el atributo 'source_nodes'
if not hasattr(final_response, 'source_nodes'):
raise AttributeError("El objeto 'final_response' no tiene un atributo 'source_nodes'.")
# Obtener los source_nodes de la respuesta
source_nodes = final_response.source_nodes
# Extraer los textos de los source nodes
contexts = []
for node in source_nodes:
if node.score <= 1: # Excluir nodos con score > 1
text = node.node.text.strip()
contexts.append(text)
if not contexts:
raise ValueError("No se encontraron textos en los source nodes.")
evaluator = FaithfulnessEvaluator(llm=llm)
eval_result = evaluator.evaluate(query=query, response=final_response.response, contexts=contexts)
print("Groundedness: " + str(eval_result.score) + " - " + str(eval_result.passing) + "\n")
print("Respuesta: " + str(final_response) + "\n\n----------")
print("Contexts: " + str(contexts) + "\n\n----------")
return float(eval_result.score)
def ans_relevancy_eval():
global final_response
global query
evaluator = AnswerRelevancyEvaluator(llm=gpt4omini)
eval_result = evaluator.evaluate(query=query, response=final_response.response)
return float(eval_result.score)
def evaluate():
global query
# Evaluaciones y escalado
faithfulness_score = round(faithfulness_eval() * 5, 1) # Redondear a un decimal
ans_relevancy_score = round(ans_relevancy_eval() * 5, 1) # Redondear a un decimal
ctx_relevancy_score = round(ctx_relevancy_eval() * 5, 1) # Redondear a un decimal
# Llamamos a save_evals_to_dataset
save_evals_to_dataset(query, faithfulness_score, ans_relevancy_score, ctx_relevancy_score)
def get_color(value):
if value <= 1.6667:
return '#f07b61' # Rojo
elif value <= 3.3334:
return '#f3e076' # Amarillo
else:
return '#84fa57' # Verde
color1 = get_color(faithfulness_score)
color2 = get_color(ans_relevancy_score)
color3 = get_color(ctx_relevancy_score)
html_output = f"""
{faithfulness_score}
Groundedness
{ans_relevancy_score}
Answer Relevance
{ctx_relevancy_score}
Context Relevance
"""
return html_output
##### Evaluate end ##########
chat_history_engine = []
result_metadata = ""
result_texts = ""
result_evals = ""
css = """
.block {
background: rgba(245, 247, 249, 0.7) !important;
}
#component-1 {
background: transparent !important;
}
.block.accordion > button:first-of-type span {
font-size: medium !important;
font-weight: bold !important;
}
.examples .block {
background: transparent !important;
}
table {
font-size: x-small !important;
}
"""
choices_with_tools = [
("[1] Guía BPL", retriever_1_tool),
("[2] Guía BPM Medicamentos", retriever_2_tool),
("[3] MANUAL DE NORMAS TÉCNICAS DE CALIDAD", retriever_3_tool),
("[4] Resolución 1160 - GMP - 2016", retriever_4_tool),
# ("[5] Resolución 2266 - DECRETO FITOTERAPÉUTICOS - 2004", retriever_5_tool),
("[6] Resolución 3619 - GLP - 2013", retriever_6_tool),
("[7] Resolución 005107 - FITOTERAPEUTICOS - 2005", retriever_7_tool),
("[8] Resolución 3690 - 2016", retriever_8_tool),
("[9] Resolución 3157 - 2018", retriever_9_tool),
("[10] Preguntas y respuestas Res. 3157 - 2018", retriever_10_tool),
("General", retriever_all),
]
# Solo extraer los nombres para mostrarlos en la interfaz
selected_choices = [label for label, _ in choices_with_tools]
choice_labels = [label for label, _ in choices_with_tools]
with gr.Blocks(theme='sudeepshouche/minimalist', css=css) as demo:
# Actualizar las choices seleccionadas
def update_selected_choices(choices):
global selected_choices
selected_choices = choices
def get_ref():
return {mkdn: gr.Markdown(result_metadata), texts: gr.HTML(str(result_texts))}
def get_logs(selected_month):
df = load_audit_trail(selected_month)
return df
def get_evals():
global result_evals
global final_response
global query
# Verificar si 'final_response' está vacío
if not final_response:
# Si no hay final_response pero hay un resultado previo en result_evals, devolverlo
if result_evals:
return {evals: gr.HTML(f"""
{result_evals}
Esta evaluación corresponde a la consulta: {query}
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 = ""
# Devolver el resultado de la evaluación
return {evals: gr.HTML(f"""{result_evals}"""), eval_accord: gr.Accordion(elem_classes="accordion", label="Evaluaciones", open=True)}
def refresh(chat_history):
global kg_data
global chat_history_engine
global result_metadata
kg_data = []
chat_history_engine = []
result_metadata = ""
chat_history = [[None, None]]
return chat_history
def summarize_assistant_messages(chat_history: List[ChatMessage]) -> List[ChatMessage]:
# Encontrar la anteúltima respuesta del asistente
assistant_messages = [msg for msg in chat_history if msg.role == MessageRole.ASSISTANT]
if len(assistant_messages) < 2:
return chat_history # No hay suficientes mensajes del asistente para resumir
anteultima_respuesta = assistant_messages[-2]
# Usar GPT-3.5 para generar un resumen de la anteúltima respuesta del asistente
prompt = Prompt(f"Responder SOLO con un resumen del siguiente texto: \n\n{anteultima_respuesta.content}")
response = llm.predict(prompt)
# Crear un nuevo ChatMessage con el resumen como contenido y el rol de asistente
summarized_message = ChatMessage(content=response, role=MessageRole.ASSISTANT)
# Reconstruir el historial de chat reemplazando la anteúltima respuesta del asistente con el resumen
new_chat_history = [msg if msg != anteultima_respuesta else summarized_message for msg in chat_history]
return new_chat_history
def respond(message, chat_history):
global chat_history_engine
global result_metadata
global result_texts
global final_response
global query
global selected_choices
# Inicializar el historial de chat si está vacío con el mensaje del usuario actual
if not chat_history:
chat_history = [[message, ""]]
else:
# Agregar el mensaje actual al historial de chat
chat_history.append([message, ""])
# Resumir los mensajes previos en chat_history_engine
chat_history_engine = summarize_assistant_messages(chat_history_engine)
# Engine:
exec(os.environ.get('engine'))
# Generar la respuesta usando el motor de chat
response = chat_engine.stream_chat(message, chat_history=chat_history_engine)
# Extraer la información de los metadatos y textos de la respuesta
metadata_info = extraer_informacion_metadata(response, max_results=10)
texts_info = extraer_textos_metadata(response, max_results=10)
if metadata_info:
result_metadata = "\n".join(metadata_info)
if texts_info:
result_texts = texts_info
# Procesar la respuesta generada y agregarla al historial del chat
for text in response.response_gen:
chat_history[-1][1] += text
yield "", chat_history # Actualiza el historial del chat y retorna el nuevo estado
# Guardar la conversación en el dataset
save_to_dataset(message, chat_history[-1][1], "no-ingresado")
final_response = response
query = message
gr.Markdown("""
# PharmaWise GMP Colombia Chat 4.7
Realiza preguntas a tus datos y obtén al final del texto las paginas y documentos utilizados generar tu responder.
""")
with gr.Row():
with gr.Column():
chatbot = gr.Chatbot(show_label=False, show_copy_button=True, ) #layout="panel"
pregunta = gr.Textbox(show_label=False, autofocus=True, placeholder="Realiza tu consulta...")
pregunta.submit(respond, [pregunta, chatbot], [pregunta, chatbot])
with gr.Row():
btn_send = gr.Button(value="Preguntar", variant="primary")
clear = gr.Button(value="Limpiar")
with gr.Row(elem_classes="examples"):
gr.Examples(label="Ejemplos", examples=["Implementación de la res. 3157 de 2018"], inputs=[pregunta])
with gr.Column():
with gr.Accordion(elem_classes="accordion", label="Bases de datos del conocimiento", open=False):
# CheckboxGroup para seleccionar bases de datos
chx = gr.CheckboxGroup(choices=choice_labels, value=choice_labels, label="Bases de datos del conocimiento")
chx.select(fn=update_selected_choices, inputs=chx)
with gr.Accordion(elem_classes="accordion", label="Referencias", open=True):
mkdn = gr.Markdown()
with gr.Row():
btn_graph = gr.Button(value="Grafo")
btn_ref = gr.Button(value="Referencias")
btn_eval = gr.Button(value="Evaluar")
eval_accord = gr.Accordion(elem_classes="accordion", label="Evaluaciones", open=False)
with eval_accord:
evals = gr.HTML()
gr.Markdown("""| **Evaluador** | **Qué mide** | **Ejemplo de uso** | **Diferencias clave** |
|-----------------------|-------------------------------------------------------|-------------------------------------------------------|--------------------------------------------------------|
| **Groundedness** | Qué tan fundamentada está la respuesta en el contexto. | ¿La respuesta está respaldada por el contexto proporcionado? | Se enfoca en la relación entre la respuesta y el contexto. |
| **Answer Relevance** | Qué tan relevante es la respuesta para la consulta. | ¿La respuesta es pertinente a lo que el usuario preguntó? | Se centra en la relevancia de la respuesta ante la consulta. |
| **Context Relevance** | Qué tan relevante es el contexto recuperado para la consulta. | ¿El contexto obtenido es relevante para la consulta del usuario? | Se enfoca en la pertinencia del contexto en relación con la consulta. |
""")
with gr.Row():
grafo = gr.Image(label="Grafo", show_share_button=False)
with gr.Accordion(elem_classes="accordion", label="Audit trail", open=False):
with gr.Row():
with gr.Column():
available_months = load_available_logs()
default_month = available_months[-1] if available_months else None
dropdown = gr.Dropdown(choices=available_months, label="Seleccionar mes", value=default_month)
btn_logs = gr.Button(value="Actualizar")
with gr.Column():
gr.Markdown()
with gr.Column():
gr.Markdown()
# Define un DataFrame con un ancho fijo para response_text
logs_df = gr.DataFrame(headers=["Marca de Tiempo", "Mensaje Usuario", "Respuesta", "Etiqueta", "Usuario"], wrap=True, line_breaks=True)
with gr.Accordion(elem_classes="accordion", label="Referencias ampliadas", open=False):
texts = gr.HTML()
btn_logs.click(fn=get_logs, inputs=[dropdown], outputs=[logs_df])
btn_ref.click(fn=get_ref, outputs=[mkdn, texts])
btn_eval.click(fn=get_evals, outputs=[evals, eval_accord])
btn_send.click(respond, [pregunta, chatbot], [pregunta, chatbot])
btn_graph.click(draw_graph, outputs=[grafo])
clear.click(refresh, inputs=[chatbot], outputs=[chatbot])
chatbot.like(print_like_dislike, None, None)
demo.queue(default_concurrency_limit=20)
demo.launch()