Evaluación macro de sistemas multiagente
Cuando un sistema agéntico falla, el problema suele ser mayor que una sola respuesta incorrecta. Una transferencia puede ocurrir demasiado tarde, un agente especialista puede pasar por alto la misma señal en muchas ejecuciones, o un proceso de revisión puede activarse para la clase de casos equivocada. Para mejorar el sistema, los equipos necesitan ver el comportamiento recurrente en toda la población de trazas.
Este manual describe un flujo de trabajo de macroevaluación para un sistema multiagente. Utilizamos un flujo de trabajo sintético de pedidos de vehículos eléctricos (EV) donde agentes especialistas manejan decisiones de precios, cumplimiento, suministro, enrutamiento de fábrica, programación y lanzamiento, mientras las condiciones del mercado y operativas cambian.
El notebook utiliza trazas sintéticas precalculadas y etiquetas de evaluación de nivel inferior guardadas, por lo que puedes ejecutar el flujo de trabajo completo sin una clave de API de OpenAI.
Aprenderás a:
- Generar o recopilar muchas ejecuciones de agentes trazadas;
- Ejecutar evaluaciones de nivel inferior en cada ejecución completada;
- Convertir cada traza en un documento compacto;
- Descubrir patrones de comportamiento recurrentes en toda la población; y
- Profundizar en un patrón de alto impacto para encontrar dónde un humano debería inspeccionar el sistema a continuación.
El objetivo no es construir una taxonomía perfecta de cada traza. El objetivo es mostrar cómo un equipo de ingeniería de IA puede pasar de miles de eventos de agentes a un pequeño número de patrones que sean comprensibles tanto para las partes interesadas técnicas como para las empresariales.
Mapa de sistema agéntico de extremo a extremo
La idea clave es que el notebook evalúa un sistema agéntico guardado, no una transcripción de chat genérica. Las entradas del escenario impulsan un enjambre de especialistas orquestado, el tiempo de ejecución emite paquetes de trazas, las etiquetas guardadas de Promptfoo se unen a las trazas normalizadas, y la capa de macroevaluación convierte esa evidencia en vistas de patrones y diagnósticos.
1. ¿Por qué macroevaluaciones?
Las evaluaciones son la forma en que los equipos de IA miden si un sistema funciona. Para una simple llamada a un modelo, una evaluación podría comparar una salida con una rúbrica o una respuesta de referencia. Para un sistema agéntico, también necesitamos evaluar si el sistema usó las herramientas correctas, delegó al especialista adecuado, se detuvo para revisar cuando el riesgo era alto y se mantuvo anclado en el contexto empresarial.
Los sistemas multiagente hacen esto más difícil porque una respuesta final es solo el último evento en un flujo de trabajo más largo. Una recomendación de lanzamiento puede parecer plausible, mientras que la traza revela que el agente de precios ignoró un incentivo, el agente de suministro pasó por alto una falta de existencias o el orquestador eludió un paso de revisión requerido.
Este notebook separa el problema en dos niveles:
- Evaluaciones de nivel inferior califican agentes individuales, transferencias, herramientas y ejecuciones completadas. En este ejemplo, Promptfoo representa esa capa de evaluación a nivel de agente al calificar si una ejecución manejó la calidad de la decisión final, la corrección de la política, el enrutamiento de especialistas, la deriva del mercado y la idoneidad de la revisión.
- Macroevaluaciones analizan muchos hallazgos de nivel inferior. Preguntan: ¿qué tipos de problemas se repiten, dónde se concentran y qué parte del flujo de trabajo del agente debemos inspeccionar primero?
Utilizaremos cuatro etiquetas orientadas al lector a lo largo del manual:
case_type: la situación comercial generada, como un pedido limpio, un bloqueo de validación, una sustitución de proveedor o una excepción de precios.run_outcome: cómo terminó la ejecución, como completada, pendiente de revisión, bloqueada o fallida.eval_finding: la señal de nivel inferior que indica lo que parecía incorrecto o arriesgado.behavior_pattern: el patrón recurrente descubierto en muchas trazas.
Un modelo mental útil es: case_type es la configuración, run_outcome es el final, eval_finding es el síntoma local y behavior_pattern es el patrón a nivel de población.
import sys
from pathlib import Path
if sys.version_info < (3, 11):
raise RuntimeError("This notebook requires Python 3.11 or newer.")
if not Path("requirements.txt").is_file():
raise FileNotFoundError("requirements.txt must be in the same folder as this notebook.")
%pip install -q --upgrade pip setuptools wheel
%pip install -q --only-binary=:all: -r requirements.txt
Configuración y materiales de datos
Instala las dependencias y luego carga el conjunto de datos sin conexión incluido en este ejemplo. Las etiquetas guardadas de Promptfoo forman parte de la carpeta de datos local, por lo que este notebook no requiere una configuración de Promptfoo separada, un artefacto de ejecución de Promptfoo o una clave de API de OpenAI.
Archivos esperados:
data/trace_results.jsonl
data/run_summary.json
data/trace_bundles.zip
data/eval_labels.jsonl
trace_bundles.zip se expande automáticamente en una caché local la primera vez que se ejecuta el notebook. Una instantánea completa de trazas de SQLite se puede colocar en data/trace_snapshot.sqlite para un enriquecimiento opcional, pero no es necesaria para el flujo de trabajo de extremo a extremo.
Si tus datos residen fuera de la carpeta de ejemplo, establece MACRO_EVALS_DATA_ROOT a ese directorio. Si las etiquetas residen por separado, establece MACRO_EVALS_LABELS_PATH.
from __future__ import annotations
import json
import os
import sqlite3
import sys
import warnings
import zipfile
from pathlib import Path
from time import perf_counter
from typing import Any
import numpy as np
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from IPython.display import Markdown, display
pd.set_option("display.max_colwidth", 180)
pd.set_option("display.max_rows", 100)
warnings.filterwarnings("ignore", message="n_jobs value 1 overridden.*")
def find_example_root(start: Path | None = None) -> Path:
start = (start or Path.cwd()).resolve()
candidates = [start, *start.parents, start / "examples/partners/macro_evals_for_agentic_systems"]
for candidate in candidates:
if (candidate / "helpers/data_prep.py").is_file() and (candidate / "helpers/macro_eval_pipeline.py").is_file():
return candidate
raise FileNotFoundError("Could not locate the macro evals example root.")
EXAMPLE_ROOT = find_example_root()
HELPERS_ROOT = EXAMPLE_ROOT / "helpers"
if str(HELPERS_ROOT) not in sys.path:
sys.path.insert(0, str(HELPERS_ROOT))
from data_prep import add_public_label_columns, build_trace_documents, load_promptfoo_label_rows, normalize_bundle
from macro_eval_pipeline import (
drill_down_topic_root_causes,
pick_focus_topic,
plot_root_cause_story,
plot_suspect_leaderboard,
plot_topic_heatmap,
plot_topic_leaderboard,
plot_topic_scatter,
plot_trace_swimlane,
run_macro_discovery,
slice_topics_by_metadata,
)
def display_path(path: Path | None) -> str:
if path is None:
return "not found"
try:
return str(path.resolve().relative_to(EXAMPLE_ROOT))
except ValueError:
return str(path)
def as_path(value: str | Path) -> Path:
path = Path(value).expanduser()
return path if path.is_absolute() else EXAMPLE_ROOT / path
def unique_paths(paths: list[Path]) -> list[Path]:
seen: set[Path] = set()
unique: list[Path] = []
for path in paths:
resolved = path.resolve()
if resolved not in seen:
seen.add(resolved)
unique.append(resolved)
return unique
def find_material(label: str, names: list[str], *, kind: str = "file", required: bool = True) -> Path | None:
checked: list[Path] = []
for root in DATA_ROOTS:
for name in names:
if not name:
continue
candidate = as_path(name) if Path(name).expanduser().is_absolute() else root / name
checked.append(candidate)
if kind == "dir":
exists = candidate.is_dir() and any(candidate.glob("*.json"))
else:
exists = candidate.is_file()
if exists:
return candidate.resolve()
if required:
checked_text = "\n".join(f"- {display_path(path)}" for path in checked)
raise FileNotFoundError(f"Missing {label}. Checked:\n{checked_text}")
return None
def ensure_trace_bundle_dir(bundle_dir: Path | None, bundle_zip: Path | None) -> Path:
if bundle_dir is not None:
return bundle_dir
if bundle_zip is None:
raise FileNotFoundError("Missing trace bundles. Expected data/trace_bundles/ or data/trace_bundles.zip.")
cache_dir = bundle_zip.parent / ".macro_eval_cache" / "trace_bundles"
marker = cache_dir / ".extracted_from_trace_bundles_zip"
if not marker.is_file() or not any(cache_dir.glob("*.json")):
cache_dir.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(bundle_zip) as archive:
for member in archive.infolist():
if member.is_dir() or not member.filename.endswith(".json"):
continue
(cache_dir / Path(member.filename).name).write_bytes(archive.read(member))
marker.write_text(str(bundle_zip.stat().st_mtime_ns), encoding="utf-8")
return cache_dir.resolve()
env_data_root = os.environ.get("MACRO_EVALS_DATA_ROOT")
DATA_ROOTS = unique_paths(
([as_path(env_data_root)] if env_data_root else [])
+ [
EXAMPLE_ROOT / "data",
]
)
RESULTS_PATH = find_material("trace results", ["trace_results.jsonl", "metadata/results.jsonl", "results.jsonl"])
SUMMARY_PATH = find_material("run summary", ["run_summary.json", "metadata/summary.json", "summary.json"])
SQLITE_PATH = find_material("optional trace snapshot", ["trace_snapshot.sqlite"], required=False)
BUNDLE_ZIP_PATH = find_material("trace bundle archive", ["trace_bundles.zip", "bundles.zip"], required=False)
BUNDLE_DIR = ensure_trace_bundle_dir(find_material("trace bundles", ["trace_bundles", "bundles"], kind="dir", required=False), BUNDLE_ZIP_PATH)
PROGRESS_PATH = find_material("run progress", ["run_progress.json", "metadata/progress.json", "progress.json"], required=False)
PROMPTFOO_LABELS_PATH = find_material(
"lower-level eval labels",
[
os.environ.get("MACRO_EVALS_LABELS_PATH", ""),
"eval_labels.jsonl",
"metadata/eval_labels.jsonl",
],
required=False,
)
DATA_ROOT = next((root for root in DATA_ROOTS if RESULTS_PATH.is_relative_to(root)), DATA_ROOTS[0])
TRACE_LIMIT = int(os.environ.get("MACRO_EVALS_TRACE_LIMIT", "0")) or None
DISCOVERY_DOC_COLUMN = "doc_structured_summary"
DISCOVERY_MIN_CLUSTER_SIZE = int(os.environ.get("MACRO_EVALS_DISCOVERY_MIN_CLUSTER_SIZE", "24"))
RANDOM_STATE = 42
resolved_paths_df = pd.DataFrame(
[
("Trace results", RESULTS_PATH),
("Run summary", SUMMARY_PATH),
("Trace bundle archive", BUNDLE_ZIP_PATH),
("Expanded trace bundles", BUNDLE_DIR),
("Optional trace snapshot", SQLITE_PATH),
("Run progress", PROGRESS_PATH),
("Lower-level eval labels", PROMPTFOO_LABELS_PATH),
],
columns=["material", "path"],
)
resolved_paths_df["path"] = resolved_paths_df["path"].map(display_path)
display(Markdown("### Data materials"))
display(resolved_paths_df)
display(Markdown(f"Example root: `{display_path(EXAMPLE_ROOT)}` \nData root: `{display_path(DATA_ROOT)}`"))
2. La simulación: pedidos de automóviles en un mundo cambiante
El negocio simulado es un flujo de trabajo de pedidos y post-configuración de vehículos eléctricos. Un cliente ha elegido una configuración de vehículo, y la empresa necesita decidir si el pedido puede proceder tal cual, necesita ajuste, debe ser redirigido, requiere sustitución o debe pausarse para revisión.
La simulación incluye los tipos de restricciones que dificultan el cumplimiento automotriz real:
- disponibilidad de componentes y sustitución de proveedores;
- capacidad de fábrica y programación de producción;
- excepciones de precios, promociones e incentivos;
- aranceles y señales de mercado desactualizadas;
- restricciones de cumplimiento regional;
- aclaración del cliente y rutas de escalada;
- umbrales de revisión de lanzamiento para casos arriesgados o ambiguos.
El enjambre de agentes se organiza en torno a esas responsabilidades comerciales. Un orquestador recibe el pedido y el entorno actual, luego delega a especialistas como validación, riesgo de suministro, planificación de adquisiciones, equilibrio de capacidad, enrutamiento de fábrica, inteligencia de mercado, precios, cumplimiento, comunicaciones con el cliente y revisión de lanzamiento.
Esto se mapea naturalmente al SDK de Agentes de OpenAI. En el SDK, un agente es la unidad central de un flujo de trabajo: empaqueta un modelo, instrucciones y comportamiento de tiempo de ejecución opcional, como herramientas, transferencias, barandillas y salidas estructuradas. La simulación sigue ese patrón:
- agentes especializados empaquetan las instrucciones y herramientas para una parte de la decisión;
- las transferencias permiten al orquestador delegar en otro agente especialista en lugar de incluir todas las responsabilidades en un solo prompt;
- las herramientas de función exponen datos de pedidos, señales de entorno y marcadores de aprobación a través de entradas y salidas estructuradas;
- las barandillas y umbrales de revisión representan flujos de validación, bloqueo y revisión humana para casos arriesgados o ambiguos;
- las salidas estructuradas hacen posible la calificación y agregación posteriores;
- las trazas conservan registros estructurados de llamadas a modelos, llamadas a herramientas, transferencias, barandillas y tramos personalizados para depuración y análisis a nivel macro.
Las evaluaciones de bajo nivel posteriores en el notebook se basan en esta historia de simulación. Si el tipo de caso dice que hay una sustitución de proveedor bajo presión arancelaria, la traza debe mostrar conciencia del riesgo de suministro, política, mercado y revisión. Si el tipo de caso es limpio, una escalada innecesida es en sí misma un hallazgo.
def read_json(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def read_jsonl(path: Path) -> list[dict[str, Any]]:
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
def result_run_id(row: dict[str, Any]) -> str | None:
if row.get("run_id"):
return str(row["run_id"])
if row.get("bundle_path"):
return Path(str(row["bundle_path"])).stem
return None
def sqlite_table_counts(db_path: Path | None) -> pd.DataFrame:
tables = [
"runs",
"configs",
"traces",
"trace_events",
"spans",
"review_packets",
"environment_events",
"environment_decisions",
]
if db_path is None:
return pd.DataFrame([{"table": table, "row_count": 0} for table in tables])
with sqlite3.connect(db_path) as conn:
existing_tables = {
row[0]
for row in conn.execute("select name from sqlite_master where type = 'table'")
}
rows = [
{
"table": table,
"row_count": conn.execute(f"select count(*) from {table}").fetchone()[0] if table in existing_tables else 0,
}
for table in tables
]
return pd.DataFrame(rows)
def load_sqlite_runs(db_path: Path | None) -> pd.DataFrame:
if db_path is None:
return pd.DataFrame()
summary_fields = [
"scenario_family",
"validation_outcome",
"review_status",
"review_decision",
"triage_outcome",
"market_regime",
"price_regime",
"schedule_regime",
"agent_version_set",
"orchestrator_mode",
"rogue_window_id",
"factory_release_state",
"trace_family",
"loop_count",
"retry_count",
"arbitration_count",
"compound_issue_count",
"specialist_activations",
"environment_event_ids",
"findings",
"failure_agent",
"error_code",
"error_message",
]
rows = []
with sqlite3.connect(db_path) as conn:
for row in conn.execute("select run_id, config_id, trace_id, status, terminal_state, started_at, ended_at, summary_json from runs"):
run_id, config_id, trace_id, status, terminal_state, started_at, ended_at, summary_json = row
summary = json.loads(summary_json or "{}")
item = {field: summary.get(field) for field in summary_fields}
item.update(
{
"run_id": run_id,
"config_id": config_id,
"trace_id": trace_id,
"sqlite_status": status,
"sqlite_terminal_state": terminal_state,
"started_at": started_at,
"ended_at": ended_at,
}
)
rows.append(item)
runs = pd.DataFrame(rows)
if runs.empty:
return runs
runs["started_at"] = pd.to_datetime(runs["started_at"], utc=True, errors="coerce")
runs["ended_at"] = pd.to_datetime(runs["ended_at"], utc=True, errors="coerce")
runs["findings_count_sqlite"] = runs["findings"].apply(lambda value: len(value or []))
runs["specialist_activation_count_sqlite"] = runs["specialist_activations"].apply(lambda value: len(value or []))
runs["environment_event_count_sqlite"] = runs["environment_event_ids"].apply(lambda value: len(value or []))
return runs
batch_summary = read_json(SUMMARY_PATH)
results_rows = read_jsonl(RESULTS_PATH)
sqlite_runs_df = load_sqlite_runs(SQLITE_PATH)
table_counts_df = sqlite_table_counts(SQLITE_PATH)
result_ids = {rid for row in results_rows if (rid := result_run_id(row))}
bundle_ids = {path.stem for path in BUNDLE_DIR.glob("*.json")}
missing_result_rows = [row for row in results_rows if result_run_id(row) is None]
def table_count(table_name: str) -> int:
rows = table_counts_df.loc[table_counts_df["table"].eq(table_name), "row_count"]
return int(rows.iloc[0]) if not rows.empty else 0
dataset_profile_df = pd.DataFrame(
[
("requested_batch_size", batch_summary.get("batch_size") or batch_summary.get("requested_runs")),
("results_rows", len(results_rows)),
("bundle_backed_result_rows", len(result_ids)),
("runner_error_rows_without_bundle", len(missing_result_rows)),
("bundle_files_available", len(bundle_ids)),
("bundle_files_not_in_results", len(bundle_ids - result_ids)),
("saved_promptfoo_label_rows", len(read_jsonl(PROMPTFOO_LABELS_PATH)) if PROMPTFOO_LABELS_PATH else 0),
("sqlite_available", SQLITE_PATH is not None),
("sqlite_runs", len(sqlite_runs_df)),
("sqlite_trace_events", table_count("trace_events")),
("sqlite_spans", table_count("spans")),
],
columns=["metric", "value"],
)
display(dataset_profile_df)
if SQLITE_PATH is not None:
display(table_counts_df)
if missing_result_rows:
display(Markdown(
f"The batch has `{len(results_rows):,}` result rows, but `{len(missing_result_rows):,}` ended before a bundle was written. "
f"The macro analysis therefore focuses on the `{len(result_ids):,}` bundle-backed traces that can be normalized and graded retrospectively."
))
if SQLITE_PATH is None:
display(Markdown(
"This packaged version omits the large SQLite mirror. The notebook uses the JSONL result rows, trace bundles, and saved Promptfoo labels for the end-to-end workflow."
))
Qué representa un paquete
En este notebook, un paquete es el conjunto de pruebas para una interacción simulada de pedido de cliente.
Imagina que un cliente ha configurado un EV y la empresa necesita decidir qué hacer a continuación. El enjambre recibe ese pedido más el mundo operativo actual: restricciones de suministro, capacidad de fábrica, promociones, incentivos, aranceles, presión de la competencia y umbrales de revisión. Los agentes luego dirigen el trabajo a través de especialistas y producen un estado final. El paquete es todo lo que necesitamos para auditar esa interacción después.
Un paquete es importante porque las macroevaluaciones necesitan la evidencia del flujo de trabajo detrás de la respuesta final. Necesitan saber qué agentes fueron consultados, qué herramientas se llamaron, qué señales ambientales estaban activas, si se requirió revisión y dónde cambió de dirección el flujo de trabajo. Con esa evidencia, podemos pasar de "¿qué sucedió en esta ejecución?" a "¿qué patrones de flujo de trabajo se repiten en muchas ejecuciones?"
def bundle_path_for_result(result_row: dict[str, Any]) -> Path | None:
raw = result_row.get("bundle_path")
if not raw:
return None
return BUNDLE_DIR / Path(str(raw)).name
def bundle_event_counts(bundle: dict[str, Any]) -> dict[str, int]:
events = bundle.get("events") or []
spans = bundle.get("spans") or []
event_types = pd.Series([event.get("event_type") or "unknown" for event in events])
span_types = pd.Series([span.get("span_type") or "unknown" for span in spans])
agents = {
event.get("agent_name")
for event in events
if event.get("agent_name")
} | {
span.get("agent_name")
for span in spans
if span.get("agent_name")
}
return {
"events": len(events),
"spans": len(spans),
"handoffs": int(event_types.eq("handoff").sum() + span_types.eq("handoff").sum()),
"tool_or_function_calls": int(event_types.eq("function").sum() + span_types.eq("function").sum()),
"status_updates": int(event_types.eq("status").sum()),
"unique_agents_seen": len(agents),
"environment_signals": len(bundle.get("environment_events") or []),
"has_review_packet": int(bool(bundle.get("review_packet"))),
}
def human_event_type(value: str) -> str:
cleaned = str(value or "").replace("product_launch", "product_launch")
return cleaned.replace("_", " ")
bundle_rows = []
sample_bundle = None
sample_result_row = None
for result_row in results_rows:
bundle_path = bundle_path_for_result(result_row)
if bundle_path is None or not bundle_path.is_file():
continue
bundle = read_json(bundle_path)
counts = bundle_event_counts(bundle)
counts["run_id"] = result_run_id(result_row)
counts["case_type"] = result_row.get("scenario_family")
counts["final_status"] = result_row.get("final_status") or result_row.get("status")
counts["review_status"] = result_row.get("review_status")
bundle_rows.append(counts)
if sample_bundle is None and result_row.get("scenario_family") != "clean_simple":
sample_bundle = bundle
sample_result_row = result_row
bundle_profile_df = pd.DataFrame(bundle_rows)
typical_bundle_df = pd.DataFrame(
[
("analyzable customer-order interactions", len(bundle_profile_df), "One completed trace bundle per simulated order interaction."),
("median normalized events per interaction", int(bundle_profile_df["events"].median()), "Status updates, handoffs, function/tool events, responses, and findings."),
("median SDK spans per interaction", int(bundle_profile_df["spans"].median()), "Lower-level SDK trace spans behind the event log."),
("median handoff records per interaction", int(bundle_profile_df["handoffs"].median()), "Delegations between orchestrator and specialist agents."),
("median tool/function calls per interaction", int(bundle_profile_df["tool_or_function_calls"].median()), "Structured reads, checks, and evaluation calls inside the run."),
("median agents observed per interaction", int(bundle_profile_df["unique_agents_seen"].median()), "How many specialist roles appear in a typical trace."),
("median environment signals per interaction", int(bundle_profile_df["environment_signals"].median()), "Tariff, incentive, stockout, promotion, competitor, launch, or schedule signals active for the order."),
("interactions with review packets", int(bundle_profile_df["has_review_packet"].sum()), "Runs where the simulated business process produced a review artifact."),
],
columns=["reader_metric", "value", "plain_english_meaning"],
)
display(typical_bundle_df)
bundle_anatomy_df = pd.DataFrame(
[
("run", "Run id, trace id, terminal state, batch metadata, and synthetic order context.", "Lets us join one interaction across tables and understand its business setup."),
("events", "A normalized event log: status updates, handoffs, tool/function activity, responses, and findings.", "This is the main evidence stream used for trace documents and AgentTrace-style diagnosis."),
("spans", "OpenAI Agents SDK trace spans for handoffs, function calls, responses, and timing.", "Gives lower-level execution structure behind the event log."),
("environment_events", "The dated world state active for the order: tariffs, incentives, stockouts, promotions, competitor pressure, launches, and schedule/capacity signals.", "Lets evals check whether the swarm reacted to the world it was given."),
("review_packet", "A simulated review artifact with findings, recommended action, allowed actions, and review status.", "Lets us evaluate whether escalation or review was appropriate."),
("snapshots", "Optional inventory, capacity, and environment snapshots.", "Provides operational context when a case depends on supply or scheduling."),
],
columns=["bundle_part", "what_it_contains", "why_it_matters_for_macro_evals"],
)
display(bundle_anatomy_df)
if sample_bundle is not None and sample_result_row is not None:
run_config = (sample_bundle.get("run") or {}).get("config") or {}
metadata = run_config.get("metadata") or {}
generation = metadata.get("generation_params") or {}
customer = run_config.get("customer") or {}
active_event_types = sorted({human_event_type(value) for value in generation.get("active_event_types") or []})
specialists = sample_result_row.get("specialist_activations") or generation.get("specialist_activations") or []
review_packet = sample_bundle.get("review_packet") or {}
example_interaction_df = pd.DataFrame(
[
("what the interaction represents", "One synthetic customer order moving through the post-configuration workflow."),
("case_type", sample_result_row.get("scenario_family")),
("synthetic customer region", customer.get("region") or "not recorded"),
("business issue cluster", generation.get("issue_cluster") or sample_result_row.get("scenario")),
("active world signals", ", ".join(active_event_types[:8]) + (" ..." if len(active_event_types) > 8 else "")),
("specialists activated", ", ".join(map(str, specialists[:8])) + (" ..." if len(specialists) > 8 else "")),
("final status", sample_result_row.get("final_status") or sample_result_row.get("status")),
("review status", sample_result_row.get("review_status") or review_packet.get("status") or "not recorded"),
("event evidence", f"{len(sample_bundle.get('events') or []):,} events and {len(sample_bundle.get('spans') or []):,} SDK spans"),
],
columns=["field", "example_value"],
)
display(example_interaction_df)
Cómo leer el perfil del conjunto de datos
El perfil del conjunto de datos nos indica la escala y la textura del proceso de negocio simulado que estamos a punto de evaluar. Cada fila analizable es una interacción de pedido de cliente con suficiente evidencia de traza para reconstruir lo que vio el enjambre de agentes, qué especialistas consultó y cómo terminó el flujo de trabajo.
El lote generado pidió al enjambre que manejara 1,000 interacciones de pedidos sintéticos. Para 992 de ellas, tenemos un paquete: un paquete de evidencia completo para calificar la ejecución, construir un documento de traza, agruparlo con ejecuciones similares e inspeccionar la ruta del agente después. Eso nos da una población lo suficientemente grande como para buscar comportamientos repetidos, al tiempo que conservamos el detalle de la traza necesario para explicar ejemplos individuales.
El paquete típico es un registro estructurado de un proceso de negocio simulado: la configuración del pedido, los eventos mundiales activos, las transferencias de especialistas, la actividad de herramientas/funciones, los artefactos de revisión y el estado terminal. Es por eso que este conjunto de datos puede admitir macroevaluaciones. Podemos evaluar decisiones individuales, y también podemos preguntar si surgen patrones de flujo de trabajo repetidos en cientos de registros de interacción ricos.
scenario_counts_df = (
pd.DataFrame(results_rows)
.assign(case_type=lambda df: df["scenario_family"].fillna("unknown"))
.groupby("case_type", as_index=False)
.size()
.rename(columns={"size": "run_count"})
.sort_values("run_count", ascending=False)
)
fig = px.bar(
scenario_counts_df,
x="run_count",
y="case_type",
orientation="h",
title="Synthetic simulation coverage by generated case type",
text="run_count",
color="run_count",
color_continuous_scale="Teal",
)
fig.update_layout(height=max(420, 28 * len(scenario_counts_df)), margin=dict(l=20, r=20, t=60, b=30))
fig.update_yaxes(title="", categoryorder="total ascending")
fig.update_xaxes(title="Run count")
fig.show()
display(scenario_counts_df.head(15))
Qué significa case_type
Un case_type es una etiqueta de escenario del generador. Describe el tipo de situación comercial que se le pidió al enjambre que manejara antes de que ocurriera cualquier evaluación o agrupación.
Ejemplos de este conjunto de datos incluyen:
clean_simple: un pedido relativamente sencillo donde el comportamiento correcto suele ser completarlo sin revisión innecesaria.validation_block_simple: una configuración tiene un problema de validación, por lo que el enjambre debe evitar un lanzamiento demasiado confiado.supplier_substitution_compound: la disponibilidad de componentes crea una decisión de sustitución, a menudo con implicaciones de enrutamiento y programación posteriores.pricing_exception_compound: los precios, incentivos o la política de margen necesitan revisión especializada.regional_compliance_compound: el pedido necesita manejo de políticas o cumplimiento regional.
El gráfico de barras anterior es una vista de cobertura. Muestra si la simulación produjo suficiente variedad para evaluar el enjambre bajo diferentes presiones comerciales. Un conjunto de datos de macroevaluación sólido necesita tanto casos ordinarios como casos de presión, porque los patrones recurrentes solo se vuelven significativos cuando podemos comparar el comportamiento en diferentes configuraciones.
CASE_TYPE_DESCRIPTIONS = {
"clean_simple": "Straightforward order; should usually complete with minimal routing.",
"validation_block_simple": "Configuration or buildability issue; should avoid unsupported release.",
"release_block_simple": "Release readiness is blocked; should defer or request review.",
"capacity_hold_simple": "Factory capacity or scheduling pressure; should route to fulfillment planning.",
"pricing_exception_compound": "Pricing, incentives, margin, or tariff pressure; should involve pricing/policy owners.",
"escalation_resume_compound": "An escalation or review flow needs to resume cleanly.",
"supplier_substitution_compound": "Supplier availability forces substitution or procurement planning.",
"regional_compliance_compound": "Regional compliance or policy constraints affect release.",
"clarification_needed_compound": "Customer intent or configuration details are ambiguous.",
"schedule_incentive_compound": "Timing, scheduling, and incentive windows interact.",
"tradeoff_recommended_compound": "The system should weigh competing business tradeoffs.",
"dual_failure_recovery_compound": "Multiple failures require coordinated recovery.",
"ambiguous_customer_intent_compound": "The customer request is underspecified or conflicting.",
"conflicting_multi_agent_compound": "Specialists may surface conflicting recommendations.",
}
case_type_guide_df = scenario_counts_df.copy()
case_type_guide_df["plain_english"] = case_type_guide_df["case_type"].map(CASE_TYPE_DESCRIPTIONS).fillna(
"Generated scenario type from the synthetic simulation."
)
display(case_type_guide_df.head(12))
La tabla anterior convierte las etiquetas del generador en lenguaje de negocios. Esto es importante porque el mismo patrón posterior puede significar cosas diferentes según la configuración. Un redireccionamiento de cumplimiento en un caso de sustitución de proveedor puede ser deseable. El mismo redireccionamiento en un caso limpio podría ser una complejidad innecesaria.
3. Evaluaciones de agentes de nivel inferior con Promptfoo
Un sistema multiagente maduro no debe depender únicamente de la inspección de la respuesta final. Cada agente lanzado suele necesitar sus propias evaluaciones: ¿este especialista utilizó la evidencia correcta, llamó a las herramientas adecuadas, respetó la política, transfirió en el momento oportuno y produjo una salida en la que el resto del sistema puede confiar?
Promptfoo desempeña ese papel en este notebook. Representa la capa de evaluación de nivel inferior que normalmente residiría junto a los agentes en un flujo de trabajo de producción. En un sistema en vivo, algunas de estas comprobaciones podrían ejecutarse en línea, otras de forma asíncrona y algunas podrían muestrearse para revisión humana. El detalle de la implementación importa menos que el contrato: cada ejecución debe llevar señales de evaluación que indiquen lo que parecía correcto, arriesgado o incorrecto a nivel de agente y de flujo de trabajo.
En este conjunto de datos, Promptfoo califica las trazas completadas con preguntas que reflejan los tipos de evaluaciones a nivel de agente que los equipos construyen para sistemas reales:
- ¿La decisión final se derivó del problema activo?
- ¿El sistema respetó las restricciones de precios, aranceles, incentivos, regionales y políticas?
- ¿El orquestador activó a los especialistas implicados por el caso?
- ¿La ejecución respondió a señales de mercado desactualizadas en lugar de actuar como si el mundo fuera estático?
- ¿La revisión o escalada fue proporcional al riesgo?
Estas comprobaciones producen eval_finding. Una evaluación de nivel inferior fallida es una señal local: una traza, una rúbrica, un síntoma. Las secciones de macroevaluación posteriores preguntan en qué se convierten esas señales locales a escala de población. ¿Se dispersan aleatoriamente o revelan patrones de comportamiento repetidos que apuntan a un agente, transferencia, herramienta o política comercial específica?
PROMPTFOO_RUBRICS = [
("final_decision_quality", "Final decision is supported by the active issues, terminal state, and agent outputs."),
("policy_compliance_correctness", "Policy, tariff, incentive, and regional compliance context is handled correctly."),
("routing_specialist_activation", "Specialist routing matches the issues present in the bundle."),
("market_drift_awareness", "Changing market conditions and dated environment signals are noticed."),
("review_appropriateness", "Review and escalation behavior is proportionate to the case risk."),
]
PROMPTFOO_ASSERTION_FALLBACK = {
f"assertion_{idx}": metric for idx, (metric, _) in enumerate(PROMPTFOO_RUBRICS, start=1)
}
def clean_metric_name(value: Any) -> str | None:
if value is None or (isinstance(value, float) and pd.isna(value)):
return None
text = str(value)
return PROMPTFOO_ASSERTION_FALLBACK.get(text, text)
def clean_metric_list(values: Any) -> list[str]:
if not isinstance(values, list):
return []
return [metric for item in values if (metric := clean_metric_name(item))]
def clean_metric_dict(values: Any) -> dict[str, Any]:
if not isinstance(values, dict):
return {}
return {clean_metric_name(key) or str(key): value for key, value in values.items()}
def clean_promptfoo_labels(labels_df: pd.DataFrame) -> pd.DataFrame:
if labels_df.empty:
return labels_df
cleaned = labels_df.copy()
# Accept older draft label files that used metric/passed/score/reason columns.
if "promptfoo_pass" not in cleaned.columns and "passed" in cleaned.columns:
cleaned["promptfoo_pass"] = cleaned["passed"]
if "promptfoo_failed_checks" not in cleaned.columns:
if "metric" in cleaned.columns:
cleaned["promptfoo_failed_checks"] = cleaned.apply(
lambda row: [] if bool(row.get("promptfoo_pass", True)) else [row.get("metric")],
axis=1,
)
else:
cleaned["promptfoo_failed_checks"] = [[] for _ in range(len(cleaned))]
if "promptfoo_score_mean" not in cleaned.columns and "score" in cleaned.columns:
cleaned["promptfoo_score_mean"] = cleaned["score"]
if "promptfoo_primary_finding" not in cleaned.columns:
if "metric" in cleaned.columns:
cleaned["promptfoo_primary_finding"] = cleaned.apply(
lambda row: None if bool(row.get("promptfoo_pass", True)) else row.get("metric"),
axis=1,
)
else:
cleaned["promptfoo_primary_finding"] = None
if "promptfoo_check_scores" not in cleaned.columns:
cleaned["promptfoo_check_scores"] = cleaned.apply(
lambda row: {row.get("metric", "unknown"): row.get("score")} if "score" in cleaned.columns else {},
axis=1,
)
if "promptfoo_rationales" not in cleaned.columns:
cleaned["promptfoo_rationales"] = cleaned.apply(
lambda row: {row.get("metric", "unknown"): row.get("reason")} if "reason" in cleaned.columns else {},
axis=1,
)
cleaned["promptfoo_pass"] = cleaned["promptfoo_pass"].astype("boolean")
cleaned["promptfoo_primary_finding"] = cleaned["promptfoo_primary_finding"].apply(clean_metric_name)
cleaned["promptfoo_failed_checks"] = cleaned["promptfoo_failed_checks"].apply(clean_metric_list)
cleaned["promptfoo_check_scores"] = cleaned["promptfoo_check_scores"].apply(clean_metric_dict)
cleaned["promptfoo_rationales"] = cleaned["promptfoo_rationales"].apply(clean_metric_dict)
return cleaned
rubric_df = pd.DataFrame(PROMPTFOO_RUBRICS, columns=["rubric", "plain_english_question"])
display(rubric_df)
promptfoo_labels_df = clean_promptfoo_labels(load_promptfoo_label_rows(PROMPTFOO_LABELS_PATH))
if promptfoo_labels_df.empty:
display(Markdown("No Promptfoo labels were found. The notebook will continue with deterministic review and runtime signals only."))
else:
pass_counts_df = (
promptfoo_labels_df["promptfoo_pass"]
.map({True: "pass", False: "fail"})
.fillna("unknown")
.value_counts()
.rename_axis("promptfoo_result")
.reset_index(name="trace_count")
)
display(Markdown(f"Loaded `{len(promptfoo_labels_df):,}` Promptfoo label rows from `{display_path(PROMPTFOO_LABELS_PATH)}`."))
display(pass_counts_df)
fig = px.pie(
pass_counts_df,
names="promptfoo_result",
values="trace_count",
title="Promptfoo grading result across bundle-backed traces",
color="promptfoo_result",
color_discrete_map={"pass": "#4daf4a", "fail": "#e41a1c", "unknown": "#999999"},
hole=0.45,
)
fig.update_traces(textposition="inside", textinfo="percent+label")
fig.show()
failed_metric_df = (
promptfoo_labels_df.explode("promptfoo_failed_checks")
.dropna(subset=["promptfoo_failed_checks"])
.groupby("promptfoo_failed_checks", as_index=False)
.size()
.rename(columns={"promptfoo_failed_checks": "rubric", "size": "failed_trace_count"})
.sort_values("failed_trace_count", ascending=False)
)
display(failed_metric_df)
if not failed_metric_df.empty:
fig = px.bar(
failed_metric_df,
x="failed_trace_count",
y="rubric",
orientation="h",
title="Which lower-level rubric failed most often?",
text="failed_trace_count",
color="failed_trace_count",
color_continuous_scale="Reds",
)
fig.update_layout(height=360, margin=dict(l=20, r=20, t=60, b=30))
fig.update_yaxes(title="", categoryorder="total ascending")
fig.update_xaxes(title="Failed traces")
fig.show()
Interpretación de las salidas de Promptfoo
El gráfico circular es el cuadro de mando de nivel inferior más simple: separa las trazas que pasaron todas las comprobaciones de la rúbrica de las trazas con al menos una comprobación fallida. En un sistema multiagente en vivo, este es el tipo de capa que nos dice qué ejecuciones merecen atención antes de realizar cualquier macroanálisis.
El gráfico de barras de rúbricas fallidas responde a una pregunta más útil: ¿qué tipos de preocupaciones de agentes o flujos de trabajo aparecen con mayor frecuencia? Para este conjunto de datos, la calidad de la decisión final es el hallazgo dominante de nivel inferior, mientras que la corrección de la política, la idoneidad de la revisión y la conciencia de la deriva del mercado también aparecen. Eso sugiere que la capa macro debería centrarse menos en errores de sintaxis aislados y más en patrones de toma de decisiones repetidos.
Este es el puente hacia las macroevaluaciones. Promptfoo asigna a cada traza etiquetas de evaluación locales. El resto del notebook pregunta cómo se organizan esas etiquetas en toda la población. En otras palabras: las evaluaciones a nivel de agente crean la señal bruta, y las macroevaluaciones convierten muchas de esas señales en un mapa del comportamiento recurrente del sistema.
4. Construir el conjunto de datos de análisis
Ahora normalizamos los paquetes de ejecución en dos tablas de análisis:
traces_df: una fila por ejecución, con metadatos, resultado, hallazgos y campos de documento.events_df: una fila por evento de traza normalizado, incluyendo transferencias, llamadas a herramientas, eventos de estado, respuestas del modelo y marcadores de revisión/hallazgo.
También construimos documentos de traza. El documento es el objeto de modelado que la sección de estilo BERTopic agrupará. El notebook utiliza doc_structured_summary porque es compacto pero aún conserva el escenario, el enrutamiento, las transiciones de estado, las transferencias, los hallazgos y el estado terminal.
La ruta de análisis pública es:
case_type -> run_outcome -> eval_finding -> behavior_pattern
Las tres primeras etiquetas se conocen antes de la agrupación. La cuarta aparece después del descubrimiento.
OUTCOME_GROUP_MAP = {
"completed": "successful_completion",
"awaiting_review": "review_escalation",
"blocked": "hard_failure",
"failed": "hard_failure",
}
SEVERITY_BY_OUTCOME = {
"successful_completion": ("low", 1.0),
"review_escalation": ("medium", 2.0),
"in_progress": ("medium", 1.5),
"blocked": ("high", 2.5),
"hard_failure": ("high", 3.0),
}
def local_bundle_path(result_row: dict[str, Any]) -> Path:
return BUNDLE_DIR / Path(str(result_row["bundle_path"])).name
def load_normalized_bundle_tables(results: list[dict[str, Any]], limit: int | None = None) -> tuple[pd.DataFrame, pd.DataFrame]:
selected_rows = [row for row in results if result_run_id(row) and row.get("bundle_path")]
if limit is not None:
selected_rows = selected_rows[:limit]
normalized = []
for record_index, result_row in enumerate(selected_rows, start=1):
bundle_path = local_bundle_path(result_row)
if not bundle_path.is_file():
continue
bundle = read_json(bundle_path)
normalized.append(normalize_bundle(bundle, result_row, record_index, bundle_path))
trace_rows = [trace_row for trace_row, _ in normalized]
event_rows = [event for _, trace_events in normalized for event in trace_events]
traces = pd.DataFrame(trace_rows)
events = pd.DataFrame(event_rows)
if not events.empty:
events["ts"] = pd.to_datetime(events["ts"], utc=True, errors="coerce")
events["ended_at"] = pd.to_datetime(events["ended_at"], utc=True, errors="coerce")
events = events.sort_values(["trace_id", "sequence_index", "ts", "event_id"]).reset_index(drop=True)
return traces, events
load_started = perf_counter()
traces_df, events_df = load_normalized_bundle_tables(results_rows, limit=TRACE_LIMIT)
print(f"Loaded {len(traces_df):,} normalized traces and {len(events_df):,} normalized events in {perf_counter() - load_started:.1f}s.")
result_metadata_cols = [
"run_id",
"market_regime",
"price_regime",
"schedule_regime",
"agent_version_set",
"orchestrator_mode",
"rogue_window_id",
"factory_release_state",
"trace_family",
"specialist_activations",
"environment_event_ids",
]
result_metadata_df = pd.DataFrame(results_rows)
result_metadata_cols = [column for column in result_metadata_cols if column in result_metadata_df.columns]
if "run_id" in result_metadata_cols:
traces_df = traces_df.merge(
result_metadata_df[result_metadata_cols].drop_duplicates(subset=["run_id"]),
on="run_id",
how="left",
suffixes=("", "_result"),
)
for column in result_metadata_cols:
if column == "run_id":
continue
result_col = f"{column}_result"
if result_col in traces_df.columns:
traces_df[column] = traces_df[column].combine_first(traces_df[result_col]) if column in traces_df.columns else traces_df[result_col]
traces_df = traces_df.drop(columns=[result_col])
for column, default in {
"market_regime": "unknown",
"price_regime": "unknown",
"schedule_regime": "unknown",
"agent_version_set": "unknown",
"orchestrator_mode": "unknown",
}.items():
if column not in traces_df.columns:
traces_df[column] = default
sqlite_enrichment_cols = [
"run_id",
"sqlite_status",
"sqlite_terminal_state",
"scenario_family",
"validation_outcome",
"review_status",
"review_decision",
"triage_outcome",
"market_regime",
"price_regime",
"schedule_regime",
"agent_version_set",
"orchestrator_mode",
"rogue_window_id",
"factory_release_state",
"trace_family",
"loop_count",
"retry_count",
"arbitration_count",
"compound_issue_count",
"findings_count_sqlite",
]
if not sqlite_runs_df.empty:
traces_df = traces_df.merge(sqlite_runs_df[sqlite_enrichment_cols], on="run_id", how="left", suffixes=("", "_sqlite"))
for column in [
"scenario_family",
"validation_outcome",
"review_status",
"review_decision",
"triage_outcome",
"market_regime",
"price_regime",
"schedule_regime",
"agent_version_set",
"orchestrator_mode",
"rogue_window_id",
"factory_release_state",
"trace_family",
"loop_count",
"retry_count",
"arbitration_count",
"compound_issue_count",
]:
sqlite_col = f"{column}_sqlite"
if sqlite_col in traces_df.columns:
traces_df[column] = traces_df[sqlite_col].combine_first(traces_df.get(column))
traces_df = traces_df.drop(columns=[sqlite_col])
traces_df["runtime_status"] = traces_df["sqlite_status"].combine_first(traces_df["runtime_status"])
traces_df["terminal_state"] = traces_df["sqlite_terminal_state"].combine_first(traces_df["terminal_state"])
traces_df["findings_count"] = traces_df["findings_count_sqlite"].combine_first(traces_df["findings_count"]).fillna(0)
else:
traces_df["findings_count"] = traces_df.get("findings_count", pd.Series(0, index=traces_df.index)).fillna(0)
traces_df["outcome_group"] = traces_df["runtime_status"].map(OUTCOME_GROUP_MAP).fillna("unknown")
traces_df["severity_label"] = traces_df["outcome_group"].map(lambda value: SEVERITY_BY_OUTCOME.get(value, ("medium", 2.0))[0])
traces_df["severity_weight"] = traces_df["outcome_group"].map(lambda value: SEVERITY_BY_OUTCOME.get(value, ("medium", 2.0))[1])
traces_df["has_failure"] = (
traces_df["outcome_group"].ne("successful_completion")
| traces_df["validation_outcome"].fillna("passed").ne("passed")
| traces_df["findings_count"].fillna(0).gt(0)
)
traces_df["impact_score"] = (
traces_df["severity_weight"].fillna(1.0)
* (1.0 + traces_df["findings_count"].fillna(0))
* (1.0 + traces_df["loop_count"].fillna(0) / 4.0)
)
documents_df = build_trace_documents(traces_df, events_df)
traces_with_docs_df = traces_df.merge(documents_df, on="trace_id", how="left")
labeled_traces_df = add_public_label_columns(traces_with_docs_df, promptfoo_labels_df=promptfoo_labels_df)
labeled_traces_df["eval_finding"] = labeled_traces_df["eval_finding"].apply(lambda value: clean_metric_name(value) or "none")
labeled_traces_df["promptfoo_failed"] = labeled_traces_df.get("promptfoo_pass").eq(False)
analysis_profile_df = pd.DataFrame(
[
("normalized_traces", len(labeled_traces_df)),
("normalized_events", len(events_df)),
("case_types", labeled_traces_df["case_type"].nunique()),
("run_outcomes", labeled_traces_df["run_outcome"].nunique()),
("eval_findings", labeled_traces_df["eval_finding"].nunique()),
("promptfoo_failed_traces", int(labeled_traces_df["promptfoo_failed"].sum())),
("failure_or_review_traces", int(labeled_traces_df["has_failure"].sum())),
],
columns=["metric", "value"],
)
display(analysis_profile_df)
display(labeled_traces_df[["run_id", "case_type", "run_outcome", "eval_finding", "market_regime", "agent_version_set", "impact_score"]].head(10))
Interpretación del perfil de análisis
El perfil anterior confirma que la capa de evaluación de nivel inferior se ha unido a la población de trazas normalizadas. Los números importantes son:
- trazas normalizadas: la población respaldada por paquetes que podemos inspeccionar;
- eventos normalizados: la evidencia a nivel de evento detrás de esas trazas;
- tipos de casos: la cobertura de escenarios producida por el generador; y
- trazas fallidas por Promptfoo o con revisión/fallo: la población de señales de nivel inferior más relevante para el descubrimiento macro.
Los recuentos exactos dependen de si ejecutas el notebook completo o estableces MACRO_EVALS_TRACE_LIMIT para una prueba rápida. Las filas de ejemplo muestran cómo el notebook simplifica los datos brutos en etiquetas legibles. Por ejemplo, un caso pricing_exception_compound que termina en revisión con un hallazgo final_decision_quality ahora es fácil de seguir a lo largo del resto del notebook.
def humanize_label(value: Any, max_len: int = 48) -> str:
if value is None or (isinstance(value, float) and pd.isna(value)):
text = "missing"
else:
text = str(value)
text = text.replace("_", " ")
return text if len(text) <= max_len else text[: max_len - 3].rstrip() + "..."
def plot_label_sankey(frame: pd.DataFrame, columns: list[str], title: str, min_count: int = 1):
working = frame[columns].copy()
for column in columns:
working[column] = working[column].fillna("missing").astype(str)
node_labels: list[str] = []
node_lookup: dict[tuple[str, str], int] = {}
sources: list[int] = []
targets: list[int] = []
values: list[int] = []
def node_id(column: str, value: str) -> int:
key = (column, value)
if key not in node_lookup:
node_lookup[key] = len(node_labels)
node_labels.append(f"{column}: {humanize_label(value)}")
return node_lookup[key]
for left, right in zip(columns[:-1], columns[1:]):
pairs = working.groupby([left, right]).size().reset_index(name="count")
pairs = pairs[pairs["count"].ge(min_count)]
for _, row in pairs.iterrows():
sources.append(node_id(left, row[left]))
targets.append(node_id(right, row[right]))
values.append(int(row["count"]))
fig = go.Figure(
data=[
go.Sankey(
arrangement="snap",
node=dict(label=node_labels, pad=14, thickness=14),
link=dict(source=sources, target=targets, value=values),
)
]
)
fig.update_layout(title=title, height=620, margin=dict(l=20, r=20, t=60, b=20))
return fig
flow_sample_df = labeled_traces_df[["case_type", "run_outcome", "eval_finding"]].copy()
plot_label_sankey(
flow_sample_df,
["case_type", "run_outcome", "eval_finding"],
"Before clustering: generated case -> run outcome -> lower-level finding",
min_count=3,
).show()
label_crosswalk_df = (
labeled_traces_df.groupby(["case_type", "run_outcome", "eval_finding"], dropna=False)
.size()
.reset_index(name="trace_count")
.sort_values("trace_count", ascending=False)
.head(18)
)
display(label_crosswalk_df)
Qué enseña el primer diagrama de Sankey
El primer diagrama de Sankey es una vista previa a la agrupación. Muestra cómo los tipos de casos generados fluyen hacia los resultados de la ejecución y los hallazgos de nivel inferior.
Léelo de izquierda a derecha:
- las bandas anchas de un
case_typesignifican que ese escenario aparece a menudo; - las divisiones en
run_outcomemuestran si ese escenario tiende a completarse, pausarse, bloquearse o fallar; - las bandas finales en
eval_findingmuestran qué rúbrica de nivel inferior o señal de tiempo de ejecución está adjunta.
Esto ya es útil para un equipo. Un lector de negocios puede preguntar si la simulación produce los tipos correctos de presión. Un ingeniero de IA puede preguntar si ciertos escenarios producen en exceso el mismo hallazgo de bajo nivel. Lo que aún no puede responder es si esos hallazgos representan el mismo patrón de comportamiento subyacente. Por eso agrupamos a continuación.
Documentos de traza: convertir ejecuciones en texto comparable
Una traza de agente en bruto es demasiado detallada para agruparla directamente. Puede contener cientos de eventos, respuestas de modelos largas, cargas útiles de herramientas y actualizaciones de estado repetidas. El paso de construcción de documentos comprime cada ejecución en una vista comparable, conservando la información que importa para las macroevaluaciones.
Un buen documento de traza incluye:
- la configuración del negocio (
case_type, ruta seleccionada, señales de entorno activas); - el resultado y la gravedad de la ejecución;
- las transferencias importantes y las activaciones de especialistas;
- marcadores de revisión/hallazgo;
- un breve resumen de la transición de estado.
La vista del documento define lo que el algoritmo de agrupación puede notar. Incluir las transferencias de agentes ayuda a la macroevaluación a descubrir patrones de enrutamiento. Incluir las señales del entorno ayuda a descubrir fallas por deriva del mercado. La calidad del documento de traza es, por lo tanto, parte del diseño de la evaluación, no un paso de limpieza mecánica.
Glosario de eventos de fallo y enfoque
Las trazas en bruto contienen muchas etiquetas a nivel de evento. Para mantener el notebook legible, no pedimos a los lectores que las aprendan todas. La sección de estilo AgentTrace se preocupa principalmente por los eventos de enfoque: momentos visibles en la traza donde el sistema parece requerir atención.
En esta simulación, las señales comunes de eventos de enfoque incluyen:
review finding: una superficie de revisión o validación registró un problema.review requiredoawaiting_review: la ejecución se detuvo porque el proceso de negocio simulado requería revisión.failedoblocked: la ejecución alcanzó un estado terminal degradado.triage routeo señales de redireccionamiento: el flujo de trabajo cambió de dirección porque otro propietario necesitaba actuar.- advertencias de herramientas o marcadores de políticas: una salida de herramienta estructurada indicaba riesgo, ambigüedad o una restricción de política.
Estas son señales de observabilidad, no prueba de la causa raíz. Le dicen al pase de diagnóstico dónde anclar su búsqueda hacia atrás.
focus_event_guide_df = pd.DataFrame(
[
("review finding", "An issue was recorded by review, validation, or a grading surface.", "Start from this when the trace has an explicit finding."),
("review required / awaiting_review", "The simulated business process paused for review.", "Check whether review was justified by the active risk."),
("failed / blocked", "The run ended in a degraded terminal state.", "Walk backward to the last handoff, tool, or specialist decision."),
("triage route / reroute", "The workflow changed ownership or path.", "Inspect whether routing matched the case type and environment signals."),
("tool warning / policy marker", "A structured tool exposed risk or policy context.", "Check whether later decisions used or ignored that signal."),
],
columns=["focus_event_signal", "meaning", "how_to_use_it"],
)
display(focus_event_guide_df)
example_candidates = labeled_traces_df[
labeled_traces_df["promptfoo_failed"].fillna(False)
& labeled_traces_df[DISCOVERY_DOC_COLUMN].fillna("").astype(str).str.len().gt(0)
]
if example_candidates.empty:
example_candidates = labeled_traces_df[labeled_traces_df[DISCOVERY_DOC_COLUMN].fillna("").astype(str).str.len().gt(0)]
example_row = example_candidates.sort_values("impact_score", ascending=False).iloc[0]
display(Markdown(
f"**Example trace document** \n"
f"`case_type={example_row['case_type']}` | `run_outcome={example_row['run_outcome']}` | "
f"`eval_finding={example_row['eval_finding']}` | `impact_score={example_row['impact_score']:.2f}`"
))
print(str(example_row[DISCOVERY_DOC_COLUMN])[:2400])
El documento de ejemplo anterior es una única traza renderizada como una narrativa compacta. Es intencionalmente más denso que la prosa, pero más fácil de comparar que un registro de eventos en bruto. Cuando adaptes este flujo de trabajo, dedica tiempo real a la construcción del documento. Los documentos mejores suelen producir patrones de comportamiento más útiles que las configuraciones de agrupación más complicadas.
5. Descubrimiento al estilo BERTopic
El pase de descubrimiento está inspirado en la familia de métodos BERTopic. La idea de alto nivel es modular:
- Representa cada documento de traza como un vector. Si el documento para la traza $i$ es $d_i$, el modelo de embeddings produce un vector $e_i = f(d_i)$.
- Reduce la geometría del vector. Un reductor como UMAP mapea $e_i$ a un punto $z_i$ de menor dimensión que conserva vecindarios locales útiles.
- Agrupa regiones densas. Un agrupador de densidad como HDBSCAN agrupa puntos cercanos y puede marcar los valores atípicos como ruido.
- Representa cada tema. Para cada clúster, calcula los términos que distinguen ese clúster del resto del corpus.
Este notebook utiliza el módulo auxiliar para mantener la implementación compacta, pero las principales ideas matemáticas son visibles:
- Una traza pertenece a un clúster $k$ cuando su vector de documento está cerca de otros vectores de traza en el espacio reducido.
- Un término es útil para etiquetar el clúster $k$ cuando aparece a menudo dentro de $k$ y con menos frecuencia en otros lugares.
- Una puntuación de término simple consciente de la clase es:
$$ score(t, k) = tf(t, k) \times \log\left(\frac{1 + N}{1 + df(t)}\right) $$
donde $tf(t, k)$ es la frecuencia del término $t$ dentro del clúster $k$, $df(t)$ es el número de clústeres/documentos donde aparece el término, y $N$ es el tamaño de la población de comparación. La implementación exacta puede variar, pero la intuición es estable: las etiquetas deben describir lo que hace que un clúster sea distintivo.
Finalmente, clasificamos los patrones por una métrica de triaje:
$$ impact_score(k) = prevalence_share(k) \times severity_weighted_prevalence(k) $$
Esta no es una fórmula de riesgo universal. Es una puntuación de priorización práctica: un patrón importa más cuando es común y grave.
discovery_input_df = labeled_traces_df.loc[
labeled_traces_df["has_failure"]
| labeled_traces_df["promptfoo_failed"].fillna(False)
| labeled_traces_df["run_outcome"].isin(["review_needed", "blocked", "runtime_error"])
].copy()
discovery_input_df = discovery_input_df.loc[
discovery_input_df[DISCOVERY_DOC_COLUMN].fillna("").astype(str).str.len().gt(0)
].copy()
if len(discovery_input_df) < 8:
broader_input_df = labeled_traces_df.loc[
labeled_traces_df[DISCOVERY_DOC_COLUMN].fillna("").astype(str).str.len().gt(0)
].copy()
if len(broader_input_df) > len(discovery_input_df):
display(Markdown(
"The current sample has very few failure/review traces, so discovery is broadened to all traces with documents."
))
discovery_input_df = broader_input_df
if len(discovery_input_df) < 2:
raise ValueError("Macro discovery needs at least two trace documents. Increase MACRO_EVALS_TRACE_LIMIT or run the full notebook.")
effective_min_cluster_size = min(DISCOVERY_MIN_CLUSTER_SIZE, max(2, len(discovery_input_df) // 4))
effective_n_neighbors = min(30, max(2, len(discovery_input_df) - 1))
print(f"Discovery input traces: {len(discovery_input_df):,}")
print(f"Discovery min_cluster_size: {effective_min_cluster_size}")
print(f"Discovery n_neighbors: {effective_n_neighbors}")
discovery_started = perf_counter()
discovery = run_macro_discovery(
discovery_input_df,
document_column=DISCOVERY_DOC_COLUMN,
min_cluster_size=effective_min_cluster_size,
n_neighbors=effective_n_neighbors,
top_n_terms=8,
random_state=RANDOM_STATE,
failure_only=False,
)
discovery_seconds = perf_counter() - discovery_started
topic_info_df = discovery.topic_info_df.copy()
work_df = discovery.trace_topic_df.copy()
work_df["behavior_pattern"] = work_df["topic_label"].fillna(work_df["topic_id"].astype(str))
topic_info_df["behavior_pattern"] = topic_info_df["topic_label"].fillna(topic_info_df["topic_id"].astype(str))
display(
pd.DataFrame(
[
("discovery_seconds", round(discovery_seconds, 2)),
("input_traces", len(discovery_input_df)),
("topics_including_noise", topic_info_df["topic_id"].nunique()),
("non_noise_patterns", int(topic_info_df["topic_id"].ne(-1).sum())),
],
columns=["metric", "value"],
)
)
display(
topic_info_df[
["topic_id", "behavior_pattern", "trace_count", "prevalence", "impact_score", "dominant_owner", "keywords_text"]
].sort_values("impact_score", ascending=False).head(12)
)
Interpretación de la salida de descubrimiento
El resumen de descubrimiento nos indica cuántas trazas se agruparon y cuántos patrones de comportamiento no ruidosos se recuperaron. Ejecutamos el descubrimiento en las trazas que ya tienen señales de fallo, revisión, tiempo de ejecución o Promptfoo porque este manual se centra en dónde el sistema necesita atención.
La tabla de temas debe leerse como un tablero de triaje:
trace_countyprevalencenos dicen con qué frecuencia aparece el patrón.severity_weighted_prevalencenos dice cuán graves suelen ser las trazas en el patrón.impact_scorecombina la prevalencia y la gravedad en una clasificación.dominant_owneres una etiqueta de propietario heurística, no una asignación.keywords_textproporciona los términos que hicieron que el patrón fuera distintivo.
Un patrón de comportamiento de alto impacto no es automáticamente un defecto. Es donde un revisor debe buscar primero porque el patrón es frecuente, consecuente o ambos.
impact_explanation_df = (
topic_info_df[topic_info_df["topic_id"].ne(-1)]
[["behavior_pattern", "trace_count", "prevalence", "severity_weighted_prevalence", "impact_score"]]
.sort_values("impact_score", ascending=False)
.head(8)
.copy()
)
impact_explanation_df["formula"] = "prevalence x severity_weighted_prevalence"
display(impact_explanation_df)
La tabla anterior hace que la puntuación de impacto sea concreta. Un patrón puede ocupar un lugar destacado porque aparece en muchas trazas, porque concentra trazas de mayor gravedad, o ambas cosas. En el contexto del configurador automotriz, esto ayuda a separar un caso excepcional raro de un comportamiento operativo recurrente que puede afectar a muchos pedidos.
leaderboard_fig = plot_topic_leaderboard(topic_info_df[topic_info_df["topic_id"].ne(-1)].copy(), top_n=10)
leaderboard_fig.update_layout(title_text="Behavior patterns by weighted impact")
leaderboard_fig.show()
scatter_df = discovery.topic_assignments.copy()
if "topic_label" in scatter_df.columns:
scatter_df["behavior_pattern"] = scatter_df["topic_label"].fillna(scatter_df["topic_id"].astype(str))
scatter_fig = plot_topic_scatter(
scatter_df,
color_col="behavior_pattern",
hover_cols=("run_id", "case_type", "run_outcome", "eval_finding"),
title="Trace map after discovery",
)
scatter_fig.update_layout(legend_title_text="Behavior pattern")
scatter_fig.show()
Interpretación de la tabla de clasificación y el mapa de trazas
La tabla de clasificación es la vista de cartera: clasifica los patrones de comportamiento por impacto ponderado. Úsala para decidir qué patrón merece la atención humana primero.
El mapa de trazas es una vista geométrica: cada punto es un documento de traza, colocado cerca de trazas con texto similar. Los puntos cercanos a menudo comparten rutas de enrutamiento, hallazgos o señales de entorno. Los colores muestran los patrones de comportamiento descubiertos. Trata el mapa como diagnóstico, no como geografía exacta. Su trabajo es revelar clústeres y valores atípicos que podrían ser difíciles de ver en las tablas.
En este conjunto de datos, patrones como redireccionamientos de cumplimiento, deriva de precios, puertas de cumplimiento y desajustes de ruedas/acabados corresponden a problemas comerciales reconocibles. Este es el primer momento en que las evaluaciones de nivel inferior se convierten en una historia a nivel macro: los comportamientos repetidos de los agentes son visibles en muchos casos.
case_pattern_df = (
work_df[work_df["topic_id"].ne(-1)]
.groupby(["case_type", "behavior_pattern"], dropna=False)
.size()
.reset_index(name="trace_count")
)
if not case_pattern_df.empty:
case_totals = case_pattern_df.groupby("case_type")["trace_count"].transform("sum")
case_pattern_df["share_within_case_type"] = case_pattern_df["trace_count"] / case_totals
display(case_pattern_df.sort_values(["share_within_case_type", "trace_count"], ascending=[False, False]).head(20))
heatmap_input_df = case_pattern_df.rename(
columns={
"case_type": "slice_value",
"share_within_case_type": "slice_share",
}
)
heatmap_input_df["lift"] = heatmap_input_df["slice_share"] / heatmap_input_df.groupby("behavior_pattern")["trace_count"].transform(lambda s: s.sum() / len(work_df))
plot_topic_heatmap(
heatmap_input_df,
row_col="behavior_pattern",
col_col="slice_value",
value_col="slice_share",
title="Behavior pattern concentration by generated case type",
top_n_rows=8,
top_n_cols=10,
).show()
pattern_eval_df = (
work_df[work_df["topic_id"].ne(-1)]
.assign(promptfoo_failed=lambda df: df["promptfoo_pass"].eq(False))
.groupby(["behavior_pattern", "eval_finding"], dropna=False)
.agg(trace_count=("trace_id", "count"), promptfoo_fail_rate=("promptfoo_failed", "mean"))
.reset_index()
.sort_values(["trace_count", "promptfoo_fail_rate"], ascending=False)
.head(20)
)
display(pattern_eval_df)
Interpretación del mapa de calor por tipo de caso
El mapa de calor pregunta: ¿qué escenarios generados concentran qué patrones de comportamiento?
Lee cada fila como un patrón de comportamiento y cada columna como un tipo de caso. Los valores más oscuros o más grandes significan que un patrón es más común dentro de esa porción del escenario. Esto ayuda a distinguir el comportamiento esperado del comportamiento sorprendente. Por ejemplo, un patrón de redireccionamiento de cumplimiento puede ser esperado en casos de sustitución de proveedores o capacidad, pero más sospechoso en casos limpios.
La tabla debajo del gráfico conecta los patrones con los hallazgos de nivel inferior. Si un patrón de comportamiento lleva repetidamente hallazgos final_decision_quality, un ingeniero de IA puede inspeccionar los prompts, los esquemas de herramientas o las políticas de transferencia. Si el patrón se asigna a un tipo de caso específico del negocio, un interesado de producto u operaciones puede preguntar si la política simulada en sí misma es realista.
Comparación de patrones entre segmentos
Este paso aparece aquí porque el descubrimiento al estilo BERTopic acaba de asignar a cada traza riesgosa un behavior_pattern. Antes de agrupar, podríamos comparar los casos generados, los resultados y los hallazgos de evaluación de nivel inferior. Después de agrupar, podemos hacer una pregunta de macroevaluación más útil: ¿dónde se concentra cada patrón de comportamiento descubierto?
Esta comparación no es una ecuación central del artículo de BERTopic. Es una capa simple de análisis de cohortes que aplicamos después de la asignación de temas. La idea es comparar dos proporciones:
- proporción general del patrón: entre todas las trazas agrupadas, ¿qué proporción pertenece a este patrón de comportamiento?
- proporción del patrón del segmento: dentro de un segmento, como
case_type = supplier_substitution_compound, ¿qué proporción pertenece a este patrón de comportamiento?
Luego calculamos:
$$ lift = \frac{slice\ pattern\ share}{overall\ pattern\ share} $$
Un lift de 1.0 significa que el patrón aparece en ese segmento con la misma frecuencia que en general. Un lift superior a 1.0 significa que el patrón se concentra en ese segmento. Un lift inferior a 1.0 significa que es menos común allí.
En las macroevaluaciones, este es el puente del descubrimiento a la acción. Un patrón de comportamiento es más fácil de investigar cuando podemos decir dónde aparece: un escenario generado, una versión de agente, un modo de orquestación, un régimen de mercado o un estado de revisión.
slice_lift_source_df = work_df[work_df["topic_id"].ne(-1)].copy()
overall_pattern_share = (
slice_lift_source_df["behavior_pattern"]
.value_counts(normalize=True)
.rename("overall_pattern_share")
.reset_index()
.rename(columns={"index": "behavior_pattern"})
)
slice_pattern_counts_df = (
slice_lift_source_df.groupby(["case_type", "behavior_pattern"], dropna=False)
.size()
.reset_index(name="trace_count")
)
slice_totals = (
slice_lift_source_df.groupby("case_type", dropna=False)
.size()
.rename("slice_total")
.reset_index()
)
slice_lift_df = (
slice_pattern_counts_df
.merge(slice_totals, on="case_type", how="left")
.merge(overall_pattern_share, on="behavior_pattern", how="left")
)
slice_lift_df["slice_pattern_share"] = slice_lift_df["trace_count"] / slice_lift_df["slice_total"]
slice_lift_df["lift"] = slice_lift_df["slice_pattern_share"] / slice_lift_df["overall_pattern_share"].replace(0, np.nan)
slice_lift_view_df = (
slice_lift_df[slice_lift_df["trace_count"].ge(5)]
.sort_values(["lift", "trace_count"], ascending=[False, False])
.head(12)
.loc[:, ["case_type", "behavior_pattern", "trace_count", "slice_pattern_share", "overall_pattern_share", "lift"]]
)
display(slice_lift_view_df)
La tabla anterior debe leerse como una cola de investigación. Destaca los patrones de comportamiento que están inusualmente concentrados en un case_type dado, al tiempo que requieren al menos un pequeño número de trazas de apoyo. Por ejemplo, si un patrón de enrutamiento es mucho más común en los casos de sustitución de proveedores que en general, eso sugiere que el equipo debe inspeccionar las herramientas de los proveedores, las transferencias de adquisiciones y la política de cumplimiento antes de tratar el patrón como un problema genérico del sistema.
plot_label_sankey(
work_df[["case_type", "run_outcome", "eval_finding", "behavior_pattern"]].copy(),
["case_type", "run_outcome", "eval_finding", "behavior_pattern"],
"After clustering: generated case -> outcome -> eval finding -> behavior pattern",
min_count=4,
).show()
public_label_view_df = (
work_df[["run_id", "case_type", "run_outcome", "eval_finding", "behavior_pattern", "impact_score"]]
.sort_values("impact_score", ascending=False)
.head(12)
)
display(public_label_view_df)
Qué añade el segundo diagrama de Sankey
El segundo diagrama de Sankey añade el behavior_pattern descubierto como paso final:
case_type -> run_outcome -> eval_finding -> behavior_pattern
Este es el movimiento clave de la macroevaluación. Las tres primeras etiquetas describen la configuración generada, el final y el síntoma local. La etiqueta final muestra si esos síntomas locales se colapsan en un número menor de patrones operativos repetidos.
Un interesado del negocio puede usar esto para preguntar: "¿Qué escenarios de pedidos están creando los problemas operativos más repetidos?". Un ingeniero de IA puede usarlo para preguntar: "¿Qué hallazgos de nivel inferior son en realidad el mismo patrón de enrutamiento o decisión?". Ambas vistas son útiles, y el Sankey les da un mapa compartido.
6. Diagnóstico al estilo AgentTrace
El descubrimiento nos dice qué se repite. El diagnóstico pregunta dónde inspeccionar primero.
Para un patrón de comportamiento seleccionado, reconstruimos un grafo de ejecución ligero:
$$ G = (V, E) $$
donde cada nodo $v \in V$ es un evento de traza normalizado y cada arista $e \in E$ vincula eventos a través del orden temporal, las transferencias, las llamadas a herramientas y el contexto de ejecución cercano. Luego elegimos un evento de enfoque, también llamado ancla. En esta simulación, un evento de enfoque suele ser un marcador de revisión/hallazgo, un estado relacionado con fallas o un evento de decisión de etapa tardía.
Desde ese ancla, el pase de diagnóstico retrocede a través del grafo y puntúa a los sospechosos anteriores. La puntuación es intencionalmente explicable:
$$ suspect_score = 0.4 \cdot proximity + 0.3 \cdot frequency + 0.2 \cdot bridge + 0.1 \cdot role $$
- Proximidad recompensa los eventos cercanos al evento de enfoque.
- Frecuencia recompensa los eventos que se repiten en las trazas muestreadas en el mismo patrón de comportamiento.
- Bridge recompensa los eventos que conectan partes del grafo de ejecución.
- Rol recompensa los eventos cuyo rol de agente/herramienta está plausiblemente relacionado con el hallazgo.
Esto no es una prueba de causalidad. Es una forma de convertir "este patrón es importante" en "inspecciona primero estos agentes, herramientas, transferencias o políticas de revisión".
focus_topic = pick_focus_topic(topic_info_df, exclude_noise=True)
focus_topic_id = focus_topic["topic_id"]
display(Markdown(
f"Investigating behavior pattern `{focus_topic['behavior_pattern']}` "
f"from topic `{focus_topic_id}` with `{int(focus_topic['trace_count'])}` traces."
))
display(focus_topic[["topic_id", "behavior_pattern", "trace_count", "prevalence", "impact_score", "dominant_owner", "keywords_text"]].to_frame("value"))
root_cause = drill_down_topic_root_causes(
discovery,
events_df=events_df,
topic_id=focus_topic_id,
top_n_traces=12,
max_depth=5,
)
def public_suspect_label(value: Any) -> str:
text = str(value or "unknown")
if text.startswith("failure: "):
return "eval/review signal: " + text.removeprefix("failure: ")
if text.startswith("handoff: "):
return "handoff involving " + text.removeprefix("handoff: ")
if text.startswith("function: "):
return "tool/function call by " + text.removeprefix("function: ")
if text.startswith("response: "):
return "agent response by " + text.removeprefix("response: ")
return text
if not root_cause.suspect_summary.empty:
suspect_display_df = root_cause.suspect_summary.head(12).copy()
suspect_display_df["reader_label"] = suspect_display_df["suspect_label"].apply(public_suspect_label)
suspect_display_df["is_eval_or_review_signal"] = suspect_display_df["reader_label"].str.startswith("eval/review signal")
display_columns = [
column
for column in [
"reader_label",
"node_kind",
"agent_name",
"tool_name",
"lane_label",
"mean_score",
"trace_coverage_share",
]
if column in suspect_display_df.columns
]
display(suspect_display_df[display_columns])
focus_signal_df = suspect_display_df[suspect_display_df["is_eval_or_review_signal"]].head(1)
operational_suspect_df = suspect_display_df[~suspect_display_df["is_eval_or_review_signal"]].head(1)
if not focus_signal_df.empty:
focus_signal = focus_signal_df.iloc[0]
operational_label = (
operational_suspect_df.iloc[0]["reader_label"]
if not operational_suspect_df.empty
else "the next highest-ranked non-review event"
)
display(Markdown(
"#### Reading the focus signal\n\n"
f"The leading signal is `{focus_signal['reader_label']}` from `{focus_signal.get('agent_name', 'unknown')}`. "
"In this simulation, a review finding means that a specialist or review surface recorded a structured issue while processing one customer order. "
"For a fulfillment-reroute pattern, that signal is best read as the point where the workflow says: this order has enough supply, routing, policy, or review risk to deserve attention. "
f"The first operational inspection target after that signal is `{operational_label}`."
))
suspect_plot_df = root_cause.suspect_summary.head(10).copy()
suspect_plot_df["suspect_label"] = suspect_plot_df["suspect_label"].apply(public_suspect_label)
plot_suspect_leaderboard(suspect_plot_df).show()
else:
display(Markdown("No repeated upstream suspects were recovered for this behavior pattern."))
Interpretación de la tabla de clasificación de sospechosos
El patrón de comportamiento de enfoque se selecciona por puntuación de impacto. Dependiendo de si ejecutas el conjunto de datos completo o una muestra más pequeña de prueba de humo, el patrón seleccionado puede diferir, pero el proceso de lectura es el mismo: comienza con el patrón de mayor impacto, luego inspecciona qué señales de revisión, transferencias, herramientas o respuestas de especialistas aparecen repetidamente cerca del evento de enfoque.
Una fila como eval/review signal: review finding no pretende ser misteriosa. En la simulación, un hallazgo de revisión es un marcador estructurado producido cuando un especialista o una superficie de revisión observa un problema que debería afectar la decisión del pedido. Es el punto final desde el que retrocedemos: el momento en que el flujo de trabajo ha acumulado suficiente evidencia para decir: "este pedido necesita atención".
Las filas más accionables son los eventos operativos alrededor de ese marcador: transferencias que involucran al orquestador, llamadas a herramientas/funciones por agentes de monitoreo u orquestación, transferencias de planificación de adquisiciones y respuestas de especialistas relacionadas. Esos son los lugares que un humano debe inspeccionar después de que la macroevaluación apunte a este patrón.
Desde una perspectiva técnica, esta salida le dice a un ingeniero de IA dónde inspeccionar:
- instrucciones de agente y contratos de herramientas para los agentes nombrados;
- reglas de transferencia alrededor de la transición repetida;
- si el sistema está registrando marcadores de revisión demasiado pronto o demasiado tarde;
- si una salida de herramienta se está ignorando o sobreponderando.
Desde una perspectiva comercial, la misma salida le dice a un interesado de operaciones o producto qué función comercial parece ser la propietaria del patrón. Un patrón de cumplimiento, precios, conformidad o aclaración debe involucrar a los propietarios comerciales correspondientes en la próxima revisión, no solo al ingeniero de prompt.
if root_cause.selected_path_nodes:
plot_root_cause_story(root_cause.__dict__, title="Representative path into the focus event").show()
if not root_cause.representative_trace_window.empty:
plot_trace_swimlane(root_cause.representative_trace_window, title="Representative focus-event window").show()
if not root_cause.suspect_summary.empty:
summary_suspects_df = root_cause.suspect_summary.copy()
summary_suspects_df["reader_label"] = summary_suspects_df["suspect_label"].apply(public_suspect_label)
review_signal_df = summary_suspects_df[summary_suspects_df["reader_label"].str.startswith("eval/review signal")].head(1)
operational_target_df = summary_suspects_df[~summary_suspects_df["reader_label"].str.startswith("eval/review signal")].head(1)
review_signal = review_signal_df.iloc[0] if not review_signal_df.empty else summary_suspects_df.iloc[0]
operational_target = operational_target_df.iloc[0] if not operational_target_df.empty else summary_suspects_df.iloc[0]
display(Markdown(
"### Diagnosis summary\n\n"
f"For behavior pattern `{focus_topic['behavior_pattern']}`, the focus signal is "
f"`{review_signal['reader_label']}`. In the auto-order simulation, this means the trace reached a review/eval checkpoint where a specialist found an issue that could affect fulfillment or release. "
f"The first operational target to inspect is `{operational_target['reader_label']}`. "
"Read the story strip and swimlane as the path into that checkpoint: which agents handled the order, which handoffs occurred before the marker, and whether the workflow used the right supply, routing, and review signals before deciding what to do next."
))
Interpretación de la tira de historia y el carril de nado
La tira de historia es un camino hacia el evento de enfoque. En esta ejecución, el evento de enfoque es un punto de control de revisión/evaluación dentro del patrón de comportamiento seleccionado. Es el proceso de negocio simulado que dice que este pedido tiene un problema que vale la pena revisar.
La vista de carril de nado mantiene más estructura temporal. Muestra la ventana circundante de eventos por carril o agente, con el evento de enfoque resaltado. Léelo de izquierda a derecha a medida que el pedido se mueve a través del enjambre:
- ¿Qué especialista manejó el pedido antes del hallazgo de la revisión?
- ¿El orquestador enrutó a través de los propietarios comerciales correctos en el momento adecuado?
- ¿Una llamada a una herramienta/función reveló información que debería haber cambiado la decisión del pedido?
- ¿La revisión ocurrió antes de que el flujo de trabajo se comprometiera con una recomendación de lanzamiento, reenvío, precios, cumplimiento o comunicación con el cliente?
Para un lector de negocios, el diagrama convierte un patrón abstracto en una historia operativa: este conjunto de pedidos alcanza repetidamente un punto de revisión similar. Para un ingeniero de IA, reduce el siguiente paso de depuración: inspeccionar la orquestación y la ruta de transferencia alrededor del marcador de revisión, especialmente el primer sospechoso no relacionado con la revisión resaltado en el resumen de diagnóstico.
7. Lo que aprendimos y qué hacer a continuación
El manual ha pasado por cuatro niveles de evidencia:
- Configuración de la simulación: el negocio generó casos de pedidos de vehículos eléctricos bajo condiciones cambiantes de suministro, precios, capacidad, cumplimiento y mercado.
- Evaluaciones de nivel inferior: Promptfoo proporcionó las señales de evaluación a nivel de agente/flujo de trabajo: calidad de la decisión, corrección de la política, enrutamiento, conocimiento del mercado y adecuación de la revisión.
- Descubrimiento macro: la agrupación al estilo BERTopic agrupó los hallazgos de nivel inferior en patrones de comportamiento recurrentes y los clasificó por impacto.
- Diagnóstico de trazas: el análisis de grafos al estilo AgentTrace inspeccionó un patrón de alto impacto e identificó sospechosos ascendentes repetidos.
Este enfoque escala al dirigir la atención humana hacia los patrones que son frecuentes y consecuentes. En lugar de leer cientos de trazas de arriba a abajo, un revisor puede comenzar con un patrón de comportamiento, inspeccionar ejemplos representativos y decidir qué agente, herramienta, transferencia o regla de negocio merece seguimiento.
Próximos pasos prácticos para un equipo de ingeniería de IA:
- promover las fallas de evaluación de nivel inferior más claras a un conjunto de regresión;
- revisar una pequeña muestra de calificaciones automatizadas para calibrar la rigurosidad de la rúbrica;
- rastrear patrones de comportamiento por versión de modelo, versión de prompt y modo de orquestación;
- asignar propietarios de negocio a los patrones de mayor impacto;
- inspeccionar los principales agentes, herramientas y transferencias sospechosos antes de cambiar el sistema.
Próximos pasos prácticos para un interesado de negocio:
- decidir si los tipos de casos generados coinciden con los riesgos operativos reales;
- verificar si los patrones de alto impacto corresponden a resultados importantes para el cliente o la operación;
- validar si los umbrales de revisión están produciendo el comportamiento comercial deseado;
- usar las vistas de Sankey y mapa de calor para priorizar qué escenarios necesitan una mejor política o diseño de proceso.
La lección central es simple: las evaluaciones a nivel de agente nos dicen qué comportamientos locales parecen riesgosos, mientras que las macroevaluaciones nos dicen en qué se convierten esos riesgos a escala del sistema.
Lectura adicional
- Documentación del SDK de agentes de OpenAI: Casos de uso del SDK de agentes
- Documentación de BERTopic: descripción general y recorrido del algoritmo
- Documentación de Promptfoo: proveedor de agentes de OpenAI y conversaciones de chat
- Artículo de AgentTrace: Causal Graph Tracing for Root Cause Analysis in Deployed Multi-Agent Systems
Colaboradores
Este manual es un esfuerzo de colaboración conjunta entre OpenAI y Slalom.