Texto como datos: tokens, vocabulario y clasificación

import re
import urllib.request
from collections import Counter
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np

# Don Quijote en el Proyecto Gutenberg (libro n.º 2000, dominio público).
# Se baja una sola vez y queda guardado en la carpeta «datos».
URL = "https://www.gutenberg.org/cache/epub/2000/pg2000.txt"
archivo = Path("datos/quijote-pg2000.txt")
if not archivo.exists():
    archivo.parent.mkdir(exist_ok=True)
    urllib.request.urlretrieve(URL, archivo)

crudo = archivo.read_text(encoding="utf-8")
# El libro está entre dos marcas: se quitan la cabecera y el pie
marca = crudo.index("*** START OF THE PROJECT GUTENBERG EBOOK")
inicio = crudo.index("\n", marca) + 1
fin = crudo.index("*** END OF")
quijote = crudo[inicio:fin].strip()

print(f"{len(crudo):,} caracteres con cabecera y pie")
print(f"{len(quijote):,} caracteres del libro")
k = quijote.index("En un lugar de la Mancha")
print(quijote[k:k + 70].replace("\n", " "))

def palabras(texto):
    """Pasa a minúsculas y devuelve la lista de palabras."""
    return re.findall(r"\w+", texto.lower())

todas = palabras(quijote)
conteo = Counter(todas)
print(f"{len(todas):,} palabras, {len(conteo):,} distintas")
print(conteo.most_common(5))

def sin_tildes(palabra):
    """Quita las tildes (pero no la ñ)."""
    return palabra.translate(str.maketrans("áéíóúü", "aeiouu"))

conteo_sin = Counter(sin_tildes(p) for p in todas)
print(f"Distintas con tildes: {len(conteo):,}   sin tildes: "
      f"{len(conteo_sin):,}")
for a, b in [("el", "él"), ("si", "sí"), ("mas", "más"),
             ("esta", "está")]:
    print(f"{a:<5} {conteo[a]:>5} veces   {b:<5} {conteo[b]:>5} veces")

frecuencias = np.array(sorted(conteo.values(), reverse=True))
rangos = np.arange(1, len(frecuencias) + 1)

print("Las 10 más frecuentes cubren el "
      f"{frecuencias[:10].sum() / len(todas):.0%} del texto")
print("Las 100 más frecuentes cubren el "
      f"{frecuencias[:100].sum() / len(todas):.0%} del texto")
una_vez = (frecuencias == 1).sum()
print(f"Aparecen una sola vez: {una_vez:,} palabras "
      f"({una_vez / len(conteo):.0%} del vocabulario)")

plt.loglog(rangos, frecuencias, color="C0", label="Don Quijote")
plt.loglog(rangos, frecuencias[0] / rangos, color="C2",
           linestyle="--", label="Zipf: $f_1$ / rango")
plt.xlabel("Rango (1 = la más frecuente)")
plt.ylabel("Veces que aparece")
plt.title("La ley de Zipf en el Quijote")
plt.legend()
plt.show()

def construir_vocabulario(conteo, minimo=2):
    """Diccionario palabra -> número, con las palabras frecuentes."""
    vocab = {"<PAD>": 0, "<UNK>": 1}
    for palabra, n in conteo.most_common():
        if n >= minimo:
            vocab[palabra] = len(vocab)
    return vocab

def codificar(tokens, vocab):
    """Cambia cada token por su número; los desconocidos, por <UNK>."""
    return [vocab.get(t, vocab["<UNK>"]) for t in tokens]

vocab = construir_vocabulario(conteo)
print(f"Vocabulario: {len(vocab):,} tokens")
frase = "Don Quijote vio un dinosaurio en la computadora"
ids = codificar(palabras(frase), vocab)
print(palabras(frase))
print(ids)

letras = sorted(set(quijote.lower()))
print(f"Caracteres distintos: {len(letras)}")
print("".join(letras[22:]))
print(len(palabras(frase)), "palabras y", len(frase), "caracteres")

def contar_pares(partes):
    """Cuántas veces aparece cada par de piezas vecinas."""
    pares = Counter()
    for piezas, n in partes.items():
        for par in zip(piezas, piezas[1:]):
            pares[par] += n
    return pares

def fusionar(piezas, par):
    """Junta en una sola pieza cada aparición del par."""
    nuevas, i = [], 0
    while i < len(piezas):
        if i + 1 < len(piezas) and (piezas[i], piezas[i + 1]) == par:
            nuevas.append(piezas[i] + piezas[i + 1])
            i += 2
        else:
            nuevas.append(piezas[i])
            i += 1
    return tuple(nuevas)

def entrenar_bpe(conteo, n_fusiones):
    """Aprende las fusiones de BPE a partir de un conteo de palabras."""
    # Cada palabra empieza partida en letras, con «_» al final
    partes = {tuple(p) + ("_",): n for p, n in conteo.items()}
    fusiones, tokens = [], []
    for _ in range(n_fusiones):
        pares = contar_pares(partes)
        if not pares:
            break
        mejor = max(pares, key=pares.get)       # el par más frecuente
        fusiones.append(mejor)
        partes = {fusionar(p, mejor): n for p, n in partes.items()}
        tokens.append(sum(len(p) * n for p, n in partes.items()))
    return fusiones, tokens

def tokenizar_bpe(palabra, fusiones):
    """Parte una palabra aplicando las fusiones en el orden aprendido."""
    piezas = tuple(palabra) + ("_",)
    for par in fusiones:
        piezas = fusionar(piezas, par)
    return list(piezas)

fusiones, tokens_bpe = entrenar_bpe(conteo, 300)
print("Primeras fusiones:")
print(["".join(par) for par in fusiones[:12]])
for p in ["quijote", "molinos", "computadora", "boliviano"]:
    print(f"{p:>12} -> {tokenizar_bpe(p, fusiones)}")

letras_texto = sum(len(p) + 1 for p in todas)    # letras más el «_»
por_palabra = np.array([letras_texto] + tokens_bpe) / len(todas)
print(f"Sin fusiones: {por_palabra[0]:.2f} tokens por palabra")
print(f"Con 300: {por_palabra[-1]:.2f} tokens por palabra")

plt.plot(range(len(por_palabra)), por_palabra, color="C0")
plt.xlabel("Número de fusiones")
plt.ylabel("Tokens por palabra")
plt.title("Más fusiones, menos tokens")
plt.show()

from transformers import AutoTokenizer

tok_gpt2 = AutoTokenizer.from_pretrained("openai-community/gpt2")
tok_bert = AutoTokenizer.from_pretrained(
    "distilbert/distilbert-base-uncased")
print(f"Vocabulario de GPT-2: {len(tok_gpt2):,}   "
      f"de DistilBERT: {len(tok_bert):,}")

es = "Mañana iré a la panadería con mi hermano."
print(" ".join(tok_gpt2.tokenize(es)))
print(" ".join(tok_bert.tokenize(es)))
ids = tok_gpt2(es)["input_ids"]
print(ids[:6])
print([tok_gpt2.decode([i]) for i in ids[:6]])

en = "Tomorrow I will go to the bakery with my brother."
nombres = ["Palabras", "Caracteres", "Nuestro BPE", "GPT-2",
           "DistilBERT"]

def contar_tokens(texto):
    """Cuántos tokens da cada tokenizador para un texto."""
    bpe = sum(len(tokenizar_bpe(p, fusiones)) for p in palabras(texto))
    return [len(palabras(texto)), len(texto), bpe,
            len(tok_gpt2.tokenize(texto)), len(tok_bert.tokenize(texto))]

n_es, n_en = contar_tokens(es), contar_tokens(en)
for nombre, a, b in zip(nombres, n_es, n_en):
    print(f"{nombre:<12} español: {a:>3}   inglés: {b:>3}")

# Solo para dibujar: barras de los dos idiomas, una al lado de la otra
y = np.arange(len(nombres))
plt.barh(y - 0.2, n_es, height=0.4, color="C0", label="Español")
plt.barh(y + 0.2, n_en, height=0.4, color="C2", label="Inglés")
plt.yticks(y, nombres)
plt.gca().invert_yaxis()
plt.xlabel("Número de tokens")
plt.title("Cinco tokenizadores")
plt.legend()
plt.show()

corte_ent = int(0.8 * len(todas))
corte_val = int(0.9 * len(todas))
entren = todas[:corte_ent]
valid = todas[corte_ent:corte_val]
prueba = todas[corte_val:]
vocab_ng = {p for p, n in Counter(entren).items() if n >= 2}

def con_unk(tokens):
    return [t if t in vocab_ng else "<UNK>" for t in tokens]

entren, valid, prueba = con_unk(entren), con_unk(valid), con_unk(prueba)
V = len(vocab_ng) + 1
print(f"Vocabulario: {V:,}   desconocidas en la validación: "
      f"{valid.count('<UNK>') / len(valid):.1%}")

def contar_ngramas(tokens, n):
    """Cuenta los n-gramas y sus contextos (los n - 1 tokens de antes)."""
    ngramas = Counter(tuple(tokens[i:i + n])
                      for i in range(len(tokens) - n + 1))
    contextos = Counter()
    for ng, c in ngramas.items():
        contextos[ng[:-1]] += c
    return ngramas, contextos

cuentas = {n: contar_ngramas(entren, n) for n in (1, 2, 3)}

def probabilidad(contexto, palabra, lam=0.6):
    """P(palabra | contexto), mezclando con un contexto más corto."""
    if not contexto:
        ngramas, _ = cuentas[1]
        return (ngramas[(palabra,)] + 1) / (len(entren) + V)
    ngramas, contextos = cuentas[len(contexto) + 1]
    c = contextos[contexto]
    directa = ngramas[contexto + (palabra,)] / c if c else 0
    # Aquí la función se llama a sí misma, con una palabra menos
    return (lam * directa
            + (1 - lam) * probabilidad(contexto[1:], palabra, lam))

ngramas2, contextos2 = cuentas[2]
bigrama = ngramas2[("don", "quijote")] / contextos2[("don",)]
unigrama = probabilidad((), "quijote")
print(f"bigrama {bigrama:.3f}   unigrama {unigrama:.4f}")
print(f"mezcla: {probabilidad(('don',), 'quijote'):.3f}")

def perplejidad(tokens, n, lam=0.6):
    log_total = 0
    for i in range(n - 1, len(tokens)):
        contexto = tuple(tokens[i - n + 1:i])
        log_total += np.log(probabilidad(contexto, tokens[i], lam))
    return np.exp(-log_total / (len(tokens) - n + 1))

for n in (1, 2):
    print(f"n = {n}: perplejidad {perplejidad(valid, n):.0f}")

print(f"Bigrama en la prueba: perplejidad {perplejidad(prueba, 2):.0f}")

for previa in ["don", "sancho"]:
    siguientes = Counter({ng[1]: c for ng, c in ngramas2.items()
                          if ng[0] == previa})
    total = contextos2[(previa,)]
    print(previa, "->", [(p, round(c / total, 2))
                         for p, c in siguientes.most_common(4)])

rng = np.random.default_rng(0)
texto = ["dijo"]
for _ in range(14):
    siguientes = [(ng[1], c) for ng, c in ngramas2.items()
                  if ng[0] == texto[-1]]
    opciones = [p for p, c in siguientes]
    pesos = np.array([c for p, c in siguientes], dtype=float)
    texto.append(rng.choice(opciones, p=pesos / pesos.sum()))
print(" ".join(texto))

import pandas as pd
from sklearn.feature_extraction.text import CountVectorizer

frases = ["el perro come pan",
          "el gato come pescado",
          "el perro y el gato duermen"]
bolsa = CountVectorizer()
X_bolsa = bolsa.fit_transform(frases)
print(pd.DataFrame(X_bolsa.toarray(),
                   columns=bolsa.get_feature_names_out()))

from sklearn.feature_extraction.text import TfidfVectorizer

tfidf = TfidfVectorizer()
X_tfidf = tfidf.fit_transform(frases)
print(pd.DataFrame(X_tfidf.toarray(),
                   columns=tfidf.get_feature_names_out()).round(2))

mini = ["muy buena la comida", "buena atención y buena comida",
        "excelente lugar, muy limpio", "comida fría y mala atención",
        "muy caro y sucio", "mala comida, nunca más"]
y_mini = np.array([1, 1, 1, 0, 0, 0])
cv_mini = CountVectorizer()
X_mini = cv_mini.fit_transform(mini).toarray()

def entrenar_nb(X, y, alpha=1):
    """Devuelve ln P(clase) y ln P(palabra | clase) para las clases 0 y 1."""
    # (y == c) vale 1 en las filas de la clase c y 0 en las demás: el
    # producto X.T @ (y == c) suma la columna de cada palabra en esas filas
    cuentas = np.array([X.T @ (y == c) for c in (0, 1)])
    # keepdims=True deja el resultado como columna, de forma (2, 1)
    totales = cuentas.sum(axis=1, keepdims=True)
    log_p_palabra = np.log((cuentas + alpha)
                           / (totales + alpha * X.shape[1]))
    # np.bincount(y) cuenta cuántas veces aparece 0, 1, ... en y
    log_prior = np.log(np.bincount(y) / len(y))
    return log_prior, log_p_palabra

def puntajes_nb(X, log_prior, log_p_palabra):
    """Una fila por texto y una columna por clase."""
    return log_prior + X @ log_p_palabra.T

prior_mini, log_p_mini = entrenar_nb(X_mini, y_mini)
print("Vocabulario:", X_mini.shape[1], "palabras")
for p in ["buena", "mala", "caro", "comida"]:
    neg, pos = np.exp(log_p_mini[:, cv_mini.vocabulary_[p]])
    print(f"{p:<7} P(· | negativa) = {neg:.3f}   "
          f"P(· | positiva) = {pos:.3f}")

from sklearn.naive_bayes import MultinomialNB

nueva = cv_mini.transform(["buena comida pero caro"]).toarray()
puntajes = puntajes_nb(nueva, prior_mini, log_p_mini)[0]
print("Puntajes (negativa, positiva):", puntajes.round(2))
prob = np.exp(puntajes) / np.exp(puntajes).sum()
print("Nuestro Naive Bayes:", prob.round(3))

nb_mini = MultinomialNB(alpha=1).fit(X_mini, y_mini)
print("MultinomialNB:      ", nb_mini.predict_proba(nueva)[0].round(3))
print("¿Mismas ln P(palabra | clase)?",
      np.allclose(nb_mini.feature_log_prob_, log_p_mini))

from datasets import load_dataset
from huggingface_hub import hf_hub_download

ruta = hf_hub_download("fancyzhx/amazon_polarity",
                       "amazon_polarity/test-00000-of-00001.parquet",
                       repo_type="dataset")
# split="train" es solo el nombre que load_dataset le da a un archivo
# suelto: nosotros mismos lo dividimos después en entrenamiento y prueba
resenas = load_dataset("parquet", data_files=ruta, split="train")
muestra = resenas.shuffle(seed=0).select(range(6000))

textos = [f"{t}. {c}" for t, c in zip(muestra["title"],
                                      muestra["content"])]
etiquetas = np.array(muestra["label"])
print(f"{len(resenas):,} reseñas; usamos {len(textos):,}")
print("Positivas:", etiquetas.sum(), "  negativas:",
      (etiquetas == 0).sum())
print(etiquetas[0], textos[0][:70])

from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline

X_ent, X_pru, y_ent, y_pru = train_test_split(
    textos, etiquetas, test_size=0.2, random_state=0, stratify=etiquetas)
X_sub, X_val, y_sub, y_val = train_test_split(
    X_ent, y_ent, test_size=0.2, random_state=0, stratify=y_ent)
print(f"Entrenamiento: {len(X_sub)}   validación: {len(X_val)}   "
      f"prueba: {len(X_pru)}")

candidatos = {
    "Naive Bayes": make_pipeline(CountVectorizer(min_df=2),
                                 MultinomialNB()),
    "Logística": make_pipeline(TfidfVectorizer(min_df=2),
                               LogisticRegression(max_iter=1000)),
}
for nombre, m in candidatos.items():
    m.fit(X_sub, y_sub)
    print(f"{nombre:<12} exactitud en la validación "
          f"{m.score(X_val, y_val):.3f}")

# nb_pipe[0] es el primer paso del pipeline (CountVectorizer)
nb_pipe = candidatos["Naive Bayes"]
C_sub = nb_pipe[0].transform(X_sub)
C_val = nb_pipe[0].transform(X_val)
prior_res, log_p_res = entrenar_nb(C_sub, y_sub)
# argmax(axis=1): en cada fila, la columna (la clase) de mayor puntaje
nuestras = puntajes_nb(C_val, prior_res, log_p_res).argmax(axis=1)
print("¿Las mismas predicciones que MultinomialNB?",
      (nuestras == nb_pipe.predict(X_val)).all())

from sklearn.metrics import confusion_matrix

modelo = make_pipeline(TfidfVectorizer(min_df=2),
                       LogisticRegression(max_iter=1000))
modelo.fit(X_ent, y_ent)
# modelo[0] es el primer paso del pipeline (TF-IDF); modelo[1], la logística
print(f"Vocabulario aprendido: {len(modelo[0].vocabulary_):,} palabras")
print(f"Exactitud en la prueba: {modelo.score(X_pru, y_pru):.3f}")
print(confusion_matrix(y_pru, modelo.predict(X_pru)))

pesos = modelo[1].coef_[0]
vocab_tfidf = modelo[0].get_feature_names_out()
orden = np.argsort(pesos)
elegidas = np.concatenate([orden[:6], orden[-6:]])
print("Más negativas:", list(vocab_tfidf[orden[:6]]))
print("Más positivas:", list(vocab_tfidf[orden[-6:][::-1]]))

# Tinta para las negativas y coral para las positivas
colores = ["C0"] * 6 + ["C1"] * 6
# Una figura más alta, para que las 12 palabras no se encimen
plt.figure(figsize=(4.8, 4.6))
plt.barh(range(12), pesos[elegidas], color=colores)
plt.yticks(range(12), vocab_tfidf[elegidas])
plt.axvline(0, color="C2", linewidth=1)
plt.xlabel("Peso en la regresión logística")
plt.title("Las palabras que más pesan")
plt.show()

nuevas = ["Great book, I loved every page.",
          "It broke after two days. Waste of money.",
          "Me encantó, es un libro excelente."]
for texto, p in zip(nuevas, modelo.predict_proba(nuevas)[:, 1]):
    # .nnz cuenta los valores que no son cero: las palabras conocidas
    conocidas = modelo[0].transform([texto]).nnz
    print(f"P(positiva) = {p:.2f}   conocidas: {conocidas:>2}   {texto}")

lote = pad_sequence(secuencias, batch_first=True,
                    padding_value=vocab["<PAD>"]).to(device)
mascara = lote != vocab["<PAD>"]
print(lote.shape)
print(lote.tolist())
print(mascara.int().tolist())

# Solo para dibujar: el lote y la máscara como dos tablas
from matplotlib.colors import ListedColormap

inverso = {i: p for p, i in vocab.items()}
# Papel para el relleno y tinta para los tokens reales
papel_tinta = ListedColormap(["C3", "C0"])
fig, (arriba, abajo) = plt.subplots(2, 1)
for eje, valores in [(arriba, lote), (abajo, mascara.int())]:
    eje.imshow(mascara.cpu(), cmap=papel_tinta, aspect="auto")
    for i in range(lote.shape[0]):
        for j in range(lote.shape[1]):
            v = valores[i, j].item()
            txt = inverso[v] if eje is arriba else str(v)
            eje.text(j, i, txt, ha="center", va="center", fontsize=12,
                     color="white" if mascara[i, j] else "C0")
    eje.set_xticks([])
    eje.set_yticks([])
    eje.grid(False)
arriba.set_title("El lote con padding")
abajo.set_title("Máscara: 1 = real, 0 = relleno")
plt.show()

x_bert = tok_bert(lote_frases, padding=True, truncation=True,
                  max_length=8, return_tensors="pt")
print(x_bert["input_ids"].shape)
for fila, mask in zip(x_bert["input_ids"], x_bert["attention_mask"]):
    print(" ".join(tok_bert.convert_ids_to_tokens(fila)))
    print(mask.tolist())