RAG: Arquitectura y diseño para sistemas de producción
La generación asistida por recuperación (RAG) ha surgido como un patrón arquitectónico dominante para construir aplicaciones sofisticadas basadas en LLM. Al fundamentar un modelo en una base de conocimiento externa y verificable, el RAG mitiga las alucinaciones, permite el acceso a datos privados o en tiempo real, y proporciona un mecanismo claro para la atribución de la fuente. Para los directores de tecnología y líderes de ingeniería, dominar la línea de procesamiento RAG no es simplemente un ejercicio académico; es un imperativo estratégico para desbloquear una IA generativa fiable y de nivel empresarial.
Este artículo proporciona una guía completa, desde cero, para diseñar e implementar un sistema RAG (Retrieval-Augmented Generation) listo para la producción. Evitaremos los frameworks de alto nivel para exponer los mecanismos fundamentales, centrándonos en las decisiones arquitectónicas, los compromisos de rendimiento y el código práctico necesario para construir una solución robusta. Implementaremos este sistema utilizando Python, aprovechando la base de datos PostgreSQL ubicua con la extensión pgvectorpgvectorextensión para la búsqueda vectorial y la API de OpenAI para sus potentes modelos.
Servicios de Ingeniería de LLM y IA
Ofrecemos una completa gama de soluciones impulsadas por la IA, que incluyen IA generativa, visión artificial, aprendizaje automático, procesamiento del lenguaje natural y automatización con IA.
La Arquitectura Central: Dos Flujos de Trabajo Distintos
Un sistema de RAG se entiende mejor como dos procesos separados, pero conectados: el Proceso de indexación sin conexióny elPipeline de inferencia en línea.
- Pipeline de indexación sin conexión: Este es un proceso preparatorio y asíncrono responsable de ingerir documentos de origen, convertirlos en un formato de búsqueda (embeddings vectoriales) y almacenarlos en una base de datos especializada. Este proceso se ejecuta cada vez que se necesita crear o actualizar la base de conocimiento.
- Pipeline de inferencia en línea: Este es el flujo de trabajo en tiempo real y orientado al usuario. Recibe una consulta del usuario, busca en la base de conocimiento indexada el contexto relevante y utiliza ese contexto junto con la consulta original para generar una respuesta fundamentada a partir de un LLM.
Las decisiones arquitectónicas clave en esta etapa incluyen:
- Modelo de Inserción: Este modelo traduce el texto en vectores de alta dimensión. La elección impacta en la calidad y el costo de la recuperación. Utilizaremos el modelo
text-embedding-3-smallde OpenAI debido a su equilibrio entre rendimiento y eficiencia de costos. - Base de Datos Vectorial: Esta base de datos debe almacenar y consultar de forma eficiente los vectores de alta dimensión. Si bien las bases de datos vectoriales dedicadas como Pinecone o Weaviate son excelentes, el uso de PostgreSQL con
pgvectorpermite a muchas organizaciones aprovechar la infraestructura y la experiencia operativa existentes, reduciendo significativamente la complejidad arquitectónica. - LLM: El componente generativo que sintetiza la respuesta final. Utilizaremos el modelo
gpt-4ode OpenAI debido a sus capacidades avanzadas de razonamiento e instrucción.
Implementación Parte 1: La tubería de indexación sin conexión
El objetivo de esta plataforma es poblar nuestra base de datos vectorial de PostgreSQL. Esto implica cargar documentos, dividirlos en fragmentos manejables, generar embeddings y almacenarlos.
1. Configuración de la base de datos con pgvector
Primero, asegúrese de tener PostgreSQL instalado con la extensiónpgvector habilitada.
-- Connect to your PostgreSQL instance and run this command
CREATE EXTENSION IF NOT EXISTS vector;
-- Create a table to store the document chunks and their embeddings
CREATE TABLE document_chunks (
id SERIAL PRIMARY KEY,
document_name TEXT NOT NULL,
chunk_text TEXT NOT NULL,
embedding VECTOR(1536) -- 1536 is the dimension for text-embedding-3-small
);
-- Create an index for efficient similarity search
-- HNSW (Hierarchical Navigable Small World) is generally preferred for its speed-accuracy trade-off.
-- The parameters lists_to_check and ef_construction are tunable for performance.
CREATE INDEX ON document_chunks
USING HNSW (embedding vector_cosine_ops);
Nota de arquitectura:
Optamos por un índice HNSW. En comparación con un índice IVFFlat, HNSW suele ofrecer un mejor rendimiento de consulta (menor latencia) a cambio de un proceso de construcción más lento e intensivo en memoria. Para la mayoría de las aplicaciones en tiempo real, esta es la compensación adecuada.
2. Carga y particionamiento de datos
La segmentación efectiva es crucial para la calidad de la recuperación. Los fragmentos que son demasiado pequeños carecen de contexto, mientras que los fragmentos que son demasiado grandes introducen ruido. El Divisor de texto recursivo para caracteresRecursiveCharacterTextSplitter
Aquí hay una implementación en Python para cargar y dividir archivos de texto.
# requirements: pip install langchain openai psycopg2-binary
import os
import openai
import psycopg2
from langchain.text_splitter import RecursiveCharacterTextSplitter
# --- Configuration ---
OPENAI_API_KEY = "YOUR_OPENAI_API_KEY"
DB_CONNECTION_STRING = "postgresql://user:password@host:port/dbname"
DOCUMENTS_PATH = "./source_documents/"
EMBEDDING_MODEL = "text-embedding-3-small"
# --- Initialize Clients ---
openai.api_key = OPENAI_API_KEY
def process_and_embed_documents():
"""
Loads documents, chunks them, generates embeddings, and stores them in PostgreSQL.
"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # The character length of each chunk
chunk_overlap=200, # The number of characters to overlap between chunks
length_function=len,
)
conn = psycopg2.connect(DB_CONNECTION_STRING)
cur = conn.cursor()
for filename in os.listdir(DOCUMENTS_PATH):
if filename.endswith(".txt"):
filepath = os.path.join(DOCUMENTS_PATH, filename)
with open(filepath, 'r') as f:
document_text = f.read()
print(f"Processing {filename}...")
chunks = text_splitter.split_text(document_text)
# Generate embeddings in batches for efficiency
response = openai.embeddings.create(
input=chunks,
model=EMBEDDING_MODEL
)
embeddings = [item.embedding for item in response.data]
# Insert into database
for i, chunk in enumerate(chunks):
cur.execute(
"INSERT INTO document_chunks (document_name, chunk_text, embedding) VALUES (%s, %s, %s)",
(filename, chunk, embeddings[i])
)
conn.commit()
cur.close()
conn.close()
print("Indexing complete.")
if __name__ == '__main__':
process_and_embed_documents()
Parte 2: Implementación: El flujo de trabajo de inferencia en línea
Este proceso se ejecuta en tiempo real cuando un usuario envía una consulta. Implica la incorporación de la consulta, la obtención del contexto relevante de la base de datos, la creación de una instrucción precisa y la llamada al modelo de lenguaje.
Servicios de Ingeniería de LLM y IA
Ofrecemos una completa gama de soluciones impulsadas por IA, que incluyen IA generativa, visión artificial, aprendizaje automático, procesamiento del lenguaje natural y automatización basada en IA.
1. Búsqueda de incrustaciones y recuperación de contexto
La consulta del usuario debe convertirse en un vector utilizando el mismo modelo de incrustación utilizado para la indexación. Luego, utilizamos este vector para realizar una búsqueda de similitud en nuestra tabla de fragmentos de documentos. El operador "= " de pgvector calcula la distancia coseno.exactamente igualModelo de incrustación utilizado para la indexación. Luego, utilizamos este vector para realizar una búsqueda de similitud en nuestrafragmentos de documentotabla.<=>operador depgvectorcalcula la distancia coseno.
import openai
import psycopg2
# --- Configuration (reuse from previous section) ---
# ...
def retrieve_context(query: str, top_k: int = 5) -> list[str]:
"""
Embeds the query and retrieves the top_k most relevant document chunks.
"""
# 1. Embed the user's query
response = openai.embeddings.create(
input=[query],
model=EMBEDDING_MODEL
)
query_embedding = response.data[0].embedding
# 2. Retrieve relevant context from PostgreSQL
conn = psycopg2.connect(DB_CONNECTION_STRING)
cur = conn.cursor()
# Find the most similar chunks using cosine distance
cur.execute(
"""
SELECT chunk_text FROM document_chunks
ORDER BY embedding <=> %s
LIMIT %s
""",
(query_embedding, top_k)
)
results = cur.fetchall()
cur.close()
conn.close()
# Return the text of the chunks
return [row[0] for row in results]
Nota sobre el rendimiento:
El parámetro LIMIT (top-k) es un parámetro crítico para ajustar. Un valor más pequeño de k es más rápido, pero puede hacer que se pierdan información relevantes. Un valor más grande de k proporciona más contexto, pero puede aumentar el ruido y los costes de los tokens de la LLM. Empezar con k=5 es una base razonable.
2. Generación de prompts mejorada
Este es el paso de "ampliación". Construimos un nuevo prompt que instruye explícitamente al LLM para que responda basándose únicamente en el contexto que acabamos de obtener. Este es el mecanismo principal para prevenir la alucinación.únicamente en el contexto que acabamos de recuperar. Este es el mecanismo principal para prevenir la alucinación.
def construct_prompt(query: str, context: list[str]) -> str:
"""
Constructs a prompt for the LLM with the retrieved context.
"""
context_str = "\n\n---\n\n".join(context)
prompt = f"""
You are a highly intelligent AI assistant. Your task is to answer the user's question based exclusively on the provided context.
- Do not use any external knowledge.
- If the answer is not present within the context, you must state: "I cannot answer this question based on the provided information."
Provided Context:
{context_str}
User's Question:
{query}
Answer:
"""
return prompt
3. Generación de la respuesta final
El paso final es enviar el prompt mejorado al LLM.
def generate_response(query: str):
"""
The main RAG pipeline function.
"""
# 1. Retrieve context
retrieved_context = retrieve_context(query, top_k=5)
# 2. Construct the prompt
final_prompt = construct_prompt(query, retrieved_context)
# 3. Generate response from LLM
response = openai.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": final_prompt}
],
temperature=0.0 # Set to 0 for deterministic, fact-based answers
)
return response.choices[0].message.content
# --- Example Usage ---
if __name__ == '__main__':
user_query = "What are the key performance considerations for the HNSW index?"
final_answer = generate_response(user_query)
print(f"Query: {user_query}\n")
print(f"Answer: {final_answer}")
Decisión arquitectónica:Establecer temperatura=0.0es crucial para los sistemas de preguntas y respuestas basados en hechos. Obliga al modelo a ser más determinista y a ceñirse estrechamente al contexto proporcionado, reduciendo las salidas creativas (y potencialmente inexactas).
Consideraciones y optimizaciones avanzadas de producción
Si bien la implementación anterior es funcional, desplegarla a gran escala requiere una mayor consideración.
- Evaluación: Un sistema RAG es tan bueno como la calidad de su recuperación. Implemente un flujo de trabajo de evaluación utilizando un conjunto de datos "de referencia" de tuplas (pregunta, respuesta esperada, contexto). Las métricas clave incluyen Precisión del contexto(¿es el contexto recuperado relevante?), Recuperación del contexto (¿se recuperó todo el contexto necesario?), y Fiabilidad(¿el resultado final se mantiene dentro del contexto?).
- Búsqueda híbrida: La búsqueda vectorial pura a veces puede fallar en consultas que contienen palabras clave, acrónimos o códigos específicos. Complementar la búsqueda vectorial con un algoritmo de búsqueda de palabras clave tradicional, como BM25, puede proporcionar un sistema de recuperación más robusto. Esto implica ejecutar dos búsquedas en paralelo y combinar los resultados.
- Reordenamiento: La recuperación inicial de los k mejores resultados está optimizada para la velocidad. Para mejorar la relevancia, se puede utilizar un modelo de reordenamiento de segunda etapa (normalmente un codificador cruzado). Este modelo toma la consulta y cada uno de los k documentos recuperados y calcula una puntuación de relevancia más precisa, reordenando los resultados antes de pasarlos al LLM.
- Escalabilidad:
- Base de datos: Para PostgreSQL, utilice la agrupación de conexiones (por ejemplo, PgBouncer) y considere las réplicas de lectura para manejar cargas de consulta elevadas.
- Inferencia: Las APIs de LLM son un cuello de botella. Implemente el almacenamiento en caché para consultas idénticas. Para un alto rendimiento, investigue la posibilidad de alojar modelos de código abierto en una infraestructura dedicada de GPU utilizando herramientas como Triton Inference Server.
Conclusión
Construir un sistema RAG desde cero revela la compleja interacción entre el procesamiento de datos, la búsqueda vectorial y el modelado del lenguaje. Al desglosar el proceso en sus componentes principales de indexación e inferencia, los líderes de ingeniería pueden tomar decisiones arquitectónicas informadas que equilibren el rendimiento, el costo y la mantenibilidad.
La pila presentada aquí—Python, pgvector, y la API de OpenAI—ofrece un punto de partida poderoso y accesible. Sin embargo, el verdadero arte de implementar RAG reside en la evaluación continua y la aplicación iterativa de técnicas avanzadas como el re-ordenamiento y la búsqueda híbrida para satisfacer las necesidades específicas de su caso de uso.