Reflection

一行サマリ

@agent.output_validator で「出力をチェックする関数」を登録し、不合格なら raise ModelRetry("理由") で LLM に 自己修正 を促す — これが PydanticAI の Reflection 機構。

ヒーロー: 通常 run と Reflection 付き run の違い

通常の Agent は LLM 呼び出し 1 回 → そのまま返す だけです。Reflection を加えると、出力を validator が検査し、不合格なら ModelRetry を投げて LLM にやり直させる ループが回ります。

概念: なぜ "自分で点検 → 自己修正" が必要か

output_type (Ch4 で扱う) で Pydantic モデルを指定すれば、LLM 応答が JSON として型に合うか は PydanticAI が自動検証します。しかし、型では表せない要件は山ほどあります。

• コンテンツ規則: 「個人情報を含めない」「90 文字以内に収める」「英語で書く」
• ビジネスルール: 「日付は今日以降にする」「金額は予算の上限を超えない」
• 整合性: 「フィールド A と B が矛盾しない」

これらを「事前に instructions で禁止する」よりも、「出した後に検査して、ダメなら理由付きで突き返す」 ほうが強力です。LLM は検査の理由を読んで、その回の応答自体を修正してきます。

コード: 3 つの代表パターン

パターン 1: 単純な NG ワード検出

最小例。応答に NG ワードが含まれていたら ModelRetry を投げます。

from pydantic_ai import Agent, ModelRetry
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    instructions='短く日本語で答えてください。',
)
 
@agent.output_validator
def no_secrets(output: str) -> str:
    forbidden = ['秘密', 'パスワード', 'API キー']
    for word in forbidden:
        if word in output:
            raise ModelRetry(
                f'出力に「{word}」が含まれています。機密情報を含まずに言い換えてください。'
            )
    return output  # OK ならそのまま返す
 
print(agent.run_sync('社内のシステム概要を 1 行で説明して').output)

ポイント:

• デコレータ 1 行 で validator を登録。元の Agent コードは変更不要
• 関数は output: str -> str 形式。OK ならそのまま返す、NG なら ModelRetry を raise
• ModelRetry(msg) の msg がそのまま LLM への 修正指示 として渡される (instructions に "前回の応答が違反しました。理由: ..." の形で追加される)

パターン 2: 構造化出力 + ビジネスルール検証

output_type で型を縛りつつ、型では表現できないルールを validator で追加検査します。

from datetime import date, timedelta
from pydantic import BaseModel
from pydantic_ai import Agent, ModelRetry
 
class Booking(BaseModel):
    customer_name: str
    date: date
    party_size: int
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    output_type=Booking,
    instructions='ユーザーの日本語文から予約情報を構造化して返してください。',
)
 
@agent.output_validator
def must_be_future(output: Booking) -> Booking:
    if output.date < date.today():
        raise ModelRetry(
            f'予約日 {output.date} が過去です。今日 ({date.today()}) 以降の日付に修正してください。'
        )
    if output.party_size < 1 or output.party_size > 20:
        raise ModelRetry(
            f'人数 {output.party_size} が範囲外です。1〜20 名で指定してください。'
        )
    return output
 
result = agent.run_sync('明日の夜、田中で 4 名予約お願いします')
print(result.output)
# 例: Booking(customer_name='田中', date=2026-05-11, party_size=4)

ポイント:

• 型エラーは PydanticAI が自動でリトライ してくれる (output_type の効能)
• 型では表現できないルール (過去日NG / 人数範囲) を validator で追加
• ModelRetry の メッセージは具体的に。「日付 2026-04-01 が過去です。今日以降に修正」のように 何が・どうなれば良いか を書くと LLM が直しやすい

パターン 3: RunContext 経由で deps を見る

外部リソース (DB / 設定 / 既存予約一覧) と整合チェックしたい場合は RunContext を取ります。(ctx, output) の 順序 に注意。

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext, ModelRetry
 
@dataclass
class BookingDeps:
    blocked_dates: set[date]  # 営業休止日
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    deps_type=BookingDeps,
    output_type=Booking,
    instructions='予約情報を構造化して返してください。',
)
 
@agent.output_validator
def not_on_blocked_date(ctx: RunContext[BookingDeps], output: Booking) -> Booking:
    if output.date in ctx.deps.blocked_dates:
        raise ModelRetry(
            f'{output.date} は営業休止日です。別の日付を提案してください。'
        )
    return output
 
deps = BookingDeps(blocked_dates={date(2026, 5, 11)})
result = agent.run_sync('明日 4 名でお願いします', deps=deps)
print(result.output)

ポイント:

• 第 1 引数に RunContext[YourDepsType]、第 2 引数に出力型
• ctx.deps で agent.run(..., deps=...) で渡したオブジェクトを参照可能
• データの中身に依存した検査 (在庫・予約済リスト・ユーザー権限) はこのパターン

リトライ回数とコスト

Reflection は便利ですが 無限ループにはなりません。

• デフォルトの retries 上限は 1 (= 最初の 1 回 + リトライ 1 回 = 計 2 回)
• 上限を超えると UnexpectedModelBehavior 例外が呼び出し側に飛ぶ
• リトライを増やすには Agent(..., retries=3) のように指定する

ただし リトライ 1 回ごとに LLM 呼び出しが発生する ため、「3 回連続でリトライ」となるとトークンコスト・レイテンシが 3〜4 倍になります。validator のメッセージを 十分に具体的に 書いて、できれば 1 回で直してもらうのが理想です。

まとめ

• @agent.output_validator で出力を検査し、raise ModelRetry("理由") で LLM に自己修正を促せる
• 型では表現できない コンテンツ規則 / ビジネスルール / データ整合性 を後段で守る
• 関数は OK なら output を返す / NG なら ModelRetry。引数順は (ctx, output)
• リトライは コスト 2〜N 倍 に直結。validator メッセージは具体的に書く

次レッスンでは、Agent の入出力を Pyright で 静的型チェック する設計 — Agent[DepsT, OutputT] のジェネリック宣言、deps_type / output_type の型推論、IDE 補完 — を扱います。