Indexación: generación de embeddings y almacenamiento de fragmentos
Generará un embedding para cada fragmento mediante la API de embeddings de OpenAI y hará upsert de los vectores resultantes con metadatos en un vector store, creando un índice consultable de sus documentos.
Indexación: generación de embeddings y almacenamiento de fragmentos es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 3 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Engineering Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Engineering Academy incluye 4 lecciones en total.
La etapa de indexación: descripción general
Después de cargar y fragmentar los documentos, llega a la etapa de indexación: convertir los fragmentos de texto en embeddings vectoriales y almacenarlos en una base de datos vectorial consultable. Este es el último paso sin conexión antes de poder responder a las consultas. La calidad de los embeddings y la eficiencia de su estrategia de almacenamiento e indexación determinan directamente la velocidad y precisión de su sistema RAG en el momento de la consulta.
Generación de embeddings mediante la API de OpenAI
El enfoque más habitual consiste en llamar a la API de embeddings de OpenAI con el texto de los fragmentos. El modelo text-embedding-3-small produce vectores de 1536 dimensiones y cuesta 0,02 $ por millón de tokens, un precio extremadamente bajo para la mayoría de las cargas de trabajo. Envíe varios textos en una sola llamada a la API (hasta 2048 entradas) para maximizar el rendimiento. La respuesta contiene un vector de embedding por cada texto de entrada, en el mismo orden.
from openai import OpenAI
client = OpenAI()
def embed_batch(texts, model='text-embedding-3-small'):
response = client.embeddings.create(
model=model,
input=texts # up to 2048 texts per call
)
return [item.embedding for item in response.data]
# Embed one batch of 100 chunk texts
texts = [chunk['text'] for chunk in chunks[:100]]
vectors = embed_batch(texts)
print(f'Embedding dimension: {len(vectors[0])}')
print(f'Embedded {len(vectors)} chunks')Procesamiento por lotes para mejorar la eficiencia
Al indexar miles de fragmentos, la eficiencia es importante. Procese los fragmentos en lotes de 100-500 para equilibrar el rendimiento y el uso de memoria. Registre su posición para poder reanudar el proceso después de un fallo sin volver a generar embeddings para los fragmentos ya procesados. Registre el progreso con regularidad. Para 100.000 fragmentos en lotes de 500, realizará 200 llamadas a la API; normalmente, esto se completa en unos minutos.
def embed_all_chunks(chunks, batch_size=200):
embedded = []
total = len(chunks)
for i in range(0, total, batch_size):
batch = chunks[i:i+batch_size]
texts = [c['text'] for c in batch]
vectors = embed_batch(texts)
for chunk, vector in zip(batch, vectors):
embedded.append({
**chunk,
'embedding': vector
})
if (i // batch_size) % 10 == 0:
print(f'Progress: {min(i+batch_size, total)}/{total}')
return embeddedGestión de los límites de velocidad durante la indexación
La API de embeddings de OpenAI tiene límites de velocidad medidos en tokens por minuto (TPM). Los trabajos de indexación grandes alcanzan estos límites y reciben RateLimitError. Implemente un retroceso exponencial con jitter: cuando se produzca un error de límite de velocidad, espere un breve intervalo aleatorio antes de volver a intentarlo y duplique la espera en cada fallo posterior. Esto distribuye los reintentos y evita que todos los trabajadores paralelos saturen la API al mismo tiempo.
import time
import random
from openai import RateLimitError
def embed_batch_with_retry(texts, max_retries=5):
for attempt in range(max_retries):
try:
return embed_batch(texts)
except RateLimitError:
if attempt == max_retries - 1:
raise
wait = (2 ** attempt) + random.uniform(0, 1)
print(f'Rate limited. Waiting {wait:.1f}s...')
time.sleep(wait)
return []Inserción de vectores en Pinecone
Después de generar los embeddings, insértelos o actualícelos en el almacén vectorial. Upsert significa insertar vectores nuevos o actualizar los existentes si ya existe el mismo ID; por diseño, es idempotente. En Pinecone, cada registro sometido a upsert contiene el ID del vector, los valores del embedding y un diccionario de metadatos con los campos que desea filtrar o mostrar más adelante. Realice el upsert en lotes de hasta 100 registros por llamada para obtener un rendimiento óptimo.
import pinecone
pc = pinecone.Pinecone(api_key='YOUR_KEY')
index = pc.Index('rag-index')
def upsert_to_pinecone(embedded_chunks, batch_size=100):
for i in range(0, len(embedded_chunks), batch_size):
batch = embedded_chunks[i:i+batch_size]
vectors = [
(
chunk['id'],
chunk['embedding'],
{
'text': chunk['text'],
'source': chunk['metadata']['source'],
'page': chunk['metadata'].get('page', 0)
}
)
for chunk in batch
]
index.upsert(vectors=vectors)
print(f'Upserted {min(i+batch_size, len(embedded_chunks))}/{len(embedded_chunks)}')Almacenamiento en pgvector
Con pgvector, inserta los embeddings directamente en una tabla de PostgreSQL mediante SQL estándar. El tipo de datos vector acepta una lista de valores flotantes de Python serializada como cadena. Después de insertar todas las filas, cree un índice HNSW para realizar consultas rápidas de vecinos más cercanos aproximados. Indexar una tabla existente con millones de filas puede tardar varios minutos, por lo que debe crear el índice después de la inserción masiva y no antes.
import psycopg2
from psycopg2.extras import execute_values
def upsert_to_pgvector(conn, embedded_chunks):
with conn.cursor() as cur:
records = [
(
chunk['id'],
chunk['text'],
chunk['metadata']['source'],
chunk['metadata'].get('page', 0),
chunk['embedding'] # list of floats
)
for chunk in embedded_chunks
]
execute_values(cur, '''
INSERT INTO document_chunks (id, text, source, page, embedding)
VALUES %s
ON CONFLICT (id) DO UPDATE
SET text = EXCLUDED.text, embedding = EXCLUDED.embedding
''', records)
conn.commit()Creación del índice HNSW
HNSW (Hierarchical Navigable Small World) es el tipo de índice que permite realizar búsquedas rápidas de vecinos más cercanos aproximados. A diferencia de la búsqueda por fuerza bruta, que compara el vector de consulta con cada vector almacenado, HNSW crea una estructura de grafo multicapa que reduce el espacio de búsqueda. El parámetro m controla cuántas conexiones tiene cada nodo (un valor más alto ofrece una mejor recuperación, pero requiere más memoria), mientras que ef_construction controla la calidad del índice durante su creación.
-- Build HNSW index after bulk insertion
CREATE INDEX CONCURRENTLY ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- Set ef_search at query time to trade recall vs speed
SET hnsw.ef_search = 100;
-- Verify index was created
SELECT indexname, indexdef
FROM pg_indexes
WHERE tablename = 'document_chunks';Diseño del esquema de metadatos
Los metadatos almacenados junto a cada vector permiten una recuperación con filtros muy potente. Diseñe el esquema de metadatos antes de indexar; añadir campos más adelante requiere volver a indexar. Incluya los campos por los que aplicará filtros (departamento, tipo de documento, intervalo de fechas), los campos que mostrará en las citas (título, página, autor) y los campos útiles para la depuración (chunk_index, total_chunks, indexed_at). Mantenga simples los valores de los metadatos: las cadenas, los números y los booleanos se indexan y filtran de forma eficiente; los objetos anidados, no.
# Well-designed metadata schema
METADATA_SCHEMA = {
# For filtering at retrieval time
'department': 'HR', # string
'doc_type': 'policy', # string
'year': 2025, # integer
'is_active': True, # boolean
# For display in citations
'title': 'Employee Handbook 2025',
'author': 'HR Team',
'page': 12,
'source': 's3://docs/handbook_2025.pdf',
# For debugging and updates
'chunk_index': 3,
'total_chunks': 24,
'indexed_at': '2025-09-01T10:00:00Z'
}Creación de puntos de control para trabajos de indexación largos
Indexar un corpus grande puede tardar horas. Un fallo a mitad del proceso desperdicia todo el progreso. Implemente un archivo de punto de control que registre qué fragmentos se han indexado correctamente. Al reiniciar, omita los fragmentos ya indexados y continúe desde el punto en que se detuvo. Esto hace que el trabajo de indexación sea idempotente y seguro de reanudar. Almacene el punto de control como un conjunto de IDs de fragmentos procesados en un archivo JSON o en una tabla de base de datos.
import json
from pathlib import Path
CHECKPOINT_FILE = '/tmp/index_checkpoint.json'
def load_checkpoint():
if Path(CHECKPOINT_FILE).exists():
return set(json.loads(Path(CHECKPOINT_FILE).read_text()))
return set()
def save_checkpoint(indexed_ids):
Path(CHECKPOINT_FILE).write_text(json.dumps(list(indexed_ids)))
def index_with_checkpoint(chunks, index):
done = load_checkpoint()
remaining = [c for c in chunks if c['id'] not in done]
print(f'Resuming: {len(done)} done, {len(remaining)} remaining')
for chunk in remaining:
upsert_to_pinecone([chunk], index)
done.add(chunk['id'])
save_checkpoint(done)Verificación de la integridad del índice
Después de indexar, verifique que todos los fragmentos se hayan incorporado al almacén vectorial. Compare el número de fragmentos producidos por el divisor con el número de vectores que indica el índice. Consulte el índice con el texto de un documento conocido y confirme que el resultado esperado aparece entre los 5 primeros. Ejecute algunas consultas conocidas de su conjunto de pruebas de referencia y compruebe que la precisión alcance el nivel esperado. Nunca dé por completo el índice sin verificarlo.
def verify_index(index, chunks, sample_size=10):
index_stats = index.describe_index_stats()
total_vectors = index_stats.total_vector_count
expected = len(chunks)
print(f'Index vectors: {total_vectors}, Expected: {expected}')
if total_vectors != expected:
print('WARNING: mismatch — some chunks may not have been indexed')
# Spot-check retrieval
import random
sample = random.sample(chunks, sample_size)
for chunk in sample:
vec = embed_batch([chunk['text']])[0]
results = index.query(vector=vec, top_k=1, include_metadata=True)
top_id = results.matches[0].id if results.matches else None
if top_id != chunk['id']:
print(f'WARNING: expected {chunk["id"]}, got {top_id}')Alternativas de embeddings locales
Para datos sensibles desde el punto de vista de la privacidad que no pueden salir de su infraestructura, utilice modelos de embeddings alojados localmente. La biblioteca sentence-transformers proporciona modelos de alta calidad como all-MiniLM-L6-v2 (384 dimensiones, 22 MB, muy rápido) y bge-large-en-v1.5 (1024 dimensiones, mejor calidad). Ejecútelos en la CPU para cargas de trabajo moderadas o en la GPU para trabajos de indexación grandes. Los modelos locales eliminan los costes de API y la salida de datos, pero requieren administrar los archivos de los modelos y los recursos de cómputo.
from sentence_transformers import SentenceTransformer
# Load once at startup
model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
def embed_locally(texts, batch_size=64):
# encode() handles batching internally
embeddings = model.encode(
texts,
batch_size=batch_size,
show_progress_bar=True,
convert_to_numpy=True
)
return embeddings.tolist() # convert numpy array to Python list
vectors = embed_locally([c['text'] for c in chunks])Comprobación rápida
Compruebe su comprensión de los conceptos de ingeniería de IA de esta lección.
Resumen de la lección
En esta lección ha aprendido a: generar embeddings en lotes con la API de OpenAI y gestionar los límites de frecuencia mediante espera exponencial, insertar o actualizar vectores en Pinecone y pgvector con metadatos, crear índices HNSW para realizar búsquedas rápidas de vecinos más cercanos aproximados y aplicar prácticas recomendadas para producción, como la reanudación basada en puntos de control, el diseño de esquemas de metadatos y la verificación de la integridad del índice. A continuación, crearemos el flujo de consultas que recupera fragmentos y genera respuestas fundamentadas.
Preguntas frecuentes
¿La lección «Indexación: generación de embeddings y almacenamiento de fragmentos» es gratis?
Sí — el texto completo de «Indexación: generación de embeddings y almacenamiento de fragmentos» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Engineering Academy, actualiza a CoddyKit PRO. El curso de AI Engineering Academy incluye 4 lecciones en total.
¿Qué aprenderé en «Indexación: generación de embeddings y almacenamiento de fragmentos»?
Generará un embedding para cada fragmento mediante la API de embeddings de OpenAI y hará upsert de los vectores resultantes con metadatos en un vector store, creando un índice consultable de sus docu… Practicas AI Engineering Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Engineering Academy?
No se requiere experiencia previa. AI Engineering Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 3 de 4.
¿Cuánto tiempo toma la lección «Indexación: generación de embeddings y almacenamiento de fragmentos»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Engineering Academy?
Sí. Cada lección de AI Engineering Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Carga de documentos y extracción de texto
- Estrategias de división: fija, por oraciones y recursiva
- Indexación: generación de embeddings y almacenamiento de fragmentos
- Consultar, recuperar y generar