門脇
本記事は、2026年2月:PythonでLLMアプリ開発を型安全に扱う ― Pydantic AI入門の続編にあたります。2月の記事では、Pydantic AIを使ってLLMの出力を型安全に扱う方法を紹介しました。
Pydantic AIでは、output_にPydanticモデルを渡すと、LLMが返したJSONがそのモデルへパースされ、型と制約が満たされているか検証されます。検証に失敗した場合はLLMの出力
しかし、この
- 一度でうまく通ったのか、何度か作り直した末に通ったのか
- 再生成で作り直したのなら、どのフィールドが、どんな理由で弾かれたのか
いずれも戻り値には残っておらず、理由も不明なまま再生成のトークンと待ち時間コストが積み重なっていきます。
このようなことから、本記事では
なお、本記事は2月の記事内容を前提として進めます。Pydantic AIの導入や基本的な使い方は改めて解説しないため、未読の方は先に目を通していただくとスムーズです。
型安全に成功しても、内部処理は見えない
まず、何が見えていないのかを整理します。
Pydantic AIのAgent.が返すのは、検証を通ったオブジェクトです。これは便利ですが、検証の過程が結果に残らないため、次のようなことがわからなくなります。
- モデルを何回呼び出したのか
- どのフィールドが、どんな理由で検証に失敗したのか
- 失敗したとき、モデルには何が返されたのか
- 再生成の分だけ増えたトークンとレイテンシはどれくらいか
おそらく小さなスクリプトなら気にならないようなことかもしれません。しかし本番で動かすとなると話が変わります。レスポンスが遅い日があっても原因を特定できず、請求額が想定と合わない理由も説明できません。
ログを仕込めばよさそうにも思えますが、リトライはPydantic AIの内部で完結するため、アプリケーション側のログには現れません。必要なのは、1回の実行を記録する仕組み、つまり
これらを確認していくために、本記事では次の2つのツールを使います。題材は2月の記事と同じく
- Pydantic AI V2 — 2026年6月に正式リリースされたバージョンを使います
(2月の記事のコードはV1系です)。 - Logfire — Pydantic AIの計測結果を受け取り、ブラウザ上で確認できます。
準備
本記事のサンプルコードはGitHubリポジトリに置いてあります。以下のように、uvで環境を作ります。
$ git clone https://github.com/kadowaki/llm-observability.git $ cd llm-observability $ uv sync
LLMの呼び出しにはOpenAI APIを使うので、APIキーを環境変数に設定します [1]。
$ export OPENAI_API_KEY='sk-...'
動作を確認したバージョンは以下のとおりです。バージョンについては、uv.に固定しているため、uv syncすると同じ組み合わせが再現されます。
Python 3.14.6
pydantic 2.13.4
pydantic-ai 2.22.0
logfire 4.39.0
Pydantic AI V2でエージェントを更新する
2月の記事では、Pydantic AIはV1系でした。その後2026年6月にV2.
V1からV2での変更点
まず、リトライの指定方法が拡張されています。
INSTRUCTIONS = "与えられたHTMLから記事の一覧情報を抽出してください。"
# 2月の記事(V1系)
agent = Agent(
"openai:gpt-5.2",
output_type=ArticleList,
instructions=INSTRUCTIONS,
retries=3,
)
# 本記事(V2系)
agent = Agent(
"openai:gpt-5.6-luna",
output_type=ArticleList,
instructions=INSTRUCTIONS,
retries={"output": 3},
)
V1ではretries=3のように整数を渡していましたが、V2では辞書でリトライの種類ごとに指定できるようになりました。"output"は出力の検証に失敗したときのリトライ回数で、"tools"
本記事では、出力の検証に注目するため{"output": 3}を指定します。retriesを指定しない場合のデフォルトは1回です。
また、これはV2への変更とは関係ありませんが、モデルもopenai:gpt-5.からopenai:gpt-5.に変更しています。2月の記事でも触れていますが、Pydantic AIではモデル名の文字列を差し替えるだけで切り替えられます。
その他の変更点については、ドキュメントAgents - Pydantic AIに記載があります。また、V1 → V2 Migration Map にも変更点がまとまっていますので、目を通してみてください。
出力モデルに「要約」を追加する
2月の記事では、記事のタイトル・
本記事では、ここにLLMに生成させるsummaryINSTRUCTIONS)example_からexample_まで共通で使うため、src/
from datetime import date
from pydantic import BaseModel, Field, field_validator
class ArticleInfo(BaseModel):
"""記事のメタデータ"""
title: str = Field(min_length=1, description="記事のタイトル")
author: str = Field(min_length=1, description="著者名")
published_date: date = Field(description="公開日")
url: str = Field(description="記事のURL(相対パス)")
# 追加: 入力に存在しないため、LLMが生成する必要がある
summary: str = Field(
min_length=10, max_length=25, description="記事内容の要約"
)
@field_validator("url")
@classmethod
def validate_url_format(cls, v: str) -> str:
"""URLが`/article/`で始まる相対パスかを検証し、不正なら例外を送出する"""
if not v.startswith("/article/"):
raise ValueError(
f"URLは'/article/'で始まる相対パスである必要があります(実際の値: {v})"
)
return v
class ArticleList(BaseModel):
"""記事一覧"""
articles: list[ArticleInfo] = Field(min_length=1)
summaryだけは、他のフィールドと性質が異なり、LLMがタイトルから内容を推測して生成するフィールドになっています。書き写すだけのフィールドはほとんど失敗しませんが、生成が必要なフィールドは制約を外しやすくなります。
上限だけでなく下限も決めておく
summaryにはmin_とmax_の両方を指定しています。
実際に、次のような挙動が起きました。
| 試行 | LLMが返した要約 | 字数 | 検証結果 |
|---|---|---|---|
| 1回目 | InterpreterPoolExecutorで並列処理を体験 | 31字 | 上限オーバー |
| 2回目 | InterpreterPoolExecutorで並列処理 | 28字 | 上限オーバー |
| 3回目 | 並列処理を体験 | 7字 | 下限割れ |
LLMが返した結果は、最初の失敗から少しずつ削って詰めていき、それでも通らないと見るや、今度は一気に7字まで削り落としていました。制約を満たそうとして、反対側に大きく振り切れてしまったようです。
この部分は、min_を入れておいたおかげで、この行き過ぎも検証で捕らえられました。下限がなければ、
型で守るというのは
入力データの取得
2月の記事ではHTMLを6件分に抜粋していましたが、連載も記事数が増えてきたため、本記事では記事一覧ページから25件を取得したものを入力としてみます。取得はsrc/data/に保存します。
$ uv run python src/fetch_source.py
以降のサンプルは、このファイルを読み込んで実行します。記事一覧は今後も増えていくため、実行時期によって件数は変わります。
instructionsには、要約の指示を追加しておきます。ここまでのコードはsrc/
INSTRUCTIONS = """\
与えられたHTMLから記事の一覧情報を抽出してください。
- タイトル、著者名、公開日、URLは、入力の表記どおりに正確に写してください
- summaryには、タイトルから読み取れる内容を25字以内の日本語で要約してください
"""
res = agent.run_sync(SAMPLE_HTML)
for article in res.output.articles:
print(f"{article.published_date} {article.author} {article.summary}")
実行すると、25件の記事が構造化されて出力されます。検証を通ったデータだけが返るので、この後の処理は型が正しい状態で行えます。
ここまでが2月の記事からの更新版です。
ただし、結果として得られたresを見ているかぎり、モデルが何回呼ばれたのかはわかりません。次章では、リトライが起きている様子を観察できるようにして見ていきます。
リトライが起きている様子を観察する
前述のコードは、25件の記事を抽出できました。しかし前述のとおり、resにはそこに至るまでの経過が残っていません。
ここではsrc/を使用して見ていきます。
Pydantic AIには、実行の途中経過をたどるためのAgent.が用意されています。run_が最終結果だけを返すのに対し、iter()は実行をノード単位で進めながら、その過程にアクセスできます。
async with agent.iter(SAMPLE_HTML) as run:
async for _node in run:
pass
usage = run.usage
res = run.result
print(f"モデルリクエスト数: {usage.requests}")
print(f"入力トークン: {usage.input_tokens}")
print(f"出力トークン: {usage.output_tokens}")
実行すると、次のように表示されます。
モデルリクエスト数: 4
入力トークン: 19866
出力トークン: 7449
25件を1回抽出しただけのはずが、モデルは4回呼ばれていました。最初の出力が検証を通らず、3回作り直していたということです。
なぜrun_sync() ではなくiter()なのか
途中経過を見たいだけならrun_の戻り値からも一部は取れます。それでもiter()を勧めるのは、失敗したときの情報が欲しいためです。
リトライの上限を超えると、Pydantic AIはUnexpectedModelBehaviorを送出します。このときrun_は結果を返さないため、そこまでに消費したトークンもわからないまま終わります。しかし、iter()とtry/を組み合わせれば、例外が発生しても使用量を確認できます。
usage = None
try:
async with agent.iter(SAMPLE_HTML) as run:
try:
async for _node in run:
pass
finally:
# 例外が発生しても、ここまでの使用量は取得できる
usage = run.usage
res = run.result
except UnexpectedModelBehavior as e:
print(f"リトライ上限に到達: {e}")
失敗した実行こそ、原因とコストを把握しておくことは大事です。最初からこの形で書いておくと、後から計測を追加する場合においても変更が少なく済みます。
検証エラーの中身を見る
実行結果からリクエスト数がわかっても、なぜ作り直したのかはまだ見えていません。エラーでリトライした時にモデルへ返された内容は、実行の履歴に残っています。
for message in run.ctx.state.message_history:
for part in message.parts:
if not isinstance(part, RetryPromptPart):
continue
print(f"--- {len(part.content)}件の検証エラーを返した ---")
for err in part.content:
loc = ".".join(str(x) for x in err["loc"])
print(f" {loc}: {err['msg']}")
print(f" 実際の値: {err['input']}")
example_を実行すると、次のような出力が得られました。
--- 3件の検証エラーを返した ---
articles.4.summary: String should have at most 25 characters
実際の値: InterpreterPoolExecutorで並列処理を体験
articles.16.summary: String should have at most 25 characters
実際の値: PythonのGILとfree threadingを解説
articles.24.summary: String should have at most 25 characters
実際の値: Cloudflare WorkersでPythonアプリ構築
--- 1件の検証エラーを返した ---
articles.4.summary: String should have at most 25 characters
実際の値: InterpreterPoolExecutorで並列処理
--- 1件の検証エラーを返した ---
articles.4.summary: String should have at least 10 characters
実際の値: 並列処理を体験
モデルに返されているのは、PydanticのValidationErrorとほぼ同じ内容です。1件ごとに次の3つが含まれています。
loc:どこで失敗したか。('articles', 4, 'summary')のようなタプルで、「 articlesの5件目(インデックスは0始まり) の summary」を意味します (出力では、読みやすさのためドット区切りに変換しています)。 msg:なぜ失敗したか( String should have at most 25 characters)input:実際に返ってきた値
3件 → 1件 → 1件と減っていき、4回目でようやく全件が通りました。前章で触れたat least 10 charactersとして記録されています。
注目したいのはinputの値です。1回目に弾かれた
なお、サンプルで示したコードについては、Pydantic AIの内部構造に依存している書き方です。バージョンが変われば動かなくなる可能性があり、実行のたびにコード修正するのも現実的ではありません。そこで次章では、こうした情報をトレースとして自動的に記録する方法を試してみます。
Logfireでトレースを見る
繰り返しになりますが、前述までの方法には以下の2つの問題がありました。
- Pydantic AIの内部構造に依存していること
- 確認したい実行ごとにコードを書き足す必要があること
Logfireを使うと、同じ情報がトレースとして自動的に記録されます。
計測を有効にする
計測を始めるには、Logfireにサインアップしてプロジェクトを作り、書き込みトークンを取得、環境変数にセットする必要があります。手順は公式ドキュメントを参照してください。また、トークンの取得についても公式ドキュメント - Write Tokensを参照してください。
$ export LOGFIRE_TOKEN='pylf_v1_...'
コード側の変更は3行だけです
import logfire
# ...(Agentの定義などは example_1.py と同じ)
# この2行を追加するだけで、エージェント実行がトレースとして記録される
logfire.configure()
logfire.instrument_pydantic_ai()
# ...(以降の実行処理も同じ)
これだけで、Pydantic AIのエージェント実行がトレースとしてLogfireに送信されるようになります。
スパンツリーで全体を見る
実行してブラウザで開くと、次のような画面が表示されます。
agent runという親スパンの下に、chat gpt-5.という子スパンが3つ並んでいます。これが
なお、リトライの回数は実行のたびに変わります。前章では4回でしたが、この実行では3回でした。LLMの出力は確定的ではないため、同じ入力でも毎回同じ結果になるとは限りません。
右側の数値にも注目してください。各スパンの入力トークンが2.、4.、6.と増えています。リトライのたびに、それまでのやりとりが会話履歴として積み上がっていくためです。このあたりは次章で詳しく見ていきます。
検証エラーがどう記録されているか
2つ目のchatスパンをクリックすると、リクエストの中身が表示されます。
Inputの欄に、前章でコードを書いて取り出したものと同じ内容が入っています。
2 validation errors:
[
{
"type": "string_too_long",
"loc": ["articles", 4, "summary"],
"msg": "String should have at most 25 characters",
"input": "InterpreterPoolExecutorで並列処理"
},
...
]
Fix the errors and try again.
Fix the errors and try again.という一文は、Pydantic AIが検証エラーに添えてモデルへ渡している指示です。つまりリトライとは、
このスパンのOutputを開くとモデルが返した25件分のデータも確認でき、どの要約がどう修正されたかをリクエストごとに追うことができます。
コードを書かずに同じ情報が得られる
前章ではRetryPromptPartを自分で取り出していましたが、トレースには同じ情報が記録されています。しかも、内部構造に依存したコードを書く必要がありません。
計測を有効にしておけば、あとから
リトライに払う代償
ここまでで、リトライが起きていることと、その理由がわかるようになりました。最後に、リトライがどれくらいのコストだったのかを見ていきます。
2件の違反のために、25件すべてを作り直している
前述のトレースをもう一度見てみますsummaryだけでした。
しかしPydantic AIのリトライは、出力全体を作り直す動作です。Outputを開くと、2回目のリクエストでもarticlesが25件返っているのが確認できます。
1回目: 25件を生成 → 2件が25字オーバー
2回目: 25件を再生成 → まだ1件が超過
3回目: 25件を再生成 → 全件通過
3回で75件分の要約が生成されましたが、直したかったのは実質2件です。残りの73件は、すでに検証を通っていたにもかかわらず作り直されています。
「エラー内容を会話に追加して、もう一度同じ依頼をする」
入力トークンと呼び出し回数
リトライにおけるコストの内訳を見ると、さらに明確になることがあります。トレースの各スパンに表示されている入力トークンは、次のようになっていました。
| リクエスト | 入力トークン |
|---|---|
| 1回目 | 2. |
| 2回目 | 4. |
| 3回目 | 6. |
入力は毎回同じHTMLのはずなのに、回を追うごとに増えています。これは、リトライがそれまでのやりとりをすべて含めて送り直しているためです。
- 1回目 :HTMLだけ
(2. 42K) - 2回目 :HTML+1回目の出力25件+検証エラー
(4. 55K) - 3回目 :HTML+1回目の出力+2回目の出力+検証エラー2回分
(6. 5K)
会話履歴に積み上がっていくことで、リトライが1回増えるたびに、
| 完了した時点 | 入力トークン累計 | 経過時間 |
|---|---|---|
| 1回目で通っていれば | 2. |
10. |
| 2回目で通っていれば | 6. |
19. |
| 3回目で通った |
13. |
28. |
呼び出しが1回から3回、つまり3倍になったとき、入力トークンは約5.
なお、個々のスパンを開くと、トークン数だけでなく推定コストも表示されます。
Tokens & cost
Input 2417 $0.0005
Output 1978 $0.0024
Speed 185 tok/s
1回あたりでは1円にも満たない金額ですが、これが積み上がるとどうなるかは、実行回数を掛け合わせてみると想像がつきます。
“あと6字”のために払っていたもの
ここまでを整理します。
最初に弾かれた要約は
その6字のために起きたことは、次のとおりでした。
- モデルの呼び出しが1回から3回に増えた
- すでに正しかった73件分の要約が、無駄に再生成された
- 入力トークンが2.
42Kから合計13. 46Kに増えた - 所要時間が10.
7秒から28. 4秒になった
いずれも、res.だけを見ていてはわからなかったものです。トレースを入れて初めて、
なお、ここで挙げた数値は特定の1回の実行のものです。LLMの出力は確定的ではないため、実行のたびに変わります。同じ入力で繰り返し実行すると、リクエスト数も消費量も次のように振れます。
リトライが3回に達した実行
Logfireは固有の仕組みではない
ここまでLogfireの画面を見てきましたが、この計測はLogfire固有の仕組みではありません。Pydantic AIの計装はOpenTelemetry
OpenTelemetryは、トレースやメトリクスを収集するための標準仕様です。特定のベンダーに依存しない形式が定められており、多くの可観測性ツールが対応しています。
logfire.が行っているのは、OpenTelemetry形式のトレースを生成し、Logfireに送信する、という2つです。そしてトレースの送信先は、差し替えることができます。
Logfire以外のツールで同じトレースを見る
たとえば、ローカルで動くotel-tuiにトレースを送ってみます。Dockerで起動し、OTLPの受信ポートを開けておきます。
$ docker run --rm -it -p 4318:4318 ymtdzzz/otel-tui
送信先の指定は環境変数で行い、Logfireへの送信を無効にします
import logfire
logfire.configure(send_to_logfire=False)
logfire.instrument_pydantic_ai()
$ export OTEL_EXPORTER_OTLP_ENDPOINT='http://localhost:4318' $ uv run python src/example_4.py
アプリケーションのコードは変えていません。表示のされ方はツールによって異なりますが、記録されている内容は同じでターミナル上に同じスパンツリーが表示されます。
左側にinvoke_と3回のchat gpt-5.が並び、Logfireで見たものと同じ構造になっています。スパン名の表示が異なるのは、Logfireが独自の表示名を付けているためで、記録されている内容は同じです。
右側には、そのスパンが持つ属性が一覧表示されています。gen_で始まる属性名が並んでいることがわかります。これはOpenTelemetryで定められたLLM向けの命名規則に沿ったもので、Pydantic AI固有のものではありません。
このように、計測する部分と、それを保存・
計測形式を標準に寄せておく
LLMアプリケーションの可観測性ツールは、ここ最近で急速に増えました。どれを使うべきかは、規模や予算、既存の監視基盤との相性によって変わります。だからこそ計測の形式は標準に寄せておく価値があります。
OpenTelemetryで記録しておけば、後からツールを乗り換えても、計測コードを書き直さずに済みます。
なお、OpenTelemetryにおけるLLM関連の属性名は、GenAI Semantic Conventionsとして仕様策定が進んでいます。仕様自体がOpenTelemetry - Documentationサイトから専用リポジトリに移されるなど、まだ動きのある段階です。
まとめ
本記事では、Pydantic AIのリトライをトレースとして可視化する方法を紹介しました。
型による検証は、LLMの出力を安全に扱うための有効な手段です。output_にPydanticモデルを渡すだけで、想定した形のデータだけを受け取れます。
一方で、検証によって発生したリトライは、最終結果からは見えません。本記事で見てきたのは、次のようなことでした。
- 25件を抽出する1回の実行で、モデルは3回呼ばれていた
- 弾かれたのは25件中2件、しかも制約を6字超えただけだった
- その2件のために、すでに正しかった73件分も作り直されていた
- リトライのたびに会話履歴が積み上がり、入力トークンは回数に比例せず増えていた
LLMは、制約を満たすように実行されますが、保証はしません。Pydantic AIは検証してリトライを行いますが、そのリトライにはコストがかかります。違反をゼロに近づけようとするより、違反は起こるものとして、起きたときの影響を小さくすることが大切です。
今回の例で言えば、25件をまとめて1回で生成していたことが、コストを大きくしていました。記事ごとにエージェントを実行する形にすれば、1件の違反が他の結果を巻き込むことはなくなり、リトライで送り直すのも1件分だけで済みます。もちろん、実行回数が25倍になることのオーバーヘッドはあるので、どちらが有利かは入力の大きさや違反の起きやすさによります。
その判断材料になるのが、ここまで見てきたトレースです。何回呼ばれ、何が弾かれ、どれだけ積み上がったのかを把握して初めて、設計を変えるかどうかを決められます。Logfireによるトレースを入れることで、こうした
型で守り、トレースで見る。この2つを組み合わせることで、LLMアプリケーションの挙動を把握できるようになります。
みなさんもぜひ試してみてください。
