RAG (Retrieval Augmented Generation)

La Retrieval Augmented Generation representa una evolución fundamental en el desarrollo de aplicaciones de inteligencia artificial, combinando la capacidad generativa de los modelos de lenguaje con sistemas de recuperación de información. Esta técnica permite que los modelos accedan a conocimiento externo específico y actualizado, superando las limitaciones de su entrenamiento original.

Fundamentos de RAG

El concepto de RAG surge de la necesidad de proporcionar a los modelos de lenguaje acceso a información que no estaba presente durante su entrenamiento o que ha sido actualizada posteriormente. En lugar de depender únicamente del conocimiento interno del modelo, RAG implementa un proceso de dos fases: primero recupera información relevante de una base de conocimiento externa, y luego utiliza esta información como contexto para generar respuestas más precisas y actualizadas.

La arquitectura RAG se basa en tres componentes principales: un sistema de indexación que procesa y almacena documentos, un mecanismo de búsqueda que recupera información relevante, y un generador que produce respuestas utilizando tanto la consulta original como la información recuperada.

from openai import OpenAI

client = OpenAI()

# Ejemplo básico de generación con contexto externo
def generar_respuesta_con_contexto(pregunta, contexto):
    prompt = f"""
    Contexto: {contexto}
    
    Pregunta: {pregunta}
    
    Responde basándote únicamente en el contexto proporcionado.
    """
    
    response = client.responses.create(
        model="gpt-4.1",
        input=prompt
    )
    
    return response.output_text

Embeddings y representación vectorial

Los embeddings constituyen el núcleo técnico de los sistemas RAG, transformando texto en representaciones vectoriales que capturan el significado semántico. OpenAI proporciona modelos de embeddings especializados que convierten documentos y consultas en vectores de alta dimensionalidad, permitiendo comparaciones matemáticas de similitud semántica.

El proceso de vectorización convierte cada fragmento de texto en un vector numérico donde la distancia entre vectores refleja la similitud conceptual entre los textos originales. Esta representación permite realizar búsquedas semánticas eficientes, encontrando contenido relevante incluso cuando no existe coincidencia exacta de palabras clave.

def obtener_embeddings(textos):
    """Genera embeddings para una lista de textos"""
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=textos
    )
    
    return [embedding.embedding for embedding in response.data]

# Ejemplo de uso
documentos = [
    "Python es un lenguaje de programación versátil",
    "Los algoritmos de machine learning requieren datos de calidad",
    "Las bases de datos relacionales utilizan SQL"
]

embeddings = obtener_embeddings(documentos)
print(f"Generados {len(embeddings)} embeddings de dimensión {len(embeddings[0])}")

Procesamiento y segmentación de documentos

La segmentación de documentos representa un paso crítico en la implementación de sistemas RAG efectivos. Los documentos largos deben dividirse en fragmentos manejables que mantengan coherencia semántica mientras optimizan la recuperación de información relevante.

Las estrategias de chunking varían según el tipo de contenido: documentos técnicos pueden segmentarse por secciones lógicas, mientras que textos narrativos requieren división por párrafos o unidades temáticas. El tamaño óptimo de cada fragmento equilibra la especificidad del contenido con la capacidad contextual del modelo generativo.

def segmentar_documento(texto, tamaño_chunk=1000, solapamiento=200):
    """Divide un documento en fragmentos con solapamiento"""
    chunks = []
    inicio = 0
    
    while inicio < len(texto):
        fin = inicio + tamaño_chunk
        
        # Buscar el final de una oración para evitar cortes abruptos
        if fin < len(texto):
            while fin > inicio and texto[fin] not in '.!?':
                fin -= 1
            fin += 1  # Incluir el punto
        
        chunk = texto[inicio:fin].strip()
        if chunk:
            chunks.append(chunk)
        
        inicio = fin - solapamiento
    
    return chunks

# Procesamiento de documento
documento = """
La inteligencia artificial ha revolucionado múltiples industrias. 
Los sistemas de machine learning procesan grandes volúmenes de datos.
Las redes neuronales imitan el funcionamiento del cerebro humano.
"""

fragmentos = segmentar_documento(documento, tamaño_chunk=100)
for i, fragmento in enumerate(fragmentos):
    print(f"Fragmento {i+1}: {fragmento}")

Búsqueda por similitud semántica

La búsqueda semántica utiliza la distancia coseno entre vectores de embeddings para identificar los fragmentos de documento más relevantes para una consulta específica. Este enfoque supera las limitaciones de la búsqueda por palabras clave, capturando relaciones conceptuales complejas.

El algoritmo de similitud calcula la proximidad entre el embedding de la consulta y los embeddings de todos los fragmentos almacenados, ordenando los resultados por relevancia semántica. Los fragmentos con mayor similitud proporcionan el contexto más apropiado para la generación de respuestas.

import numpy as np

def calcular_similitud_coseno(vector1, vector2):
    """Calcula la similitud coseno entre dos vectores"""
    dot_product = np.dot(vector1, vector2)
    norma1 = np.linalg.norm(vector1)
    norma2 = np.linalg.norm(vector2)
    return dot_product / (norma1 * norma2)

def buscar_fragmentos_relevantes(consulta, fragmentos, embeddings_fragmentos, top_k=3):
    """Encuentra los fragmentos más relevantes para una consulta"""
    # Generar embedding de la consulta
    embedding_consulta = obtener_embeddings([consulta])[0]
    
    # Calcular similitudes
    similitudes = []
    for i, embedding_fragmento in enumerate(embeddings_fragmentos):
        similitud = calcular_similitud_coseno(embedding_consulta, embedding_fragmento)
        similitudes.append((i, similitud, fragmentos[i]))
    
    # Ordenar por similitud descendente
    similitudes.sort(key=lambda x: x[1], reverse=True)
    
    return similitudes[:top_k]

# Ejemplo de búsqueda
consulta = "¿Cómo funcionan las redes neuronales?"
resultados = buscar_fragmentos_relevantes(consulta, fragmentos, embeddings)

for indice, similitud, fragmento in resultados:
    print(f"Similitud: {similitud:.3f} - {fragmento}")

Implementación de sistema RAG completo

Un sistema RAG funcional integra todos los componentes anteriores en un flujo cohesivo: indexación de documentos, búsqueda semántica y generación contextual. La implementación debe manejar eficientemente el almacenamiento de embeddings y optimizar las consultas para respuestas en tiempo real.

La arquitectura completa incluye mecanismos de actualización incremental de la base de conocimiento, gestión de metadatos de documentos y estrategias de caché para mejorar el rendimiento. El sistema debe balancear precisión en la recuperación con velocidad de respuesta.

class SistemaRAG:
    def __init__(self):
        self.client = OpenAI()
        self.fragmentos = []
        self.embeddings = []
        self.metadatos = []
    
    def indexar_documento(self, texto, metadatos=None):
        """Indexa un documento completo en el sistema"""
        fragmentos = segmentar_documento(texto)
        embeddings = obtener_embeddings(fragmentos)
        
        for fragmento, embedding in zip(fragmentos, embeddings):
            self.fragmentos.append(fragmento)
            self.embeddings.append(embedding)
            self.metadatos.append(metadatos or {})
    
    def consultar(self, pregunta, top_k=3):
        """Realiza una consulta RAG completa"""
        # Buscar fragmentos relevantes
        resultados = buscar_fragmentos_relevantes(
            pregunta, self.fragmentos, self.embeddings, top_k
        )
        
        # Construir contexto
        contexto = "\n\n".join([fragmento for _, _, fragmento in resultados])
        
        # Generar respuesta
        prompt = f"""
        Contexto relevante:
        {contexto}
        
        Pregunta: {pregunta}
        
        Proporciona una respuesta precisa basada en el contexto.
        """
        
        response = self.client.responses.create(
            model="gpt-4.1",
            input=prompt
        )
        
        return {
            'respuesta': response.output_text,
            'fuentes': [f"Fragmento {i}" for i, _, _ in resultados],
            'similitudes': [sim for _, sim, _ in resultados]
        }

# Uso del sistema RAG
rag = SistemaRAG()

# Indexar documentos
documento_ia = """
La inteligencia artificial comprende algoritmos que permiten a las máquinas 
realizar tareas que tradicionalmente requieren inteligencia humana. 
Los modelos de machine learning aprenden patrones de los datos.
"""

rag.indexar_documento(documento_ia, {'fuente': 'manual_ia.pdf'})

# Realizar consulta
resultado = rag.consultar("¿Qué es machine learning?")
print(f"Respuesta: {resultado['respuesta']}")
print(f"Fuentes: {resultado['fuentes']}")

Optimización y mejores prácticas

La optimización de sistemas RAG requiere atención a múltiples aspectos: calidad de la segmentación, relevancia de la búsqueda y coherencia de la generación. Las métricas de evaluación incluyen precisión en la recuperación, relevancia del contexto y calidad de las respuestas generadas.

Las mejores prácticas incluyen experimentación con diferentes tamaños de fragmento, ajuste de parámetros de similitud y implementación de filtros de calidad para el contenido recuperado. El monitoreo continuo del rendimiento permite identificar oportunidades de mejora en la pipeline RAG.

def evaluar_calidad_rag(sistema_rag, preguntas_test, respuestas_esperadas):
    """Evalúa la calidad de un sistema RAG"""
    resultados = []
    
    for pregunta, respuesta_esperada in zip(preguntas_test, respuestas_esperadas):
        resultado = sistema_rag.consultar(pregunta)
        
        # Evaluar similitud entre respuesta generada y esperada
        embeddings_eval = obtener_embeddings([
            resultado['respuesta'], 
            respuesta_esperada
        ])
        
        similitud = calcular_similitud_coseno(embeddings_eval[0], embeddings_eval[1])
        
        resultados.append({
            'pregunta': pregunta,
            'similitud_respuesta': similitud,
            'fuentes_utilizadas': len(resultado['fuentes'])
        })
    
    return resultados

# Configuración avanzada para diferentes tipos de contenido
def configurar_rag_especializado(tipo_contenido):
    """Configura parámetros RAG según el tipo de contenido"""
    configuraciones = {
        'tecnico': {
            'tamaño_chunk': 800,
            'solapamiento': 100,
            'top_k': 5
        },
        'narrativo': {
            'tamaño_chunk': 1200,
            'solapamiento': 200,
            'top_k': 3
        },
        'codigo': {
            'tamaño_chunk': 500,
            'solapamiento': 50,
            'top_k': 2
        }
    }
    
    return configuraciones.get(tipo_contenido, configuraciones['tecnico'])