Arquitectura de un Sistema RAG Two-Stage: Guía Técnica

Arquitectura de un Sistema RAG Two-Stage: Guía Técnica

¿Qué es un sistema RAG Two-Stage?

Un sistema RAG Two-Stage es una arquitectura que separa claramente la etapa de recuperación de contexto de la etapa de generación de respuesta, permitiendo personalización más fina, modularidad y mayor control sobre las respuestas generadas.

• Stage 1: Retrieval: Se obtiene información contextual relevante (con o sin embeddings)
• Stage 2: Generation: Se utiliza un LLM para generar una respuesta o estructura, utilizando el contexto anterior

Flujo general del sistema

Usuario → Parser JSON → Vector DB / Contexto → LLM → Respuesta estructurada → Ejecución backend

Componentes del sistema

3.1. Input del usuario

Entrada libre en lenguaje natural. Ejemplo:

"¿Cuáles fueron las ventas totales de la sucursal Abasto esta semana?"

3.2. Módulo de Parsing semántico

Este módulo convierte la pregunta en una estructura tipo JSON:

{
  "funcion": "obtenerVentasSucursal",
  "parametros": {
    "sucursal": "Abasto",
    "fechaInicio": "2024-07-22",
    "fechaFin": "2024-07-28"
  },
  "resumen": "Ventas totales en la sucursal Abasto esta semana"
}

Este parsing puede hacerse con:

• Un LLM instruido con un prompt especializado
• Un modelo finetuneado para structured output
• Expresiones regulares (para campos simples)
• Combinación híbrida

3.3. Vectorización y base de datos de embeddings

El corpus o base de conocimiento se vectoriza usando modelos como:

• all-MiniLM-L6-v2
• text-embedding-ada-002
• instructor-xl (de HuggingFace)

Y se guarda en:

• FAISS
• ChromaDB
• Pinecone (cloud)
• Weaviate

3.4. Recuperación (Retriever)

Consulta vectorial:

1. Se vectoriza la pregunta del usuario
2. Se calcula similitud con la base de embeddings
3. Se selecciona top-k documentos

3.5. Razonamiento (Reader / Generador)

Se forma el prompt:

Contexto:
[documento1]
[documento2]

Pregunta:
¿...?

Instrucción:
Responde en formato JSON: {"funcion": "...", "parametros": {..}, "resumen": "..."}

Este paso puede hacerse con:

• OpenAI (gpt-4, gpt-3.5-turbo)
• HuggingFace (mistral, llama, zephyr)
• Modelos localizados con transformers + vllm / llama.cpp

3.6. Composición de la respuesta final

Se recibe la estructura JSON, y se ejecuta en el backend la función correspondiente. Por ejemplo:

if datos["funcion"] == "obtenerVentasSucursal":
    return obtenerVentasSucursal(**datos["parametros"])

Estructura JSON esperada para ejecución de funciones

Modelo de salida estándar para integración backend:

{
  "funcion": "nombre_funcion",
  "parametros": {
    "clave1": "valor1",
    "clave2": "valor2"
  },
  "resumen": "explicación humana"
}

Consideraciones en la implementación de cada componente

5.1. Parsing e interpretación de intenciones

• Usa ejemplos y prompts pocos (few-shot) si no tenés entrenamiento
• Evitá confiar solo en regex si el input es ambiguo o complejo

Ejemplo de prompt para GPT:

Convierte la siguiente pregunta en JSON con nombre de función, parámetros clave-valor, y resumen:

Pregunta: ¿Cuál es la ganancia total de hoy?
Respuesta esperada: {
  "funcion": "obtenerGananciaTotal",
  "parametros": {
    "fecha": "2024-07-23"
  },
  "resumen": "Ganancia total de hoy"
}

5.2. Embeddings y bases de datos vectoriales

Instalar FAISS:

pip install faiss-cpu

Crear vector DB:

from sentence_transformers import SentenceTransformer
import faiss

model = SentenceTransformer("all-MiniLM-L6-v2")
vectores = model.encode(lista_textos)
index = faiss.IndexFlatL2(len(vectores[0]))
index.add(vectores)

5.3. Llamadas API y frameworks útiles

Para acceso LLM:

• openai, transformers, llama-cpp, vllm

Para parsing con LLM:

• guidance, outlines, jsonformer

Para gestión de RAG:

• langchain, llama-index, Haystack, r2r

5.4. Comparativas de herramientas Python

Función	Regex	Embeddings
Extraer campos	✅	❌
Detectar intención	❌ limitado	✅ con LLM
Generalizar	❌	✅
Contexto externo	❌	✅

Buenas prácticas de diseño e implementación

• Separar cada etapa en funciones reutilizables
• Controlar que el JSON generado esté validado antes de ejecutarlo
• Limitar el contexto a 3-5 documentos relevantes
• Caching de embeddings para mejorar velocidad
• Usar logs para comparar pregunta vs. función ejecutada
• Medir calidad de parseo: exactitud y cobertura

Ejemplo completo de flujo

# Paso 1: Parsing de pregunta
respuesta_llm = {
  "funcion": "ventasSucursal",
  "parametros": {
    "sucursal": "Abasto",
    "fechaInicio": "2024-07-01",
    "fechaFin": "2024-07-07"
  },
  "resumen": "Ventas en Abasto primera semana de julio"
}

# Paso 2: Ejecutar función
def ventasSucursal(sucursal, fechaInicio, fechaFin):
    # consulta SQL
    pass

# Paso 3: Llamar y retornar
output = ventasSucursal(**respuesta_llm["parametros"])

Ejemplos prácticos en Python

1. 🔍 Parser de Preguntas a JSON estructurado

from transformers import pipeline

# Modelo de clasificación o prompting adaptado
qa_parser = pipeline("text2text-generation", model="google/flan-t5-large")

def parse_question_to_json(pregunta: str) -> dict:
    prompt = f"Extraé la intención, función y parámetros de esta pregunta: {pregunta}"
    respuesta = qa_parser(prompt)[0]['generated_text']
    
    # Suponiendo salida en formato similar a:
    # {"funcion": "obtenerGananciaTotal", "parametros": {"fecha": "2025-07-23"}, "resumen": "Solicita la ganancia total de hoy."}
    return eval(respuesta)  # ⚠️ Eval solo si tenés confianza en la fuente. Reemplazar con `json.loads` si es posible.

2. 🔎 Retriever con generación dinámica de SQL

import sqlite3  # o `psycopg2`, `mysql.connector`, etc.

def ejecutar_sql(query: str, parametros=()):
    conn = sqlite3.connect("bd.db")
    cursor = conn.cursor()
    cursor.execute(query, parametros)
    resultados = cursor.fetchall()
    conn.close()
    return resultados

def retriever_obtenerGananciaTotal(parametros):
    fecha = parametros.get("fecha")
    query = "SELECT SUM(ganancia) FROM ventas WHERE fecha = ?"
    return ejecutar_sql(query, (fecha,))

3. ⚙️ Selector de función (etapa de ejecución)

funciones_disponibles = {
    "obtenerGananciaTotal": retriever_obtenerGananciaTotal,
    "ventasPorSucursal": lambda p: ejecutar_sql(
        "SELECT * FROM ventas WHERE sucursal = ? AND fecha BETWEEN ? AND ?",
        (p["sucursal"], p["fechaInicio"], p["fechaFin"])
    )
}

def ejecutar_funcion(json_input):
    funcion = json_input.get("funcion")
    parametros = json_input.get("parametros", {})
    if funcion in funciones_disponibles:
        return funciones_disponibles[funcion](parametros)
    else:
        return {"error": f"Función '{funcion}' no definida"}

4. 💬 Manejo de historial y continuidad de conversación

historial_conversaciones = {}

def guardar_historial(usuario_id, pregunta, respuesta, contexto):
    if usuario_id not in historial_conversaciones:
        historial_conversaciones[usuario_id] = []
    historial_conversaciones[usuario_id].append({
        "pregunta": pregunta,
        "respuesta": respuesta,
        "contexto": contexto
    })

def obtener_contexto(usuario_id):
    return historial_conversaciones.get(usuario_id, [])[-3:]  # últimos 3 intercambios

5. � Chunking eficiente (para tu base de conocimiento)

def chunk_text(texto, max_palabras=100):
    palabras = texto.split()
    chunks = []
    for i in range(0, len(palabras), max_palabras):
        chunk = " ".join(palabras[i:i+max_palabras])
        chunks.append(chunk)
    return chunks

Recomendación: No cortes frases o párrafos a la mitad. Considerá dividir por puntuación.

6. ✅ Control de calidad de embeddings

from sentence_transformers import SentenceTransformer, util

modelo = SentenceTransformer("all-MiniLM-L6-v2")

def validar_calidad_embedding(texto, umbral=0.7):
    emb = modelo.encode(texto, convert_to_tensor=True)
    referencia = modelo.encode("Referencia de calidad deseada", convert_to_tensor=True)
    similitud = util.pytorch_cos_sim(emb, referencia).item()
    return similitud >= umbral

Recursos y librerías recomendadas

• sentence-transformers
• faiss
• openai
• langchain
• Haystack
• guidance
• outlines
• jsonformer