Creación de un flujo de trabajo estable de Fable 5 Traces en Colab: llamadas a herramientas de análisis, auditoría de datos y líneas de base de capacitación
rprint(Panel.fit(“[bold]Línea de base 1: predecir el tipo de salida a partir del contexto utilizando Python Naive Bayes puro[/bold]”)) model_artifacts = {} classifier_df = df.dropna(subconjunto=[“output_type”]).copiar() clasificador_df = clasificador_df[
classifier_df[“output_type”].astype(str).str.len() > 0 ].copy() si classifier_df[“output_type”].nunique() >= 2 y len(classifier_df) >= 30: X_text = ( classifier_df[“context”]
.fillna(“”) .astype(str) .map(texto lambda: texto[:12000]) .tolist() ) y = clasificador_df[“output_type”].astype(str).tolist() train_indices, test_indices = estratificado_train_test_indices(y, test_size=0.2, seed=SEED) X_train = [X_text[i] para i en train_indices]y_train = [y[i] para i en train_indices]X_test = [X_text[i] para i en test_indices]y_test = [y[i] para i en test_indices]salida_tipo_clasificador = PureMultinomialNB( max_features=20000, min_df=2, alfa=1.0, ) salida_tipo_clasificador.fit(X_train, y_train) predicciones = salida_tipo_clasificador.predict(X_test) salida_tipo_metrics, salida_report_df = evaluar_predicciones(y_test, predicciones) salida_matrix_df = confusión_matrix_df(y_test, predicciones) salida_tipo_métricas[“train_rows”] = len(X_train) salida_tipo_métricas[“test_rows”] = len(X_test) tipo_salida_métricas[“vocab_size”] = len(output_type_classifier.vocab) rprint(“[bold]Informe del clasificador de tipo de salida:[/bold]”) display(output_report_df) display(output_matrix_df) output_report_df.to_csv(OUT_DIR / “output_type_classifier_report.csv”, index=False) output_matrix_df.to_csv(OUT_DIR / “output_type_confusion_matrix.csv”) top_token_records = []
para etiqueta en output_type_classifier.labels: para token, margen en output_type_classifier.top_tokens_for_class(label, n=25): top_token_records.append( { “label”: label, “token”: token, “score_margin”: margin, } ) pd.DataFrame(top_token_records).to_csv( OUT_DIR / “output_type_top_tokens.csv”, index=False, ) con open( OUT_DIR / “output_type_classifier_metrics.json”, “w”, encoding=”utf-8″, ) como archivo: json.dump(output_type_metrics, file, sure_ascii=False, indent=2) model_artifacts[“output_type_classifier_metrics”] = str( OUT_DIR / “output_type_classifier_metrics.json”) model_artifacts[“output_type_classifier_report”] = str( OUT_DIR / “tipo_salida_clasificador_report.csv”) model_artifacts[“output_type_confusion_matrix”] = str( OUT_DIR / “output_type_confusion_matrix.csv”) model_artifacts[“output_type_top_tokens”] = str( OUT_DIR / “output_type_top_tokens.csv”) más: rprint( “[yellow]Saltando el clasificador tipo_salida porque hay muy pocas ” “clases o filas.[/yellow]” ) tipo_salida_metrics = {} rprint(Panel.fit(“[bold]Línea de base 2: predecir nombre_herramienta a partir del contexto utilizando Python Naive Bayes puro[/bold]”)) herramienta_clasificador_df = df[
df[“output_type”].eq(“uso_herramienta”) y df[“tool_name”].fillna(“”).astype(str).str.len().gt(0) ].copy() si len(tool_classifier_df) >= 50 y tool_classifier_df[“tool_name”].nunique() >= 2: top_tools = herramienta_clasificador_df[“tool_name”].value_counts().head(12).index.tolist() herramienta_clasificador_df[“tool_label”] = herramienta_clasificador_df[“tool_name”].donde( herramienta_clasificador_df[“tool_name”].isin(top_tools), “__OTHER__”, ) y_tool = herramienta_clasificador_df[“tool_label”].astype(str).tolist() X_tool_text = ( herramienta_clasificador_df[“context”]
.fillna(“”) .astype(str) .map(texto lambda: texto[:12000]) .tolist() ) if len(set(y_tool)) >= 2: train_indices, test_indices = estratified_train_test_indices(y_tool, test_size=0.2, seed=SEED) X_train = [X_tool_text[i] para i en train_indices]y_train = [y_tool[i] para i en train_indices]X_test = [X_tool_text[i] para i en test_indices]y_test = [y_tool[i] para i en test_indices]tool_classifier = PureMultinomialNB( max_features=20000, min_df=2, alpha=1.0, ) tool_classifier.fit(X_train, y_train) tool_predictions = tool_classifier.predict(X_test) tool_metrics, tool_report_df = evalua_predicciones(y_test, tool_predictions) tool_matrix_df = confusión_matrix_df(y_test, herramientas_predicciones) herramientas_métricas[“train_rows”] = len(X_train) herramienta_métricas[“test_rows”] = len(X_test) herramienta_métricas[“vocab_size”] = len(tool_classifier.vocab) rprint(“[bold]Informe clasificador de herramientas:[/bold]”) display(tool_report_df) display(tool_matrix_df) tool_report_df.to_csv(OUT_DIR / “tool_name_classifier_report.csv”, index=False) tool_matrix_df.to_csv(OUT_DIR / “tool_name_confusion_matrix.csv”) top_tool_token_records = []
para etiqueta en tool_classifier.labels: para token, margen en tool_classifier.top_tokens_for_class(label, n=25): top_tool_token_records.append( { “label”: etiqueta, “token”: token, “score_margin”: margen, } ) pd.DataFrame(top_tool_token_records).to_csv( OUT_DIR / “tool_name_top_tokens.csv”, index=False, ) con open( OUT_DIR / “tool_name_classifier_metrics.json”, “w”, encoding=”utf-8″, ) como archivo: json.dump(tool_metrics, file, sure_ascii=False, indent=2) model_artifacts[“tool_name_classifier_metrics”] = str( OUT_DIR / “nombre_herramienta_clasificador_metrics.json”) model_artifacts[“tool_name_classifier_report”] = str( OUT_DIR / “nombre_herramienta_clasificador_report.csv”) model_artifacts[“tool_name_confusion_matrix”] = str( OUT_DIR / “nombre_herramienta_confusion_matrix.csv”) model_artifacts[“tool_name_top_tokens”] = str( OUT_DIR / “nombre_herramienta_top_tokens.csv”) más: rprint(“[yellow]Saltar el clasificador de herramientas porque las etiquetas se colapsaron en una clase.[/yellow]”) tool_metrics = {} más: rprint( “[yellow]Omitir el clasificador de herramientas porque hay muy pocas filas o ” “clases de herramientas de uso de herramientas.[/yellow]” ) herramientas_métricas = {} rprint(Panel.fit(“[bold]Creación de un sencillo asistente de búsqueda de palabras clave[/bold]”)) def search_rows(palabra clave, límite=5, search_cols=(“context”, “cot”, “completion”, “text_payload”)): palabra clave = str(keyword).lower() máscara = pd.Series(False, index=df.index) para la columna en search_cols: máscara = máscara | ( df[column]
.fillna(“”) .astype(str) .str.lower() .str.contains(re.escape(palabra clave), regex=True) ) hits = df[mask].head(límite) resultados = []
para _, fila en hits.iterrows(): resultados.append( { “uid”: fila.get(“uid”), “sesión”: fila.get(“sesión”), “tipo_salida”: fila.get(“tipo_salida”), “nombre_herramienta”: fila.get(“nombre_herramienta”), “context_preview”: vista previa_texto(row.get(“context”), 400), “payload_preview”: vista previa_texto(row.get(“text_payload”), 400), } ) devuelve resultados ejemplo_queries = [
“Bash”,
“Write”,
“browser”,
“test”,
“README”,
]
search_demo = { consulta: search_rows(query, limit=2) para consulta en example_queries } con open( OUT_DIR / “keyword_search_demo.json”, “w”, encoding=”utf-8″, ) como archivo: json.dump(search_demo, file, sure_ascii=False, indent=2) rprint(“[bold]Resultados de búsqueda de palabras clave de ejemplo:[/bold]”) rprint(safe_json_dumps(search_demo, max_chars=5000)) resumen = { “dataset_id”: DATASET_ID, “flat_jsonl_filename”: FLAT_JSONL_FILENAME, “output_directory”: str(OUT_DIR), “repo_file_summary”: file_summary, “rows”: int(len(df)), “columnas”: lista (df.columns), “distribución_tipo_salida”: ( df[“output_type”]
.fillna(“faltante”) .value_counts() .to_dict() ), “top_tools”: ( df.loc[df[“output_type”].eq(“tool_use”), “tool_name”].replace(“”, “unknown”) .value_counts() .head(20) .to_dict() ), “top_source_roots”: ( df[“source_root”]
.fillna(“desconocido”) .value_counts() .head(20) .to_dict() ), “length_summary”: { columna: { “media”: float(df[column].media()), “mediana”: flotante(df[column].mediana()), “p90”: flotante(df[column].cuantil(0.90)), “p95”: flotante(df[column].cuantil(0,95)), “max”: int(df[column].max()), } para la columna en [
“context_chars”,
“cot_chars”,
“completion_chars”,
“text_payload_chars”,
]
}, “posibles_filas_secretas”: int(df[“possible_secret_anywhere”].sum()), “plots”: plot_paths, “model_artifacts”: model_artifacts, “safe_exports”: { “train”: str(OUT_DIR / “fable5_no_cot_chat_train.jsonl”), “validation”: str(OUT_DIR / “fable5_no_cot_chat_validation.jsonl”), “test”: str(OUT_DIR / “fable5_no_cot_chat_test.jsonl”), }, “analysis_files”: { “csv”: str(OUT_DIR / “fable5_analysis_index.csv”), “pickle”: str(OUT_DIR / “fable5_analysis_index.pkl”), “keyword_search_demo”: str(OUT_DIR / “keyword_search_demo.json”), }, } con open( OUT_DIR / “analysis_summary.json”, “w”, encoding=”utf-8″, ) como archivo: json.dump(clean_for_json(summary), file, sure_ascii=False, indent=2, default=str) FENCE = chr(96) * 3 report_md = ( “# Informe del tutorial avanzado de Fable 5 Traces\n\n” “## Conjunto de datos\n\n” f”- Conjunto de datos: `{DATASET_ID}`\n” f”- JSONL plano: `{FLAT_JSONL_FILENAME}`\n” f”- Filas cargadas: `{len(df):,}`\n” f”- Sesiones de origen únicas: `{df[‘session’].nunique(dropna=True):,}`\n” f”- Modelos únicos: `{df[‘model’].nunique(dropna=True):,}`\n\n” “## Nota de seguridad importante\n\n” “Este tutorial trata el conjunto de datos como telemetría del agente. Previsualiza y analiza comandos, ” “llamadas a herramientas, ediciones de archivos y transcripciones de texto, pero nunca ejecuta comandos que se encuentran dentro ” “de los rastros.\n\n” f”Posibles patrones similares a secretos detectados: `{int(df[‘possible_secret_anywhere’].sum()):,}` filas.\n” “Exporta redactar patrones comunes tipo token/clave API.\n\n” “## Distribución del tipo de salida\n\n” f”{FENCE}json\n” f”{json.dumps(clean_for_json(summary[‘output_type_distribution’]), sangría=2, asegurar_ascii=False)}\n” f”{FENCE}\n\n” “## Herramientas principales\n\n” f”{FENCE}json\n” f”{json.dumps(clean_for_json(summary[‘top_tools’]), sangría=2, asegurar_ascii=False)}\n” f”{FENCE}\n\n” “## Archivos guardados\n\n” “- `analysis_summary.json`\n” “- `fable5_analysis_index.csv`\n” “- `fable5_analysis_index.pkl`\n” “- `fable5_no_cot_chat_train.jsonl`\n” “- `fable5_no_cot_chat_validation.jsonl`\n” “- `fable5_no_cot_chat_test.jsonl`\n” “- trazar archivos PNG\n” “- métricas del clasificador de referencia, cuando hay suficientes filas/clases disponibles\n\n” “## Próximos pasos recomendados\n\n” “1. Inspeccione `fable5_no_cot_chat_train.jsonl` antes de realizar cualquier ajuste.\n” “2. Tenga en cuenta la licencia del conjunto de datos antes de entrenar o redistribuir el modelo.\n” “3. Evite el entrenamiento directamente en salidas de terminal sin procesar sin filtrado adicional de privacidad y seguridad.\n” “4. Comience con la exportación de chat sin CoT a menos que su investigación requiera explícitamente supervisión de rastreo de razonamiento.\n” ) con open( OUT_DIR / “REPORT.md”, “w”, encoding=”utf-8″, ) como archivo: file.write(report_md) rprint( Panel.fit( f”[bold green]Tutorial completo.[/bold green]\n\n” f”Artefactos guardados en:\n{OUT_DIR}\n\n” f”Archivos clave:\n” f”- {OUT_DIR / ‘REPORT.md’}\n” f”- {OUT_DIR / ‘analysis_summary.json’}\n” f”- {OUT_DIR / ‘fable5_no_cot_chat_train.jsonl’}\n” f”- {OUT_DIR / ‘fable5_analysis_index.csv’}”, title=”Listo”, ) ) display( pd.DataFrame( { “artefacto”: [
“Report”,
“Summary JSON”,
“No-CoT train export”,
“No-CoT validation export”,
“No-CoT test export”,
“Analysis CSV”,
“Analysis pickle”,
“Keyword search demo”,
]”camino”: [
str(OUT_DIR / “REPORT.md”),
str(OUT_DIR / “analysis_summary.json”),
str(OUT_DIR / “fable5_no_cot_chat_train.jsonl”),
str(OUT_DIR / “fable5_no_cot_chat_validation.jsonl”),
str(OUT_DIR / “fable5_no_cot_chat_test.jsonl”),
str(OUT_DIR / “fable5_analysis_index.csv”),
str(OUT_DIR / “fable5_analysis_index.pkl”),
str(OUT_DIR / “keyword_search_demo.json”),
]} ) )