Lab práctico · Semana 2: Datos para modelos fundacionales, embeddings y almacenes vectoriales

Embeddings con Titan y búsqueda semántica en Amazon S3 Vectors

Qué vas a construir

Un pequeño almacén vectorial sin servidores ni coste fijo con Amazon S3 Vectors, alimentado con embeddings de Amazon Titan Text Embeddings V2, con metadatos filtrables, un registro de estado en DynamoDB y un proceso de sincronización incremental que solo vuelve a vectorizar lo que cambia.

flowchart LR
  C["Corpus: políticas internas (JSON)"] --> SY["sync.py: detecta cambios por hash"]
  SY <-->|"hash y versión por documento"| DDB["DynamoDB: estado de indexación"]
  SY -->|"solo lo nuevo o cambiado"| T["Titan Text Embeddings V2 (512 dimensiones)"]
  T --> S3V["S3 Vectors: índice politicas (coseno)"]
  SY -->|"borrados"| S3V
  Q["buscar.py: pregunta + filtros"] --> T2["Titan V2 (misma configuración)"]
  T2 --> S3V
  S3V --> R["Top-K chunks con distancia y metadatos"]

Cubre las Skills del task statement 1.4: arquitectura con DynamoDB para metadatos y un almacén vectorial (1.4.1), marco de metadatos para filtrar (1.4.2) y mantenimiento incremental con detección de cambios (1.4.5). También verás en la práctica el efecto de las dimensiones y de la métrica de distancia.

Antes de empezar

• CloudShell en eu-central-1. Si S3 Vectors no estuviera disponible en tu región, usa eu-west-1 (cambia AWS_REGION).
• Titan Text Embeddings V2 está disponible In-Region en Fráncfort e Irlanda: se invoca con el ID amazon.titan-embed-text-v2:0, sin perfil.

Paso 1: crea el bucket vectorial, el índice y la tabla de estado

export AWS_REGION=eu-central-1
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
export VBUCKET=lab04-vectores-$ACCOUNT_ID
mkdir -p ~/lab04 && cd ~/lab04

aws s3vectors create-vector-bucket --vector-bucket-name $VBUCKET

Crea el índice. Piensa bien cada parámetro: ni la dimensión, ni la métrica, ni las claves no filtrables se pueden cambiar después.

aws s3vectors create-index --vector-bucket-name $VBUCKET --index-name politicas \
  --data-type float32 --dimension 512 --distance-metric cosine \
  --metadata-configuration '{"nonFilterableMetadataKeys": ["texto"]}'

aws s3vectors get-index --vector-bucket-name $VBUCKET --index-name politicas

• 512 dimensiones: Titan V2 admite 1.024, 512 o 256. Con 512 guardas la mitad que con 1.024 (en el paso 5 comprobarás el efecto en la calidad).
• Coseno: la métrica habitual para texto; Titan V2 normaliza los vectores por defecto.
• texto no filtrable: el texto del chunk se devuelve con cada resultado, pero no gasta el límite de 2 KB de metadatos filtrables.
• En get-index fíjate en el modo del índice (ENHANCED en buckets creados desde el 30/09/2026): filtra antes de buscar y mejora el recall de las consultas filtradas.

Crea la tabla de DynamoDB que guardará el estado de cada documento:

aws dynamodb create-table --table-name lab04-estado-indexacion \
  --attribute-definitions AttributeName=doc_id,AttributeType=S \
  --key-schema AttributeName=doc_id,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST
aws dynamodb wait table-exists --table-name lab04-estado-indexacion

Paso 2: el corpus

Crea corpus.json: políticas ficticias de una empresa con oficinas en España y Francia. Fíjate en los metadatos: departamento, país, año y si está vigente. Hay dos versiones de la política de teletrabajo (2025, derogada, y 2026, vigente).

[
  {"doc_id": "viajes-alojamiento", "texto": "El importe máximo de alojamiento es de 120 euros por noche en España y 160 euros en el extranjero. Las reservas se hacen con la agencia autorizada.", "departamento": "viajes", "pais": "ES", "anio": 2026, "vigente": true},
  {"doc_id": "viajes-dietas", "texto": "La dieta diaria nacional es de 45 euros y la internacional de 70 euros. Los tiques de comidas se suben a la aplicación de gastos en 15 días.", "departamento": "viajes", "pais": "ES", "anio": 2026, "vigente": true},
  {"doc_id": "viajes-dietas-fr", "texto": "En Francia la dieta diaria es de 55 euros. Las comidas con clientes requieren la aprobación previa del responsable.", "departamento": "viajes", "pais": "FR", "anio": 2026, "vigente": true},
  {"doc_id": "viajes-tren", "texto": "Para trayectos de menos de 600 kilómetros se prioriza el tren frente al avión, en clase turista.", "departamento": "viajes", "pais": "ES", "anio": 2026, "vigente": true},
  {"doc_id": "teletrabajo-2025", "texto": "El personal puede teletrabajar un máximo de dos días por semana, previa autorización del responsable.", "departamento": "rrhh", "pais": "ES", "anio": 2025, "vigente": false},
  {"doc_id": "teletrabajo-2026", "texto": "El personal puede teletrabajar hasta tres días por semana. Los martes son presenciales para todos los equipos.", "departamento": "rrhh", "pais": "ES", "anio": 2026, "vigente": true},
  {"doc_id": "vacaciones", "texto": "Cada empleado dispone de 23 días laborables de vacaciones al año. Se solicitan en el portal del empleado con un mes de antelación.", "departamento": "rrhh", "pais": "ES", "anio": 2026, "vigente": true},
  {"doc_id": "formacion", "texto": "La empresa financia cursos y certificaciones técnicas relacionadas con el puesto hasta 1.500 euros anuales por persona.", "departamento": "rrhh", "pais": "ES", "anio": 2026, "vigente": true},
  {"doc_id": "portatil-perdida", "texto": "Si pierdes el portátil o te lo roban, avisa en menos de 24 horas al centro de soporte para bloquear el equipo y revocar el acceso.", "departamento": "it", "pais": "ES", "anio": 2026, "vigente": true},
  {"doc_id": "contrasenas", "texto": "Las contraseñas deben tener al menos 14 caracteres y el acceso a las aplicaciones corporativas requiere autenticación multifactor.", "departamento": "it", "pais": "ES", "anio": 2026, "vigente": true},
  {"doc_id": "ia-generativa", "texto": "Está prohibido introducir datos de clientes en herramientas de IA generativa no aprobadas por el departamento de seguridad.", "departamento": "it", "pais": "ES", "anio": 2026, "vigente": true},
  {"doc_id": "gastos-reembolso", "texto": "Los gastos aprobados se reembolsan en la nómina del mes siguiente a su validación por el departamento financiero.", "departamento": "finanzas", "pais": "ES", "anio": 2026, "vigente": true}
]

Paso 3: sincronización incremental con detección de cambios

Crea sync.py. Para cada documento calcula un hash del contenido y de los metadatos; si coincide con el guardado en DynamoDB, lo salta. Solo vectoriza lo nuevo o modificado y borra los vectores de los documentos que ya no están en el corpus. Es la misma lógica que aplica una sincronización incremental de Knowledge Bases.

import hashlib
import json
import os
import boto3

REGION = os.environ.get("AWS_REGION", "eu-central-1")
VBUCKET = os.environ["VBUCKET"]
INDICE = "politicas"
DIMENSIONES = 512

bedrock = boto3.client("bedrock-runtime", region_name=REGION)
s3v = boto3.client("s3vectors", region_name=REGION)
tabla = boto3.resource("dynamodb", region_name=REGION).Table("lab04-estado-indexacion")

def embedding(texto):
    r = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=json.dumps({"inputText": texto, "dimensions": DIMENSIONES, "normalize": True}),
    )
    cuerpo = json.loads(r["body"].read())
    return cuerpo["embedding"], cuerpo["inputTextTokenCount"]

def huella(doc):
    return hashlib.sha256(json.dumps(doc, sort_keys=True, ensure_ascii=False).encode()).hexdigest()

def sincronizar(ruta="corpus.json"):
    corpus = json.load(open(ruta, encoding="utf-8"))
    estado = {i["doc_id"]: i for i in tabla.scan()["Items"]}
    nuevos, tokens, saltados = [], 0, 0

    for doc in corpus:
        h = huella(doc)
        if estado.get(doc["doc_id"], {}).get("hash") == h:
            saltados += 1
            continue
        vector, n = embedding(doc["texto"])
        tokens += n
        metadatos = {k: v for k, v in doc.items() if k != "doc_id"}  # incluye 'texto' (no filtrable)
        nuevos.append({"key": doc["doc_id"], "data": {"float32": vector}, "metadata": metadatos})
        version = int(estado.get(doc["doc_id"], {}).get("version", 0)) + 1
        tabla.put_item(Item={"doc_id": doc["doc_id"], "hash": h, "version": version})

    for i in range(0, len(nuevos), 500):  # PutVectors admite hasta 500 vectores por llamada
        s3v.put_vectors(vectorBucketName=VBUCKET, indexName=INDICE, vectors=nuevos[i:i + 500])

    ids_corpus = {d["doc_id"] for d in corpus}
    borrados = [k for k in estado if k not in ids_corpus]
    if borrados:
        s3v.delete_vectors(vectorBucketName=VBUCKET, indexName=INDICE, keys=borrados)
        for k in borrados:
            tabla.delete_item(Key={"doc_id": k})

    print(f"Vectorizados: {len(nuevos)} | sin cambios: {saltados} | borrados: {len(borrados)} | tokens: {tokens}")

if __name__ == "__main__":
    sincronizar()

export VBUCKET
python3 sync.py
python3 sync.py

La primera ejecución vectoriza los 12 documentos; la segunda, ninguno («sin cambios: 12»). Eso es una actualización incremental: no pagas embeddings ni escrituras por lo que no ha cambiado.

Paso 4: búsqueda semántica con filtros

Crea buscar.py:

import json
import os
import sys
import boto3
from sync import embedding, VBUCKET, INDICE, REGION

s3v = boto3.client("s3vectors", region_name=REGION)

def buscar(pregunta, filtro=None, k=3):
    vector, _ = embedding(pregunta)
    params = dict(vectorBucketName=VBUCKET, indexName=INDICE, queryVector={"float32": vector},
                  topK=k, returnDistance=True, returnMetadata=True)
    if filtro:
        params["filter"] = filtro
    r = s3v.query_vectors(**params)
    print(f"\n? {pregunta}   filtro={json.dumps(filtro, ensure_ascii=False) if filtro else '-'}")
    for v in r["vectors"]:
        m = v["metadata"]
        print(f"  {v['distance']:.3f}  {v['key']:<20} [{m['departamento']}, {m['pais']}, {m['anio']}, vigente={m['vigente']}]")

if __name__ == "__main__":
    buscar("¿Cuánto me pagan por comer fuera cuando viajo?")
    buscar("¿Cuántos días puedo trabajar desde casa?")
    buscar("¿Cuántos días puedo trabajar desde casa?", filtro={"vigente": True})
    buscar("¿Cuánto me pagan por comer fuera cuando viajo?",
           filtro={"$and": [{"pais": {"$eq": "FR"}}, {"departamento": {"$in": ["viajes", "finanzas"]}}]})
    buscar("Me han robado el ordenador del trabajo, ¿qué hago?")

python3 buscar.py

Qué observar:

• «¿Cuánto me pagan por comer fuera…?» encuentra las dietas aunque no comparten palabras: eso es búsqueda semántica.
• La distancia (métrica coseno) es menor cuanto más parecido: el primero es el más cercano.
• Sin filtro, la pregunta del teletrabajo puede devolver también la política derogada de 2025. Con {"vigente": true} desaparece. Este es el motivo de diseñar metadatos filtrables: sin ellos, el modelo podría responder con una norma derogada.
• El filtro compuesto ($and, $eq, $in) restringe a Francia.
• «Me han robado el ordenador» encuentra la política del portátil.

Paso 5: ¿cuántas dimensiones necesito?

Compara la ordenación de resultados con 256, 512 y 1.024 dimensiones, calculando el coseno en memoria (sin crear más índices):

import json
import math
import boto3

bedrock = boto3.client("bedrock-runtime", region_name="eu-central-1")
corpus = json.load(open("corpus.json", encoding="utf-8"))

def emb(texto, dim):
    r = bedrock.invoke_model(modelId="amazon.titan-embed-text-v2:0",
                             body=json.dumps({"inputText": texto, "dimensions": dim, "normalize": True}))
    return json.loads(r["body"].read())["embedding"]

def coseno(a, b):
    return sum(x * y for x, y in zip(a, b)) / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))

pregunta = "¿Qué hago si pierdo el portátil?"
for dim in (256, 512, 1024):
    q = emb(pregunta, dim)
    ranking = sorted(((coseno(q, emb(d["texto"], dim)), d["doc_id"]) for d in corpus), reverse=True)[:3]
    tam_kb = dim * 4 / 1024
    print(f"{dim:>5} dims ({tam_kb:.0f} KB por vector float32): " + ", ".join(f"{i} ({s:.3f})" for s, i in ranking))

Guárdalo como dimensiones.py y ejecútalo con python3 dimensiones.py. En un corpus así de pequeño, las tres configuraciones suelen coincidir en el primer resultado. La diferencia importa a escala: 10 millones de vectores × 1.024 dimensiones × 4 bytes ≈ 41 GB, frente a ≈ 10 GB con 256. Menos dimensiones = menos almacenamiento, menos memoria de índice (en OpenSearch o pgvector) y consultas más rápidas; decide midiendo el recall con tus preguntas.

Paso 6: un cambio, un alta y una baja

Simula la vida real del corpus. Edita corpus.json:

1. Cambia la dieta nacional de 45 a 50 euros en viajes-dietas.
2. Añade un documento nuevo:

{"doc_id": "bicicleta", "texto": "La empresa ofrece aparcamiento para bicicletas y una ayuda de 200 euros anuales para quien acuda al trabajo en bicicleta.", "departamento": "rrhh", "pais": "ES", "anio": 2026, "vigente": true}

3. Borra el documento teletrabajo-2025 (ya está derogado).

Después:

python3 sync.py
python3 buscar.py
aws dynamodb scan --table-name lab04-estado-indexacion \
  --query "Items[?version.N!='1'].[doc_id.S,version.N]" --output table

Deberías ver «Vectorizados: 2 | sin cambios: 10 | borrados: 1», la nueva dieta en las búsquedas y la versión 2 de viajes-dietas en DynamoDB. En producción, este script sería una Lambda disparada por eventos de S3 vía EventBridge (cambios casi en tiempo real) o por EventBridge Scheduler (refresco programado).

Comprueba que funciona

• list-vectors muestra los vectores esperados:

aws s3vectors list-vectors --vector-bucket-name $VBUCKET --index-name politicas --query "vectors[].key"

• Las búsquedas encuentran resultados por significado y los filtros excluyen lo no vigente o de otro país.
• La segunda sincronización sin cambios no vectoriza nada.

Limpieza

En orden de dependencias: índice → bucket vectorial; y la tabla de DynamoDB.

aws s3vectors delete-index --vector-bucket-name $VBUCKET --index-name politicas
aws s3vectors delete-vector-bucket --vector-bucket-name $VBUCKET
aws dynamodb delete-table --table-name lab04-estado-indexacion
rm -rf ~/lab04

Comprueba que no queda nada:

aws s3vectors list-vector-buckets --query "vectorBuckets[].vectorBucketName"

Preguntas para pensar como arquitecto