Guía paso a paso para crear un canal completo de detección y redacción de PII con el filtro de privacidad OpenAI

En este tutorial, creamos una canalización completa de estilo producción para detectar y redactar información de identificación personal utilizando el filtro de privacidad OpenAI. Comenzamos configurando el entorno y cargando un modelo de clasificación de tokens que identifica múltiples categorías de datos confidenciales, incluidos nombres, correos electrónicos, números de teléfono, direcciones y secretos. Luego diseñamos funciones auxiliares para normalizar etiquetas, extraer tramos estructurados y transformar los resultados del modelo sin procesar en formatos utilizables. A partir de ahí, implementamos un sistema de redacción configurable que nos permite reemplazar entidades sensibles con marcadores de posición significativos, preservando la privacidad y brindando claridad contextual. A lo largo del proceso, probamos la canalización en ejemplos seleccionados, convertimos los resultados en marcos de datos estructurados y preparamos el sistema para el procesamiento por lotes y el uso en el mundo real.

!pip install -q -U los transformadores aceleran la antorcha pandas matplotlib huggingface_hub import os, re, json, time, textwrap, advertencias de pathlib import Ruta de las colecciones import Counter import pandas as pd import matplotlib.pyplot as plt import torch from transformadores import AutoTokenizer, AutoModelForTokenClassification, pipeline advertencias.filterwarnings(“ignore”) MODEL_ID = “openai/privacy-filter” OUT_DIR = Ruta(“/content/privacy_filter_outputs”) OUT_DIR.mkdir(parents=True, exist_ok=True) dispositivo = 0 si torch.cuda.is_available() más -1 torch_dtype = torch.bfloat16 si torch.cuda.is_available() si no torch.float32 print(“Dispositivo:”, “GPU” si torch.cuda.is_available() si no “CPU”) print(“Tipo de antorcha:”, tipo_antorcha) print(“Modelo:”, MODEL_ID)

Instalamos todas las bibliotecas necesarias y configuramos el entorno de ejecución de la canalización. Configuramos la selección de dispositivos e inicializamos rutas para almacenar salidas. También imprimimos los detalles del sistema para confirmar que todo está listo antes de cargar el modelo.

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID) model = AutoModelForTokenClassification.from_pretrained( MODEL_ID, torch_dtype=torch_dtype, device_map=”auto” if torch.cuda.is_available() else Ninguno) clasificador = pipeline( task=”token-classification”, model=model, tokenizer=tokenizer, aggregation_strategy=”simple”, dispositivo=dispositivo si no torch.cuda.is_available() else Ninguno ) LABEL_MASKS = { “número_cuenta”: “[ACCOUNT_NUMBER]”, “dirección_privada”: “[PRIVATE_ADDRESS]”, “correo electrónico_privado”: “[PRIVATE_EMAIL]”, “persona_privada”: “[PRIVATE_PERSON]”, “teléfono_privado”: “[PRIVATE_PHONE]”, “url_privada”: “[PRIVATE_URL]”, “fecha_privada”: “[PRIVATE_DATE]”, “secreto”: “[SECRET]” }

Instalamos todas las bibliotecas necesarias y configuramos el entorno de ejecución de la canalización. Configuramos la selección de dispositivos e inicializamos rutas para almacenar salidas. También imprimimos los detalles del sistema para confirmar que todo está listo antes de cargar el modelo.

def normalize_label(etiqueta): etiqueta = etiqueta.replace(“B-“, “”).replace(“I-“, “”).replace(“E-“, “”).replace(“S-“, “”) return label.strip() def detect_pii(texto): raw = clasificador(texto) abarca = []
para el elemento sin formato: etiqueta = normalize_label(item.get(“entity_group”, item.get(“entity”, “”))) si etiqueta == “O” o no etiqueta: continuar spans.append({ “label”: etiqueta, “score”: float(item[“score”]), “texto”: elemento[“word”]”inicio”: int(elemento[“start”]), “fin”: int(elemento[“end”]) }) abarca = ordenado (extensión, clave = lambda x: (x[“start”]x[“end”])) return abarca def redact_text(text, spans, min_score=0.50, mode=”typed”): filtrado = [s for s in spans if s[“score”] >= min_score]filtrado = ordenado (filtrado, clave = lambda x: x[“start”]reverso=Verdadero) redactado = texto para intervalo en filtrado: reemplazo = LABEL_MASKS.get(intervalo[“label”]”[PII]”) si modo == “escribió” más “[REDACTED]”redactado = redactado[:span[“start”]]+ reemplazo + redactado[span[“end”]:]return redactado def Privacy_report(texto, min_score=0.50): spans = detect_pii(texto) redactado = redact_text(texto, spans, min_score=min_score) return { “original_text”: texto, “redacted_text”: redactado, “span_count”: len([s for s in spans if s[“score”] >= min_score]), “abarca”: [s for s in spans if s[“score”] >= puntuación_min]}

Definimos funciones auxiliares para normalizar etiquetas y extraer intervalos de PII de las predicciones del modelo. Implementamos una función de redacción que reemplaza segmentos sensibles según umbrales de confianza. Combinamos todo en una única función de informes que devuelve resultados estructurados.

textos_muestra = [
“My name is Alice Smith and my email is [email protected]. Llámeme al +1 415 555 0189.”, “El paciente Rohan Mehta me visitó el 11 de abril de 2025 y vive en 221B Baker Street, Londres.”, “Utilice la clave API sk-test-51HxYzDemoSecret987 y envíe la factura a [email protected].”, “El sitio web público es https://example.com, pero el portal privado de Jane Doe es https://jane-private.example.net.”, “El número de cuenta 123456789012 se vinculó a Ahmed Khan el 12 de marzo de 2024.”, “Esta oración no tiene información privada y en su mayoría debe permanecer sin cambios”. ]informes = []
para i, texto en enumerar (sample_texts, 1): informe = informe_privacidad (texto, puntuación mínima = 0,50) informe[“example_id”] = i informes.append(informe) para r en informes: print(“\n” + “=” * 100) print(“Ejemplo:”, r[“example_id”]) imprimir(“Original:”, r[“original_text”]) print(“Redactado:”, r[“redacted_text”]) print(“Intervalos detectados:”) print(json.dumps(r[“spans”]sangría=2, asegurar_ascii=False)) filas = []
para r en informes: para s en r[“spans”]: filas.append({ “example_id”: r[“example_id”]”etiqueta”: s[“label”]”puntuación”: s[“score”]”texto_detectado”: s[“text”]”inicio”: s[“start”]”fin”: s[“end”]”texto_original”: r[“original_text”]”texto_redactado”: r[“redacted_text”]
}) df = pd.DataFrame(filas) mostrar(df)

Creamos entradas de muestra y las ejecutamos a través del proceso para probar la detección y redacción. Recopilamos resultados estructurados e imprimimos texto original y redactado para comparar. También convertimos los resultados en un marco de datos para facilitar el análisis.

json_path = OUT_DIR / “privacy_filter_reports.json” csv_path = OUT_DIR / “privacy_filter_spans.csv” con open(json_path, “w”, encoding=”utf-8″) como f: json.dump(reports, f, indent=2, sure_ascii=False) df.to_csv(csv_path, index=False) print(“\nJSON guardado:”, json_path) print(“CSV guardado:”, csv_path) if len(df): label_counts = df[“label”].value_counts() plt.figure(figsize=(10, 5)) label_counts.plot(kind=”bar”) plt.title(“Categorías de PII detectadas”) plt.xlabel(“Categoría de PII”) plt.ylabel(“Recuento de intervalos detectados”) plt.xticks(rotation=35, ha=”right”) plt.tight_layout() plt.show() plt.figure(figsize=(10, 5)) gl[“score”].plot(kind=”hist”, bins=10) plt.title(“Distribución de confianza de detección”) plt.xlabel(“Puntuación de confianza”) plt.ylabel(“Frecuencia”) plt.tight_layout() plt.show() def compare_thresholds(texto, umbrales=(0,30, 0,50, 0,70, 0,90)): abarca = detect_pii(texto) resultados = []
para umbral en umbrales: mantenido = [s for s in spans if s[“score”] >= umbral]results.append({ “umbral”: umbral, “span_count”: len(kept), “redacted_text”: redact_text(texto, intervalos, min_score=umbral) }) return pd.DataFrame(resultados) umbral_demo = comparar_umbrales(textos_de muestra[0]) mostrar(threshold_demo)

Guardamos los resultados procesados ​​en formatos JSON y CSV para su persistencia y reutilización. Visualizamos las categorías de PII detectadas y las distribuciones de confianza mediante gráficos. También analizamos cómo el cambio de umbrales afecta el comportamiento de detección y redacción.

long_document = “”” Transcripción de atención al cliente: Agente: Hola, ¿puedo confirmar su nombre? Cliente: Mi nombre es PSP. Agente: Gracias. ¿Podría confirmar su correo electrónico? Cliente: [email protected]. Agente: ¿Y su número de teléfono? Cliente: +91 xxxxx xxxxx. Agente: Su dirección de servicio es 45 MG Road, Bengaluru, Karnataka. Cliente: Sí. Además, mi correo electrónico de respaldo es [email protected]. Agente: No comparta contraseñas ni OTP. Cliente: El token temporal que recibí es ghp_demoSecretToken123456. “”” informe_largo = informe_privacidad(documento_largo, puntuación mínima=0.50) print(“\nREDACCIÓN DE DOCUMENTOS LARGO”) print(“=” * 100) print(informe_largo)[“redacted_text”]) print(“\nTramos estructurados:”) print(json.dumps(long_report[“spans”]sangría=2, asegurar_ascii=False)) def pii_audit_table(textos, min_score=0.50): audit_rows = []
para idx, texto en enumerar (textos, 1): resultado = informe_privacidad (texto, puntuación mínima = puntuación mínima) etiquetas = Contador ([s[“label”] para s en resultado[“spans”]]) audit_rows.append({ “id”: idx, “original_chars”: len(texto), “redacted_chars”: len(resultado[“redacted_text”]), “span_count”: resultado[“span_count”]”labels_found”: dict(etiquetas), “redacted_text”: resultado[“redacted_text”]
}) devolver pd.DataFrame(audit_rows) audit_df = pii_audit_table(sample_texts + [long_document]min_score=0.50) display(audit_df) audit_path = OUT_DIR / “privacy_filter_audit.csv” audit_df.to_csv(audit_path, index=False) print(“CSV de auditoría guardado:”, audit_path) custom_text = input(“\nIngrese su propio texto para la redacción de PII, o presione Enter para omitir:\n”) si texto_personalizado.strip(): informe_personalizado = informe_privacidad(texto_personalizado, puntuación mínima = 0,50) print(“\nOriginal:”) print(informe_personalizado[“original_text”]) print(“\nRedactado:”) print(informe_personalizado[“redacted_text”]) print(“\nSpans:”) print(json.dumps(informe_personalizado[“spans”]sangría=2, asegurar_ascii=False)) else: print(“Se omitió entrada personalizada.”) print(“\nTutorial completo.”)

Probamos el proceso en un documento más largo y realista para evaluar su solidez. Generamos un resumen estilo auditoría que muestra recuentos y categorías de PII detectada. También permitimos la entrada personalizada del usuario para que podamos ejecutar el filtro de privacidad de forma interactiva.

En conclusión, desarrollamos un flujo de trabajo de filtrado de privacidad sólido y extensible que va más allá de la simple detección. Evaluamos sistemáticamente las predicciones de los modelos, aplicamos umbrales de confianza y comparamos diferentes estrategias de redacción para comprender su impacto. También generamos informes estructurados, visualizamos patrones de detección y exportamos resultados en formatos JSON y CSV para auditoría e integración posterior. Este enfoque nos permite incorporar salvaguardias de privacidad confiables en los canales de datos, garantizando que la información confidencial se identifique de manera constante y se maneje de manera responsable, manteniendo al mismo tiempo la usabilidad de los datos subyacentes.

Consulte los códigos completos aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 130.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