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

Pipeline de documentos y audio listo para un modelo fundacional

Qué vas a construir

Un pipeline que convierte contenido «en bruto» en texto limpio, validado, enriquecido con metadatos y en el formato que espera Amazon Bedrock Knowledge Bases:

flowchart LR
  PDF["PDF de política (S3)"] --> TX["Textract: LAYOUT + TABLES"]
  AUD["Audio de una reunión (S3)"] --> TR["Transcribe es-ES con diarización"]
  TX --> N["Normalizar y trocear por secciones"]
  TR --> N
  N --> V["Validar: longitud, idioma, PII"]
  V -->|"no válido"| Q["S3: cuarentena/"]
  V -->|"válido"| C["Comprehend: entidades + ocultar PII"]
  C --> B["Nova Micro: título, resumen y categoría (JSON)"]
  B --> OUT["S3: procesado/ (.txt + .metadata.json)"]
  V --> CW["CloudWatch: métricas de calidad"]

Practicarás las cuatro Skills del task statement 1.3: validar la calidad (con métricas de CloudWatch), procesar texto y audio, formatear la entrada para Bedrock y mejorar la calidad (normalización, entidades, reformateo con un FM).

Antes de empezar

• Haz antes el lab-01-cuenta-y-bedrock.
• CloudShell en eu-central-1. Todos los servicios del lab (Textract, Transcribe, Comprehend, Polly, Bedrock) están disponibles en Fráncfort.
• Usarás Amazon Polly solo para generar un audio de prueba en español (no forma parte del temario, pero evita que tengas que subir un fichero propio).

Paso 1: bucket y documentos de prueba

export AWS_REGION=eu-central-1
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
export BUCKET=lab03-docs-$ACCOUNT_ID-$AWS_REGION
aws s3 mb s3://$BUCKET --region $AWS_REGION
mkdir -p ~/lab03 && cd ~/lab03

Genera un PDF de una página con una política ficticia. El script construye el PDF a mano (sin librerías externas):

# genera_pdf.py
def esc(t):
    return t.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")

lineas = [
    (18, "Política de viajes 2026"),
    (12, ""),
    (14, "1. Ámbito de aplicación"),
    (11, "Esta política se aplica a todo el personal de Ejemplo S.L. que viaje por trabajo."),
    (11, "Contacto de viajes: Marta Gómez, marta.gomez@ejemplo.com, teléfono 612 345 678."),
    (12, ""),
    (14, "2. Alojamiento"),
    (11, "El importe máximo por noche es de 120 euros en España y 160 euros en el extranjero."),
    (11, "Las reservas se hacen con la agencia autorizada con al menos 7 días de antelación."),
    (12, ""),
    (14, "3. Dietas"),
    (11, "Dieta diaria nacional: 45 euros. Dieta diaria internacional: 70 euros."),
    (11, "Los tiques deben subirse a la aplicación de gastos en un plazo de 15 días."),
]
contenido = "BT /F1 12 Tf 50 790 Td 20 TL\n"
for tam, txt in lineas:
    contenido += f"/F1 {tam} Tf ({esc(txt)}) Tj T*\n"
contenido += "ET"
datos = contenido.encode("latin-1")

objetos = [
    b"<< /Type /Catalog /Pages 2 0 R >>",
    b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
    b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 595 842] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>",
    b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica /Encoding /WinAnsiEncoding >>",
    b"<< /Length " + str(len(datos)).encode() + b" >>\nstream\n" + datos + b"\nendstream",
]
pdf = b"%PDF-1.4\n"
posiciones = []
for i, obj in enumerate(objetos, 1):
    posiciones.append(len(pdf))
    pdf += f"{i} 0 obj\n".encode() + obj + b"\nendobj\n"
inicio_xref = len(pdf)
pdf += f"xref\n0 {len(objetos) + 1}\n0000000000 65535 f \n".encode()
for p in posiciones:
    pdf += f"{p:010d} 00000 n \n".encode()
pdf += f"trailer\n<< /Size {len(objetos) + 1} /Root 1 0 R >>\nstartxref\n{inicio_xref}\n%%EOF\n".encode()
open("politica-viajes.pdf", "wb").write(pdf)
print("PDF generado:", len(pdf), "bytes")

python3 genera_pdf.py
aws s3 cp politica-viajes.pdf s3://$BUCKET/entrada/politica-viajes.pdf

Genera también un audio corto en español con Polly y súbelo:

aws polly synthesize-speech --output-format mp3 --voice-id Lucia --language-code es-ES \
  --text "Buenos días. En la reunión de hoy acordamos que el límite de alojamiento en el extranjero sube a ciento sesenta euros por noche a partir de enero. Laura Pérez enviará el acta el viernes." \
  reunion.mp3
aws s3 cp reunion.mp3 s3://$BUCKET/entrada/reunion.mp3

Paso 2: extraer el texto con estructura (Textract)

Usa AnalyzeDocument (síncrona: vale para una página) con LAYOUT, que devuelve títulos, cabeceras de sección y párrafos, y TABLES:

aws textract analyze-document \
  --document "{\"S3Object\": {\"Bucket\": \"$BUCKET\", \"Name\": \"entrada/politica-viajes.pdf\"}}" \
  --feature-types '["LAYOUT", "TABLES"]' > textract.json

python3 -c "
import json, collections
b = json.load(open('textract.json'))['Blocks']
print(collections.Counter(x['BlockType'] for x in b))
"

Verás bloques PAGE, LINE, WORD y los de maquetación: LAYOUT_TITLE, LAYOUT_SECTION_HEADER, LAYOUT_TEXT… Esa estructura es la que permite trocear por secciones en lugar de por un número fijo de tokens.

Paso 3: transcribir el audio (Transcribe)

aws transcribe start-transcription-job \
  --transcription-job-name lab03-reunion \
  --language-code es-ES \
  --media MediaFileUri=s3://$BUCKET/entrada/reunion.mp3 \
  --output-bucket-name $BUCKET --output-key transcripciones/reunion.json \
  --settings ShowSpeakerLabels=true,MaxSpeakerLabels=2

# Espera a que termine (normalmente menos de un minuto)
aws transcribe get-transcription-job --transcription-job-name lab03-reunion \
  --query TranscriptionJob.TranscriptionJobStatus --output text

Repite el último comando hasta ver COMPLETED. Después:

aws s3 cp s3://$BUCKET/transcripciones/reunion.json .
python3 -c "import json; print(json.load(open('reunion.json'))['results']['transcripts'][0]['transcript'])"

Paso 4: normalizar, validar, enriquecer y formatear

Crea pipeline.py. Hace, para cada documento, lo que haría una Lambda por paso en una máquina de Step Functions:

import json
import os
import re
import unicodedata
import boto3

REGION = "eu-central-1"
BUCKET = os.environ["BUCKET"]

s3 = boto3.client("s3", region_name=REGION)
comprehend = boto3.client("comprehend", region_name=REGION)
bedrock = boto3.client("bedrock-runtime", region_name=REGION)
cloudwatch = boto3.client("cloudwatch", region_name=REGION)

# ---------- 1. Extraer secciones del resultado de Textract (chunking por estructura) ----------
def secciones_textract(ruta):
    bloques = json.load(open(ruta))["Blocks"]
    por_id = {b["Id"]: b for b in bloques}
    def texto_de(bloque):
        ids = [i for r in bloque.get("Relationships", []) if r["Type"] == "CHILD" for i in r["Ids"]]
        return " ".join(por_id[i].get("Text", "") for i in ids if por_id[i]["BlockType"] == "LINE")
    secciones, actual = [], {"titulo": "Introducción", "texto": ""}
    for b in bloques:
        if b["BlockType"] in ("LAYOUT_TITLE", "LAYOUT_SECTION_HEADER"):
            if actual["texto"].strip():
                secciones.append(actual)
            actual = {"titulo": texto_de(b), "texto": ""}
        elif b["BlockType"] == "LAYOUT_TEXT":
            actual["texto"] += texto_de(b) + "\n"
    if actual["texto"].strip():
        secciones.append(actual)
    return secciones

# ---------- 2. Normalizar ----------
def normalizar(texto):
    texto = unicodedata.normalize("NFC", texto)
    texto = re.sub(r"<[^>]+>", " ", texto)          # restos de HTML
    texto = re.sub(r"[ \t]+", " ", texto)            # espacios repetidos
    return texto.strip()

# ---------- 3. Validar y publicar métricas de calidad ----------
def validar(texto):
    motivos = []
    if len(texto) < 40:
        motivos.append("demasiado_corto")
    idioma = comprehend.detect_dominant_language(Text=texto[:4000])["Languages"][0]
    if idioma["LanguageCode"] not in ("es", "en"):
        motivos.append("idioma_no_soportado")
    return motivos, idioma["LanguageCode"]

def metrica(nombre, valor, fuente):
    cloudwatch.put_metric_data(
        Namespace="Lab03/CalidadDatos",
        MetricData=[{"MetricName": nombre, "Dimensions": [{"Name": "Fuente", "Value": fuente}],
                     "Value": valor, "Unit": "Count"}],
    )

# ---------- 4. Ocultar PII y extraer entidades (Comprehend) ----------
def ocultar_pii(texto, idioma):
    entidades = comprehend.detect_pii_entities(Text=texto, LanguageCode=idioma)["Entities"]
    for e in sorted(entidades, key=lambda e: e["BeginOffset"], reverse=True):
        texto = texto[:e["BeginOffset"]] + f"[{e['Type']}]" + texto[e["EndOffset"]:]
    return texto, len(entidades)

def entidades(texto, idioma):
    ents = comprehend.detect_entities(Text=texto, LanguageCode=idioma)["Entities"]
    return sorted({e["Text"] for e in ents if e["Type"] in ("ORGANIZATION", "LOCATION", "DATE") and e["Score"] > 0.8})

# ---------- 5. Reformatear y generar metadatos con un FM barato ----------
def metadatos_llm(texto):
    prompt = ("Devuelve SOLO un JSON con las claves titulo (máx. 8 palabras), resumen (una frase) "
              "y categoria (una de: viajes, rrhh, finanzas, otros) para este texto:\n\n" + texto[:6000])
    r = bedrock.converse(
        modelId="eu.amazon.nova-micro-v1:0",
        messages=[{"role": "user", "content": [{"text": prompt}]}],
        inferenceConfig={"maxTokens": 200, "temperature": 0},
    )
    salida = r["output"]["message"]["content"][0]["text"]
    salida = salida.strip()
    if salida.startswith("`"):  # quita un posible bloque de código Markdown
        salida = salida.strip("`").removeprefix("json").strip()
    return json.loads(salida)

# ---------- 6. Escribir en el formato de Knowledge Bases ----------
def guardar(nombre, texto, meta):
    s3.put_object(Bucket=BUCKET, Key=f"procesado/{nombre}.txt", Body=texto.encode("utf-8"))
    s3.put_object(Bucket=BUCKET, Key=f"procesado/{nombre}.txt.metadata.json",
                  Body=json.dumps({"metadataAttributes": meta}, ensure_ascii=False).encode("utf-8"))

def procesar(nombre, texto, fuente):
    texto = normalizar(texto)
    motivos, idioma = validar(texto)
    if motivos:
        s3.put_object(Bucket=BUCKET, Key=f"cuarentena/{nombre}.txt", Body=texto.encode("utf-8"))
        metrica("DocumentosRechazados", 1, fuente)
        print(f"  ✗ {nombre}: cuarentena ({', '.join(motivos)})")
        return
    texto_limpio, n_pii = ocultar_pii(texto, idioma)
    meta = metadatos_llm(texto_limpio)
    meta.update({"idioma": idioma, "fuente": fuente, "entidades": ", ".join(entidades(texto_limpio, idioma))[:200]})
    guardar(nombre, texto_limpio, meta)
    metrica("DocumentosValidos", 1, fuente)
    metrica("EntidadesPIIOcultadas", n_pii, fuente)
    print(f"  ✓ {nombre}: {n_pii} PII ocultadas | {meta}")

if __name__ == "__main__":
    print("PDF por secciones:")
    for i, sec in enumerate(secciones_textract("textract.json")):
        procesar(f"politica-viajes-s{i}", f"{sec['titulo']}\n{sec['texto']}", "textract")
    print("Audio transcrito:")
    transcripcion = json.load(open("reunion.json"))["results"]["transcripts"][0]["transcript"]
    procesar("reunion", transcripcion, "transcribe")
    print("Documento defectuoso:")
    procesar("vacio", "<p>ok</p>", "manual")

export BUCKET
python3 pipeline.py

Qué observar:

• El PDF sale troceado por secciones («Ámbito de aplicación», «Alojamiento», «Dietas»), cada una con su título: es chunking por estructura.
• En la sección «Ámbito de aplicación», el correo y el teléfono aparecen como [EMAIL] y [PHONE].
• Si Textract no reconoce algún encabezado como LAYOUT_SECTION_HEADER, verás menos secciones: es normal y te muestra por qué conviene revisar el resultado del parsing antes de indexar.
• Nova Micro genera titulo, resumen y categoria en JSON: es mejorar la entrada con un FM y crear metadatos para filtrar.
• El documento vacio acaba en cuarentena/ y suma una métrica de rechazo.

Paso 5: revisa las métricas de calidad y crea una alarma

aws cloudwatch list-metrics --namespace Lab03/CalidadDatos --output table

aws cloudwatch put-metric-alarm --alarm-name lab03-rechazos \
  --namespace Lab03/CalidadDatos --metric-name DocumentosRechazados \
  --dimensions Name=Fuente,Value=manual \
  --statistic Sum --period 300 --evaluation-periods 1 \
  --threshold 1 --comparison-operator GreaterThanOrEqualToThreshold \
  --treat-missing-data notBreaching

En un pipeline real, esta alarma avisaría (SNS) de que una fuente ha empezado a enviar basura, antes de que llegue al índice.

Comprueba el resultado final en S3:

aws s3 ls s3://$BUCKET/procesado/
aws s3 cp s3://$BUCKET/procesado/politica-viajes-s0.txt.metadata.json -

Paso 6 (lectura): lo mismo con Glue Data Quality para datos tabulares

Si en vez de documentos recibieras un CSV con metadatos del corpus (una fila por documento), la validación declarativa se haría con AWS Glue Data Quality sobre la tabla del Data Catalog. Un conjunto de reglas DQDL equivalente sería:

Rules = [
    IsComplete "documento_id",
    IsUnique "documento_id",
    ColumnValues "idioma" in ["es", "en"],
    ColumnLength "texto" > 40,
    Completeness "categoria" > 0.95
]

Los resultados se publican en CloudWatch y EventBridge, y en trabajos ETL puedes separar los registros que fallan. No lo ejecutamos para no crear tablas ni trabajos de Glue que no vamos a reutilizar.

Comprueba que funciona

• procesado/ contiene tres secciones del PDF y la reunión, cada una con su .metadata.json.
• cuarentena/ contiene vacio.txt.
• La PII de la sección de contacto está enmascarada.
• CloudWatch muestra las métricas DocumentosValidos, DocumentosRechazados y EntidadesPIIOcultadas.

Limpieza

aws cloudwatch delete-alarms --alarm-names lab03-rechazos
aws transcribe delete-transcription-job --transcription-job-name lab03-reunion
aws s3 rm s3://$BUCKET --recursive
aws s3 rb s3://$BUCKET
rm -rf ~/lab03

Las métricas personalizadas no se pueden borrar: dejan de facturarse al dejar de recibir datos y caducan solas.

Preguntas para pensar como arquitecto