大規模なマルチエージェントシステムの評価（48 分読了）
TLDR AI は、マルチエージェントシステムの評価において、個々の応答ではなくトレーシングデータから再発するパターンを特定し、技術者とビジネス担当者の両方が理解できる診断へ繋げるマクロ評価ワークフローを提案している。
キーポイント
マクロ評価の必要性
単一の不良応答ではなく、ハンドオフの遅れや専門家の信号見落としなど、システム全体の再発する振る舞いをトレーシングデータから可視化するアプローチが重要である。
合成データによる評価ワークフロー
EV 注文処理を想定した合成シナリオと事前計算済みトレーシングデータを用い、API キーなしで低レベル評価からマクロパターン発見までの完全なワークフローを実行可能にする。
2 レベル評価アーキテクチャ
個々のエージェントやツールの使用を判定する「低レベル評価」と、それらを統合してビジネス文脈でのシステム挙動を診断する「マクロ評価」の二段構えで構成される。
パターン発見と人間による検査
数千件のイベントから理解可能な少数のパターンへ集約し、特にインパクトの高いパターンの特定箇所を人間が重点的に検査するべきポイントとして提示する。
評価の二段階アプローチ
エージェントシステムの評価を、個別の要素（ツール使用、ルーティング等）を検証する「Lower-level evals」と、反復パターンや集中箇所を分析する「Macro evals」の2段階に分割して行う。
4 つの評価ラベル
ケース設定（case_type）、実行結果（run_outcome）、局所的な問題信号（eval_finding）、および集団レベルでのパターン（behavior_pattern）という4つのラベルを用いて評価データを構造化する。
外観と内部プロセスの乖離
最終回答が正しく見えても、その背後にあるトレーシングデータには重要なステップの省略や誤ったツール使用が含まれている可能性があり、単なる出力評価だけでは不十分である。
影響分析・編集コメントを表示
影響分析
この記事は、単発の LLM 応答評価から、複雑なエージェントシステム全体の挙動を定量的・定性的に分析する新しいパラダイムを示唆しています。特に、技術チームとビジネス層の間で共通言語として機能する「パターンベース」の評価指標の重要性を強調しており、実社会での大規模マルチエージェント導入における品質保証（QA）の標準化に大きな影響を与える可能性があります。
編集コメント
エージェントシステムの信頼性を高めるためには、個々の応答の精度だけでなく、システム全体のフローとエラーパターンの可視化が不可欠です。このマクロ評価のアプローチは、実運用レベルでの AI エンジニアリング標準を確立する上で重要な指針となります。
エージェントシステムが失敗した場合、その問題は単一の不適切な応答よりもはるかに大きいことが多い。ハンドオフが遅すぎたり、専門家のエージェントが多数の実行で同じシグナルを見逃したり、レビュープロセスが誤ったクラスのカスタマーケースに対してトリガーされたりする可能性がある。システムを改善するためには、チームはすべてのトレースの集団全体にわたって反復して現れる行動を見る必要がある。
このクックブックでは、マルチエージェントシステムのマクロ評価ワークフローについて解説します。ここでは、専門家のエージェントが価格設定、コンプライアンス、サプライ、工場ルーティング、スケジューリング、リリース決定を処理し、市場および運用条件が変化する合成されたEV注文ワークフローを使用します。
このノートブックは、事前に計算された合成トレースと保存された低レベルの評価ラベルを使用するため、OpenAI APIキーなしで完全なワークフローを実行できます。
あなたは以下を学ぶことになります:
- 多数のトレース付きエージェント実行を生成または収集すること;
- 各完了した実行に対して低レベルの評価を実行すること;
- 各トレースをコンパクトなドキュメントに変換すること;
- 集団全体にわたって反復して現れる行動パターンを発見すること; そして
- 高いインパクトを持つ1つのパターンに深く入り込み、人間が次にシステムを検査すべき場所を見つけること。
目標は、すべてのトレースの完璧な分類体系を構築することではありません。目標は、AIエンジニアリングチームが数千件のエージェントイベントから、技術的およびビジネスの関係者の両方が理解できる少数のパターンへと移行する方法を示すことです。
核となる考え方は、このノートブックが汎用的なチャット記録ではなく、保存されたエージェント型システムを評価する点にあります。シナリオ入力が調整された専門家の群れを駆動し、ランタイムはトレースバンドルを出力し、保存された Promptfoo のラベルが正規化されたトレースに結合され、マクロ評価層がこの証拠をパターンと診断ビューに変換します。
評価（evals）とは、AI チームがシステムが機能しているかどうかを測定する方法です。単純なモデル呼び出しの場合、評価は 1 つの出力をルブリックや参照回答と比較するものになります。エージェント型システムにおいては、システムが適切なツールを使用したか、適切な専門家に委譲したか、リスクが高い際にレビューのために一時停止したか、ビジネス文脈に根ざしていたかも評価する必要があります。
マルチエージェントシステムはこの作業をより困難にします。なぜなら、最終的な回答は長いワークフローにおける最後のイベントに過ぎないからです。リリース推奨事項が妥当に見える一方で、トレースを見ると、価格設定エージェントがインセンティブを無視したか、供給エージェントが在庫切れを見逃したか、調整者が必要なレビューステップを迂回していたことが明らかになる場合があります。
このノートブックは問題を 2 つのレベルに分割します：
- 下位レベルの評価では、個々のエージェント、ハンドオフ、ツール、完了した実行を採点します。本例では、Promptfoo がそのエージェントレベル評価層の役割を果たし、実行が最終的な意思決定の質、ポリシーの正しさ、専門家のルーティング、市場のドリフト、レビューの適切性をどのように扱ったかを採点します。
- マクロ評価は、多数の低レベルな知見を横断的に眺めます。そこでは、「どのような種類の問題が繰り返されるのか」「それらがどこに集中しているのか」「エージェントワークフローのどの部分を最初に精査すべきか」という問いを立てます。
本クックブック全体を通じて、4 つの読者向けラベルを使用します:
- case_type: 生成されたビジネス状況（例：クリーンな注文、検証ブロック、サプライヤー置換、価格例外など）
- run_outcome: ランの終了状態（例：完了、レビュー待ち、ブロック済み、失敗）
- eval_finding: 何かが間違っているかリスクがあることを示す低レベルのシグナル
- behavior_pattern: 多数のトレースにわたって発見される反復するパターン
有用なメンタルモデルは以下の通りです：case_type は設定（setup）、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 は、ノートブックが初めて実行された際に自動的にローカルキャッシュに展開されます。オプションの拡張のために data/trace_snapshot.sqlite に完全な 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
必ず JSON 形式で返してください。translation フィールドのみ。他のフィールド (technical_terms 等) は一切追加しないこと — 余計なフィールドを書こうとして本文翻訳がトークン上限で打ち切られる事故を防ぐため:
{"translation": "翻訳全文"}
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)}"))
シミュレーションされるビジネスは、電気自動車（EV）の注文および設定完了後のワークフローです。顧客が車両の設定を選択し、企業はその注文を現状のまま進めるか、調整が必要か、再ルーティングすべきか、代替品が必要か、あるいはレビューのために一時停止すべきかを決定する必要があります。
このシミュレーションには、実際の自動車納品を困難にするような制約の種類が含まれています:
- 部品の可用性とサプライヤーの代替;
- 工場の生産能力と生産スケジュール;
- 価格例外、プロモーション、インセンティブ;
- 関税および時系列市場シグナル;
- 地域ごとのコンプライアンス制約;
- 顧客による確認要求およびエスカレーションパス;
- リスクのあるまたは曖昧なケースに対するリリースレビューの閾値。
エージェント群は、これらの業務責任を中心に構成されています。オーケストレーターが注文と現在の環境を受け取り、検証、供給リスク、調達計画、容量バランス調整、工場ルーティング、市場インテリジェンス、価格設定、コンプライアンス、顧客コミュニケーション、リリースレビューといった専門家に委任します。
これは OpenAI Agents SDK に自然にマッピングされます。SDK においてエージェントはワークフローの中核単位であり、モデル、指示、およびツール、ハンドオフ、ガードレール、構造化出力などのオプションのランタイム動作をパッケージ化します。シミュレーションはこのパターンに従います：
- 専門的なエージェントが意思決定の一部に対する指示とツールをパッケージ化し；
- ハンドオフにより、オーケストレーターはすべての責任を一つのプロンプトに詰め込むのではなく、別の専門エージェントへ委任できます；
- 関数型ツール（function tools）は、構造化された入力と出力を通じて注文データ、環境信号、承認マーカーを公開します；
- ガードレールとレビュー閾値は、リスクのあるまたは曖昧なケースに対する検証、ブロック、および人間によるレビューフローを表します；
- 構造化出力により、下流の採点と集計が可能になります；
- トレース（traces）は、デバッグやマクロレベル分析のためにモデル呼び出し、ツール呼び出し、ハンドオフ、ガードレール、カスタムスパンの構造化記録を保持します。
ノートブックの後半にある低レベルの評価は、このシミュレーションの物語に基づいています。ケースタイプに tariff pressure（関税圧力）下でのサプライヤー置換があると記載されている場合、トレースには供給、政策、市場、およびレビューリスクへの認識が示されるべきです。一方、ケースタイプがクリーンな場合、不要なエスカレーション自体が発見事項となります。
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."
))
What One Bundle Represents
In this notebook, a bundle is the evidence packet for one simulated customer-order interaction.
Imagine one customer has configured an EV and the business needs to decide what to do next. The swarm receives that order plus the current operating world: supply constraints, factory capacity, promotions, incentives, tariffs, competitor pressure, and review thresholds. The agents then route work through specialists and produce a final state. The bundle is everything we need to audit that interaction afterward.
バundler が重要なのは、マクロ評価において最終回答の背後にあるワークフローのエビデンスが必要だからです。どのエージェントが参照され、どのツールが呼び出され、どのような環境シグナルが有効だったか、レビューが必要であったかどうか、そしてワークフローがどのように方向転換したかを把握する必要があります。そのエビデンスがあれば、「この 1 回の実行で何が起こったのか」から「多くの実行にわたって繰り返されるワークフローパターンは何か」へと議論を進展させることができます。
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(
[
("分析可能な顧客注文インタラクション", len(bundle_profile_df), "シミュレーションされた各注文インタラクションごとに 1 つの完了したトレースバンドル。"),
("インタラクションあたりの中央値正規化イベント数", int(bundle_profile_df["events"].median()), "ステータス更新、ハンドオフ、関数/ツールイベント、レスポンス、および発見事項。"),
("インタラクションあたりの中央値 SDK スパン数", int(bundle_profile_df["spans"].median()), "イベントログの背後にある低レベルの SDK トレーススパン。"),
("インタラクションあたりの中央値ハンドオフ記録数", int(bundle_profile_df["handoffs"].median()), "オーケストレーターと専門エージェント間の委任。"),
("インタラクションあたりの中央値ツール/関数呼び出し数", int(bundle_profile_df["tool_or_function_calls"].median()), "実行内の構造化された読み取り、チェック、および評価呼び出し。"),
("インタラクションで観測された中央値エージェント数", int(bundle_profile_df["unique_agents_seen"].median()), "典型的なトレースに現れる専門役割の数。"),
("インタラクションあたりの中央値環境シグナル数", int(bundle_profile_df["environment_signals"].median()), "注文に対して有効な関税、インセンティブ、在庫切れ、プロモーション、競合他社、ローンチ、またはスケジュールのシグナル。"),
("レビューパケットを含むインタラクション数", int(bundle_profile_df["has_review_packet"].sum()), "シミュレーションされたビジネスプロセスがレビューアーティファクトを生成した実行。"),
],
columns=["reader_metric", "value", "plain_english_meaning"]
原文を表示
When an agentic system fails, the problem is often larger than a single bad response. A handoff may happen too late, a specialist agent may miss the same signal across many runs, or a review process may trigger for the wrong class of cases. To improve the system, teams need to see recurring behavior across the whole population of traces.
This cookbook walks through a macro-eval workflow for a multi-agent system. We use a synthetic EV order workflow where specialist agents handle pricing, compliance, supply, factory routing, scheduling, and release decisions while market and operational conditions change.
The notebook uses precomputed synthetic traces and saved lower-level eval labels, so you can run the full workflow without an OpenAI API key.
You will learn how to:
- Generate or collect many traced agent runs;
- Run lower-level evals on each completed run;
- Turn each trace into a compact document;
- Discover recurring behavior patterns across the population; and
- Drill into one high-impact pattern to find where a human should inspect the system next.
The goal is not to build a perfect taxonomy of every trace. The goal is to show how an AI engineering team can move from thousands of agent events to a small number of patterns that are understandable by both technical and business stakeholders.
The key idea is that the notebook evaluates a saved agentic system, not a generic chat transcript. Scenario inputs drive an orchestrated specialist swarm, the runtime emits trace bundles, saved Promptfoo labels are joined to normalized traces, and the macro-eval layer turns that evidence into pattern and diagnosis views.
Evals are how AI teams measure whether a system is working. For a simple model call, an eval might compare one output against a rubric or reference answer. For an agentic system, we also need to evaluate whether the system used the right tools, delegated to the right specialist, paused for review when risk was high, and stayed grounded in the business context.
Multi-agent systems make this harder because a final answer is only the last event in a longer workflow. A release recommendation can look plausible while the trace reveals that the pricing agent ignored an incentive, the supply agent missed a stockout, or the orchestrator routed around a required review step.
This notebook separates the problem into two levels:
- Lower-level evals grade individual agents, handoffs, tools, and completed runs. In this example, Promptfoo stands in for that agent-level eval layer by grading whether a run handled final decision quality, policy correctness, specialist routing, market drift, and review appropriateness.
- Macro evals look across many lower-level findings. They ask: which kinds of problems repeat, where do they concentrate, and which part of the agent workflow should we inspect first?
We will use four reader-facing labels throughout the cookbook:
- case_type: the generated business situation, such as a clean order, a validation block, a supplier substitution, or a pricing exception.
- run_outcome: how the run ended, such as completed, awaiting review, blocked, or failed.
- eval_finding: the lower-level signal that says what seemed wrong or risky.
- behavior_pattern: the recurring pattern discovered across many traces.
A useful mental model is: case_type is the setup, run_outcome is the ending, eval_finding is the local symptom, and behavior_pattern is the population-level 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.txtInstall the dependencies, then load the offline dataset bundled with this example. The saved Promptfoo labels are part of the local data folder, so this notebook does not require a separate Promptfoo config, Promptfoo run artifact, or OpenAI API key.
Expected files:
data/trace_results.jsonl
data/run_summary.json
data/trace_bundles.zip
data/eval_labels.jsonltrace_bundles.zip is expanded automatically into a local cache the first time the notebook runs. A full SQLite trace snapshot can be placed at data/trace_snapshot.sqlite for optional enrichment, but it is not required for the end-to-end workflow.
If your data lives outside the example folder, set MACRO_EVALS_DATA_ROOT to that directory. If labels live separately, set 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)}`"))The simulated business is an EV order and post-configuration workflow. A customer has chosen a vehicle configuration, and the company needs to decide whether the order can proceed as-is, needs adjustment, should be rerouted, requires substitution, or should pause for review.
The simulation includes the kinds of constraints that make real automotive fulfillment hard:
- component availability and supplier substitution;
- factory capacity and production scheduling;
- pricing exceptions, promotions, and incentives;
- tariffs and dated market signals;
- regional compliance constraints;
- customer clarification and escalation paths;
- release review thresholds for risky or ambiguous cases.
The agent swarm is organized around those business responsibilities. An orchestrator receives the order and current environment, then delegates to specialists such as validation, supply risk, procurement planning, capacity balancing, factory routing, market intelligence, pricing, compliance, customer communications, and release review.
This maps naturally to the OpenAI Agents SDK. In the SDK, an agent is the core unit of a workflow: it packages a model, instructions, and optional runtime behavior such as tools, handoffs, guardrails, and structured outputs. The simulation follows that pattern:
- specialized agents package the instructions and tools for one part of the decision;
- handoffs let the orchestrator delegate to another specialist agent instead of stuffing every responsibility into one prompt;
- function tools expose order data, environment signals, and approval markers through structured inputs and outputs;
- guardrails and review thresholds represent validation, blocking, and human-review flows for risky or ambiguous cases;
- structured outputs make downstream grading and aggregation possible;
- traces preserve structured records of model calls, tool calls, handoffs, guardrails, and custom spans for debugging and macro-level analysis.
The low-level evals later in the notebook are grounded in this simulation story. If the case type says there is a supplier substitution under tariff pressure, the trace should show awareness of supply, policy, market, and review risk. If the case type is clean, unnecessary escalation is itself a finding.
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."
))What One Bundle Represents
In this notebook, a bundle is the evidence packet for one simulated customer-order interaction.
Imagine one customer has configured an EV and the business needs to decide what to do next. The swarm receives that order plus the current operating world: supply constraints, factory capacity, promotions, incentives, tariffs, competitor pressure, and review thresholds. The agents then route work through specialists and produce a final state. The bundle is everything we need to audit that interaction afterward.
A bundle matters because macro evals need the workflow evidence behind the final answer. They need to know which agents were consulted, which tools were called, which environment signals were active, whether review was required, and where the workflow changed direction. With that evidence, we can move from “what happened in this one run?” to “which workflow patterns repeat across many runs?”
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"</sp
関連記事
今日のまとめ
AI日報で今日の重要ニュースをまとめ読み