Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| from huggingface_hub import InferenceClient | |
| import chromadb | |
| from langchain_community.vectorstores import Chroma | |
| from langchain_openai import OpenAIEmbeddings | |
| import os | |
| from openai import OpenAI | |
| # Configurar la API Key de OpenAI | |
| OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") | |
| # Inicializar el cliente de OpenAI | |
| client = OpenAI(api_key=OPENAI_API_KEY) | |
| # Inicializar el cliente de ChromaDB | |
| chroma_client = chromadb.PersistentClient(path="chroma_db_crminbox/content/chroma_db") # Ajusta la ruta según tu entorno | |
| # Cargar la base de datos de Chroma como un vector store | |
| vectorstore = Chroma( | |
| client=chroma_client, | |
| collection_name="docs", # Nombre de la colección en Chroma | |
| embedding_function=OpenAIEmbeddings(model="text-embedding-3-small", openai_api_key=OPENAI_API_KEY) | |
| ) | |
| # Crear un retriever | |
| retriever = vectorstore.as_retriever() | |
| # Función para obtener extractos relevantes | |
| def obtener_extractos(pregunta): | |
| docs_relevantes = retriever.invoke(pregunta) | |
| return [(doc.page_content, doc.metadata.get("url", "URL no disponible")) for doc in docs_relevantes] | |
| def extract_unique_citations_paragraph(response): | |
| """Extrae todas las URL y títulos únicos de un objeto Response, sin duplicados.""" | |
| citations = [] | |
| seen = set() # Para evitar duplicados (usaremos la URL como clave única) | |
| for item in getattr(response, "output", []): | |
| # Solo buscamos en mensajes del asistente | |
| if getattr(item, "type", None) == "message": | |
| for block in getattr(item, "content", []) or []: | |
| if getattr(block, "type", None) == "output_text": | |
| for ann in getattr(block, "annotations", []) or []: | |
| if getattr(ann, "type", None) == "url_citation": | |
| url = getattr(ann, "url", None) | |
| title = getattr(ann, "title", None) | |
| # Evita duplicados basados en URL | |
| if url and url not in seen: | |
| seen.add(url) | |
| citations.append({ | |
| "title": title, | |
| "url": url, | |
| "start": getattr(ann, "start_index", None), | |
| "end": getattr(ann, "end_index", None) | |
| }) | |
| return citations | |
| def respond(message, history: list[tuple[str, str]], domain_table): | |
| """Genera una respuesta basada en el historial y documentos relevantes.""" | |
| # --- Limpiar el DataFrame o lista de dominios --- | |
| if domain_table is None: | |
| allowed_domains = [] | |
| elif hasattr(domain_table, "values"): # DataFrame | |
| allowed_domains = [d for d in domain_table.iloc[:, 0].dropna().tolist() if d] | |
| else: # lista de listas | |
| allowed_domains = [d[0] for d in domain_table if d and d[0]] | |
| system_message = '''Eres un agente especializado con alta competencia en investigación médica. Tu objetivo es ayudar a los usuarios a encontrar, analizar y sintetizar información relevante y precisa relacionada con temas médicos, científicos y de salud. Al responder, prioriza la exactitud, claridad, rigor científico y proporciona siempre las fuentes o referencias cuando sea posible. Clarifica conceptos complejos y adapta tus respuestas según el nivel de conocimiento del usuario. | |
| Para tu investigación usarás siempre tu herrammienta de búsqueda en línea. | |
| # Detalles adicionales | |
| - Mantente actualizado en cuanto a literatura y evidencia médica reciente (hasta tu fecha de conocimiento). | |
| - No proporciones diagnósticos médicos personalizados ni recomendaciones clínicas específicas, pero sí puedes guiar sobre dónde encontrar información fiable. | |
| - Estructura las respuestas inicialmente explicando el razonamiento y la evidencia antes de presentar conclusiones o resúmenes. | |
| # Pasos sugeridos | |
| 1. Analiza la consulta médica o científica recibida. | |
| 2. Identifica las fuentes de información y evidencia relevantes. | |
| 3. Sintetiza la información resaltando los hallazgos clave y referencias. | |
| 4. Presenta la respuesta, primero explicando el razonamiento y las evidencias encontradas, y luego dando la conclusión o resumen. | |
| # Formato de salida | |
| La respuesta debe estar bien estructurada, comenzando con la explicación y razonamiento seguido por las conclusiones o respuestas directas. Si es pertinente, incluye referencias en formato [autor, año] o enlaces. | |
| # Ejemplo | |
| Consulta del usuario: "¿Cuáles son los tratamientos más eficaces para la diabetes tipo 2 según la evidencia actual?" | |
| Respuesta: | |
| Primero, revisé guías internacionales y revisiones sistemáticas recientes para identificar las opciones terapéuticas recomendadas para diabetes tipo 2. Diversas fuentes coinciden en que la metformina sigue siendo el fármaco de primera elección debido a su eficacia, seguridad y bajo coste ([ADA, 2023]). Además, existen nuevas familias farmacológicas como los inhibidores de SGLT2 y los agonistas del GLP-1, que han demostrado beneficios cardiovasculares adicionales en estudios recientes ([Zinman et al., 2015]; [Marso et al., 2016]). La selección del tratamiento debe individualizarse según perfil del paciente y comorbilidades. | |
| En conclusión, el abordaje terapéutico debe basarse en guías actuales, iniciando generalmente con metformina y considerando otras opciones según características individuales y evidencia más reciente. | |
| # Notas | |
| - Si la consulta es ambigua o muy general, clarifica la pregunta antes de responder. | |
| - No inventes evidencia ni referencias; si la información es incierta, indícalo claramente. | |
| - Si el usuario requiere información técnica (por ejemplo, sobre mecanismos moleculares), adapta el nivel de detalle según lo solicitado. | |
| Recuerda: Tu enfoque principal es brindar investigación y síntesis experta en el ámbito médico.''' | |
| # Obtener documentos relevantes desde ChromaDB | |
| contexto = ""#obtener_extractos(message) | |
| # Construir el mensaje del sistema con el contexto directamente incluido | |
| system_message_final = f"""{system_message} | |
| Información relevante extraída de los documentos, en caso de que estos documentos tenga la informacion que necesitas, no olvides tomar | |
| el historial de conversacion con el usuario: | |
| {contexto} | |
| """ | |
| messages = [{"role": "system", "content": system_message}] | |
| # Agregar historial del chat | |
| for val in history: | |
| if val[0]: | |
| messages.append({"role": "user", "content": val[0]}) | |
| if val[1]: | |
| messages.append({"role": "assistant", "content": val[1]}) | |
| # Agregar la nueva pregunta del usuario | |
| messages.append({"role": "user", "content": message}) | |
| # --- Construir dinámicamente la tool --- | |
| web_search_tool = { | |
| "type": "web_search", | |
| "search_context_size": "high", | |
| } | |
| if allowed_domains: # 👈 Solo agregar si hay dominios | |
| web_search_tool["filters"] = {"allowed_domains": allowed_domains} | |
| # --- Configurar include dinámico --- | |
| include_params = ["web_search_call.action.sources"] if allowed_domains else [] | |
| # --- Llamada a la API con streaming --- | |
| with client.responses.stream( | |
| model="gpt-5-mini", | |
| input=messages, | |
| tools=[web_search_tool], | |
| include=include_params, # 👈 también condicional | |
| ) as stream: | |
| response = "" | |
| for event in stream: | |
| if event.type == "response.output_text.delta": | |
| response += event.delta | |
| yield response | |
| elif event.type == "response.completed": | |
| response_stream = stream.get_final_response() | |
| citations = extract_unique_citations_paragraph(response_stream) | |
| response += "\n\n🔗 Fuentes citadas:" | |
| for c in citations: | |
| response += f"\n• {c['title'] or 'Sin título'} → {c['url']}" | |
| yield response | |
| break | |
| def add_domain(domains, new_domain): | |
| # Si domains está vacío o None, inicializarlo como lista vacía | |
| if domains is None: | |
| domains = [] | |
| # Si llega como DataFrame (Gradio lo hace internamente) | |
| elif hasattr(domains, "values"): | |
| domains = domains.values.tolist() | |
| # Si llega como lista con una fila vacía, limpiarla | |
| domains = [d for d in domains if d and d[0]] | |
| # Agregar el nuevo dominio si no está vacío | |
| if new_domain and new_domain.strip(): | |
| domains.append([new_domain.strip()]) | |
| return domains, "" # limpia el textbox | |
| with gr.Blocks() as demo: | |
| gr.Markdown("### 🧠 Chat Médico AI") | |
| gr.Markdown("Este es un chatbot especializdo en la búsqueda de información médica con fuentes verídicas. Puedes colocar que busque sólo en dominios específicos.") | |
| with gr.Row(): | |
| new_domain = gr.Textbox(label="Búscar en sitios web específicos (opcional)") | |
| add_button = gr.Button("➕ Añadir") | |
| domain_table = gr.Dataframe( | |
| headers=["Dominios"], | |
| row_count=(1, "dynamic"), | |
| col_count=(1, "fixed"), | |
| label="Sitios web permitidos" | |
| ) | |
| custom_chatbot = gr.Chatbot(height=500, label="Chat Médico AI") | |
| chat = gr.ChatInterface( | |
| fn=respond, | |
| chatbot=custom_chatbot, | |
| additional_inputs=[domain_table], | |
| ) | |
| add_button.click(add_domain, [domain_table, new_domain], [domain_table, new_domain]) | |
| if __name__ == "__main__": | |
| demo.launch() | |