"""
=========================================================================
=======
20_rag_supremo.py - EL RAG FINAL INTEGRADO (TODO EN UNO)
=========================================================================
=======
MAPA DEL SISTEMA (Flujo de la Información):
------------------------------------------------
Imagina que este script es un HOSPITAL INTELIGENTE:
1. RECEPCIÓN (Input Usuario):
- El paciente llega y dice: "Me duele el pecho".
2. TRIAJE (Routing - Script 19):
- Un enfermero (Gemma 2B) dice: la categoría CARDIOLOGÍA
- Decide buscar SOLO en el departamento de Cardio.
3. INVESTIGACIÓN (HyDE - Script 17 - Opcional): El experto alucina la
respuesta.
- Antes de mirar los libros, el médico imagina qué síntomas debería
tener.
- "Si le duele el pecho, buscaré 'infarto', 'angina', 'presión'..."
4. BÚSQUEDA (Retrieval - Script 12):
- El becario corre al archivo de Cardiología y trae 10 libros
posibles.
5. REVISIÓN (Reranking - Script 14):
- El Jefe de Servicio (Cross-Encoder) lee los 10 libros a fondo.
- "Este libro habla de dolor de pecho por amor... ¡DESCARTADO!"
- Se queda con los 3 mejores.
6. INFORME (Generación - Script 15):
- El especialista redacta la respuesta final para el paciente
basándose
en esos 3 libros.
-------------------------------------------------------------------------
-------
"""
import os
import chromadb
from sentence_transformers import SentenceTransformer, CrossEncoder
from openai import OpenAI
import logging
from dotenv import load_dotenv
import time
import utils
ENV_PATH = utils.project_root() / ".env"
load_dotenv(dotenv_path=ENV_PATH)
utils.setup_logging() # Configura formato bonito de logs
logger = [Link]("rag_avanzado")
# RUTAS
DB_DIR = str(utils.project_root() / "chroma_db")
[Link](f"Ruta Base de Datos: {DB_DIR}")
COLLECTION_NAME = [Link]("COLLECTION_NAME", "coleccion_pdfs")
[Link](f"COLECCION ES ESTA '{COLLECTION_NAME}' lista en {DB_DIR}")
# MODELOS UNIFICADOS
LLM_API_KEY = [Link]("LLM_API_KEY", "ollama")
LLM_BASE_URL = [Link]("LLM_BASE_URL", "[Link]
MODELO_LLM = [Link]("MODELO_LLM", "deepseek-r1:8b")
# Usamos el modelo rápido para Routing
MODELO_ROUTER = [Link]("MODELO_FAST", "deepseek-r1:1.5b")
MODELO_EMBEDDINGS = [Link]("MODELO_EMBEDDINGS", "Qwen/Qwen3-Embedding-
0.6B")
MODELO_RERANKER = [Link]("MODELO_RERANKER", "BAAI/bge-reranker-v2-m3")
# ROL DE EXPERTO PARA HYDE (Mejorado del Script 17)
SYSTEM_PROMPT_HYDE = """Eres un redactor médico experto escribiendo para
un manual clínico técnico.
Tu objetivo NO es responder al usuario, sino generar un párrafo teórico
denso y rico en terminología médica precisa sobre el tema de la pregunta.
Usa palabras como 'patología', 'sintomatología', 'tratamiento
farmacológico', 'etiología', etc."""
# CATEGORÍAS (Deben coincidir con ingesta)
CATEGORIAS_VALIDAS = [
"Salud Mental",
"Respiratorias",
"Enfermedades Crónicas"
]
#
=========================================================================
===
# 1. COMPONENTE: ROUTER INTELIGENTE (Del Script 19)
#
=========================================================================
===
def router_inteligente(pregunta, client_llm):
"""Clasifica la pregunta en una categoría usando un modelo ligero."""
start_t = [Link]()
prompt = f"""Clasificador Médico. Categorías válidas:
{', '.join(CATEGORIAS_VALIDAS)}
PREGUNTA: "{pregunta}"
INSTRUCCIONES:
- Responde SOLO con el nombre de la categoría.
- Si no encaja, responde "General".
CATEGORÍA:"""
try:
resp = client_llm.[Link](
model=MODELO_ROUTER,
messages=[{"role": "user", "content": prompt}],
temperature=0.0
)
cat = [Link][0].[Link]()
# Limpieza básica
for c in CATEGORIAS_VALIDAS:
if [Link]() in [Link]():
[Link](f" Router: '{pregunta}' -> [{c}]
({[Link]()-start_t:.2f}s)")
return c
[Link](f" Router: '{pregunta}' -> [General] ({[Link]()-
start_t:.2f}s)")
return None
except Exception as e:
[Link](f"Error Router: {e}")
return None
#
=========================================================================
===
# 2. COMPONENTE: HyDE (Del Script 17)
#
=========================================================================
===
def generar_hyde(pregunta, client_llm, categoria=None):
"""Genera un documento hipotético para mejorar la búsqueda
vectorial."""
[Link](" Generando HyDE...")
try:
contexto_extra = f" sobre {categoria}" if categoria else ""
# User Prompt más específico (del Script 17) + Contexto de
Categoría
usuario = f"""Escribe un breve fragmento de manual
médico{contexto_extra} que explique teóricamente el siguiente concepto o
duda.
PREGUNTA USUARIO: {pregunta}
FRAGMENTO TÉCNICO:"""
resp = client_llm.[Link](
model=MODELO_LLM, # Usamos el modelo listo para redactar
messages=[
{"role": "system", "content": SYSTEM_PROMPT_HYDE},
{"role": "user", "content": usuario}
],
temperature=0.7,
max_tokens=250
)
return [Link][0].[Link]
except Exception as e:
[Link](f"HyDE fallo, usando pregunta original: {e}")
return pregunta # Fallback
#
=========================================================================
===
# 3. COMPONENTE: RETRIEVAL + RERANKING (Del Script 15)
#
=========================================================================
===
def recuperar_avanzado(pregunta, col, model_emb, reranker,
filtro_categoria=None, usar_hyde=False, client_llm=None):
# A) QUERY EXPANSION (HyDE)
consulta_vector = pregunta
if usar_hyde and client_llm:
hipotesis = generar_hyde(pregunta, client_llm, filtro_categoria)
consulta_vector = hipotesis # Buscamos con la alucinación experta
[Link](f" (HyDE) Buscando vectores de:
'{hipotesis[:50]}...'")
# B) RETRIEVAL (Bi-Encoder) con FILTRO
q_emb = utils.generar_embeddings(model_emb, [consulta_vector])
filtros_db = None # Por defecto SIN filtro (None, no {})
if filtro_categoria:
filtros_db = {"category": filtro_categoria}
[Link](f" Búsqueda Filtrada: {filtros_db}")
res = [Link](
query_embeddings=q_emb,
n_results=10, # Traemos candidatos de sobra
where=filtros_db
)
docs = res['documents'][0]
metas = res['metadatas'][0]
if not docs: return [], []
# C) RERANKING (Cross-Encoder)
# Siempre rerankeamos contra la PREGUNTA ORIGINAL (no la de HyDE)
pares = [[pregunta, doc] for doc in docs]
scores = [Link](pares)
datos_reranked = sorted(zip(scores, docs, metas), key=lambda x: x[0],
reverse=True)
# Top-3 final
return [(d, m) for s, d, m in datos_reranked[:3]]
#
=========================================================================
===
# 4. COMPONENTE: GENERACIÓN (Del Script 15)
#
=========================================================================
===
def generar_respuesta_stream(pregunta, docs_finales, client_llm):
# Formato de contexto con [DOC X] para que el LLM cite correctamente
# Basado en mejores practicas de [Link]
contexto = ""
for i, (doc, meta) in enumerate(docs_finales, 1):
contexto += f"[DOC {i}] (Fuente: {meta['source']})\n{doc}\n\n"
#
=========================================================================
# SYSTEM PROMPT MEJORADO (Basado en [Link])
#
=========================================================================
# Estructura:
# 1. HIGH-LEVEL INSTRUCTIONS (comportamiento fundamental)
# 2. TONE & PERSONALITY (cómo comunicarse)
# 3. RAG SPECIFICS (citar fuentes, admitir ignorancia)
system_prompt = """# INSTRUCCIONES DEL SISTEMA
Eres un asistente medico experto y empatico que responde preguntas
basandose UNICAMENTE en los documentos proporcionados.
# COMPORTAMIENTO
- Usa SOLO la informacion de los documentos para responder
- Cita las fuentes usando [DOC 1], [DOC 2], etc. al final de cada
afirmacion
- Si la informacion NO esta en los documentos, di: "No tengo informacion
sobre eso"
- NUNCA recetes medicamentos ni des consejos medicos especificos
- NO inventes informacion
# TONO
- Profesional pero empatico y accesible
- Respuestas claras y bien estructuradas
- Si hay multiples fuentes relevantes, mencionalas todas
# EJEMPLO DE RESPUESTA CORRECTA
"Segun los documentos, la depresion puede causar fatiga y perdida de
interes [DOC 1].
El tratamiento suele incluir terapia psicologica [DOC 2]."
"""
user_prompt = f"""# DOCUMENTOS RECUPERADOS (Ordenados por relevancia,
DOC 1 = mas relevante)
{contexto}
# PREGUNTA DEL USUARIO
{pregunta}
# TU RESPUESTA (cita las fuentes con [DOC X]):"""
stream = client_llm.[Link](
model=MODELO_LLM,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
stream=True,
temperature=0.3
)
for chunk in stream:
content = [Link][0].[Link] # contenido del chunk
if content:
yield content # Genera el stream , va mostrando el resultado
#
=========================================================================
===
# MAIN
#
=========================================================================
===
def main():
print("\n" + "="*80)
print(" RAG SUPREMO: Routing + HyDE + Reranking + Chat")
print("="*80)
# 1. SETUP
if not [Link](DB_DIR):
[Link]("No hay base de datos.")
return
client_db = [Link](path=DB_DIR)
col = client_db.get_collection(COLLECTION_NAME)
[Link]("Cargando modelos... (Esto puede tardar un poco)")
model_emb = SentenceTransformer(MODELO_EMBEDDINGS)
reranker = CrossEncoder(MODELO_RERANKER)
client_llm = OpenAI(base_url=LLM_BASE_URL, api_key=LLM_API_KEY)
# Configuración de Usuario
usar_hyde = input("¿Activar HyDE (Query Expansion)? (s/n): ").lower()
== 's'
while True:
pregunta = input("\n Pregunta Médica (o 'salir'): ")
if [Link]() in ['salir', 'exit']: break
t_total = [Link]()
# PASO 1: ROUTING
t1 = [Link]()
categoria = router_inteligente(pregunta, client_llm)
tiempo_routing = [Link]() - t1
# PASO 2 & 3: RETRIEVAL AVANZADO
t2 = [Link]()
resultados = recuperar_supremo(
pregunta, col, model_emb, reranker,
filtro_categoria=categoria,
usar_hyde=usar_hyde,
client_llm=client_llm
)
tiempo_retrieval = [Link]() - t2
if not resultados:
print(" No encontre informacion relevante.")
continue
[Link](f" Tiempos: Routing={tiempo_routing:.2f}s |
Retrieval+Rerank={tiempo_retrieval:.2f}s")
# PASO 4: GENERACIÓN
print(f"\n Generando respuesta (Fuente: {categoria if categoria
else 'Global'})...\n")
full_res = ""
for chunk in generar_respuesta_stream(pregunta, resultados,
client_llm):
print(chunk, end="", flush=True)
full_res += chunk
print("\n" + "-"*80)
if __name__ == "__main__":
main()