Misión de texto y audio

El problema

Una tienda en línea recibe mensajes de soporte y quiere enviarlos al equipo correcto. Cada mensaje trata de uno de cuatro temas: envío, pago, devolución o cuenta. Algunos clientes mencionan primero un tema que ya quedó resuelto y después el que sigue pendiente. La etiqueta es siempre el tema pendiente.

El organizador entrega 400 mensajes etiquetados y 400 mensajes de prueba sin etiqueta. Guarda las 400 etiquetas y califica una sola vez.

Mensajes y partición

La celda plegada simula el concurso. generar elige un tema, una queja de ese tema y rellena sus huecos con palabras de RELLENO. Con probabilidad 0.2 antepone una queja de otro tema seguida de «eso ya quedó resuelto, pero». Las etiquetas de prueba quedan en respuestas_organizador; ninguna decisión las usa hasta la última celda del texto.

import numpy as np

TEMAS = ["envío", "pago", "devolución", "cuenta"]
QUEJAS = [
    [
        "mi {cosa} no ha llegado",
        "el {cosa} sigue en camino desde hace {n} días",
        "el rastreo de mi {cosa} no se actualiza",
        "el repartidor no pasó {cuando}",
        "el {cosa} llegó a otra dirección",
        "la entrega se retrasó {n} días",
        "la paquetería dice que entregó el {cosa} y no lo tengo",
        "nadie me entregó el {cosa} {cuando}",
    ],
    [
        "me cobraron dos veces {cuando}",
        "la tarjeta fue rechazada al pagar el {cosa}",
        "el cargo del {cosa} aparece duplicado",
        "no puedo pagar con {medio}",
        "el cobro no coincide con el precio del {cosa}",
        "el pago con {medio} quedó pendiente",
        "me cobraron el envío que era gratis",
        "la factura del {cosa} tiene un monto equivocado",
    ],
    [
        "quiero devolver el {cosa}",
        "el {cosa} llegó roto y quiero cambiarlo",
        "cómo solicito un reembolso",
        "la talla no me queda y quiero cambiarla",
        "ya envié la devolución y no recibo mi dinero",
        "el reembolso no aparece en mi {destino}",
        "el {cosa} no es el que pedí y lo quiero regresar",
        "cuánto tarda el reembolso del {cosa}",
    ],
    [
        "no puedo iniciar sesión {cuando}",
        "olvidé mi contraseña",
        "quiero cambiar el correo de mi cuenta",
        "mi cuenta está bloqueada",
        "no me llega el código de verificación",
        "quiero borrar mi cuenta",
        "la app me saca de mi sesión",
        "alguien entró a mi cuenta sin permiso",
    ],
]
RELLENO = {
    "cosa": ["pedido", "paquete", "producto", "libro", "celular", "reloj"],
    "cuando": ["hoy", "ayer", "esta semana", "desde el lunes"],
    "medio": ["tarjeta", "tarjeta de débito", "transferencia"],
    "destino": ["tarjeta", "cuenta bancaria"],
    "n": ["dos", "tres", "cinco", "ocho"],
}
SALUDOS = ["", "hola, ", "buenas tardes, ", "urgente: "]
DESPEDIDAS = ["", ", gracias", ", saludos", ", me pueden ayudar"]


def rellenar(plantilla, rng):
    return plantilla.format(**{hueco: rng.choice(opciones) for hueco, opciones in RELLENO.items()})


def generar(n, rng, p_dos_temas=0.2):
    etiquetas = np.repeat(np.arange(4), n // 4)
    rng.shuffle(etiquetas)
    mensajes = []
    for tema in etiquetas:
        queja = rellenar(rng.choice(QUEJAS[tema]), rng)
        if rng.random() < p_dos_temas:
            otro = rng.choice([t for t in range(4) if t != tema])
            queja = rellenar(rng.choice(QUEJAS[otro]), rng) + ", eso ya quedó resuelto, pero " + queja
        mensajes.append(rng.choice(SALUDOS) + queja + rng.choice(DESPEDIDAS))
    return mensajes, etiquetas


rng = np.random.default_rng(0)
mensajes, etiquetas = generar(400, rng)
mensajes_prueba, respuestas_organizador = generar(400, rng)

mensajes es una lista de textos y etiquetas un arreglo de ids de tema, de 0 a 3 en el orden de TEMAS. np.bincount cuenta cuántas veces aparece cada id.

Pregunta. Escribe cuántos mensajes tiene cada tema. ¿Dos mensajes del mismo tema comparten siempre alguna palabra de contenido?

La partición se hace antes de construir el vocabulario. train_test_split separa 100 mensajes de validación, y stratify=etiquetas conserva en ambas partes la proporción de cada tema.

from sklearn.model_selection import train_test_split

m_ent, m_val, y_ent, y_val = train_test_split(
    mensajes, etiquetas, test_size=100, stratify=etiquetas, random_state=0
)
print(len(m_ent), len(m_val), np.bincount(y_val))

300 100 [25 25 25 25]

Quedan 300 mensajes para entrenar y 100 para decidir, 25 por tema. Un error de validación vale 0.01 de exactitud.

El vocabulario sale solo de los 300 mensajes de entrenamiento. tokenizar pasa el texto a minúsculas y extrae sus palabras con re.findall(r"\w+", …), y Counter cuenta cada token. vocab pone <pad> en el id 0 y <unk> en el id 1, y codificar convierte un texto en su lista de ids.

import re
from collections import Counter


def tokenizar(texto):
    return re.findall(r"\w+", texto.lower())


conteo = Counter(token for mensaje in m_ent for token in tokenizar(mensaje))
vocab = ["<pad>", "<unk>"] + sorted(conteo)
indice = {palabra: i for i, palabra in enumerate(vocab)}


def codificar(texto):
    return [indice.get(token, 1) for token in tokenizar(texto)]  # palabra nueva -> 1, <unk>


print("vocabulario:", len(vocab))

vocabulario: 130

El vocabulario tiene 128 palabras más <pad> y <unk>.

Pregunta. ¿Cuántos tokens de validación recibirán el id de <unk>? Escribe también cuántos tokens esperas en el mensaje más largo.

longitudes = [len(tokenizar(mensaje)) for mensaje in m_ent]
desconocidos = sum(codificar(mensaje).count(1) for mensaje in m_val)

print("tokens por mensaje:", min(longitudes), int(np.median(longitudes)), max(longitudes))
print("tokens <unk> en validación:", desconocidos)

tokens por mensaje: 4 9 29
tokens <unk> en validación: 0

Baseline TF-IDF

El baseline tiene dos niveles. DummyClassifier predice siempre el mismo tema. El Pipeline de TF-IDF con regresión logística ajusta el vectorizador solo con m_ent. make_pipeline encadena el vectorizador y el clasificador, y baseline[0] recupera el primer paso. tokenizer=tokenizar reutiliza la tokenización del vocabulario, y token_pattern=None desactiva el patrón propio del vectorizador.

Pregunta. ¿Qué exactitud de validación tendrá el dummy? ¿La de TF-IDF quedará cerca de su exactitud de entrenamiento?

from sklearn.dummy import DummyClassifier
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline

dummy = DummyClassifier(strategy="most_frequent").fit(m_ent, y_ent)
baseline = make_pipeline(
    TfidfVectorizer(tokenizer=tokenizar, token_pattern=None),
    LogisticRegression(max_iter=1000),
)
baseline.fit(m_ent, y_ent)

print("columnas TF-IDF:", len(baseline[0].vocabulary_))
print("dummy    val:", round(dummy.score(m_val, y_val), 3))
print("TF-IDF   ent:", round(baseline.score(m_ent, y_ent), 3))
print("TF-IDF   val:", round(baseline.score(m_val, y_val), 3))

columnas TF-IDF: 128
dummy    val: 0.25
TF-IDF   ent: 0.95
TF-IDF   val: 0.91

Red con embeddings

La red tiene la arquitectura del clasificador de reseñas con embeddings, ahora con cuatro salidas: nn.Embedding de dimensión 16, promedio con máscara sobre las posiciones reales y una capa lineal. Se entrena con el lote completo durante 100 épocas con Adam. Cada 25 épocas se registran pérdida y exactitud en entrenamiento y validación.

lote convierte cada mensaje en un tensor de ids. pad_sequence rellena los más cortos con padding_value=0, el id de <pad>, y batch_first=True deja un mensaje por fila.

import torch
from torch import nn
from torch.nn.utils.rnn import pad_sequence


def lote(textos):
    ids = [torch.tensor(codificar(texto)) for texto in textos]
    return pad_sequence(ids, batch_first=True, padding_value=0)


X_ent_ids, X_val_ids = lote(m_ent), lote(m_val)

En ClasificadorMensajes, padding_idx=0 fija en ceros el vector de <pad> y no lo actualiza. La máscara ids != 0 marca las posiciones reales, y el promedio suma solo esas posiciones y divide entre cuántas hay.

class ClasificadorMensajes(nn.Module):
    def __init__(self, palabras, dimension, temas):
        super().__init__()
        self.embedding = nn.Embedding(palabras, dimension, padding_idx=0)
        self.salida = nn.Linear(dimension, temas)

    def forward(self, ids):
        mascara = (ids != 0).unsqueeze(-1).float()  # (mensajes, posiciones, 1) para multiplicar cada vector
        mensaje = (self.embedding(ids) * mascara).sum(dim=1) / mascara.sum(dim=1)
        return self.salida(mensaje)

medir evalúa sin calcular gradientes, dentro de torch.no_grad(), y devuelve la pérdida y la exactitud. nn.CrossEntropyLoss recibe los cuatro logits de cada mensaje y el id de su tema.

y_ent_t, y_val_t = torch.tensor(y_ent), torch.tensor(y_val)
criterio = nn.CrossEntropyLoss()


def medir(modelo, X, y):
    modelo.eval()
    with torch.no_grad():
        logits = modelo(X)
    return round(criterio(logits, y).item(), 4), round((logits.argmax(dim=1) == y).float().mean().item(), 3)

entrenar fija la semilla con torch.manual_seed, de modo que cada semilla da otra inicialización de la red. Cada época hace un solo paso de Adam con los 300 mensajes.

def entrenar(semilla, epocas=100):
    torch.manual_seed(semilla)
    modelo = ClasificadorMensajes(len(vocab), 16, 4)
    optimizador = torch.optim.Adam(modelo.parameters(), lr=0.02)
    historia = []
    for epoca in range(1, epocas + 1):
        modelo.train()
        optimizador.zero_grad()
        criterio(modelo(X_ent_ids), y_ent_t).backward()
        optimizador.step()
        if epoca % 25 == 0:  # fila: época, pérdida y exactitud de entrenamiento, luego de validación
            historia.append((epoca, *medir(modelo, X_ent_ids, y_ent_t), *medir(modelo, X_val_ids, y_val_t)))
    return modelo, historia

Pregunta. Escribe la forma de X_val_ids y la de la salida de la red. ¿Superará la red la exactitud de validación del baseline? ¿La pérdida de validación bajará hasta la época 100?

red, historia = entrenar(semilla=0)
print(X_ent_ids.shape, X_val_ids.shape, red(X_val_ids).shape)

print("época  p_ent   p_val  ex_ent  ex_val")
for epoca, p_ent, ex_ent, p_val, ex_val in historia:
    print(f"{epoca:5d} {p_ent:.4f} {p_val:.4f} {ex_ent:.3f}  {ex_val:.3f}")

torch.Size([300, 29]) torch.Size([100, 27]) torch.Size([100, 4])
época  p_ent   p_val  ex_ent  ex_val
   25 0.4025 0.4274 0.913  0.880
   50 0.1449 0.2234 0.937  0.880
   75 0.0857 0.2560 0.993  0.890
  100 0.0571 0.2794 0.997  0.890

La red se entrena otra vez con entrenar(1) y entrenar(2), y se compara la exactitud de validación en la época 100.

Pregunta. Con otras dos semillas, ¿la exactitud de validación de la red varía más o menos que su distancia al baseline?

Errores y evaluación final

La matriz de confusión cuenta, para cada tema real (fila), cuántos mensajes recibió cada tema predicho (columna). Los mensajes con dos temas contienen la palabra «pero»; ninguna queja de un solo tema la usa. groupby("dos_temas")["error"].agg(["size", "sum"]) cuenta, en cada grupo, los mensajes y los errores.

Pregunta. ¿Los errores del baseline se reparten entre los mensajes de uno y de dos temas, o se concentran en un grupo?

import pandas as pd
from sklearn.metrics import confusion_matrix

pred_val = baseline.predict(m_val)
print(pd.DataFrame(confusion_matrix(y_val, pred_val), index=TEMAS, columns=TEMAS))

lectura = pd.DataFrame(
    {
        "dos_temas": ["pero" in tokenizar(mensaje) for mensaje in m_val],
        "error": pred_val != y_val,
    }
)
print(lectura.groupby("dos_temas")["error"].agg(["size", "sum"]))

            envío  pago  devolución  cuenta
envío          25     0           0       0
pago            0    22           3       0
devolución      3     0          22       0
cuenta          0     1           2      22
           size  sum
dos_temas
False        81    0
True         19    9

La siguiente celda toma los mensajes mal clasificados y aplica el baseline solo a la parte anterior a «pero», la queja resuelta. También cuenta cuántos de esos mensajes falla la red de semilla 0. np.flatnonzero devuelve las posiciones donde la condición es verdadera.

Pregunta. ¿En cuántos errores coincide la predicción del mensaje completo con la de su queja resuelta?

errores = np.flatnonzero(pred_val != y_val)
resueltas = [m_val[i].split(" pero ")[0] for i in errores]
pred_resueltas = baseline.predict(resueltas)
pred_red = red(X_val_ids).argmax(dim=1).numpy()

print("coinciden con la queja resuelta:", int((pred_resueltas == pred_val[errores]).sum()), "de", len(errores))
print("también los falla la red:", int((pred_red[errores] != y_val[errores]).sum()), "de", len(errores))
for i in errores[:3]:
    print(f"{TEMAS[y_val[i]]} -> {TEMAS[pred_val[i]]} | {m_val[i]}")

coinciden con la queja resuelta: 9 de 9
también los falla la red: 8 de 9
cuenta -> devolución | el producto no es el que pedí y lo quiero regresar, eso ya quedó resuelto, pero olvidé mi contraseña, me pueden ayudar
pago -> devolución | hola, cuánto tarda el reembolso del paquete, eso ya quedó resuelto, pero la factura del celular tiene un monto equivocado
devolución -> envío | hola, el repartidor no pasó esta semana, eso ya quedó resuelto, pero la talla no me queda y quiero cambiarla, me pueden ayudar

El modelo final es el baseline reentrenado con los 400 mensajes etiquetados, sin cambiar ningún parámetro. El organizador lo califica con las etiquetas que guardó.

Pregunta. Escribe la exactitud esperada en prueba y la exactitud que esperas en cada grupo de mensajes.

final = make_pipeline(
    TfidfVectorizer(tokenizer=tokenizar, token_pattern=None),
    LogisticRegression(max_iter=1000),
).fit(mensajes, etiquetas)
pred_prueba = final.predict(mensajes_prueba)

acierto = pred_prueba == respuestas_organizador
dos_temas = np.array(["pero" in tokenizar(mensaje) for mensaje in mensajes_prueba])
print(f"exactitud en prueba: {acierto.mean():.3f}")
print(f"un tema:   {acierto[~dos_temas].mean():.3f} ({(~dos_temas).sum()} mensajes)")
print(f"dos temas: {acierto[dos_temas].mean():.3f} ({dos_temas.sum()} mensajes)")

exactitud en prueba: 0.902
un tema:   1.000 (318 mensajes)
dos temas: 0.524 (82 mensajes)

Laboratorio de audio

La misma lógica se aplica a sonidos. Cada grabación sintética dura 0.5 s a 8000 Hz y pertenece a una de tres clases. Todas usan una frecuencia base f entre 500 y 700 Hz:

Cada grabación lleva ruido gaussiano con desviación estándar 0.5, 1 o 2. La celda plegada genera 300 grabaciones con generar_audio, que devuelve las señales, sus clases y su nivel de ruido. Después las parte con train_test_split, igual que los mensajes, y separa 90 grabaciones de validación estratificadas por clase. El tercer arreglo, ruidos, se reparte junto con cada señal. Las partes quedan en s_ent/s_val (señales), c_ent/c_val (clases) y r_ent/r_val (ruido).

SONIDOS = ["silbato", "timbre", "claxon"]
FS = 8000
t = np.arange(int(0.5 * FS)) / FS


def sonido(clase, rng):
    f = rng.uniform(500, 700)
    grave = np.sin(2 * np.pi * f * t)
    agudo = np.sin(2 * np.pi * 1.25 * f * t)
    if clase == 0:
        return np.sin(2 * np.pi * 3 * f * t)
    if clase == 1:
        return np.where((t // 0.1) % 2 == 0, grave, agudo)
    return (grave + agudo) / np.sqrt(2)


def generar_audio(n, rng):
    clases = np.repeat(np.arange(3), n // 3)
    rng.shuffle(clases)
    ruidos = rng.choice([0.5, 1.0, 2.0], size=n)
    senales = np.stack([sonido(c, rng) + rng.normal(0, r, t.size) for c, r in zip(clases, ruidos)])
    return senales, clases, ruidos


senales, clases, ruidos = generar_audio(300, np.random.default_rng(1))
s_ent, s_val, c_ent, c_val, r_ent, r_val = train_test_split(
    senales, clases, ruidos, test_size=90, stratify=clases, random_state=0
)

senales tiene forma (300, 4000): cada fila es una grabación de 4000 muestras, y np.bincount(clases) da [100 100 100]: 100 por clase, en el orden de SONIDOS. Quedan 210 grabaciones para entrenar y 90 para validar, y np.bincount(c_val) da [30 30 30]: 30 por clase.

spectrogram corta cada señal en ventanas de nperseg=256 muestras y calcula la potencia de cada frecuencia en cada ventana. np.array_split reparte los índices de frecuencia en 32 grupos casi iguales, y energia_bandas suma la potencia de cada grupo.

from scipy.signal import spectrogram


def energia_bandas(senales, n_bandas=32):
    _, _, potencia = spectrogram(senales, fs=FS, nperseg=256)
    grupos = np.array_split(np.arange(potencia.shape[1]), n_bandas)  # eje 1: frecuencias
    return np.stack([potencia[:, g, :].sum(axis=1) for g in grupos], axis=1)

energia_bandas agrupa las 129 frecuencias del espectrograma en 32 bandas de unos 125 Hz. El baseline de audio promedia la potencia de cada banda sobre todas las ventanas y toma el logaritmo: un vector de 32 números por grabación.

Pregunta. Escribe la forma de energia_bandas(s_ent). ¿Qué par de clases confundirá el baseline? ¿Los errores crecerán con el ruido?

def espectro_medio(senales):
    return np.log10(energia_bandas(senales).mean(axis=-1))  # promedio sobre las ventanas


print(s_ent.shape, energia_bandas(s_ent).shape, espectro_medio(s_ent).shape)
dummy_audio = DummyClassifier(strategy="most_frequent").fit(s_ent, c_ent)
print("dummy val:", round(dummy_audio.score(s_val, c_val), 3))

audio_medio = LogisticRegression(max_iter=2000).fit(espectro_medio(s_ent), c_ent)
pred_audio = audio_medio.predict(espectro_medio(s_val))
print("espectro medio val:", round((pred_audio == c_val).mean(), 3))
print(pd.DataFrame(confusion_matrix(c_val, pred_audio), index=SONIDOS, columns=SONIDOS))
print(pd.DataFrame({"ruido": r_val, "error": pred_audio != c_val}).groupby("ruido")["error"].agg(["size", "sum"]))

(210, 4000) (210, 32, 17) (210, 32)
dummy val: 0.333
espectro medio val: 0.678
         silbato  timbre  claxon
silbato       30       0       0
timbre         0      19      11
claxon         0      18      12
       size  sum
ruido
0.5      28    9
1.0      32   10
2.0      30   10

con_variacion añade a cada banda la desviación estándar de su energía logarítmica entre ventanas.

Pregunta. ¿Cuántas columnas tendrá la nueva representación? ¿Qué celdas de la matriz de confusión cambiarán?

def con_variacion(senales):
    energia = energia_bandas(senales)
    return np.hstack([np.log10(energia.mean(axis=-1)), np.log10(energia).std(axis=-1)])


audio_var = LogisticRegression(max_iter=2000).fit(con_variacion(s_ent), c_ent)
pred_var = audio_var.predict(con_variacion(s_val))
print(con_variacion(s_val).shape)
print("con variación val:", round((pred_var == c_val).mean(), 3))
print(pd.DataFrame(confusion_matrix(c_val, pred_var), index=SONIDOS, columns=SONIDOS))

(90, 64)
con variación val: 0.989
         silbato  timbre  claxon
silbato       30       0       0
timbre         0      29       1
claxon         0       0      30

Práctica

Traza guiada

Antes de procesar, decide si final asigna el mismo tema a los dos mensajes y cuál sería la etiqueta correcta de cada uno.

casos = [
    "me cobraron dos veces, eso ya quedó resuelto, pero olvidé mi contraseña",
    "olvidé mi contraseña, eso ya quedó resuelto, pero me cobraron dos veces",
]

Comprueba con final.predict(casos). Si las predicciones coinciden, calcula X = final[0].transform(casos) y cuenta las posiciones en que difieren las dos filas de TF-IDF con (X[0] != X[1]).nnz.

Cláusula final

Escribe ultima_clausula(texto), que devuelva la parte del mensaje después del último «pero», o el mensaje completo si no lo contiene. Entrena el mismo Pipeline con [ultima_clausula(m) for m in m_ent] y evalúa en validación con la misma transformación. Antes de procesar, predice cuántos errores quedarán en cada grupo de lectura. Después escribe una queja nueva de un solo tema que contenga «pero» y revisa qué recibe el modelo.

Concurso de tres temas

Escribe un generador para mensajes a la biblioteca de una escuela con tres temas (préstamo, horario y multa), seis quejas por tema y la misma regla de dos temas. Genera 300 mensajes etiquetados y 300 de prueba con np.random.default_rng(2). Produce:

1. la partición estratificada con 75 mensajes de validación y el conteo de <unk>
2. el dummy y el baseline TF-IDF
3. ClasificadorMensajes con 3 salidas y tres semillas
4. la matriz de confusión y los errores por grupo de mensajes
5. la exactitud en prueba del modelo elegido

Comprueba con assert que la salida de la red tenga forma (75, 3) y que ningún id de validación sea mayor o igual que len(vocab).

Reto opcional

Añade a sonido una cuarta clase, sirena: una frecuencia que sube de f a 1.25 f durante la grabación, con np.sin(2 * np.pi * np.cumsum(frecuencia) / FS). Antes de procesar, predice con qué clase la confundirá espectro_medio y si con_variacion la separa. Comprueba con las dos matrices de confusión de 4 × 4.

Errores que revisar

• Si ajustas el vocabulario o el TfidfVectorizer con validación o con prueba, el conteo de <unk> y la exactitud de validación dejan de medir mensajes nuevos.
• Si dos mensajes con las mismas palabras en distinto orden producen filas idénticas en la representación ((X[0] != X[1]).nnz == 0), la representación no ve el orden. TF-IDF y el promedio de embeddings pierden ese orden igual.
• Si la diferencia entre la red y el baseline es menor que la variación entre semillas, esa diferencia no decide el modelo.
• Si la pérdida de validación sube mientras la de entrenamiento baja, la red memoriza el entrenamiento. Compara la exactitud de validación en las épocas registradas antes de alargar el entrenamiento.
• Si spectrogram devuelve una forma inesperada, revisa el eje: con un arreglo (n, muestras) produce (n, frecuencias, ventanas).

Cierre

Escribe el reporte de cuatro líneas que acompañaría el envío: modelo elegido, exactitud esperada, tipo de mensaje que concentra los errores y el cambio de representación que corrigió el error análogo en audio.

Lista de comprobación

• ¿Por qué el vocabulario y el vectorizador se ajustan solo con los mensajes de entrenamiento?
• ¿Por qué TF-IDF da la misma fila a dos mensajes con las mismas palabras en distinto orden?
• ¿Por qué el promedio de embeddings de la red tiene la misma limitación?
• ¿Por qué el espectro medio no separa timbre de claxon y la desviación entre ventanas sí?
• ¿Qué evidencia indica que los errores no vienen de palabras desconocidas ni del ruido?

Documentación oficial

• TfidfVectorizer: parámetros tokenizer y token_pattern, y el vocabulario aprendido con fit.
• nn.Embedding: tabla de embeddings, padding_idx y formas de entrada y salida.
• confusion_matrix: filas como clase real y columnas como clase predicha.
• scipy.signal.spectrogram: ventanas, nperseg y los ejes de frecuencia y tiempo de la salida.