Fundamentos de Agentic RAG

Agentic RAG combina técnicas de recuperación de información con modelos generativos de IA mediante agentes inteligentes. Este enfoque optimiza el flujo de trabajo:

• Fase de almacenamiento: Preprocesamiento de documentos mediante limpieza y segmentación
• Fase de consulta: Recuperación contextualizada y generación de respuestas coherentes

Los sistemsa RAG tradicionales persentan limitaciones en calidad de consultas y robustez. Agentic RAG incorpora:

• Reformulación de consultas
• Recuperación multimodal
• Evaluación de calidad
• Decisión condicional

Arquitectura del Sistema

Componentes Principales

• Motor de Búsqueda: Recuperación híbrida (keywords + vectores)
• Agantes Especializados: Reformulador, Evaluador, Generador
• Control de Flujo: Agente condicional para gestión de rutas

# Estructura de directorios
sistema_rag/
├── nucleo/
│   ├── agente_principal.py
│   ├── motor_busqueda.py
│   ├── reformulador.py
│   └── evaluador_calidad.py
├── datos/
│   ├── indice_vectorial
│   └── textos.pkl

Implementación Clave

Agente Condicional Principal

class AgenteFlujoCondicional(BaseAgent):
    def __init__(self, **kwargs):
        self.reformulador = AgenteReformulador()
        self.motor_busqueda = MotorBusquedaLocal()
        self.evaluador = AgenteEvaluadorCalidad()
        super().__init__(sub_agentes=[
            self.reformulador,
            self.motor_busqueda,
            self.evaluador
        ], **kwargs)

    async def ejecutar_flujo(self, contexto):
        # Reformulación de consulta
        consulta_reformulada = await self.reformulador.procesar(contexto)
        
        # Búsqueda local
        resultados = await self.motor_busqueda.recuperar(consulta_reformulada)
        
        # Evaluación de calidad
        evaluacion = await self.evaluador.analizar(resultados)
        
        # Decisión condicional
        if evaluacion['aprobado']:
            return await self.generador.respuesta_directa(resultados)
        else:
            resultados_web = await self.busqueda_web.ejecutar(consulta_reformulada)
            return await self.generador.respuesta_combinada(resultados, resultados_web)

Recuperación Híbrida

class MotorBusquedaHibrido:
    def __init__(self, modelo_embeddings='BGE-M3'):
        self.modelo_embed = cargar_modelo(modelo_embeddings)
        self.indice = FAISS.leer_indice('datos/indice_vectorial')
    
    def buscar_vectores(self, consulta, top_n=5):
        vector = self.modelo_embed.codificar([consulta])[0]
        puntuaciones, indices = self.indice.buscar(vector, top_n)
        return [self.textos[i] for i in indices]
    
    def buscar_keywords(self, consulta, top_n=5):
        tokens = segmentar_texto(consulta)
        return algoritmo_bm25(tokens, top_n)
    
    def recuperar(self, consulta):
        resultados_vec = self.buscar_vectores(consulta)
        resultados_kw = self.buscar_keywords(consulta)
        return combinar_resultados(resultados_vec, resultados_kw)

Evaluación de Calidad

plantilla_evaluacion = """
Evalúe los resultados en:
1. Relevancia (0-10)
2. Precisión legal (0-10)
3. Cobertura (0-10)
4. Utilidad (0-10)

Calificación total: {total}/40
Decisión: {estado} (Aprobado si >=32)
"""

agente_evaluador = AgenteLLM(
    modelo="deepseek-chat",
    instrucciones=plantilla_evaluacion,
    parametros={"temperatura": 0.2, "max_tokens": 1024}
)

Flujo de Trabajo Optimizado

1. Reformulación de consulta inicial
2. Recuperación multimodal (keywords + vectores)
3. Evaluación con umbral del 80%
4. Generación de respuesta directa o con refuerzo web

Ventajas Prácticas

• Reformulación automática mejora precisión en 40%
• Modelo en embudo reduce latencia en respuestas válidas
• Evaluación de calidad garantiza confiabilidad
• Integración de fuentes externas como respaldo