El proyecto con kit

Antes de escribir una línea del pipeline necesitas lo que en C01 ya existe: el gateway lleva semanas escribiendo logs y las fotos viven en S3. compose.yaml levanta dos contenedores (el S3 local y Label Studio) y mundo/reiniciar.py --todo crea el esqueleto de tu pipeline, tu config.yaml y genera el mundo: descarga Fashion-MNIST, entrena el modelo en producción y sube 24 500 fotos.

python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

Un entorno aislado con las versiones fijadas del dojo.

La salida depende de tu corrida.

docker compose up -d

Si un puerto está ocupado, cambia el lado izquierdo en compose.yaml («9100:9000» o «8090:8080») y la dirección que le corresponde en config.yaml (s3.endpoint o labelstudio.url).

La salida depende de tu corrida.

python mundo/reiniciar.py --todo

≈ 1–2 min. Crea pipeline/, config.yaml y el mundo. La descarga de Fashion-MNIST solo pasa la primera vez.

  pipeline/: esqueleto de los 7 pasos
  config.yaml: valores originales
  data/pipeline y data/etiquetado: borrados
  descargando train-images-idx3-ubyte…
  descargando train-labels-idx1-ubyte…
  descargando t10k-images-idx3-ubyte…
  descargando t10k-labels-idx1-ubyte…
  entrenando el modelo en producción (9 categorías, sin Bolsas)…
  subiendo 24,500 fotos a S3 local…
✓ mundo listo: 24,000 requests en 60 días, 407 correcciones, golden set de 500
✓ listo

python verificar.py mundo

El primer check: los servicios y los datos están arriba.

✓ mundo          S3, Label Studio y los datos del mundo están arriba

1/1 en verde

python -c "import duckdb; print(duckdb.sql(\"DESCRIBE SELECT * FROM read_parquet('data/logs/*/*.parquet', hive_partitioning = true)\"))"

Mira el esquema del log antes de usarlo: top_k es una lista de structs {clase, score} y fecha sale de la carpeta.

┌────────────────────┬──────────────────────────────────────┬─────────┬─────────┬─────────┬─────────┐
│    column_name     │             column_type              │  null   │   key   │ default │  extra  │
│      varchar       │               varchar                │ varchar │ varchar │ varchar │ varchar │
├────────────────────┼──────────────────────────────────────┼─────────┼─────────┼─────────┼─────────┤
│ request_id         │ VARCHAR                              │ YES     │ NULL    │ NULL    │ NULL    │
│ ts                 │ TIMESTAMP WITH TIME ZONE             │ YES     │ NULL    │ NULL    │ NULL    │
│ model_version      │ VARCHAR                              │ YES     │ NULL    │ NULL    │ NULL    │
│ cuenta_id          │ INTEGER                              │ YES     │ NULL    │ NULL    │ NULL    │
│ top_k              │ STRUCT(clase TINYINT, score FLOAT)[] │ YES     │ NULL    │ NULL    │ NULL    │
│ img_sha256_prefijo │ VARCHAR                              │ YES     │ NULL    │ NULL    │ NULL    │
│ foto_key           │ VARCHAR                              │ YES     │ NULL    │ NULL    │ NULL    │
│ latencia_ms        │ FLOAT                                │ YES     │ NULL    │ NULL    │ NULL    │
│ fecha              │ DATE                                 │ YES     │ NULL    │ NULL    │ NULL    │
└────────────────────┴──────────────────────────────────────┴─────────┴─────────┴─────────┴─────────┘

Tres fuentes, una llave: request_id. Con DuckDB lees los Parquet directo, sin cargarlos antes. Del log sacas el top-1, su score y el margen (top-1 menos top-2). Las correcciones pueden ser varias por anuncio: con QUALIFY row_number() … = 1 te quedas con la más reciente antes de unir, para no multiplicar filas. Logs con publicaciones van con JOIN (todo request publicado tiene log); las correcciones van con LEFT JOIN (la mayoría no tiene). Al final cuentas filas: un pipeline de datos que no cuenta, pierde datos sin enterarse.

WITH logs AS (
  SELECT request_id, ts, cuenta_id, img_sha256_prefijo, foto_key,
         top_k[1].clase AS top1,
         top_k[1].score AS top1_score,
         top_k[2].clase AS top2,
         top_k[1].score - top_k[2].score AS margen
  FROM read_parquet('data/logs/*/*.parquet', hive_partitioning = true)
),
ultima_correccion AS (   -- puede haber varias por request: cuenta solo la más reciente
  SELECT request_id, categoria_corregida, ts_correccion
  FROM read_parquet('data/producto/correcciones.parquet')
  QUALIFY row_number() OVER (PARTITION BY request_id ORDER BY ts_correccion DESC) = 1
)
SELECT l.*, p.ts_publicacion, p.categoria_final, p.eleccion, p.titulo,
       c.categoria_corregida, c.ts_correccion
FROM logs l
JOIN read_parquet('data/producto/publicaciones.parquet') p USING (request_id)
LEFT JOIN ultima_correccion c USING (request_id)

Va dentro de un COPY (…) TO 'data/pipeline/senales.parquet' (FORMAT parquet). Las listas de DuckDB empiezan en 1.

n_logs = con.execute("SELECT count(*) FROM read_parquet('data/logs/*/*.parquet')").fetchone()[0]
n_pub = con.execute("SELECT count(*) FROM 'data/producto/publicaciones.parquet'").fetchone()[0]
n, unicos = con.execute("SELECT count(*), count(DISTINCT request_id) FROM 'data/pipeline/senales.parquet'").fetchone()
print(f"logs: {n_logs:,} · publicaciones: {n_pub:,} · señales: {n:,} ({unicos:,} request_id únicos)")
print(f"se unió el {n / n_pub:.1%} de las publicaciones")

En DuckDB 2.0 las relaciones ya no tienen fetchone(): usa con.execute(…).fetchone().

python pipeline/1_senales.py

logs: 24,000 · publicaciones: 24,000 · señales: 24,000 (24,000 request_id únicos)
se unió el 100.0% de las publicaciones

python verificar.py senales

✓ senales        24,000 requests, una fila cada uno, con la última corrección

1/1 en verde

No todo lo que eligió quien vende sirve. Entra con su fuente: correccion si moderación la corrigió (la última manda); busqueda si quien vende buscó la categoría en vez de aceptar la sugerencia (señal fuerte: el modelo no la convenció); sugerencia_madura si aceptó una sugerencia y pasaron 30 días sin corrección. Las sugerencias aceptadas de los últimos 30 días todavía no cuentan: aún pueden corregirse. Las cuentas del holdout (eleccion = 'manual') se van a otro archivo: nunca se mezclan con el entrenamiento.

SELECT request_id, img_sha256_prefijo,
       coalesce(categoria_corregida, categoria_final) AS label,
       CASE WHEN categoria_corregida IS NOT NULL THEN 'correccion'
            WHEN eleccion = 'busqueda' THEN 'busqueda'
            ELSE 'sugerencia_madura' END AS fuente
FROM 'data/pipeline/senales.parquet'
WHERE eleccion <> 'manual'
  AND (categoria_corregida IS NOT NULL
       OR eleccion = 'busqueda'
       OR (eleccion LIKE 'sugerencia_%'
           AND ts_publicacion <= TIMESTAMPTZ '2026-10-31T00:00:00Z' - INTERVAL 30 DAY))

El corte y los 30 días salen de config.yaml (corte, debil.dias_maduracion); aquí están escritos para leerlo fácil.

python pipeline/2_debil.py

┌───────────────────┬───────┐
│      fuente       │   n   │
│      varchar      │ int64 │
├───────────────────┼───────┤
│ sugerencia_madura │ 10586 │
│ busqueda          │  2537 │
│ correccion        │   383 │
└───────────────────┴───────┘

holdout: 253 labels

python verificar.py debil

La auditoría usa la verdad del simulador: en una empresa sería una muestra revisada a mano.

✓ debil          13,506 labels débiles, 2.0% mal según la auditoría

1/1 en verde

El presupuesto es fijo, así que cada foto tiene que valer. Primero se sortea un 25 % al azar de todo el tráfico: si se sorteara al final, ya sin los casos difíciles, saldría optimista. Luego se llenan los cupos en orden: nuevas (categorías que negocio acaba de agregar: 10 %), luego desacuerdos (una persona eligió otra cosa que el top-1: 15 %), luego incertidumbre (top-1 < 0.5 o margen < 0.15) Lo que no alcance en un cupo se llena con incertidumbre. El holdout no se toca.

s = s.sort_values("request_id")                  # DuckDB no garantiza el orden: así la muestra es reproducible
s = s[s.eleccion != "manual"]                    # el holdout no se toca
humano = s.categoria_corregida.fillna(s.categoria_final.where(s.eleccion == "busqueda"))

grupos = {
    "nueva": s.categoria_final.isin(c["nuevas"]["clases"]) | s.categoria_corregida.isin(c["nuevas"]["clases"]),
    "desacuerdo": humano.notna() & (humano != s.top1),
    "incertidumbre": (s.top1_score < k["umbral_top1"]) | (s.margen < k["umbral_margen"]),
}

humano = lo que dijo una persona sin influencia de la sugerencia: la corrección o, si no hay, la búsqueda.

# La aleatoria primero, de todo el tráfico: si se sorteara al final, ya sin los casos difíciles, saldría optimista.
elegidos = dict.fromkeys(rng.choice(s.request_id.to_numpy(), cupo["aleatoria"], replace=False), "aleatoria")
for motivo in ["nueva", "desacuerdo", "incertidumbre"]:
    candidatos = s.request_id[grupos[motivo] & ~s.request_id.isin(list(elegidos))]
    n = min(cupo[motivo], len(candidatos))
    if motivo == "incertidumbre":                # lo que no alcanzó en los otros cupos se va a incertidumbre
        n = min(len(candidatos), k["tamano"] - len(elegidos))
    elegidos |= dict.fromkeys(rng.choice(candidatos.to_numpy(), n, replace=False), motivo)

Los cupos salen de config.yaml (cola.*). Una foto entra con un solo motivo: el primero que la elige.

python pipeline/3_cola.py

motivo
incertidumbre    400
aleatoria        200
desacuerdo       120
nueva             80
total: 800

python verificar.py cola-mezcla cola-nuevas

✓ cola-mezcla    incertidumbre 400 · aleatoria 200 · desacuerdo 120 · nueva 80
✓ cola-nuevas    79 ejemplos de bolsas (cupo 80)

2/2 en verde

Primero el bucket: se crea, se le pone el bloqueo de acceso público completo (los cuatro interruptores) y una regla de ciclo de vida que vence lo que esté bajo cola/ a los 90 días. Luego cada foto: se baja del bucket del producto, se le quita todo metadato y se sube con el request_id como nombre. Para quitar el EXIF: ImageOps.exif_transpose aplica la rotación antes de perder el tag de orientación, y Image.frombytes crea una imagen nueva solo con los píxeles. Al final se borra lo que sobre de otras corridas: el bucket solo tiene la cola de este mes.

cli.put_public_access_block(Bucket=B, PublicAccessBlockConfiguration={
    "BlockPublicAcls": True, "IgnorePublicAcls": True, "BlockPublicPolicy": True, "RestrictPublicBuckets": True})
cli.put_bucket_lifecycle_configuration(Bucket=B, LifecycleConfiguration={"Rules": [{
    "ID": "cola-vence", "Status": "Enabled", "Filter": {"Prefix": "cola/"},
    "Expiration": {"Days": 90}}]})

Los 90 días salen de config.yaml (s3.dias_vencimiento).

def limpiar(jpg: bytes) -> bytes:
    im = ImageOps.exif_transpose(Image.open(io.BytesIO(jpg)))   # aplica la rotación antes de perder el tag
    limpio = Image.frombytes(im.mode, im.size, im.tobytes())   # una imagen nueva: solo píxeles, cero metadatos
    out = io.BytesIO()
    limpio.save(out, "JPEG", quality=92)
    return out.getvalue()

save() de un JPEG no copia el EXIF si no se lo pasas, pero sí el comentario del archivo (y otros formatos copian más): reconstruir desde los píxeles no deja nada.

python pipeline/4_copiar.py

copiadas: 800 · borradas de corridas anteriores: 0

curl -s -o /dev/null -w '%{http_code}\n' http://localhost:9100/etiquetado-restringido/cola/$(python -c "import pandas as pd; print(pd.read_parquet('data/pipeline/cola.parquet').request_id.iloc[0])").jpg

Pedir una foto sin credenciales: tiene que dar 403.

403

python verificar.py copia-exif bucket-seguro

✓ copia-exif     800 fotos, todas sin metadatos
✓ bucket-seguro  privado, sin acceso público y con vencimiento de 90 días

2/2 en verde

La guía v1 se escribió antes de que existiera «bolsas» y no dice nada de los casos límite. Escribe guia/guia-v2.md: copia la v1, agrega ## bolsas y una sección de casos límite con una regla para cada par que se confunde, con este formato: - regla: camisas-vs-playeras — … (también camisas-vs-sueteres y sueteres-vs-abrigos). Cambia etiquetado.guia_version en config.yaml. Después, el paso 5 escribe una tarea por foto con la versión de la guía y marca al azar el 20 % como doble. Para ver el efecto de la guía, corre primero con la v1 y luego con la v2.

## bolsas
Bolsas de mano, bolsos, mochilas pequeñas y bolsas tote. No es ropa: si la foto muestra una prenda y una bolsa, manda la prenda.

## Casos límite

- regla: camisas-vs-playeras — Si se ven cuello y botones al frente, es camisa. Manga corta sin botones es playera, aunque tenga cuello.
- regla: camisas-vs-sueteres — Botones al frente = camisa. Tejido grueso sin abertura al frente = suéter.
- regla: sueteres-vs-abrigos — Si se abre al frente de arriba abajo, es abrigo. Si se pone por la cabeza, es suéter.

Lo que la v2 agrega a la v1. El proveedor simulado busca exactamente «## bolsas» y «regla: ».

t = pd.read_parquet(SAL / "copiadas.parquet")
t["guia_version"] = c["guia_version"]      # cada label recuerda con qué versión de la guía se hizo
t["doble"] = False
t.loc[rng.choice(t.index, round(len(t) * c["frac_doble"]), replace=False), "doble"] = True
t.to_parquet(ETQ / "tareas-proveedor.parquet", index=False)

python pipeline/5_proveedor.py && python mundo/anotar.py

Con la guía v2 en config.yaml.

tareas: 800 · dobles: 160 · guía: guia-v2
✓ el proveedor devolvió 960 anotaciones (160 dobles) con guia-v2

python verificar.py guia

✓ guia           guia-v2: las 10 categorías y los 3 casos límite

1/1 en verde

Sobre las 160 dobles calculas el acuerdo bruto y el kappa, global y por clase. scikit-learn no trae kappa por clase: se calcula «uno contra el resto», preguntando para cada categoría si A y B coinciden en «es esta / no es esta». Los desacuerdos van a Label Studio como tareas: la imagen viaja como URL prefirmada (presigned URL) del bucket restringido: vence en 7 días, el máximo con la firma SigV4, y el bucket sigue privado. En AWS, si firmas con credenciales temporales de un rol, la URL muere cuando mueren esas credenciales. En Label Studio (http://localhost:8090, usuario taller@ejemplo.com / taller-etiquetas): crea un proyecto, pega guia/labelstudio-config.xml en Settings → Labeling Interface → Code, importa data/etiquetado/adjudicar-tareas.json, decide cada caso y exporta en JSON_MIN a data/etiquetado/adjudicacion-export.json. Con la guía v1 el kappa se cae justo donde la guía no decide: en la prueba del dojo, camisas 0.09, suéteres 0.09, playeras 0.29 y bolsas 0.54; con la v2, la más baja es camisas con 0.82.

dobles = a.pivot(index="request_id", columns="anotador", values="clase").dropna().astype(int)
por_clase = {}
for i, nombre in enumerate(CLASES):
    k = cohen_kappa_score(dobles.A == i, dobles.B == i)   # uno contra el resto
    por_clase[nombre] = None if np.isnan(k) else round(float(k), 3)

tareas = [{"data": {
    "image": cli.generate_presigned_url("get_object", Params={"Bucket": B, "Key": f"cola/{rid}.jpg"}, ExpiresIn=7 * 24 * 3600),
    "request_id": rid, "anotador_a": CLASES[f.A], "anotador_b": CLASES[f.B], "guia_version": version,
}} for rid, f in desacuerdos.iterrows()]

Todo lo que va en data se ve en la interfaz ($anotador_a, $anotador_b) y regresa tal cual en el export.

python pipeline/6_acuerdo.py

Corre primero con la guía v1 y luego con la v2: compara camisas, suéteres y bolsas.

{
  "guia_version": "guia-v2",
  "n_dobles": 160,
  "acuerdo_bruto": 0.931,
  "kappa": 0.922,
  "kappa_por_clase": {
    "playeras": 0.937,
    "pantalones": 0.944,
    "sueteres": 0.853,
    "vestidos": 0.953,
    "abrigos": 0.972,
    "sandalias": 0.902,
    "camisas": 0.819,
    "tenis": 1.0,
    "bolsas": 0.955,
    "botines": 1.0
  }
}
desacuerdos para adjudicar: 11

python verificar.py acuerdo

✓ acuerdo        kappa global 0.922 con 160 dobles · guía guia-v2

1/1 en verde

Los labels del mes: lo que dijo A en las sencillas y, en las dobles, el acuerdo o tu adjudicación. Antes de guardarlos, se cruzan los hashes de contenido contra el manifest del golden set: si alguna foto del golden aparece, el paso se detiene, porque entrenar con ella haría que el golden deje de medir. Por último, el reporte: la exactitud del modelo en producción medida contra cada fuente. Ahí ves el feedback loop: contra las sugerencias aceptadas sale ≈ 0.89; contra el holdout, ≈ 0.81.

golden = pd.read_parquet(DATA / "golden" / "manifest.parquet")
fuga = set(labels.img_sha256_prefijo) & set(golden.img_sha256.str[:16])
if fuga:
    sys.exit(f"✗ {len(fuga)} imágenes del golden set están en los labels del mes: el golden dejaría de medir")

El hash se calcula una vez, en el gateway, sobre los píxeles originales, y viaja como columna: quitar el EXIF o recomprimir no cambia la identidad. Si recalculas el hash de la copia, no coincide (recomprimir cambia algunos píxeles).

reporte = {"exactitud": {
    "vs_labels_debiles": {f: exactitud(g) for f, g in debil.groupby("fuente")},
    "vs_holdout": exactitud(pd.read_parquet(SAL / "holdout.parquet").merge(s)),
    "vs_labels_mes": exactitud(labels.merge(s)),
    "vs_golden": exactitud(golden),
}}

Para el golden se baja cada foto de s3://golden-set y se predice con mundo/modelo.py, como bajarías el modelo del registry.

python pipeline/7_cierre.py

Tus números de labels del mes y de adjudicados pueden variar un poco según cómo adjudiques. Esta salida es con el experto simulado.

{
  "exactitud": {
    "vs_labels_debiles": {
      "busqueda": 0.575,
      "correccion": 0.055,
      "sugerencia_madura": 0.89
    },
    "vs_holdout": 0.81,
    "vs_labels_mes": 0.411,
    "vs_golden": 0.732
  },
  "n": {
    "labels_mes": 800,
    "adjudicados": 11,
    "golden": 500
  },
  "guia_version": "guia-v2"
}

python verificar.py

Los 12 checks.

✓ mundo          S3, Label Studio y los datos del mundo están arriba
✓ senales        24,000 requests, una fila cada uno, con la última corrección
✓ debil          13,506 labels débiles, 2.0% mal según la auditoría
✓ cola-mezcla    incertidumbre 400 · aleatoria 200 · desacuerdo 120 · nueva 80
✓ cola-nuevas    79 ejemplos de bolsas (cupo 80)
✓ copia-exif     800 fotos, todas sin metadatos
✓ bucket-seguro  privado, sin acceso público y con vencimiento de 90 días
✓ guia           guia-v2: las 10 categorías y los 3 casos límite
✓ acuerdo        kappa global 0.922 con 160 dobles · guía guia-v2
✓ labels-mes     800 labels (11 adjudicados) con guia-v2
✓ golden         500 imágenes congeladas, ninguna en la cola ni en los labels del mes
✓ reporte        holdout 0.81 · golden 0.732 · labels del mes 0.411

12/12 en verde