智能体系统的宏观评估：从海量轨迹到可操作模式

本教程面向有编程基础的开发者，深入讲解多智能体系统的宏观评估方法。通过一个合成EV订单工作流，展示了如何从海量智能体轨迹中提取可操作的行为模式。内容涵盖低级评估（Promptfoo）、轨迹文档构建、BERTopic风格聚类发现以及AgentTrace风格根因诊断。读者将学会如何将数千个智能体事件转化为少量高影响力的模式，并定位系统需要优先检查的环节。教程提供完整代码、数据准备步骤和优化建议，帮助团队系统化改进智能体系统。

技术背景

多智能体系统在复杂业务场景中日益普及，但评估其整体性能面临巨大挑战。传统评估仅关注最终输出，忽略了系统内部复杂的交互和决策过程。当系统出现问题时，往往不是单一错误，而是重复出现的模式，如交接延迟、专家遗漏信号或审查流程不当。宏观评估旨在通过分析大量轨迹，识别这些重复模式，从而指导系统优化。本教程基于OpenAI Agents SDK和Promptfoo，结合BERTopic和AgentTrace技术，提供了一套完整的宏观评估工作流。

核心原理

宏观评估的核心思想是将智能体系统的评估分为两个层次：低级评估和宏观评估。低级评估对单个智能体、交接、工具调用和运行结果进行评分，使用Promptfoo等工具实现。宏观评估则聚合大量低级评估结果，通过聚类和诊断发现重复的行为模式。关键概念包括：case_type（业务场景）、run_outcome（运行结果）、eval_finding（低级信号）和behavior_pattern（群体级模式）。工作流包括：生成轨迹、低级评估、构建轨迹文档、聚类发现模式、深入诊断高影响模式。

架构分析

系统架构分为四层：模拟层生成合成业务场景和轨迹；评估层使用Promptfoo进行低级评估；分析层构建轨迹文档并进行BERTopic风格聚类；诊断层使用AgentTrace风格图分析定位根因。数据流为：场景输入驱动智能体群体，运行时产生追踪包，Promptfoo标签与追踪连接，宏观评估层将证据转化为模式和诊断视图。关键组件包括：编排器、专家智能体、工具函数、审查机制和追踪系统。

完整实战教程

当智能体系统失败时，问题往往比单个错误响应更大。交接可能发生得太晚，专家智能体可能在多次运行中遗漏同一信号，或者审查流程可能针对错误的案例类别触发。为了改进系统，团队需要观察整个轨迹群体中的重复行为。

本教程介绍了一个多智能体系统的宏观评估工作流程。我们使用一个合成的EV订单工作流，其中专家智能体处理定价、合规、供应、工厂路由、调度和发布决策，同时市场和运营条件不断变化。

该笔记本使用预计算的合成轨迹和保存的低级评估标签，因此您无需OpenAI API密钥即可运行完整工作流程。

您将学习如何：

  1. 生成或收集许多跟踪的智能体运行；
  2. 对每次完成的运行进行低级评估；
  3. 将每个轨迹转换为紧凑文档；
  4. 发现整个群体中的重复行为模式；以及
  5. 深入分析一个高影响模式，以找到人类下一步应检查系统的位置。

目标不是构建每个轨迹的完美分类法。目标是展示AI工程团队如何从数千个智能体事件转变为少量模式，这些模式对技术和业务利益相关者都易于理解。

端到端智能体系统图谱

End-to-end macro evals architecture关键思想是，笔记本评估的是一个已保存的智能体系统，而不是通用的聊天记录。场景输入驱动一个编排好的专家群体，运行时发出追踪包，保存的Promptfoo标签与规范化追踪连接，宏观评估层将这些证据转化为模式和诊断视图。

1. 为什么需要宏观评估？

评估是 AI 团队衡量系统是否正常工作的方式。对于简单的模型调用，评估可能会将单个输出与评分标准或参考答案进行比较。对于智能体系统，我们还需要评估系统是否使用了正确的工具、是否委派给了正确的专家、在风险高时是否暂停审查，以及是否始终基于业务上下文。

多智能体系统使这变得更加困难，因为最终答案只是更长工作流中的最后一个事件。一个看似合理的发布建议，其追踪记录可能显示定价智能体忽略了激励措施，供应智能体错过了缺货情况，或者编排器绕过了必需的审查步骤。

本笔记本将问题分为两个层面：

在整个手册中，我们将使用四个面向读者的标签：

一个有用的心智模型是：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. 模拟：变化世界中的汽车订单

模拟的业务是电动汽车订单及后配置工作流。客户已选择车辆配置，公司需要决定订单是否可以按原样继续、需要调整、应重新路由、需要替换，或应暂停审查。

模拟包含使真实汽车履约变得困难的各类约束：

智能体群围绕这些业务职责组织。编排器接收订单和当前环境，然后委派给验证、供应风险、采购规划、产能平衡、工厂路由、市场情报、定价、合规、客户沟通和发布审查等专家。

这自然映射到 OpenAI Agents SDK。在 SDK 中，智能体是工作流的核心单元：它封装模型、指令和可选运行时行为，如工具、交接、护栏和结构化输出。模拟遵循该模式：

笔记本后面的低级评估基于此模拟故事。如果案例类型表明在关税压力下存在供应商替换，追踪应显示对供应、政策、市场和审查风险的意识。如果案例类型是干净的，不必要的升级本身就是一个发现。

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."
    ))

一个 Bundle 代表什么

在本笔记本中，一个 bundle 是模拟客户订单交互的证据包。

想象一个客户配置了一辆电动汽车，企业需要决定下一步该做什么。蜂群接收到该订单以及当前运营环境：供应限制、工厂产能、促销、激励措施、关税、竞争压力和审查阈值。然后，代理将工作路由给专家并产生最终状态。Bundle 是我们之后审计该交互所需的一切。

Bundle 之所以重要，是因为宏观评估需要最终答案背后的工作流证据。它们需要知道咨询了哪些代理、调用了哪些工具、哪些环境信号处于激活状态、是否需要进行审查，以及工作流在何处改变了方向。有了这些证据，我们就能从“这次运行发生了什么？”转向“哪些工作流模式在多次运行中重复出现？”

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)

如何阅读数据集概览

数据集概览告诉我们即将评估的模拟业务流程的规模和纹理。每个可分析的行都是一次客户订单交互，包含足够的痕迹证据，以重建代理群体所见、咨询了哪些专家以及工作流如何结束。

生成的批次要求代理群体处理1,000次合成订单交互。其中992次我们有一个捆绑包：一个完整的证据包，用于评估运行、构建跟踪文档、将其与类似运行聚类，并随后检查代理路径。这为我们提供了足够大的总体来寻找重复行为，同时保留解释单个示例所需的跟踪细节。

典型的捆绑包是模拟业务流程的结构化记录：订单设置、活跃的世界事件、专家交接、工具/功能活动、审查工件和终止状态。这就是为什么该数据集可以支持宏观评估。我们可以评估单个决策，也可以询问在数百个丰富的交互记录中是否出现重复的工作流模式。

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))

case_type 的含义

case_type 是生成器给出的场景标签。它描述了在执行任何评估或聚类之前，群体被要求处理的业务情境类型。

该数据集中的示例包括：

上面的条形图是覆盖视图。它显示了模拟是否产生了足够的多样性，以评估群体在不同业务压力下的表现。一个强大的宏观评估数据集需要同时包含普通案例和压力案例，因为只有当我们能够比较不同设置下的行为时，重复出现的模式才有意义。

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 评估

一个成熟的多智能体系统不应仅依赖最终答案的检查。每个启动的智能体通常都需要自己的评估：这个专家是否使用了正确的证据、调用了正确的工具、遵守了策略、在正确的时间进行了交接，并产生了系统其余部分可以信任的输出？

Promptfoo 在这个笔记本中扮演了这个角色。它代表了低层级的评估层，在生产工作流中通常会与智能体并排放置。在实时系统中，其中一些检查可能在线运行，一些可能异步运行，还有一些可能抽样供人工审查。实现细节不如契约重要：每次运行都应携带评估信号，说明在智能体和工作流层面哪些看起来正确、有风险或错误。

在这个数据集中，Promptfoo 使用与团队为真实系统构建的智能体级评估相呼应的问题来对完成的轨迹进行评分：

这些检查产生 eval_finding。一个失败的低层级评估是一个局部信号：一个轨迹、一个规则、一个症状。后面的宏观评估部分会问这些局部信号在群体规模上会变成什么。它们是随机分散的，还是揭示了指向特定智能体、交接、工具或业务策略的重复行为模式？

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 输出

饼图是最简单的底层记分卡：它将通过所有规则检查的轨迹与至少有一项检查失败的轨迹分开。在实时多智能体系统中，这种层级告诉我们哪些运行在宏观分析之前值得关注。

失败规则条形图回答了一个更有用的问题：哪种智能体或工作流问题最常出现？对于这个数据集，最终决策质量是主要的底层发现，而策略正确性、审查适当性和市场漂移意识也出现。这表明宏观层应更少关注孤立的语法错误，而更多关注重复的决策模式。

这是通往宏观评估的桥梁。Promptfoo 为每个轨迹提供局部评估标签。笔记本的其余部分询问这些标签如何在整个群体中组织。换句话说：智能体级评估创建原始信号，而宏观评估将许多这样的信号转化为系统重复行为的地图。

4. 构建分析数据集

现在我们将运行捆绑包规范化为两个分析表：

我们还构建跟踪文档。该文档是 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))

解读分析档案

上述档案确认了较低层级的评估层已加入归一化追踪群体。关键数字包括：

具体计数取决于您是运行完整笔记本还是设置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)

第一个桑基图的启示

第一个桑基图是聚类前的视图。它展示了生成的用例类型如何流向运行结果和更底层的发现。

从左到右阅读：

这对团队已经很有用。业务读者可以询问模拟是否产生了正确的压力类型。AI 工程师可以询问某些场景是否过度产生相同的底层发现。但它还不能回答这些发现是否代表相同的底层行为模式。这就是我们接下来进行聚类的原因。

追踪文档：将运行转化为可比较的文本

原始代理追踪过于详细，无法直接聚类。它可能包含数百个事件、长模型响应、工具负载和重复的状态更新。文档构建步骤将每次运行压缩为可比较的视图，同时保留对宏观评估至关重要的信息。

一个好的追踪文档包括：

文档视图定义了聚类算法允许注意的内容。包含代理交接有助于宏观评估发现路由模式。包含环境信号有助于发现市场漂移失败。因此，追踪文档的质量是评估设计的一部分，而不是机械的清理步骤。

失败与焦点事件术语表

原始跟踪记录包含许多事件级标签。为了让笔记本保持可读性，我们不要求读者全部掌握它们。AgentTrace风格的部分主要关注焦点事件：跟踪记录中系统似乎需要关注的可见时刻。

在此模拟中，常见的焦点事件信号包括：

这些是可观测性信号，而非根本原因的证据。它们告诉诊断过程在哪里锚定其向后搜索。

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 方法家族的启发。其高层思想是模块化的：

  1. 将每个追踪文档表示为向量。 如果追踪 $i$ 的文档是 $d_i$，嵌入模型生成向量 $e_i = f(d_i)$。
  2. 降低向量几何维度。 诸如 UMAP 的降维器将 $e_i$ 映射到低维点 $z_i$，保留有用的局部邻域。
  3. 聚类密集区域。 诸如 HDBSCAN 的密度聚类器将邻近点分组，并可将离群点标记为噪声。
  4. 表示每个主题。 对于每个聚类，计算将该聚类与语料库其余部分区分开的术语。

本笔记本使用辅助模块以保持实现紧凑，但主要的数学思想是可见的：

$$

score(t, k) = tf(t, k) \times \log\left(\frac{1 + N}{1 + df(t)}\right)

$$

其中 $tf(t, k)$ 是术语 $t$ 在聚类 $k$ 内的词频，$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 信号的轨迹上运行发现，因为本教程侧重于系统需要关注的区域。

主题表应作为分流板来阅读：

高影响力的行为模式并不自动意味着缺陷。它是审查者应首先查看的地方，因为该模式频繁、后果严重，或两者兼有。

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()

解读排行榜与追踪图谱

排行榜是组合视图：它按加权影响对行为模式进行排序。用它来决定哪种模式应首先获得人工关注。

追踪图谱是几何视图：每个点代表一个追踪文档，放置在文本相似的追踪点附近。邻近的点通常共享路由路径、发现结果或环境信号。颜色显示已发现的行为模式。将图谱视为诊断工具，而非精确的地理位置。它的作用是揭示在表格中可能难以看到的聚类和异常值。

在此数据集中，诸如履约改道、定价漂移、合规门禁以及车轮/装饰件不匹配等模式，对应着可识别的业务问题。这是低级评估首次成为宏观层面的故事：重复的代理行为在众多案例中可见。

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 论文中的核心公式。它是在主题分配之后应用的一个简单的队列分析层。其思路是比较两个占比：

然后计算：

$$

提升度 = \frac{切片模式占比}{整体模式占比}

$$

提升度为 1.0 表示该模式在该切片中出现的频率与整体出现的频率大致相同。提升度高于 1.0 表示该模式集中在该切片中。提升度低于 1.0 表示该模式在该切片中较少见。

在宏观评估中，这是从发现到行动的桥梁。当我们能说出行为模式出现在哪里时（生成的场景、代理版本、编排模式、市场状态或审查状态），就更容易对其进行调查。

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 =

  1. 4 \cdot proximity +
  2. 3 \cdot frequency +
  3. 2 \cdot bridge +
  4. 1 \cdot role

    5. $$

这不是因果关系的证明。它是一种将“这种模式很重要”转化为“首先检查这些代理、工具、交接或审查策略”的方法。

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 这样的行并不神秘。在模拟中，审查发现是一个结构化标记，当专家或审查表面观察到影响订单决策的问题时产生。这是我们向后追踪的终点：工作流积累了足够证据说“此订单需要关注”的时刻。

更具可操作性的行是围绕该标记的操作事件：涉及编排器的交接、监控或编排代理的工具/函数调用、采购计划交接以及相关的专家响应。在宏观评估指向此模式后，这些是人类应该检查的地方。

从技术角度来看，此输出告诉AI工程师在哪里检查：

从业务角度来看，相同的输出告诉运营或产品利益相关者哪个业务功能似乎拥有该模式。履约、定价、合规或澄清模式应让相应的业务负责人参与下一次审查，而不仅仅是提示工程师。

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."
    ))

解读故事条和泳道图

故事条是通往焦点事件的路径。在此次运行中，焦点事件是所选行为模式内的一个审查/评估检查点。它表示模拟业务流程认为该订单存在值得审查的问题。

泳道图保留了更多的时间结构。它按泳道或代理显示周围的事件窗口，并突出显示焦点事件。从左到右阅读，可以看到订单在群体中的流转过程：

对于业务读者来说，该图将抽象模式转化为可操作的故事：这批订单反复到达类似的审查点。对于AI工程师来说，它缩小了下一步调试的范围：检查审查标记周围的编排和交接路径，尤其是诊断摘要中标记的第一个非审查可疑点。

7. 我们学到了什么以及下一步该做什么

本手册贯穿了四个层次的证据：

  1. 模拟设置： 业务在供应、定价、容量、合规和市场条件变化下生成电动汽车订单案例。
  2. 低层级评估： Promptfoo 提供代理/工作流级别的评估信号：决策质量、策略正确性、路由、市场意识和审查适当性。
  3. 宏观发现： BERTopic 风格的聚类将低层级发现分组为重复出现的行为模式，并按影响排序。
  4. 追踪诊断： AgentTrace 风格的图分析检查一个高影响的模式，并识别出重复出现的上游嫌疑对象。

这种方法通过将人类注意力引导到既频繁又重要的模式上来扩展。审阅者无需从头到尾阅读数百条追踪记录，而是可以从一个行为模式开始，检查代表性示例，并决定哪个代理、工具、交接或业务规则需要跟进。

AI 工程团队的实用后续步骤：

业务利益相关者的实用后续步骤：

核心教训很简单：代理级评估告诉我们哪些局部行为看起来有风险，而宏观评估告诉我们这些风险在系统规模上会变成什么。

进一步阅读

贡献者

本手册是 OpenAI 和 Slalom 之间的联合合作成果。

常见问题

常见错误包括：数据路径配置错误、依赖版本不兼容、轨迹文档构建不当导致聚类效果差、聚类参数设置不合理、忽略离群点等。排查方法：检查数据文件是否存在、确保Python版本和依赖正确、调整文档构建策略、使用UMAP和HDBSCAN参数调优、结合可视化工具验证聚类结果。生产环境优化：使用SQLite存储轨迹、并行处理、增量评估、模型版本追踪、设置合理的评估阈值。

企业应用扩展

本教程可扩展至其他智能体系统，如客服、金融风控等。可进一步探索：在线评估与离线评估结合、多版本模型对比、行为模式随时间演变分析、自动生成回归测试用例、集成更多评估指标（如延迟、成本）。此外，可结合强化学习优化智能体策略，或使用大语言模型自动生成诊断报告。

总结

本教程面向有编程基础的开发者，深入讲解多智能体系统的宏观评估方法。通过一个合成EV订单工作流，展示了如何从海量智能体轨迹中提取可操作的行为模式。内容涵盖低级评估（Promptfoo）、轨迹文档构建、BERTopic风格聚类发现以及AgentTrace风格根因诊断。读者将学会如何将数千个智能体事件转化为少量高影响力的模式，并定位系统需要优先检查的环节。教程提供完整代码、数据准备步骤和优化建议，帮助团队系统化改进智能体系统。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部模型评估与微调教程