Agent系统的宏观评估
来源: https://developers.openai.com/cookbook/examples/partners/macro_evals_for_agentic_systems/macro_evals_for_agentic_systems 抓取时间: 2026-07-21 16:21:59
当一个Agent系统出现故障时，问题往往比单次错误响应更严重。交接可能发生得太晚，专家Agent可能在多次运行中遗漏相同的信号，或者审查流程可能针对错误的案例类型触发。为了改进系统，团队需要看到整个追踪群体中的重复行为。
本指南演示了一个多Agent系统的宏观评估工作流程。我们使用一个合成的电动汽车订单工作流程，其中专家Agent处理定价、合规、供应、工厂路由、调度和发布决策，同时市场和运营条件不断变化。
本笔记本使用预先计算的合成追踪和保存的低级评估标签，因此您无需OpenAI API密钥即可运行完整的工作流程。
您将学习如何：
- 生成或收集许多追踪的Agent运行；
- 对每个完成的运行运行低级评估；
- 将每个追踪转换为紧凑的文档；
- 在整个群体中发现重复的行为模式；以及
- 深入研究一个高影响力的模式，找出人类接下来应该检查系统的哪个部分。
目标不是为每个追踪构建完美的分类法。目标是展示AI工程团队如何从数千个Agent事件转变为技术和业务利益相关者都能理解的少量模式。
端到端Agent系统映射
核心思想是，笔记本评估一个已保存的Agent系统，而不是通用的聊天记录。场景输入驱动一个编排的专家群体，运行时发出追踪包，保存的Promptfoo标签与标准化追踪连接，宏观评估层将该证据转换为模式和诊断视图。
1. 为什么需要宏观评估？
评估是AI团队衡量系统是否正常工作的方式。对于简单的模型调用，评估可能将一个输出与评分标准或参考答案进行比较。对于Agent系统，我们还需要评估系统是否使用了正确的工具、委派给了正确的专家、在风险较高时暂停审查以及始终立足于业务上下文。
多Agent系统使这变得更加困难，因为最终答案只是更长工作流程中的最后一个事件。发布建议可能看起来合理，而追踪显示定价Agent忽略了激励措施，供应Agent错过了缺货，或者编排器绕过了必要的审查步骤。
本笔记本将问题分为两个层次：
- 低级评估 对个体Agent、交接、工具和已完成的运行进行评分。在这个例子中，Promptfoo通过评估运行是否处理了最终决策质量、策略正确性、专家路由、市场漂移和审查适当性来代表那个Agent级别的评估层。
- 宏观评估 在许多低级发现中进行观察。它们询问：哪些类型的问题重复出现，它们集中在哪里，以及我们应该首先检查Agent工作流程的哪个部分？
我们将在整个指南中使用四个面向读者的标签：
case_type：生成的业务情况，例如干净的订单、验证阻止、供应商替换或定价异常。run_outcome：运行如何结束，例如完成、等待审查、阻止或失败。eval_finding：说明什么看起来错误或有风险的低级信号。behavior_pattern：在许多追踪中发现的重复模式。
一个有用的心理模型是：case_type是设置，run_outcome是结局，eval_finding是局部症状，而behavior_pattern是群体层面的模式。
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
设置和数据材料
安装依赖项，然后加载与本例捆绑的离线数据集。保存的Promptfoo标签是本地数据文件夹的一部分，因此本笔记本不需要单独的Promptfoo配置、Promptfoo运行构件或OpenAI API密钥。
预期文件：
data/trace_results.jsonl
data/run_summary.json
data/trace_bundles.zip
data/eval_labels.jsonl
trace_bundles.zip会在笔记本首次运行时自动展开到本地缓存中。完整的SQLite追踪快照可以放在data/trace_snapshot.sqlite处以备可选的丰富，但端到端工作流程不需要它。
如果您的数据位于示例文件夹之外，请将MACRO_EVALS_DATA_ROOT设置为该目录。如果标签单独存放，请设置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. 模拟：变化世界中的汽车订单
模拟的业务是电动汽车订单和配置后工作流程。客户选择了车辆配置，公司需要决定订单是否可以按原样继续、需要调整、应该重新路由、需要替换还是应该暂停审查。
模拟包括使真实汽车履改变得困难的各种约束：
- 组件可用性和供应商替换；
- 工厂产能和生产调度；
- 定价异常、促销和激励措施；
- 关税和过时的市场信号；
- 区域合规约束；
- 客户澄清和升级路径；
- 风险或模糊案例的发布审查阈值。
Agent群体围绕这些业务职责组织。编排器接收订单和当前环境，然后委派给专家，如验证、供应风险、采购规划、产能平衡、工厂路由、市场情报、定价、合规、客户沟通和发布审查。
这自然映射到OpenAI Agents SDK。在SDK中，Agent是工作流程的核心单元：它打包模型、指令和可选的运行时行为，如工具、交接、护栏和结构化输出。模拟遵循该模式：
- 专家Agent 打包决策的一部分的指令和工具；
- 交接 让编排器委派给另一个专家Agent，而不是将每个职责塞进一个提示中；
- 函数工具 通过结构化的输入和输出公开订单数据、环境信号和批准标记；
- 护栏和审查阈值 表示针对风险或模糊案例的验证、阻止和人工审查流程；
- 结构化输出 使下游评分和聚合成为可能；
- 追踪 保留模型调用、工具调用、交接、护栏和自定义跨度的结构化记录，用于调试和宏观级别分析。
笔记本后面的低级评估基于这个模拟故事。如果案例类型表示存在关税压力下的供应商替换，追踪应该显示对供应、策略、市场和审查风险的意识。如果案例类型是干净的，不必要的升级本身就是一个发现。
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."
))
一个包代表什么
在本笔记本中，包是一个模拟客户订单交互的证据包。
想象一下，一个客户配置了一辆电动汽车，企业需要决定下一步做什么。群体接收该订单加上当前运营世界：供应约束、工厂产能、促销、激励措施、关税、竞争对手压力和审查阈值。然后Agent通过专家路由工作并产生最终状态。包是我们事后审计该交互所需的一切。
包很重要，因为宏观评估需要最终答案背后的工作流程证据。它们需要知道咨询了哪些Agent、调用了哪些工具、哪些环境信号处于活动状态、是否需要审查，以及工作流程在哪里改变了方向。有了这些证据，我们可以从"这次运行中发生了什么？"转向"哪些工作流程模式在多次运行中重复出现？"
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)
如何阅读数据集概况
数据集概况告诉我们我们即将评估的模拟业务流程的规模和纹理。每个可分析的行都是一个客户订单交互，具有足够的追踪证据来重建Agent群体看到的内容、咨询了哪些专家，以及工作流程如何结束。
生成的批次要求群体处理1,000个合成订单交互。对于其中的992个，我们有一个包：用于对运行评分、构建追踪文档、将其与类似运行聚类以及事后检查Agent路径的完整证据包。这给了我们一个足够大的群体来寻找重复行为，同时仍然保留解释单个示例所需的追踪细节。
典型的包是模拟业务流程的结构化记录：订单设置、活动世界事件、专家交接、工具/函数活动、审查构件和终端状态。这就是为什么这个数据集可以支持宏观评估。我们可以评估单个决策，也可以询问重复的工作流程模式是否在数百个丰富的交互记录中出现。
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))
casetype的含义
case_type是生成器的场景标签。它描述了在任何评估或聚类发生之前，群体被要求处理的业务情况类型。
该数据集中的示例包括：
clean_simple：相对直接的订单，正确的行为通常是完成而不需要不必要的审查。validation_block_simple：配置存在验证问题，因此群体应避免过于自信的发布。supplier_substitution_compound：组件可用性产生替换决策，通常伴随下游路由和调度影响。pricing_exception_compound：定价、激励或保证金策略需要专家审查。regional_compliance_compound：订单需要区域策略或合规处理。
上面的条形图是覆盖率视图。它显示模拟是否产生了足够的多样性来在不同的业务压力下评估群体。一个强大的宏观评估数据集既需要普通案例也需要压力案例，因为只有当我们可以比较不同设置下的行为时，重复模式才变得有意义。
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))
上表将生成器标签转换为业务语言。这很重要，因为相同的后期模式可能根据设置意味着不同的事情。供应商替换案例中的履行重新路由可能是可取的。干净案例中的相同重新路由可能是不必要的复杂性。
3. 使用Promptfoo进行低级Agent评估
成熟的多Agent系统不应仅依赖于最终答案检查。每个已启动的Agent通常都需要自己的评估：这位专家是否使用了正确的证据、调用了正确的工具、尊重了策略、在正确的时间交接，并产生了系统其余部分可以信任的输出？
Promptfoo在本笔记本中扮演那个角色。它代表通常位于生产工作流程中Agent旁边的低级评估层。在实时系统中，这些检查中的一些可能在线运行，一些可能异步运行，一些可能被采样以供人工审查。实现细节不如合同重要：每次运行都应携带评估信号，说明在Agent和工作流程级别上看起来正确、有风险或错误的内容。
在这个数据集中，Promptfoo用反映团队为真实系统构建的那种Agent级评估的问题对已完成的追踪进行评分：
- 最终决策是否从活跃问题中得出？
- 系统是否尊重定价、关税、激励、区域和策略约束？
- 编排器是否激活了案例隐含的专家？
- 运行是否响应了过时的市场信号，而不是表现得好像世界是静态的？
- 审查或升级是否与风险相称？
这些检查产生eval_finding。失败的低级评估是一个局部信号：一个追踪，一个评分标准，一个症状。后面的宏观评估部分询问那些局部信号在群体规模上变成了什么。它们是随机分散的，还是揭示了指向特定Agent、交接、工具或业务策略的重复行为模式？
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()
解释Promptfoo输出
饼图是最简单的低级记分卡：它将通过所有评分标准检查的追踪与至少有一个失败检查的追踪分开。在实时多Agent系统中，这是那种告诉我们在进行任何宏观分析之前哪些运行值得关注的层。
失败评分标准条形图回答了一个更有用的问题：哪种Agent或工作流程问题最常出现？对于这个数据集，最终决策质量是主要的低级发现，而策略正确性、审查适当性和市场漂移意识也出现了。这表明宏观层应该更少关注孤立的语法错误，更多关注重复的决策模式。
这是通往宏观评估的桥梁。Promptfoo为每个追踪提供本地评估标签。笔记本的其余部分询问这些标签如何在整个群体中组织。换句话说：Agent级评估创建原始信号，宏观评估将许多这样的信号转变为重复系统行为的地图。
4. 构建分析数据集
现在我们将运行包标准化为两个分析表：
traces_df：每次运行一行，包含元数据、结果、发现和文档字段。events_df：每个标准化追踪事件一行，包括交接、工具调用、状态事件、模型响应和审查/发现标记。
我们还构建追踪文档。文档是BERTopic风格部分将聚类的建模对象。笔记本使用doc_structured_summary，因为它紧凑但仍保留场景、路由、状态转换、交接、发现和终端状态。
公共分析路径是：
case_type -> run_outcome -> eval_finding -> behavior_pattern
前三个标签在聚类前已知。第四个在发现后出现。
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))
解释分析概况
上面的概况确认低级评估层已加入到标准化追踪群体中。重要数字是：
- normalized_traces：我们可以检查的包支持的群体；
- normalized_events：这些追踪背后的事件级证据；
- case_types：生成器产生的场景覆盖率；以及
- Promptfoo失败或携带审查/失败的追踪：与宏观发现最相关的低级信号群体。
确切的计数取决于您是运行完整笔记本还是为冒烟测试设置MACRO_EVALS_TRACE_LIMIT。示例行显示笔记本如何将原始数据简化为可读标签。例如，一个以final_decision_quality发现结束审查的pricing_exception_compound案例现在很容易在笔记本的其余部分中跟踪。
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)
第一个桑基图教给我们什么
第一个桑基图是聚类前视图。它显示生成的案例类型如何流入运行结果和低级发现。
从左到右阅读：
- 来自
case_type的宽带意味着该场景经常出现； - 分成
run_outcome显示该场景是否倾向于完成、暂停、阻止或失败； - 最终进入
eval_finding的带显示附加了哪个低级评分标准或运行时信号。
这对团队已经很有用。业务读者可以询问模拟是否产生正确类型的压力。AI工程师可以询问某些场景是否过度产生相同的低级发现。它还不能回答的是，那些发现是否代表相同的底层行为模式。这就是为什么我们接下来进行聚类。
追踪文档：将运行转化为可比较的文本
原始Agent追踪太详细，无法直接聚类。它可能包含数百个事件、长模型响应、工具负载和重复的状态更新。文档构建步骤将每次运行压缩成可比较的视图，同时保留对宏观评估重要的信息。
好的追踪文档包括：
- 业务设置（
case_type、选定路线、活动环境信号）； - 运行结果和严重性；
- 重要的交接和专家激活；
- 审查/发现标记；
- 简短的状态转换摘要。
文档视图定义了聚类算法允许注意到什么。包括Agent交接有助于宏观评估发现路由模式。包括环境信号有助于它发现市场漂移故障。因此，追踪文档的质量是评估设计的一部分，而不是机械的清理步骤。
故障和焦点事件词汇表
原始追踪包含许多事件级标签。为了保持笔记本可读，我们不要求读者学习所有这些。AgentTrace风格部分主要关心焦点事件：追踪中系统似乎需要关注的可见时刻。
在这个模拟中，常见的焦点事件信号包括：
review finding：审查或验证表面记录了问题。review required或awaiting_review：运行暂停，因为模拟的业务流程需要审查。failed或blocked：运行达到降级的终端状态。triage route或重新路由信号：工作流程改变方向，因为另一个所有者需要采取行动。- 工具警告或策略标记：结构化工具输出指示风险、模糊性或策略约束。
这些是可观察性信号，而不是根本原因的证明。它们告诉诊断通道在哪里锚定其向后搜索。
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])
上面的示例文档是单个追踪渲染为紧凑的叙述。它有意比散文更密集，但比原始事件日志更容易比较。当您调整此工作流程时，请花实际时间在文档构建上。更好的文档通常比更复杂的聚类设置产生更有用的行为模式。
5. BERTopic风格发现
发现传递受BERTopic方法系列的启发。高级思想是模块化的：
- 将每个追踪文档表示为向量。 如果追踪$i$的文档是$d_i$，嵌入模型产生向量$e_i = f(d_i)$。
- 减少向量几何。 减少器如UMAP将$e_i$映射到保留有用局部邻域的低维点$z_i$。
- 聚类密集区域。 密度聚类器如HDBSCAN对附近的点进行分组，并可以将异常值标记为噪声。
- 表示每个主题。 对于每个聚类，计算区分该聚类与语料库其余部分的术语。
本笔记本使用辅助模块来保持实现紧凑，但主要的数学思想是可见的：
- 当追踪的文档向量在减少的空间中靠近其他追踪向量时，该追踪属于聚类$k$。
- 当术语经常出现在$k$内而在其他地方出现较少时，该术语对于标记聚类$k$是有用的。
- 一个简单的类感知术语分数是：
$$ score(t, k) = tf(t, k) \times \log\left(\frac{1 + N}{1 + df(t)}\right) $$
其中$tf(t, k)$是聚类$k$内术语$t$的术语频率，$df(t)$是术语出现的聚类/文档数量，$N$是比较群体大小。确切的实现可能会有所不同，但直觉是稳定的：标签应该描述是什么使聚类与众不同。
最后，我们通过分诊指标对模式进行排名： $$ impact_score(k) = prevalence_share(k) \times severity_weighted_prevalence(k) $$
这不是一个普遍的风险公式。它是一个实用的优先级分数：当一个模式既常见又重要时，它更重要。
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)
)
解释发现输出
发现摘要告诉我们聚类了多少追踪，以及恢复了多少非噪声行为模式。我们在已经有失败、审查、运行时或Promptfoo信号的追踪上运行发现，因为本指南专注于系统需要关注的地方。
主题表应该作为分诊板阅读：
trace_count和prevalence告诉我们模式出现的频率。severity_weighted_prevalence告诉我们模式中的追踪往往有多严重。impact_score将普遍性和严重性组合成排名。dominant_owner是启发式所有者标签，而不是分配。keywords_text给出使模式与众不同的术语。
高影响力的行为模式不一定是缺陷。它是审查者应该首先查看的地方，因为模式频繁、重要，或两者兼而有之。
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)
上表使影响评分具体化。一个模式可以排名很高，因为它出现在许多追踪中，因为它集中了更高严重性的追踪，或两者兼而有之。在汽车配置器设置中，这有助于将罕见的边缘情况与可能影响许多订单的重复操作行为分开。
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()
解释排行榜和追踪地图
排行榜是投资组合视图：它按加权影响对行为模式进行排名。用它来决定哪种模式应该首先引起人类关注。
追踪地图是几何视图：每个点是一个追踪文档，放在具有类似文本的追踪附近。附近的点通常共享路由路径、发现或环境信号。颜色显示发现的行为模式。将地图视为诊断，而不是精确的地理。它的工作是揭示可能在表格中难以看到的聚类和异常值。
在这个数据集中，诸如履行重新路由、定价漂移、合规门和车轮/装饰不匹配等模式对应于可识别的业务问题。这是低级评估变成宏观级别故事的第一个时刻：重复的Agent行为在许多案例中可见。
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)
解释案例类型热图
热图询问：哪些生成的场景集中了哪些行为模式？
将每一行作为行为模式阅读，将每一列作为案例类型。较暗或较大的值意味着模式在该场景切片中更常见。这有助于区分预期行为和令人惊讶的行为。例如，履行重新路由模式可能在供应商替换或产能案例中是预期的，但在干净案例中更可疑。
图表下方的表格将模式连接回低级发现。如果一个行为模式反复携带final_decision_quality发现，AI工程师可能会检查提示、工具模式或交接策略。如果模式映射到特定于业务的案例类型，产品或运营利益相关者可以询问模拟的策略本身是否现实。
跨切片比较模式
这一步出现在这里，因为BERTopic风格的发现刚刚给每个有风险的追踪一个behavior_pattern。在聚类之前，我们可以比较生成的案例、结果和低级评估发现。聚类之后，我们可以问一个更有用的宏观评估问题：每个发现的行为模式集中在哪里？
这种比较不是BERTopic论文中的核心方程。它是我们在主题分配后应用的简单队列分析层。思想是比较两个份额：
- 整体模式份额：在所有聚类的追踪中，有多少份额属于这种行为模式？
- 切片模式份额：在一个切片内，例如
case_type = supplier_substitution_compound，有多少份额属于这种行为模式？
然后我们计算： $$ lift = \frac{slice\ pattern\ share}{overall\ pattern\ share} $$
1.0的提升意味着模式在该切片中出现的频率与它整体出现的频率大致相同。高于1.0的提升意味着模式集中在该切片中。低于1.0的提升意味着它在那里不太常见。
在宏观评估中，这是从发现到行动的桥梁。当我们可以说行为模式出现在哪里时，它更容易调查：生成的场景、Agent版本、编排模式、市场制度或审查状态。
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)
上表应该作为调查队列阅读。它突出了在给定case_type中异常集中的行为模式，同时要求至少少量支持追踪。例如，如果路由模式在供应商替换案例中比在整体中更常见，这表明团队应该在将模式视为通用系统问题之前检查供应商工具、采购交接和履行政策。
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)
第二个桑基图增加了什么
第二个桑基图增加了发现的behavior_pattern作为最终步骤：
case_type -> run_outcome -> eval_finding -> behavior_pattern
这是关键的宏观评估举措。前三个标签描述了生成的设置、结局和局部症状。最终标签显示那些局部症状是否崩溃成少量重复的操作模式。
业务利益相关者可以用它来问，"哪些订单场景正在创造最多的重复运营问题？" AI工程师可以用它来问，"哪些低级发现实际上是相同的路由或决策模式？" 两种视图都有用，桑基图给它们一个共享的地图。
6. AgentTrace风格诊断
发现告诉我们什么重复。诊断询问首先检查哪里。
对于选定的行为模式，我们重建一个轻量级的执行图： $$ G = (V, E) $$
其中每个节点$v \in V$是标准化追踪事件，每个边$e \in E$通过时间顺序、交接、工具调用和附近的执行上下文链接事件。然后我们选择一个焦点事件，也称为锚。在这个模拟中，焦点事件通常是审查/发现标记、与故障相关的状态或后期决策事件。
从该锚点，诊断传递向后遍历图并对上游嫌疑犯评分。分数有意是可解释的： $$ suspect_score = 0.4 \cdot proximity + 0.3 \cdot frequency + 0.2 \cdot bridge + 0.1 \cdot role $$
- 接近度 奖励靠近焦点事件的事件。
- 频率 奖励在相同行为模式的采样追踪中反复出现的事件。
- 桥梁 奖励连接执行图各部分的事件。
- 角色 奖励其Agent/工具角色与发现合理相关的事件。
这不是因果关系的证明。这是一种将"这个模式很重要"转变为"首先检查这些Agent、工具、交接或审查策略"的方式。
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."))
解释嫌疑犯排行榜
焦点行为模式通过影响评分选择。根据您是运行完整数据集还是较小的冒烟测试样本，选择的模式可能不同，但阅读过程是相同的：从最高影响力的模式开始，然后检查哪些审查信号、交接、工具或专家响应在焦点事件附近反复出现。
诸如eval/review signal: review finding之类的行并不意味着神秘。在模拟中，审查发现是当专家或审查表面观察到可能影响订单决策的问题时产生的结构化标记。这是我们向后追踪的终点：工作流程积累了足够证据说"这个订单需要关注"的时刻。
更具可操作性的行是该标记周围的操作事件：涉及编排器的交接、监控或编排Agent的工具/函数调用、采购计划交接和相关的专家响应。这些是宏观评估指向该模式后人类应该检查的地方。
从技术角度来看，此输出告诉AI工程师在哪里检查：
- 命名Agent的Agent指令和工具契约；
- 重复转换周围的交接规则；
- 系统是否过早或过晚记录审查标记；
- 工具输出是否被忽略或过度加权。
从业务角度来看，相同的输出告诉运营或产品利益相关者哪个业务功能似乎拥有该模式。履行、定价、合规或澄清模式应将相应的业务所有者带入下一次审查，而不仅仅是提示工程师。
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."
))
解释故事条和泳道
故事条是进入焦点事件的路径。在这次运行中，焦点事件是选定行为模式内的审查/评估检查点。这是模拟的业务流程说这个订单有值得审查的问题。
泳道视图保留了更多的时间结构。它按泳道或Agent显示事件的周围窗口，焦点事件突出显示。从左到右阅读，因为订单在群体中移动：
- 在审查发现之前，哪位专家处理了订单？
- 编排器是否在正确的时间路由通过正确的业务所有者？
- 工具/函数调用是否公开了应该改变订单决策的信息？
- 审查是否发生在工作流程承诺发布、重新路由、定价、合规或客户沟通建议之前？
对于业务读者，图表将抽象模式转变为操作故事：这组订单反复到达类似的审查点。对于AI工程师，它缩小了下一步调试步骤：检查审查标记周围的编排和交接路径，特别是诊断摘要中突出显示的第一个非审查嫌疑犯。
7. 我们学到了什么以及下一步做什么
本指南通过了四个级别的证据：
- 模拟设置：业务在不断变化的供应、定价、产能、合规和市场条件下生成电动汽车订单案例。
- 低级评估：Promptfoo提供了Agent/工作流程级别的评估信号：决策质量、策略正确性、路由、市场意识和审查适当性。
- 宏观发现：BERTopic风格的聚类将低级发现分组为重复的行为模式，并按影响排名。
- 追踪诊断：AgentTrace风格的图分析检查了一个高影响力的模式，并识别了重复的上游嫌疑犯。
这种方法通过将人类注意力引导到既频繁又重要的模式上来扩展。审查者不必从头到尾阅读数百个追踪，而是可以从行为模式开始，检查代表性示例，并决定哪个Agent、工具、交接或业务规则值得跟进。
AI工程团队的实际下一步：
- 将最清晰的低级评估失败提升到回归套件中；
- 审查一小部分自动化等级以校准评分标准严格性；
- 按模型版本、提示版本和编排模式跟踪行为模式；
- 将业务所有者分配给最高影响力的模式；
- 在改变系统之前检查顶级嫌疑犯Agent、工具和交接。
业务利益相关者的实际下一步：
- 决定生成的案例类型是否与真实的运营风险匹配；
- 检查高影响力模式是否对应于重要的客户或运营结果；
- 验证审查阈值是否产生预期的业务行为；
- 使用桑基图和热图视图优先考虑哪些场景需要更好的策略或流程设计。
核心教训很简单：Agent级评估告诉我们哪些本地行为看起来有风险，而宏观评估告诉我们这些风险在系统规模上变成了什么。
扩展阅读
- OpenAI Agents SDK文档：Agents SDK use cases
- BERTopic文档：overview和algorithm walkthrough
- Promptfoo文档：OpenAI Agents provider和chat conversations
- AgentTrace论文：Causal Graph Tracing for Root Cause Analysis in Deployed Multi-Agent Systems
贡献者
本指南是OpenAI和Slalom之间的联合协作成果。