En este tutorial, desarrollamos un flujo de trabajo de análisis de sentimientos de un extremo a otro utilizando el conjunto de datos de revisión de películas grandes IMDb de NLP de Stanford y comparamos el aprendizaje automático clásico con el ajuste fino del transformador con parámetros eficientes. Comenzamos estableciendo un entorno reproducible y auditando el conjunto de datos para determinar el orden de las clases, el sesgo en la duración de las revisiones, las fugas de duplicados y los artefactos de preprocesamiento antes de entrenar una línea de base sólida de TF-IDF y regresión logística. Luego ajustamos DistilBERT con LoRA a través de PEFT, lo evaluamos usando precisión, macro-F1, ROC-AUC, matrices de confusión y curvas ROC, y examinamos la selección de umbral y la calibración de probabilidad a través del error de calibración esperado y el análisis de confiabilidad. Más allá de las métricas de titulares, investigamos los errores de confianza, el rendimiento en la duración de las revisiones, la prominencia de la oclusión a nivel de palabra y el truncamiento de cabeza versus cola para comprender cómo el modelo alcanza sus predicciones y dónde las limitaciones del contexto largo afectan el rendimiento. Finalmente, utilizamos la división de IMDb sin etiquetar para un pseudoetiquetado basado en la confianza, comparamos el modelo semisupervisado resultante con nuestra línea de base y guardamos el transformador fusionado para una inferencia de sentimiento reutilizable.
importar importlib.util, subproceso, sys, os, tiempo, aleatorio, advertencias, inspeccionar, hashlib advertencias.filterwarnings("ignore") os.environ["TOKENIZERS_PARALLELISM"] = "false" os.environ["WANDB_DISABLED"] = "true" _REQUIRED = { "transformers": "transformers", "datasets": "datasets", "peft": "peft", "accelerate": "accelerate", "sklearn": "scikit-learn", } _missing = [pkg para mod, paquete en _REQUIRED.items() si importlib.util.find_spec(mod) es Ninguno] si _missing: print(f"Instalando: {', '.join(_missing)} …") subprocess.run([sys.executable, "-m", "pip", "install", "-q", *_missing], check=True) print("Listo. (Si las importaciones fallan a continuación, reinicie el tiempo de ejecución y vuelva a ejecutar.)n") importar numpy como np importar pandas como pd importar antorcha importar matplotlib.pyplot como plt desde conjuntos de datos importar load_dataset desde sklearn.feature_extraction.text importar TfidfVectorizer desde sklearn.linear_model importar LogisticRegression desde sklearn.pipeline importar make_pipeline desde sklearn.metrics importar (accuracy_score, f1_score, roc_auc_score, Classification_report, confusion_matrix, roc_curve) desde transformadores importar (AutoTokenizer, AutoModelForSequenceClassification, TrainingArguments, Trainer, DataCollatorWithPadding, EarlyStoppingCallback, set_seed) de peft import LoraConfig, get_peft_model, TaskType def _disable_torchao_probe(): parcheado =[]intente: importar peft.import_utils como _piu _piu.is_torchao_available = lambda: Falso patched.append("peft.import_utils") excepto Excepción: pase para _name, _mod en la lista(sys.modules.items()): if _name.startswith("peft") y hasattr(_mod, "is_torchao_available"): _mod.is_torchao_available = lambda: False patched.append(_name) devuelve el intento parcheado: importar torchao como _tao _v = getattr(_tao, "__version__", "?") if tuple(int(x) for x in _v.split(".")[:2]) < (0, 16): print(f"[compat] torchao {_v} < 0.16 -> deshabilitar la sonda torchao de PEFT: " f"{', '.join(_disable_torchao_probe())}") excepto excepción: _disable_torchao_probe() SEED = 42 MODEL_NAME = "distilbert-base-uncased" MAX_LEN = 256 N_TRAIN = 5000 N_EVAL = 2000 N_UNSUP = 3000 EPOCHS = 2 BATCH = 16 LR = 3e-4 FULL_RUN = Falso si FULL_RUN: N_TRAIN, N_EVAL, EPOCHS = 25000, 25000, 3 set_seed(SEED); semilla.aleatoria(SEMILLA); np.random.seed(SEMILLA); torch.manual_seed(SEED) DISPOSITIVO = "cuda" if torch.cuda.is_available() else "cpu" print("=" * 79) print(f"device={DEVICE} | torch={torch.__version__} | " f"gpu={torch.cuda.get_device_name(0) if DEVICE=='cuda' else 'n/a'}") print("=" * 79) t0 = time.time() raw = load_dataset("stanfordnlp/imdb") print(raw, f"nloaded in {time.time()-t0:.1f}sn") print("— ejemplo (truncado) —") print("etiqueta:", raw["tren"][0]["etiqueta"], "|", sin procesar["tren"][0]["text"][:300], "…n") first_labels = np.array(raw["train"]["label"][:5]) last_labels = np.array(raw["train"]["label"][-5:]) print(f"TRAP #1 – orden dividido: primeras 5 etiquetas {first_labels}, " f"últimas 5 etiquetas {last_labels} -> SIEMPRE mezclar antes del submuestreo." len(test_full)))) print(f" después de barajar+submuestra: train balance = " f"{np.bincount(train_ds['label'])}, eval balance = {np.bincount(eval_ds['label'])}") lens = np.array([len(t.split()) for t in train_full["text"]]) q = np.percentile(lente, [50, 75, 90, 95, 99]) print(f"nTRAP #2 – longitud (palabras): mediana={q[0]:.0f}p75={q[1]:.0f}p90={q[2]:.0f} "f"p95={q[3]:.0f}p99={q[4]:.0f} max={lens.max()}") print(f" ~{(lens > MAX_LEN*0.75).mean()*100:.1f}% de las revisiones excede MAX_LEN={MAX_LEN} " f"tokens (palabras aproximadas->factor de tokens 1.3). La sección 9 mide lo que cuesta.") h_tr = {hashlib.md5(t.encode()).hexdigest() for t in raw["train"]["text"]} h_te = {hashlib.md5(t.encode()).hexdigest() for t in raw["test"]["text"]} print(f"nTRAP #3 – fuga: {len(h_tr & h_te)} revisiones duplicadas exactas en " f"train/test; {len(raw['train'])-len(h_tr)} duplicados dentro del propio tren.") def clean return t.reemplazar("
", " ").reemplazar("
", " ").strip() plt.figure(figsize=(11, 3.2)) plt.subplot(1, 2, 1) plt.hist(np.clip(lens, 0, 1000), bins=60) plt.axvline(MAX_LEN, ls="–", color="k", label=f"MAX_LEN={MAX_LEN}") plt.title("Longitud de la revisión (palabras, recortadas a 1000)"); plt.legend() plt.subplot(1, 2, 2) plt.bar(["neg", "pos"], np.bincount(raw["train"]["label"])) plt.title("Equilibrio de clases de tren (perfectamente equilibrado)") plt.tight_layout(); plt.show()
Configuramos el entorno Colab, instalamos las bibliotecas necesarias, aplicamos la corrección de compatibilidad PEFT-torchao y configuramos semillas deterministas para experimentos reproducibles. Cargamos el conjunto de datos de Stanford IMDb, mezclamos y submuestreamos las divisiones de entrenamiento y prueba, e inspeccionamos el equilibrio de clases, las distribuciones de la duración de la revisión, las fugas duplicadas y los artefactos HTML. También visualizamos la duración de las revisiones y las frecuencias de las etiquetas para comprender la estructura del conjunto de datos antes de construir cualquier modelo.
print("n" + "=" * 79 + "n3. TF-IDF BASELINEn" + "=" * 79) Xtr = [clean Xte = [clean t0 = time.time() tfidf_clf = make_pipeline( TfidfVectorizer(ngram_range=(1, 2), min_df=2, max_features=300_000, sublinear_tf=True, strip_accents="unicode"), LogisticRegression(C=8.0, max_iter=2000, n_jobs=-1), ) tfidf_clf.fit(Xtr, ytr) p_tfidf = tfidf_clf.predict_proba(Xte)[:, 1] acc_tfidf = precision_score(yte, p_tfidf > 0.5) auc_tfidf = roc_auc_score(yte, p_tfidf) print(f"entrenado en {time.time()-t0:.1f}s -> acc={acc_tfidf:.4f} auc={auc_tfidf:.4f}") vec, lr = tfidf_clf.steps[0][1], tfidf_clf.pasos[1][1]hazañas, coefs = np.array(vec.get_feature_names_out()), lr.coef_[0]order = np.argsort(coefs) print("nlos n-gramas más NEGATIVOS:", ", ".join(feats[order[:12]])) print("los n-gramas más POSITIVOS:", ", ".join(feats[order[-12:]][::-1])) print("n" + "=" * 79 + "n4. LoRA FINE-TUNINGn" + "=" * 79) tok = AutoTokenizer.from_pretrained(MODEL_NAME) def tokenize(batch): return tok([clean tr_tok = (train_ds.map(tokenize, batched=True, remove_columns=["text"]) .rename_column("label", "labels")) ev_tok = (eval_ds.map(tokenize, loteed=True, remove_columns=["text"]) .rename_column("label", "labels")) base = AutoModelForSequenceClassification.from_pretrained( MODEL_NAME, num_labels=2, id2label={0: "NEGATIVO", 1: "POSITIVO"}, label2id={"NEGATIVO": 0, "POSITIVO": 1},) lora_cfg = LoraConfig( task_type=TaskType.SEQ_CLS, r=16, lora_alpha=32, lora_dropout=0.05, target_modules=["q_lin", "v_lin"], module_to_save=["pre_classifier", "classifier"],) intente: model = get_peft_model(base, lora_cfg) excepto ImportError como e: _disable_torchao_probe() print(f"[compat] reintento después de falla de la sonda backend: {e}") model = get_peft_model(base, lora_cfg) model.print_trainable_parameters() def compute_metrics(eval_pred): logits, etiquetas = eval_pred probs = torch.softmax(torch.tensor(logits), dim=-1).numpy()[:, 1] preds = (probs > 0.5).astype(int) return {"accuracy": precision_score(labels, preds), "f1_macro": f1_score(labels, preds, Average="macro"), "roc_auc": roc_auc_score(labels, probs)} _ta = inspect.signature(TrainingArguments.__init__).parameters _eval_key = "eval_strategy" if "eval_strategy" in _ta else "evaluación_strategy" ta_kwargs = dict(output_dir="./imdb_lora", learning_rate=LR, per_device_train_batch_size=BATCH, per_device_eval_batch_size=BATCH * 2, num_train_epochs=EPOCHS, Weight_decay=0.01, Warmup_ratio=0.06, logging_steps=50, save_strategy="epoch", save_total_limit=1, load_best_model_at_end=True, metric_for_best_model="accuracy", fp16=(DEVICE == "cuda"), report_to="none", seed=SEED,) ta_kwargs[_eval_key] = "epoch" _tr = inspect.signature(Trainer.__init__).parameters _tok_key = "processing_class" if "processing_class" in _tr else "tokenizer" trainer = Trainer( model=model, args=TrainingArguments(**ta_kwargs), train_dataset=tr_tok, eval_dataset=ev_tok, data_collator=DataCollatorWithPadding(tok), compute_metrics=compute_metrics, callbacks=[EarlyStoppingCallback(early_stopping_patience=2)], **{_tok_key: tok}, ) t0 = time.time() trainer.train() print(f"nafinado en {(time.time()-t0)/60:.1f} min")
Entrenamos una sólida línea de base de TF-IDF y Regresión Logística e inspeccionamos los n-gramas positivos y negativos más influyentes para establecer un punto de referencia interpretable. Luego tokenizamos las revisiones de IMDb y configuramos DistilBERT con adaptadores LoRA que actualizan solo un pequeño subconjunto de parámetros del modelo mientras mantienen la columna vertebral en gran medida congelada. Usamos Hugging Face Trainer con relleno dinámico, parada temprana, precisión mixta y múltiples métricas de evaluación para ajustar el transformador de manera eficiente.
print("n" + "=" * 79 + "n5. EVALUACIÓNn" + "=" * 79) pred_out = trainer.predict(ev_tok) p_lora = torch.softmax(torch.tensor(pred_out.predictions), dim=-1).numpy()[:, 1] y_true = np.array(pred_out.label_ids) yhat = (p_lora > 0.5).astype(int) print(classification_report(y_true, yhat, target_names=["neg", "pos"], dígitos=4)) cm = confusion_matrix(y_true, yhat) fig, ax = plt.subplots(1, 2, figsize=(11, 4)) ax[0].imshow(cm, cmap="Blues") para i en el rango(2): para j en el rango(2): ax[0].text(j, i, cm[i, j], ha="centro", va="centro", tamaño de fuente=14) hacha[0].set_xticks([0, 1], ["pred neg", "pred pos"]) eje[0].set_yticks([0, 1], ["verdadero neg", "verdadero pos"]); hacha[0].set_title("Matriz de confusión") para nombre, p en [("TF-IDF", p_tfidf), ("DistilBERT+LoRA", p_lora)]: fpr, tpr, _ = roc_curve(y_true, p) ax[1].plot(fpr, tpr, label=f"{nombre} (AUC={roc_auc_score(y_true, p):.4f})") hacha[1].plot([0, 1], [0, 1], "k–", lw=0.8) eje[1].set_xlabel("FPR"); hacha[1].set_ylabel("TPR"); hacha[1].set_title("República de China"); hacha[1].legend() plt.tight_layout(); plt.show() print("n" + "=" * 79 + "n6. UMBRAL Y CALIBRACIÓNn" + "=" * 79) ths = np.linspace(0.05, 0.95, 91) acc = [(y_true == (p_lora > t)).mean() for t in ths] best_t = esto[int(np.argmax(accs))] print(f"[correo electrónico protegido] = {accs[45]:.4f} | mejor umbral = {best_t:.2f} -> acc = {max(accs):.4f}") def expected_calibration_error(probs, etiquetas, n_bins=10): """ECE: |confianza – precisión| promediado sobre contenedores de confianza.[],[]para lo, hola en zip(bins[:-1], bins[1:]): m = (conf > lo) & (conf <= hola) si m.sum() == 0: continuar ece += m.mean() * abs(conf[m].mean() – correcto[m].mean()) xs.append(conf[m].mean()); ys.append(correct[m].mean()) return ece, np.array(xs), np.array(ys) ece, cx, cy = expected_calibration_error(p_lora, y_true) print(f"Error de calibración esperado = {ece:.4f} (0 = perfectamente calibrado)") plt.figure(figsize=(9, 3.2)) plt.subplot(1, 2, 1); plt.plot(ths, accs); plt.axvline(best_t, ls="–", color="r") plt.xlabel("umbral"); plt.ylabel("precisión"); plt.title("Barrido de umbral") plt.subplot(1, 2, 2); plt.plot([0.5, 1], [0.5, 1], "k–", lw=0.8) plt.plot(cx, cy, "o-"); plt.xlabel("confianza media"); plt.ylabel("precisión empírica") plt.title(f"Diagrama de confiabilidad (ECE={ece:.3f})") plt.tight_layout(); plt.mostrar()
Evaluamos el modelo DistilBERT-LoRA ajustado utilizando métricas de clasificación, una matriz de confusión y curvas ROC mientras comparamos directamente su rendimiento ROC-AUC con la línea de base TF-IDF. Analizamos los umbrales de clasificación para determinar si el límite de probabilidad predeterminado de 0,5 ofrece la mejor precisión en nuestro conjunto de evaluación. También calculamos el error de calibración esperado y construimos un diagrama de confiabilidad para medir qué tan cerca corresponde la confianza prevista del modelo a su corrección real.
print("n" + "=" * 79 + "n7. ANÁLISIS DE ERRORESn" + "=" * 79) err = pd.DataFrame({ "text": eval_ds["text"], "y": y_true, "p_pos": p_lora, "n_words": [len(t.split()) for t in eval_ds["text"]], }) err["pred"] = (err.p_pos > 0.5).astype(int) err["correct"] = err.pred == err.y err["confidence"] = np.maximum(err.p_pos, 1 – err.p_pos) print("— 3 errores más CONFIDENTES (donde el modelo es claramente incorrecto) —") for _, r in err[~err.correct].nlargest(3, "confianza").iterrows(): print(f"n[true={'pos' if ry else 'neg'} pred={'pos' if r.pred else 'neg'} " f"conf={r.confidence:.3f} palabras={r.n_words}]") print(clean(r.text)[:400].replace("n", " "), "…") err["bucket"] = pd.qcut(err.n_words, 4, etiquetas=["short", "med", "long", "v.long"]) by_len = err.groupby("bucket", observe=True).agg(acc=("correct", "mean"), n=("correct", "size")) print("n— precisión por duración de la revisión (el truncamiento perjudica las revisiones largas) —") print(by_len.to_string()) print("n" + "=" * 79 + "n8. OCCLUSION SALIENCYn" + "=" * 79) infer_model = model.merge_and_unload() infer_model.to(DEVICE).eval() @torch.no_grad() def predict_proba(textos, bs=64): fuera =[]para i en rango(0, len(textos), bs): enc = tok([clean max_length=MAX_LEN, padding=True, return_tensors="pt").to(DEVICE) out.append(torch.softmax(infer_model(**enc).logits, dim=-1)[:, 1].cpu().numpy()) return np.concatenate(out) def occlusion(text, max_words=60): palabras = clean(texto).split()[:max_words] base = predict_proba([" ".join(palabras)])[0]variantes = [" ".join(palabras[:i] + palabras[i + 1:]) para i en el rango(len(palabras))] drop = predict_proba(variants) devuelve palabras, base – drop, muestra base = err[err.correct].nlargest(1, "confidence").iloc[0]palabras, contrib, base_p = occlusion(sample.text) print(f"P(positive) para el extracto completo = {base_p:.3f} " f"(true label = {'pos' if sample.y else 'neg'})n") top = np.argsort(np.abs(contrib))[-15:] plt.figure(figsize=(7, 5)) plural empuja NEGATIVO") plt.tight_layout(); plt.show() print("n" + "=" * 79 + "n9. HEAD vs TAIL TRUNCATIONn" + "=" * 79) probe = err.nlargest(600, "n_words") W = 180 head_txt = [" ".join(clean tail_txt = [" ".join(clean yp = probe.y.values acc_head = ((predict_proba(head_txt) > 0.5).astype(int) == yp).mean() acc_tail = ((predict_proba(tail_txt) > 0.5).astype(int) == yp).mean() print(f"en las reseñas más largas de {len(probe)}, usando solo {W} palabras:") print(f" primeras {W} palabras -> acc {acc_head:.4f}") print(f" últimas {W} palabras -> acc {acc_tail:.4f}") print(" Conclusión práctica: si la cola gana, alimente cabeza+cola al modelo o " "aumente MAX_LEN, en lugar de truncar ciegamente desde la izquierda.")
Examinamos las predicciones incorrectas más confiables del modelo y las revisiones grupales por extensión para identificar patrones de falla relacionados con el truncamiento y ejemplos difíciles. Fusionamos los adaptadores LoRA en el modelo subyacente y aplicamos una oclusión de dejar una palabra fuera para estimar qué palabras impulsan las predicciones individuales hacia un sentimiento positivo o negativo. Luego comparamos predicciones basadas en las partes inicial y final de revisiones largas para determinar dónde reside la información de sentimiento más sólida.
print("n" + "=" * 79 + "n10. PSEUDO-LABELLINGn" + "=" * 79) unsup = raw["unsupervised"].shuffle(seed=SEED).select(range(N_UNSUP)) p_uns = predict_proba(unsup["text"]) keep = (p_uns > 0.95) | (p_uns < 0.05) pl_texts = [clean pl_labels = (p_uns[keep] > 0.5).astype(int) print(f"kept {keep.sum()}/{N_UNSUP} pseudoetiquetas en conf>0.95 " f"(balance: {np.bincount(pl_labels)})") aug = make_pipeline( TfidfVectorizer(ngram_range=(1, 2), min_df=2, max_features=300_000, sublinear_tf=True, strip_accents="unicode"), LogisticRegression(C=8.0, max_iter=2000, n_jobs=-1), ).fit(Xtr + pl_texts, np.concatenate([ytr, pl_labels])) acc_aug = precision_score(yte, aug.predict(Xte)) print(f"TF-IDF baseline: {acc_tfidf:.4f}") print(f"TF-IDF + pseudoetiquetas: {acc_aug:.4f} (Δ {acc_aug-acc_tfidf:+.4f})") print("Advertencia: las ganancias son limitado por el maestro La autoformación también amplifica " "los prejuicios del maestro: valide siempre con datos limpios y reservados.") print("n" + "=" * 79 + "n11. SAVE & INFERn" + "=" * 79) SAVE_DIR = "./imdb-distilbert-lora-merged" infer_model.save_pretrained(SAVE_DIR); tok.save_pretrained(SAVE_DIR) print(f"saved fusioned model to {SAVE_DIR}/ (load with " f"AutoModelForSequenceClassification.from_pretrained('{SAVE_DIR}'))") demos = [ "Una clase magistral en tensión. El acto final dejó a todo el teatro en silencio.", "Dos horas que nunca volveré. Actuación rígida, trama incoherente.", "No es el desastre que prometió el tráiler, pero tampoco llega nunca.", ] for d, p in zip(demos, predict_proba(demos)): print(f" P(pos)={p:.3f} -> {'POSITIVO' if p > 0.5 else 'NEGATIVO'} | {d}") print("n" + "=" * 79) print(f"SUMMARY (n_train={N_TRAIN}, n_eval={N_EVAL}, max_len={MAX_LEN})") print("=" * 79) print(pd.DataFrame([ {"model": "TF-IDF + LogReg", "accuracy": acc_tfidf, "roc_auc": auc_tfidf}, {"model": "TF-IDF + pseudo-labels", "accuracy": acc_aug, "roc_auc": float("nan")}, {"model": "DistilBERT + LoRA", "accuracy": precision_score(y_true, yhat), "roc_auc": roc_auc_score(y_true, p_lora)}, ]).to_string(index=False)) print(""" PRÓXIMOS EXPERIMENTOS – Establezca FULL_RUN = True para el punto de referencia real de 25k/25k (~40 min en un T4). – Cambie MODEL_NAME a 'roberta-base' (target_modules=['query','value']) o 'answerdotai/ModernBERT-base' para una ventana de contexto de 8k, sin truncamiento. – Truncamiento de cabeza y cola: primeros 128 + últimos 128 tokens, motivados por la sección 9. – Eliminación del rango r de LoRA en {4, 8, 16, 64} y precisión del gráfico frente a parámetros entrenables – Reemplace la pseudoetiqueta del maestro con un conjunto e itere el autoentrenamiento – Empuje al Hub: huggingface_hub.login() y luego infer_model.push_to_hub(…).
Usamos el transformador ajustado para generar pseudoetiquetas de alta confianza para ejemplos de la división sin etiquetar de IMDb y agregamos estos ejemplos al corpus de entrenamiento TF-IDF. Comparamos el clasificador aumentado con la línea de base original para medir si el autoentrenamiento semisupervisado mejora la precisión predictiva. Finalmente, guardamos el modelo DistilBERT y el tokenizador combinados, ejecutamos inferencia de sentimientos en revisiones personalizadas y resumimos el rendimiento de todos los modelos desarrollados a lo largo del tutorial.
En conclusión, desarrollamos un riguroso proceso de clasificación de sentimientos que va mucho más allá del simple ajuste de un transformador y la precisión de los informes. Establecimos una línea de base competitiva de TF-IDF, entrenamos a DistilBERT de manera eficiente con LoRA y evaluamos tanto la calidad predictiva como la confiabilidad de la probabilidad mientras identificamos cómo la duración de la revisión, el truncamiento y los errores de alta confianza influyen en el desempeño en el mundo real. También interpretamos predicciones individuales a través de prominencia basada en oclusión, probamos si la información de sentimiento se concentra cerca del principio o al final de revisiones largas y ampliamos el aprendizaje supervisado con pseudoetiquetas de alta confianza del conjunto de datos sin etiquetar.
Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 150.000 ml y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
¿Necesita asociarse con nosotros para promocionar su repositorio de GitHub O su página principal de Hugging O su lanzamiento de producto O seminario web, etc.? Conéctate con nosotros
A Sana Hassan, pasante de consultoría en Marktechpost y estudiante de doble titulación en IIT Madras, le apasiona aplicar la tecnología y la inteligencia artificial para abordar los desafíos del mundo real. Con un gran interés en resolver problemas prácticos, aporta una nueva perspectiva a la intersección de la IA y las soluciones de la vida real.
