RAG con pgvector: Arquitectura a Escala
Retrieval-Augmented Generation (RAG) se ha convertido en el patrón dominante para aplicaciones de IA que necesitan conocimiento actualizado y específico. Pero pasar de un prototype a un sistema de producción con millones de documentos no es trivial. Aquí es donde pgvector y PostgreSQL brillan.
El Problema de Escalar RAG
Cuando construimos un RAG inicial, todo parece simple:
# Prototype: 1,000 documents, todo en memoria
documents = load_documents("/data/*.pdf")
embeddings = model.encode(documents)
query = "¿Cuál es el proceso de aprobación?"
results = search_similar(query, embeddings, k=5)
response = llm.generate(query, results)
En producción te enfrentarás a: memoria agotada, latencia inaceptable, costos exponenciales de embeddings y dificultad para actualizar el conocimiento.
Por qué pgvector + PostgreSQL
La combinación de pgvector con PostgreSQL ofrece ventajas únicas:
Arquitectura Base de un RAG Escalable
Esquema de Base de Datos
-- Tabla de documentos con vector embeddings
CREATE TABLE documents (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
title TEXT NOT NULL,
content TEXT NOT NULL,
embedding vector(1536), -- OpenAI text-embedding-ada-002
metadata JSONB,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- Índice vectorial con HNSW
CREATE INDEX documents_embedding_idx
ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- Particionamiento por fecha para 10M+ documentos
CREATE TABLE documents_partitioned (
id UUID,
title TEXT NOT NULL,
content TEXT NOT NULL,
embedding vector(1536),
metadata JSONB,
created_at TIMESTAMPTZ,
updated_at TIMESTAMPTZ,
PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (created_at);
-- Crear particiones mensuales
CREATE TABLE documents_2026_07 PARTITION OF documents_partitioned
FOR VALUES FROM ('2026-07-01') TO ('2026-08-01');
CREATE TABLE documents_2026_08 PARTITION OF documents_partitioned
FOR VALUES FROM ('2026-08-01') TO ('2026-09-01');
Pipeline de Ingesta
Extraer y procesar documentos
Chunking inteligente
El chunking es crítico para la calidad del RAG. No uses chunks fijos de 500 tokens.
from langchain.text_splitter import RecursiveCharacterTextSplitter
def smart_chunk(content, max_length=1000, overlap=200):
splitter = RecursiveCharacterTextSplitter(
chunk_size=max_length,
chunk_overlap=overlap,
separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]
)
return splitter.create_documents([content])
# Resultado: chunks que respetan párrafos y oraciones
chunks = smart_chunk(document.content)
Embedding y batch insertion
Procesa en batches de 100-500 embeddings para optimizar costos y latencia.
async def batch_insert_embeddings(chunks, batch_size=100):
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i+batch_size]
embeddings = await embed_batch([c.page_content for c in batch])
await pg_pool.execute("""
INSERT INTO documents (title, content, embedding, metadata)
SELECT $1, $2, $3, $4
""", *[(c.metadata.get('title'), c.page_content, e, c.metadata)
for c, e in zip(batch, embeddings)])
Optimizaciones de Rendimiento
1. HNSW Tuning
HNSW (Hierarchical Navigable Small World) es el algoritmo de búsqueda vectorial por defecto en pgvector. Los parámetros clave:
-- m = número máximo de conexiones por nodo (16-64 default: 16)
-- ef_construction = calidad del índice al construir (40-128 default: 64)
CREATE INDEX docs_hnsw ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 32, ef_construction = 128);
Mayor m y ef_construction = mejor precisión pero más memoria y build time lento. Para producción: m=32, ef_construction=128.
2. Eficiencia en Queries
Usa SET hnsw.ef_search para balance entre precisión y latencia:
async def search_similar(query_embedding, k=5, ef_search=40):
await pg_pool.execute(f"SET hnsw.ef_search = {ef_search}")
results = await pg_pool.fetch("""
SELECT id, title, content,
1 - (embedding <=> $1) as similarity
FROM documents
ORDER BY embedding <=> $1
LIMIT $2
""", query_embedding, k)
return results
3. Prefiltering con Metadata
Filtra antes de la búsqueda vectorial para reducir el espacio de búsqueda:
async def search_with_filters(query_embedding, filters):
# Filtro SQL antes del ranking vectorial
where_clause = " AND ".join([
f"metadata->{key} = '{value}'"
for key, value in filters.items()
])
results = await pg_pool.fetch(f"""
SELECT id, title, content,
1 - (embedding <=> $1) as similarity
FROM documents
WHERE {where_clause}
ORDER BY embedding <=> $1
LIMIT $2
""", query_embedding, 10)
return results
# Uso: buscar solo en documentos de 2026
results = await search_with_filters(
query_embedding,
{"year": "2026", "department": "engineering"}
)
Sharding Horizontal para Multi-Tenant
Para sistemas SaaS con múltiples tenants, el sharding horizontal es crítico:
class TenantRouter:
def __init__(self, config):
self.pools = {
tenant_id: create_pool(config[tenant_id]["database_url"])
for tenant_id in config
}
async def search(self, tenant_id, query_embedding):
pool = self.pools[tenant_id]
results = await pool.fetch("""
SELECT id, title, content,
1 - (embedding <=> $1) as similarity
FROM documents
ORDER BY embedding <=> $1
LIMIT 5
""", query_embedding)
return results
# Uso en endpoint de RAG multi-tenant
router = TenantRouter(tenant_config)
results = await router.search("tenant-123", query_embedding)
Sharding horizontal = mejor aislamiento y escalabilidad, pero mayor complejidad de gestión. Considéralo solo cuando el aislamiento de datos es crítico (compliance, regulaciones).
Monitorización y Observabilidad
Métricas clave
from prometheus_client import Counter, Histogram
search_latency = Histogram(
'rag_search_latency_seconds',
'Latencia de búsqueda vectorial',
['tenant_id']
)
similarity_scores = Histogram(
'rag_similarity_scores',
'Distribución de similitud de resultados',
['tenant_id']
)
document_count = Gauge(
'rag_document_count',
'Total de documentos indexados',
['tenant_id']
)
Logging estructurado
import structlog
logger = structlog.get_logger()
async def search_with_logging(query, tenant_id):
start_time = time.time()
results = await search_similar(query_embedding)
latency = time.time() - start_time
logger.info(
"rag_search_completed",
tenant_id=tenant_id,
query_length=len(query),
results_count=len(results),
avg_similarity=np.mean([r['similarity'] for r in results]),
latency_ms=latency * 1000
)
return results
Arquitectura de Recuperación Híbrida
Combina búsqueda vectorial con búsqueda tradicional (BM25/keyword) para mejores resultados:
async def hybrid_search(query, tenant_id):
# 1. Búsqueda vectorial (semantic)
vector_results = await search_similar(
embed(query),
k=10,
tenant_id=tenant_id
)
# 2. Búsqueda keyword (exact match)
keyword_results = await pg_pool.fetch("""
SELECT id, title, content,
ts_rank_cd(text_search, query) as rank
FROM documents,
plainto_tsquery('spanish', $1) query
WHERE text_search @@ query
ORDER BY rank DESC
LIMIT 10
""", query, tenant_id)
# 3. Reciprocal Rank Fusion (RRF)
def reciprocal_rank_fusion(vector_res, keyword_res, k=60):
scores = {}
for rank, doc in enumerate(vector_res):
scores[doc['id']] = 1 / (k + rank + 1)
for rank, doc in enumerate(keyword_res):
scores[doc['id']] += 1 / (k + rank + 1)
return sorted(scores.items(), key=lambda x: x[1], reverse=True)
final_results = reciprocal_rank_fusion(vector_results, keyword_results)
return final_results[:5]
En mi experiencia, la búsqueda híbrida mejora la relevancia en 30-40% vs búsqueda puramente vectorial, especialmente para queries con entidades específicas o términos técnicos.
Lecciones Aprendidas
- Chunking inteligente importa más que el modelo de embeddings. Respetar la estructura del documento (párrafos, secciones) mejora la calidad drásticamente.
- Prefilter con metadata reduce latencia significativamente. Filtra antes de buscar vectorialmente cuando tengas millones de documentos.
- HNSW tuning es un trade-off constante. No uses ef_search=100 para todas las queries. Adáptalo según el caso de uso.
- Hybrid search = resultados más robustos. Combina búsqueda semántica con keyword para cubrir gaps de ambos enfoques.
- Monitorización desde el inicio. Sin métricas, no puedes optimizar. Mide latencia, similitud de resultados y coverage de retrieval.
Conclusión
RAG con pgvector y PostgreSQL te permite escalar desde miles hasta millones de documentos sin abandonar el ecosistema SQL que ya conoces. La combinación de partitioning, sharding y optimizaciones de HNSW crea una arquitectura que crece con tu aplicación.
Si estás migrando de una vector database especializada, pgvector te ofrece la ventaja de mantener tus datos relacionales junto a tus embeddings en un solo sistema. Menos complejidad de infraestructura, mejor consistencia de datos y costos predecibles.
¿Qué sigue? En el próximo post de la serie "Arquitectura de Software Avanzada", exploraremos SOLID en microservicios: cuándo aplicar principios de diseño y cuándo el pragmatismo gana.
¿Tienes experiencia con RAG en producción? ¿Qué desafíos encontraste al escalar? Comparte en los comentarios.