Airbnb、大規模生成AI評価のフレームワークを公開
本文の状態
日本語全文を表示中
詳細モードで約20分の本文を読めます。
同じ出来事の情報源
この情報源を基点に整理
Pydantic Blog
Airbnb は生成 AI の大規模評価における独自のプレイブックを公開し、コードによる客観的チェック、LLM による解釈、人間の検証という 3 レイヤー構成と、失敗事例の分析に基づく評価器構築の具体的な手順を提示した。
Continue in AI NEW LAB
このニュースを、実務の判断につなげる
AI NEW LABで、試したことや先に確認したい条件を共有できます。まずはログインなしで読めます。
AI NEW LABで論点を見るAI深層分析を開く2026年8月7日 05:16
AI深層分析
キーポイント
評価フレームワークの 3 レイヤー構造
Airbnb は評価を「コードで特定可能な客観的チェック」「解釈が必要な LLM 判定」「真偽や不一致を解決する人間レビュー」の 3 つの役割分担レイヤーに分割している。
評価器構築前の事前分析プロセス
Airbnb は約 100 個の出力とトレースを読み込み、実際に発見された失敗モードを分類してから評価器を作成することを推奨しており、抽象的な「有用性」スコアは避けるべきとしている。
Pydantic AI と Logfire を活用した実装
記事では Pydantic AI と Logfire を用いて、サポートエージェントの評価から失敗の分析、人間による校正、プロンプト改善提案までのワークフローを構築する具体的なコード例を示している。
評価器数の最適化に関する指針
Airbnb は 20 個のノイズの多い評価器よりも、3〜5 個のよく調整された評価器を持つことを推奨しており、評価セットは小さく保つべきであるとしている。
コードによる評価の強制
Pydantic AI は出力形式を検証して失敗を防ぎ、Logfire はツールの呼び出しを追跡することで、モデルのプロセスを監視する。
重要な引用
Its framework has three layers: Programmatic checks for failures code can identify exactly. LLM judges for qualities that require interpretation. Human review for ground truth, disputed cases, and judge calibration.
A generic 'helpfulness' score written before anyone has seen the failures mostly measures the author's imagination.
Airbnb's rule of thumb is three to five well-calibrated judges rather than 20 noisy ones.
This code gives the evaluation two contracts to enforce before adding a judge.
編集コメントを表示
編集コメント
Airbnb が公開した評価プレイブックは、単なるツールの紹介ではなく、生成 AI の品質保証における「人間と機械の役割分担」という本質的な課題への回答を示している。Pydantic AI や Logfire を用いた具体的な実装例が示されているため、現場での導入検討において即座に参照可能な価値がある。
Source Article
元記事を日本語で読む
本文に関係しない購読案内、埋め込み通知、サイト内プロモーションは除いています。
Braintrust Week
これは 5 本立てシリーズの 4 日目です。まずは Score を自由に読み進めてください。
Airbnb は、大規模な生成 AI の評価におけるベストプラクティスを公開しました。そのアプローチはまず、約 100 件の出力やトレースを手動で確認し、実際に発見した失敗事例に基づいて評価器(evaluator)を構築することから始まります。このフレームワークには 3 つの層があります。
コードが明確に特定できる失敗に対するプログラムによるチェック。
解釈が必要な品質を判断するための LLM 判定(LLM judges)。
真偽の確認、論争のあるケース、および判定器の較正を行うための人間によるレビューです。
これらの層は役割分担を示しています。コードは客観的な質問に答え、判定器が解釈を担当し、人間が「良い状態」の定義と較正を行います。確認された本番環境での失敗事例は、新たなテストケースとして蓄積されます。
昨日は、既存の Braintrust 評価を再構築することなく Logfire に送信する方法を紹介しました。今日は Pydantic AI と Logfire を活用して、Airbnb のワークフローをゼロから構築します。具体的には、サポートエージェントの評価実行、Evals ワークスペースでの失敗事例の調査、人間レビューによる判定器の較正、そして Logfire のオプティマイザーを用いた次のプロンプト改善案の提案までを一連で行います。
各質問に最適な評価器を選ぶ
プログラムによるチェック:システムが客観的な契約(contract)に従ったかを確認します。オフラインの全ケースで Pydantic の出力検証、カスタム評価器、およびアジェンシーの軌跡チェックを実施し、本番環境でもコストが許容できる場合のみ適用してください。
LLM 判定:回答は証拠に忠実でしょうか?オフラインの全ケースでは、1 つの狭義なルブリック(評価基準)と記録された理由を用いた LLMJudge を使用し、その後本番トラフィックをサンプリングして検証します。
人間によるレビュー:評価基準は専門家の判断と一致しているか?
実行アノテーション、アノテーションキュー、ホストされたデータセットを活用して「ゴールドセット」を調整し、意見の相違を解決し、本番環境のサンプルを検証しましょう。
Airbnb では、評価器(evaluator)を作成する前に、約 100 の事例を実行し、出力とトレースを確認することを推奨しています。失敗例を誰も見ていない段階で「有用性」といった一般的なスコアを設定しても、それは作者の想像力を測るものに過ぎません。
まずは同じ手順から始めましょう:
- プロトタイプに計装(インストルメント)を行い、50〜100 の代表的な入力を実行する
- 検索やツール呼び出しを含む完全なトレースを読み込む
- 繰り返し発生する失敗モードを分類する
- 各防止すべき失敗に対して評価器を一つ作成する
- 評価セットは小さく保つ。Airbnb の経験則では、20 ものノイズの多い判定者よりも、3〜5 人のよく調整された判定者のほうが優れています。
テスト対象システムの構築
エージェント、評価、観測機能のパッケージをインストールします:
pip install logfire pydantic-evals "pydantic-ai-slim[openai]"
OPENAI_API_KEY を設定し、support_agent.py を作成します:
from dataclasses import dataclass, field
from typing import Literal
import logfire
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
logfire.configure()
logfire.instrument_pydantic_ai()
POLICIES = {
'returns': 'Unused items can be returned within 30 days. Refunds take 5 to 7 business days.',
'cancellations': 'An order can be cancelled before it ships. Shipped orders must use the return process.',
}
class SupportAnswer(BaseModel):
text: str
cited_policy_ids: list[str]
action: Literal['answer', 'escalate']
class SupportResult(BaseModel):
answer: SupportAnswer
evidence: dict[str, str]
@dataclass
class SupportDeps:
evidence: dict[str, str] = field(default_factory=dict)
support_agent = Agent(
'openai:gpt-5-mini',
deps_type=SupportDeps,
output_type=SupportAnswer,
system_prompt=(
'Before answering, call lookup_policy with the customer question. '
'Answer only from the policies it returns and cite every policy used by ID. '
'Escalate when the returned policies do not answer the question.'
),
)
@support_agent.tool
def lookup_policy(ctx: RunContext[SupportDeps], question: str) -> dict[str, str]:
"""Return support policies relevant to the customer's question."""
words = question.lower()
evidence: dict[str, str] = {}
if 'cancel' in words or 'ship' in words:
evidence['cancellations'] = POLICIES['cancellations']
if 'return' in words or 'refund' in words:
evidence['returns'] = POLICIES['returns']
ctx.deps.evidence.update(evidence)
return evidence
async def answer_support_question(question: str) -> SupportResult:
deps = SupportDeps()
result = await support_agent.run(question, deps=deps)
return SupportResult(answer=result.output, evidence=deps.evidence)
このコードは、評価プロセスに 2 つの契約(コントラクト)を適用し、その後に判定者(ジャッジ)を導入する仕組みを提供します。Pydantic AI は、SupportAnswer を返す前に検証を行い、モデルが有効な出力を生成できない場合は再試行するかエラーを発生させます。Logfire は、エージェントのポリシーツール呼び出しとモデル呼び出しを同じトレースとして記録します。これらはいずれも文章の忠実性を直接証明するものではありませんが、別のモデルに問いかけることなく、特定の種類の失敗全体を排除する役割を果たします。
レイヤー 1:コードから何が判明するかを確認する
quality.py を作成します。最初の評価器は、回答が取得した証拠を引用しているか、およびポリシーが見つからない場合にエージェントがエスカレーション(上級者への報告)を行っているかを検証します。2 つ目の評価器は最終的な文章ではなくトレースデータから、ポリシー検索が実際に実行されたことを確認します。
from dataclasses import dataclass
from pydantic_evals.evaluators import (
EvaluationReason,
Evaluator,
EvaluatorContext,
LLMJudge,
ToolCorrectness,
)
from support_agent import SupportResult
@dataclass
class EvidenceContract(Evaluator[object, SupportResult, object]):
def evaluate(
self, ctx: EvaluatorContext[object, SupportResult, object]
) -> EvaluationReason:
cited = set(ctx.output.answer.cited_policy_ids)
available = set(ctx.output.evidence)
missing = sorted(cited - available)指定されたポリシーIDが見つからない場合:
return EvaluationReason(
value=False,
reason=f'Unknown policy IDs: {", ".join(missing)}',
)
出力された回答が"escalate(エスカレーション)"アクションである場合:
if available:
return EvaluationReason(
value=False,
reason='The agent escalated when relevant policies were available.',
)
return EvaluationReason(value=True)
該当するポリシーが見つからなかったにもかかわらずエスカレーションしなかった場合:
return EvaluationReason(
value=False,
reason='The answer did not escalate when no policy was found.',
)
引用元が明記されていない証拠を使用した場合:
return EvaluationReason(
value=False,
reason='The answer used evidence without citing it.',
)
上記の条件に該当しない場合:
return EvaluationReason(value=True)
used_policy_lookup = ToolCorrectness(
expected_tools=['lookup_policy'],
evaluation_name='used_policy_lookup',
)
これらのチェックは高速で決定論的であり、モデル呼び出しを行いません。オフラインのケースでは常に実行してください。本番環境ではバックグラウンドで実行し、可能な限り対象となるすべてのトラフィックに適用します。
cited_policy_ids に未知の文字列が含まれているかどうかを LLM が判断してはなりません。
used_policy_lookup のチェックは、エージェントがポリシー検索ツールを正確に 1 回呼び出し、予期しないツールを呼び出していないという必須のステップを証明するものです。しかし、これだけで全体の軌跡が正しく効率的だったことを証明できるわけではありません。誤ったツール引数の使用、順序違いの呼び出し、不要な再試行といった他の失敗モードについては、それぞれ専用のアジェンティック評価器を追加する必要があります。
ToolCorrectness は、Logfire で確認する OpenTelemetry トレースと同じものを評価します。
レイヤー 2: 判定者に 1 つの仕事だけ任せる
決定論的なチェックで引用が存在することは検証できますが、回答がその引用されたポリシーから導き出されているかどうかまでは判断できません。そこで、この質問にのみ答えるための判定者を 1 人追加します。
FAITHFULNESS_RUBRIC = """
answer.text のすべての事実上の主張が、
answer.cited_policy_ids に記載された evidence エントリによって裏付けられている場合にのみ合格とする。
正確な言い換えは合格とする。
追加された期限、資格ルール、保証、または例外がある場合は不合格とする。
"""
faithfulness_judge = LLMJudge(
rubric=FAITHFULNESS_RUBRIC,
model='openai:gpt-5.2',
assertion={
'evaluation_name': 'faithful_to_policy',
'include_reason': True,
},
)
この判定者は、合格・不合格の判断と、その理由を返します。このルブリックでは、1 から 10 のスコアよりも、二値の結果の方が明確な判断基準となります。また、理由が記載されていることで、予期せぬ結果の原因究明が可能になります。
Airbnb では、各評価項目ごとに専用の評価器と判定プロンプトを用意することを推奨しています。忠実性と簡潔性にはそれぞれ独立した判定者を割り当て、各ルブリックを個別にデバッグ・調整できるようにしましょう。
3 つの自動チェックをすべて実行する
eval_support.py を作成します。
import asyncio
import logfire
from pydantic_evals import Case, Dataset
from quality import EvidenceContract, faithfulness_judge, used_policy_lookup
from support_agent import SupportResult, answer_support_question
dataset = Dataset[str, SupportResult, None](
name='support-policy-agent',
cases=[
Case(name='return_window', inputs='Can I return an unused item after 20 days?'),
Case(name='refund_timing', inputs='How quickly will my refund arrive?'),
Case(name='cancel_unshipped', inputs='Can I cancel an order that has not shipped?'),
Case(name='return_after_shipping', inputs='Can I return an unused item after it ships?'),
Case(name='unknown_policy', inputs='Do you offer price matching?'),
],
evaluators=[
EvidenceContract(),
used_policy_lookup,
faithfulness_judge,
],
)
def pass_rate(report, evaluation_name: str) -> float:
values = [
case.assertions[evaluation_name].value
for case in report.cases
if evaluation_name in case.assertions
]
if not values:
raise RuntimeError(f'No results found for {evaluation_name!r}.')
return sum(values) / len(values)
async def main() -> None:
report = await dataset.evaluate(
answer_support_question,
name='baseline',
)
report.print(include_reasons=True)assert pass_rate(report, 'EvidenceContract') == 1.0
assert pass_rate(report, 'used_policy_lookup') == 1.0
assert pass_rate(report, 'faithful_to_policy') == 1.0
print(logfire.url_from_eval(report))
if __name__ == '__main__':
asyncio.run(main())
これらの 5 つのケースは、評価パイプラインが正常に動作し、有用な結果を記録していることを確認するものです。ただし、これだけではエージェントの品質を判断するには不十分です。最初の 100 件のトレースを読み込んで見つけた失敗事例に置き換え、新たな回帰現象が発生したらそれを実際の回帰テストとして追加してください。
実行後は Evals ワークスペースで実験として表示されます。次のプロンプトやモデルバージョンとの比較が可能になり、スコアが変化した背後にあるケースを開いて詳細を確認できます。また、結果をクリックすれば完全なトレースを追跡することも可能です。各評価器に対して個別の閾値を設定し、無関係なチェックを平均化して一つの数値にするのではなく、それぞれの基準で判断しましょう。このように記述されたアサーションにより、スクリプトは継続的インテグレーション(CI)ゲートとして機能します。
「AI Evaluations > Experiments」を開き、ベースラインを見つけて「Review results」を選択してください。「Overview」から始め、すべてのケースが完了したことを確認し、アサーションとタスクエラーの合計をチェックしてから、各評価器をスキャンします。アサーションバーは、合格と失敗のバランスを示しています。数値スコアも記録している場合は、ヒストグラムが表示され、平均値が弱いテールや複数の明確なクラスタを隠していないかを確認できます。集計結果は「どこを見るべきか」を教えてくれるものであり、「何を修正すべきか」を直接示すものではありません。

faithful_to_policy 以外のレビューケースを選択し、"要レビュー" または "失敗" にフィルターしてください。証拠は以下の順序で確認します。
タスクが受け取った入力内容を確認する。
モデルが返した出力内容を精査する。
選択された評価者の結果と理由を読む。
出力だけでは結果の背景が説明できない場合は、ライブビューでトレースを開く。

悪い回答に対して正確な評価がついている場合、それはエージェント側の問題です。一方、妥当な回答なのに予想外のスコアが付いた場合は、評価基準(ルブリック)か評価者の側に問題があります。システムプロンプトを編集する前に、まずこの境界線を特定して診断してください。
レイヤー 3:人間を較正セットとして活用する
較正されていない判定者は、単なる別のモデルの出力に過ぎず、真の正解(グラウンドトゥルース)ではありません。Airbnb では、良質な回答と悪質な回答の両方を含む 50〜100 例からなるゴールドセットを準備し、評価者間で一致率が 80 台後半または 90% に達するまで合意形成を行った上で、大規模な判定に活用することを推奨しています。Google の評価ガイドラインも同様のループを提案しており、まず自身がサンプルを評価し、次に判定モデルで同じサンプルを評価して結果を比較し、評価基準を改善するという手順です。
Logfire では、完全なインタラクションを見ながらこれらのラベルを作成できます。まずは「顧客の質問に答えるが、ポリシーの詳細を捏造しない」など、一つの評価基準を統一します。評価者が同じルールを適用して初めて、その判定結果は再利用可能な較正データとなります。
個別の実行では以下を実行してください:
AI 評価 > アノテーションを開く。
サポートエージェントを選択し、「アノテーションへ進む」をクリックします。
実行中のランを開き、入力内容、最終出力、モデル呼び出し、ツール呼び出し、およびトレースを確認します。
「注釈」を選択し、「合格」「中立」「不合格」のいずれかを選定してください。カテゴリが適用される場合は失敗を分類し、エージェントが返すべき回答が明確な場合は「期待する出力」欄に修正後の回答を追加します。
判断理由をコメント欄に記載し、"unsupported-claim"や"missing-escalation"などのタグを付与した上で「保存」をクリックしてください。Logfire は自動的に次のキュー待ちランへ進みます。最後のランの場合はボタンが「保存して閉じる」と表示されます。

体系的なキャリブレーション用パスとして、50〜100 件のランを注釈キューに格納してください。ラン注釈機能はベータ版であり、注釈キューの利用には Logfire の Design Partner プランへの加入が必要です。
各注釈は、自動化された評価器の結果と併せて、そのインタラクションに対するスコアとして紐付けられます。レビュー担当者が新たな失敗を確認した場合は、そのトレースを「ライブビュー」からホストされたデータセットへ追加し、次のオフライン実験で再テストできるようにします。詳細な手順については、人間によるレビューガイドをご覧ください(ラン注釈、注釈キュー、エンドユーザーフィードバックについて解説されています)。
専門家によってラベル付けが完了したセットは、ローカルまたはホストされたデータセットに保存された出力とラベルとして保持してください。サポートエージェントを再実行することなく保存済みの結果を再生し、判定器(judge)を適用して、レビュー担当者との一致率を測定します。
from pydantic_evals import Case, Dataset
from quality import faithfulness_judge
from support_agent import SupportAnswer, SupportResultdef saved_result(answer: str, evidence: str) -> SupportResult:
return SupportResult(
answer=SupportAnswer(
text=answer,
cited_policy_ids=['returns'],
action='answer',
),
evidence={'returns': evidence},
)
gold_set = Dataset(
name='faithfulness-judge-gold-set',
cases=[
Case(
name='accurate_paraphrase',
inputs=saved_result(
'You can return an unused item within 30 days.',
'Unused items can be returned within 30 days.',
),
metadata={'human_faithful': True},
),
Case(
name='invented_window',
inputs=saved_result(
'You can return an unused item within 60 days.',
'Unused items can be returned within 30 days.',
),
metadata={'human_faithful': False},
),
# Add 48 to 98 reviewed examples, including difficult failures.
],
evaluators=[faithfulness_judge],
)
report = gold_set.evaluate_sync(lambda result: result, name='judge-calibration')
agreement = sum(
case.metadata['human_faithful'] == case.assertions['faithful_to_policy'].value
for case in report.cases
) / len(report.cases)
print(f'Agreement: {agreement:.1%}')
2 つのケースはスニペットを短く保つために活用します。本番用のゴールドセットは、50〜100 件のバランスの取れた例から構築しましょう。そこには正解となる回答、明確な失敗事例、そして境界線が困難なケースが含まれている必要があります。セットが増えるにつれて、混同行列(confusion matrix)や Cohen's kappa を追跡してください。専門家の間で意見が割れる場合は、自動化する前にその相違点を解消するか、評価基準をより絞り込みます。
判定者(judge)とレビュアーの判断が食い違った場合、その理由を読み込んで原因を分類します。
- 評価基準に曖昧さがある:ルールを客観的に観察可能な形にし、具体例を追加する。
- 判定者に文脈不足がある:必要なソース資料を含める。
- 人間のラベル付けが誤りだった:ゴールドセットを修正し、その理由も記録する。
- そのケースは本質的に主観的である:無理に自動化しようとせず、人間による判断に残す。
ドメインが変わったとき、新たな失敗モードが発生したとき、あるいは判定モデルを変更したときは、このキャリブレーションを再実行してください。
証拠を次のプロンプト変更へつなぐ
3 つのレイヤーによって何が失敗し、その診断が信頼できるかが明確になります。Logfire のオプティマイザーは、次に何を修正すべきかを決定するステップを短縮します。
サポートエージェントには観測機能(instrumentation)が実装されているため、オフラインでの評価実行と本番トラフィックの両方がエージェントの実行履歴として記録されます。Open Agents を開き、対象のエージェントを選択して「Optimize」を開きます。オプティマイザーは最近の実行履歴を確認し、例外を発生させた実行を優先的に取り上げます。そしてプロンプトの修正案を 1 つ提案し、その根拠となるトレース(traces)も併せて提示します。ただし、この修正は自動的に適用されるわけではありません。

これは同じレビューループの一部として活用してください:
評価結果の失敗事例と人間の注釈を活用し、改善が必要な行動を特定します。
エージェント向けの最適化提案を作成します。
プロンプトの変更点を確認し、引用された実行履歴を開きます。もし証拠がプロンプトの問題ではなく、不正確な評価器、文脈の欠如、あるいはインフラの不具合を示している場合は、その提案は却下してください。
承認された編集を候補バージョンに適用し、同じデータセットで再実行して、Evals ワークスペース上でベースラインと比較します。
最適化ツールは 3 つのレイヤーを基に変更提案を作成しますが、最終的な判断はレビュー担当者が行います。引用された証拠を読み込み、その編集が実際のエージェントの失敗に対処しているか、あるいは欠陥のある評価器を満たすようにプロンプトを学習させていないかを必ず確認してください。プロンプト最適化の手順書には、この提案とレビューの流れの詳細が記載されています。
本番環境でも同じレイヤーを活用する
オフライン実験では「この変更をリリースしても安全か?」と問いかけます。一方、オンライン評価は実際のトレースにスコアを付け、「実トラフィックで機能するか?」を検証します。
Pydantic Evals を使用すれば、ライブ関数に対して同じ評価器を紐付けることが可能です。本例では、Airbnb の 5% のプロダクションサンプルをモデル判定に適用しつつ、対象となるすべてのトラフィックには低コストのチェックを実行する構成を示しています。オンライン評価はバックグラウンドで実行され、並列処理の制限により処理がドロップされた際に、その情報をコールバックが記録します。
import logfire
from pydantic_evals.evaluators import EvaluatorContext
from pydantic_evals.online import OnlineEvaluator, evaluate
品質評価の仕組みを構築する
品質保証のための証拠契約(EvidenceContract)や忠実性判定(faithfulness_judge)、ポリシー参照(used_policy_lookup)といった関数をインポートし、サポート質問への回答を行う answer_support_question 関数を用意します。
評価コンテキストを受け取る record_evaluation_drop 関数は、並列処理の上限に達した際にオンライン評価がスキップされたことを警告ログとして記録します。
この関数を用いて、answer_support_question の実行結果を評価するプロセスを設定します。ここでは3つのオンライン評価器(OnlineEvaluator)を並列で適用します。
1 つ目は証拠契約(EvidenceContract)を使用し、サンプル率 1.0%、最大並列数 100 で動作します。
2 つ目はポリシー参照(used_policy_lookup)を使用し、同様にサンプル率 1.0%、最大並列数 100 です。
3 つ目は忠実性判定(faithfulness_judge)を使用しますが、ここではサンプリング率を 5% に抑え、最大並列数を 5 に設定しています。いずれも上限に達した場合は record_evaluation_drop が呼び出されます。
これらは「support-policy-agent」を対象として実行され、引数の抽出と結果の記録が有効化されています。
評価の結果は、OpenTelemetry の gen_ai.evaluation.result イベントとして出力されます。Logfire の「Open AI Evaluations > Live Monitoring」画面でリアルタイムに監視可能です。各評価結果は、それを生成した本番環境のトレースと常に紐付けられています。
これにより、以下の運用ループが形成されます。
- リアルタイムモニタリングで合格率和評価器のエラーを確認する。
- 失敗した結果を開き、関連するトレースを精査する。
- 失敗事例や評価者の不一致、ランダムなサンプルを人間によるレビューへ送る。
- 確認された新たな失敗パターンをオフラインデータセットに追加する。
- システムを更新し、実験を再実行してベースラインと比較する。
Logfire はスコアごとの課金ではないため、サンプリングは評価モデルの運用コストとレビューの処理能力が主な考慮事項であり、評価回数の制限とは関係ありません。ラッパーをデプロイする前に、実際のトラフィックに対して並行処理の上限値をロードテストしておく必要があります。
実践的な運用サイクル
このサポートエージェントの場合、合理的な開始サイクルは以下の通りです。
すべてのプルリクエストで:すべての専門モデルを実行
原文を表示
Braintrust Week
This is day four of a five-post series. Start with Score freely.
Airbnb recently published its playbook for evaluating generative AI at scale. Start by reading roughly 100 outputs and traces, then build evaluators for the failures you actually find. Its framework has three layers:
Programmatic checks for failures code can identify exactly.
LLM judges for qualities that require interpretation.
Human review for ground truth, disputed cases, and judge calibration.
The layers are a division of labor. Code answers objective questions, judges handle interpretation, and humans define and calibrate what good means. Confirmed production failures become new test cases.
Yesterday, we showed how to send existing Braintrust evals to Logfire without rewriting them. Today, we will build Airbnb's workflow end to end with Pydantic AI and Logfire: evaluate a support agent, inspect failures in the Evals workspace, calibrate the judge with human review, and use Logfire's optimizer to propose the next prompt improvement.
#Choose the right evaluator for each question
Programmatic checks: Did the system obey an objective contract? Use Pydantic output validation, custom evaluators, and agentic trajectory checks on every offline case, and in production when the check is cheap enough.
LLM judges: Is the answer faithful to the evidence? Use LLMJudge with one narrow rubric and a recorded reason on every offline case, then sample production traffic.
Human review: Does the rubric match expert judgment? Use run annotations, annotation queues, and hosted datasets to calibrate a gold set, resolve disagreements, and review a production sample.
Airbnb recommends running roughly 100 examples and reading the outputs and traces before writing evaluators. A generic "helpfulness" score written before anyone has seen the failures mostly measures the author's imagination.
Start the same way:
Instrument the prototype and run 50 to 100 representative inputs.
Read the complete traces, including retrieval and tool calls.
Classify the recurring failure modes.
Write one evaluator for each failure worth preventing.
Keep the set small. Airbnb's rule of thumb is three to five well-calibrated judges rather than 20 noisy ones.
#Build the system under test
Install the agent, evaluation, and observability packages:
pip install logfire pydantic-evals "pydantic-ai-slim[openai]"
Set OPENAI_API_KEY, then create support_agent.py:
from dataclasses import dataclass, field
from typing import Literal
import logfire
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
logfire.configure()
logfire.instrument_pydantic_ai()
POLICIES = {
'returns': 'Unused items can be returned within 30 days. Refunds take 5 to 7 business days.',
'cancellations': 'An order can be cancelled before it ships. Shipped orders must use the return process.',
}
class SupportAnswer(BaseModel):
text: str
cited_policy_ids: list[str]
action: Literal['answer', 'escalate']
class SupportResult(BaseModel):
answer: SupportAnswer
evidence: dict[str, str]
@dataclass
class SupportDeps:
evidence: dict[str, str] = field(default_factory=dict)
support_agent = Agent(
'openai:gpt-5-mini',
deps_type=SupportDeps,
output_type=SupportAnswer,
system_prompt=(
'Before answering, call lookup_policy with the customer question. '
'Answer only from the policies it returns and cite every policy used by ID. '
'Escalate when the returned policies do not answer the question.'
),
)
@support_agent.tool
def lookup_policy(ctx: RunContext[SupportDeps], question: str) -> dict[str, str]:
"""Return support policies relevant to the customer's question."""
words = question.lower()
evidence: dict[str, str] = {}
if 'cancel' in words or 'ship' in words:
evidence['cancellations'] = POLICIES['cancellations']
if 'return' in words or 'refund' in words:
evidence['returns'] = POLICIES['returns']
ctx.deps.evidence.update(evidence)
return evidence
async def answer_support_question(question: str) -> SupportResult:
deps = SupportDeps()
result = await support_agent.run(question, deps=deps)
return SupportResult(answer=result.output, evidence=deps.evidence)
This code gives the evaluation two contracts to enforce before adding a judge. Pydantic AI validates SupportAnswer before returning it and retries or raises an error if the model cannot produce valid output. Logfire records the agent's policy-tool call and model call in the same trace. Neither establishes that the prose is faithful, but both rule out entire classes of failure without asking another model.
#Layer 1: check what code can know
Create quality.py. The first evaluator verifies that answers cite retrieved evidence and that the agent escalates when no policy is found. The second verifies from the trace, rather than the final prose, that the policy lookup ran.
from dataclasses import dataclass
from pydantic_evals.evaluators import (
EvaluationReason,
Evaluator,
EvaluatorContext,
LLMJudge,
ToolCorrectness,
)
from support_agent import SupportResult
@dataclass
class EvidenceContract(Evaluator[object, SupportResult, object]):
def evaluate(
self, ctx: EvaluatorContext[object, SupportResult, object]
) -> EvaluationReason:
cited = set(ctx.output.answer.cited_policy_ids)
available = set(ctx.output.evidence)
missing = sorted(cited - available)
if missing:
return EvaluationReason(
value=False,
reason=f'Unknown policy IDs: {", ".join(missing)}',
)
if ctx.output.answer.action == 'escalate':
if available:
return EvaluationReason(
value=False,
reason='The agent escalated when relevant policies were available.',
)
return EvaluationReason(value=True)
if not available:
return EvaluationReason(
value=False,
reason='The answer did not escalate when no policy was found.',
)
if not cited:
return EvaluationReason(
value=False,
reason='The answer used evidence without citing it.',
)
return EvaluationReason(value=True)
used_policy_lookup = ToolCorrectness(
expected_tools=['lookup_policy'],
evaluation_name='used_policy_lookup',
)
These checks are fast, deterministic, and make no model calls. Run them on every offline case. In production, run them in the background and apply them to all eligible traffic when practical. An LLM should never adjudicate whether cited_policy_ids contains an unknown string.
The used_policy_lookup check proves one required step: the agent called the policy lookup tool exactly once and called no unexpected tools. It does not prove that the entire trajectory was correct or efficient. Add separate agentic evaluators for other observed failure modes, such as wrong tool arguments, calls in the wrong order, or unnecessary retries. ToolCorrectness evaluates the same OpenTelemetry trace you inspect in Logfire.
#Layer 2: give the judge one job
A deterministic check can verify that a citation exists, but it cannot establish whether the answer follows from the cited policy. Add one judge to answer only that question:
FAITHFULNESS_RUBRIC = """
Pass only when every factual claim in answer.text is supported by an
evidence entry named in answer.cited_policy_ids. Accurate paraphrases pass.
Added deadlines, eligibility rules, guarantees, or exceptions fail.
"""
faithfulness_judge = LLMJudge(
rubric=FAITHFULNESS_RUBRIC,
model='openai:gpt-5.2',
assertion={
'evaluation_name': 'faithful_to_policy',
'include_reason': True,
},
)
The judge returns a pass/fail assertion and a reason. For this rubric, a binary result has a clearer decision boundary than a 1-to-10 score, and the reason makes a surprising result debuggable.
Airbnb recommends a separate evaluator and judge prompt for each dimension. Give faithfulness and concision their own judges so you can debug and calibrate each rubric independently.
#Run all three automated checks
Create eval_support.py:
import asyncio
import logfire
from pydantic_evals import Case, Dataset
from quality import EvidenceContract, faithfulness_judge, used_policy_lookup
from support_agent import SupportResult, answer_support_question
dataset = Dataset[str, SupportResult, None](
name='support-policy-agent',
cases=[
Case(name='return_window', inputs='Can I return an unused item after 20 days?'),
Case(name='refund_timing', inputs='How quickly will my refund arrive?'),
Case(name='cancel_unshipped', inputs='Can I cancel an order that has not shipped?'),
Case(name='return_after_shipping', inputs='Can I return an unused item after it ships?'),
Case(name='unknown_policy', inputs='Do you offer price matching?'),
],
evaluators=[
EvidenceContract(),
used_policy_lookup,
faithfulness_judge,
],
)
def pass_rate(report, evaluation_name: str) -> float:
values = [
case.assertions[evaluation_name].value
for case in report.cases
if evaluation_name in case.assertions
]
if not values:
raise RuntimeError(f'No results found for {evaluation_name!r}.')
return sum(values) / len(values)
async def main() -> None:
report = await dataset.evaluate(
answer_support_question,
name='baseline',
)
report.print(include_reasons=True)
assert pass_rate(report, 'EvidenceContract') == 1.0
assert pass_rate(report, 'used_policy_lookup') == 1.0
assert pass_rate(report, 'faithful_to_policy') == 1.0
print(logfire.url_from_eval(report))
if __name__ == '__main__':
asyncio.run(main())
These five cases confirm that the evaluation pipeline runs and records useful results. They are too small to establish the agent's quality. Replace them with the failures you found while reading the first 100 traces, then add real regressions as they appear.
The run now appears as an experiment in the Evals workspace. You can compare it with the next prompt or model version, open the cases behind a changed score, and follow any result into its complete trace. Set a threshold for each evaluator rather than averaging unrelated checks into one number. The assertions make the script usable as a continuous integration (CI) gate.
Open AI Evaluations > Experiments, find baseline, and select Review results. Start on Overview: confirm that every case completed, check the assertion and task-error totals, then scan each evaluator. Assertion bars show the balance of passes and failures. If you also record numeric scores, histograms show whether an average hides a weak tail or several distinct clusters. The aggregate tells you where to look, not what to change.

Select Review cases beside faithful_to_policy, then filter to Needs review or Failed. Read the evidence in this order:
Confirm the input the task received.
Inspect the output it returned.
Read the selected evaluator's result and reason.
Open the trace in Live view when the output alone does not explain the result.

A bad answer paired with an accurate evaluation points to the agent. A reasonable answer paired with a surprising score points to the rubric or evaluator. Diagnose that boundary before editing the system prompt.
#Layer 3: make humans the calibration set
An uncalibrated judge is another model output, not ground truth. Airbnb recommends a gold set of 50 to 100 examples containing both good and bad outputs, and reaching agreement in the high 80s or 90s before using a judge at scale. Google's evaluator guidance recommends the same loop: rate a sample yourself, run the judge on the same sample, compare, and refine the rubric.
You can create those labels in Logfire while looking at the complete interaction. First, agree on one criterion, such as "The answer resolves the customer's question without inventing policy details." A verdict becomes reusable calibration data only when reviewers apply the same rule.
For individual runs:
Open AI Evaluations > Annotations.
Choose the support agent and select Proceed to annotate.
Open a run and inspect its input, final output, model calls, tool calls, and trace.
Select Annotate, choose Pass, Neutral, or Fail, and classify the failure when a category applies. Add the corrected response in Expected output when you know what the agent should have returned.
Explain the verdict in Comment, add tags such as unsupported-claim or missing-escalation, then select Save. Logfire advances to the next queued run; on the final run, the button reads Save and close.

For a systematic calibration pass, put 50 to 100 runs in an annotation queue. Run annotations are in beta, and annotation queues are available on the Logfire Design Partner plan.
Each annotation becomes a score attached to the interaction alongside the automated evaluator results. When a reviewer confirms a new failure, add its trace to a hosted dataset from Live view so the next offline experiment tests it again. The human review guide covers run annotations, annotation queues, and end-user feedback in more depth.
Once an expert has labeled a set, keep each reviewed output and label in a local or hosted dataset. Replay those saved results without rerunning the support agent, apply the judge, and measure how often it agrees with the reviewer:
from pydantic_evals import Case, Dataset
from quality import faithfulness_judge
from support_agent import SupportAnswer, SupportResult
def saved_result(answer: str, evidence: str) -> SupportResult:
return SupportResult(
answer=SupportAnswer(
text=answer,
cited_policy_ids=['returns'],
action='answer',
),
evidence={'returns': evidence},
)
gold_set = Dataset(
name='faithfulness-judge-gold-set',
cases=[
Case(
name='accurate_paraphrase',
inputs=saved_result(
'You can return an unused item within 30 days.',
'Unused items can be returned within 30 days.',
),
metadata={'human_faithful': True},
),
Case(
name='invented_window',
inputs=saved_result(
'You can return an unused item within 60 days.',
'Unused items can be returned within 30 days.',
),
metadata={'human_faithful': False},
),
# Add 48 to 98 reviewed examples, including difficult failures.
],
evaluators=[faithfulness_judge],
)
report = gold_set.evaluate_sync(lambda result: result, name='judge-calibration')
agreement = sum(
case.metadata['human_faithful'] == case.assertions['faithful_to_policy'].value
for case in report.cases
) / len(report.cases)
print(f'Agreement: {agreement:.1%}')
The two cases keep the snippet short. Build the real gold set from 50 to 100 balanced examples with good answers, clear failures, and difficult boundaries. Track a confusion matrix or Cohen's kappa as the set grows. If experts disagree, resolve the disagreement or narrow the rubric before automating it.
When the judge disagrees with a reviewer, read the reason and classify the cause:
The rubric is ambiguous: make the rule observable and add examples.
The judge lacks context: include the source material it needs.
The human label is wrong: correct the gold set and record why.
The case is genuinely subjective: keep it with humans instead of forcing automation.
Re-run this calibration when the domain changes, a new failure mode appears, or you switch judge models.
#Turn the evidence into the next prompt change
The three layers establish what failed and whether the diagnosis is trustworthy. Logfire's optimizer shortens the next step: deciding what to change.
Because the support agent is instrumented, both its offline eval executions and its production traffic appear as agent runs. Open Agents, select the support agent, and open Optimize. The optimizer reviews recent runs, prioritizes runs that raised exceptions, proposes one prompt edit, and cites the traces behind it. It does not apply the edit automatically.

Use it as part of the same review loop:
Use evaluator failures and human annotations to identify the behavior that needs work.
Generate an optimization proposal for the agent.
Read the prompt diff and open the cited runs. Reject the proposal if the evidence points to a bad evaluator, missing context, or an infrastructure failure instead of the prompt.
Apply the accepted edit to a candidate version, rerun the same dataset, and compare it with the baseline in the Evals workspace.
The optimizer turns the three layers into a change proposal, but the decision stays with the reviewer. Read the cited evidence and make sure the edit addresses a real agent failure rather than teaching the prompt to satisfy a flawed evaluator. The prompt optimization walkthrough shows the proposal and review flow in more detail.
#Use the same layers in production
Offline experiments ask, "Is this change safe to ship?" Online evaluations score live traces and ask, "Does it hold up on real traffic?"
Pydantic Evals can attach the same evaluators to a live function. This example adapts Airbnb's 5% production sample to the model judge while applying cheap checks to all eligible traffic. Online evaluation runs in the background; the callback records any work dropped when a concurrency limit is full.
import logfire
from pydantic_evals.evaluators import EvaluatorContext
from pydantic_evals.online import OnlineEvaluator, evaluate
from quality import EvidenceContract, faithfulness_judge, used_policy_lookup
from support_agent import answer_support_question
def record_evaluation_drop(_: EvaluatorContext) -> None:
logfire.warning('Online evaluation dropped because its concurrency limit was reached')
evaluated_answer_support_question = evaluate(
OnlineEvaluator(
evaluator=EvidenceContract(),
sample_rate=1.0,
max_concurrency=100,
on_max_concurrency=record_evaluation_drop,
),
OnlineEvaluator(
evaluator=used_policy_lookup,
sample_rate=1.0,
max_concurrency=100,
on_max_concurrency=record_evaluation_drop,
),
OnlineEvaluator(
evaluator=faithfulness_judge,
sample_rate=0.05,
max_concurrency=5,
on_max_concurrency=record_evaluation_drop,
),
target='support-policy-agent',
extract_args=True,
record_return=True,
)(answer_support_question)
The evaluator results are emitted as OpenTelemetry gen_ai.evaluation.result events. Open AI Evaluations > Live Monitoring in Logfire. Each result stays linked to the production trace that produced it. That creates an operating loop:
Watch pass rates and evaluator errors in Live Monitoring.
Open a failed result and inspect its trace.
Send failures, disagreements, and a random sample to human review.
Add confirmed new failure modes to the offline dataset.
Update the system, rerun the experiment, and compare it with the baseline.
Logfire does not charge per score, so sampling is about judge-model spend and review capacity rather than an evaluation meter. Load-test the concurrency limits against your traffic before deploying the wrapper.
#A practical cadence
For this support agent, a reasonable starting cadence is:
Every pull request: Run all pro
関連記事
今日のまとめ
AIデイリーブリーフで今日の重要ニュースをまとめ読み