Chuyển đến nội dung chính
友田 陽大
Pydantic & type-safe validation
Python
Pydantic
PydanticAI
AIエージェント
LLM
型安全
可観測性

PydanticAI 実践ガイド:型安全なAIエージェントを本番運用する(構造化出力・ツール・DI・可観測性)

PydanticAI公式ドキュメントに忠実に、Agentの作り方・output_typeによる型安全な構造化出力・@agent.toolとtool_plain・deps_typeの依存性注入・output_validatorとModelRetryの自己修復・run_streamのストリーミング・Logfire可観測性・Temporal/DBOSによる耐久実行まで、本番品質のAIエージェント設計を実コードで解説します。

Published
Reading time
15 min read
Author
友田 陽大
Chia sẻ

導入:LLM アプリの「文字列地獄」から抜け出す

LLM を組み込んだアプリケーションは、放っておくと文字列地獄になります。プロンプトという文字列を渡し、自由文という文字列が返り、それを正規表現や json.loads で恐る恐るパースし、形が崩れていれば実行時に落ちる。テストは「だいたい合っていそう」で済ませ、本番で何が起きたかはログを grep して推測する——これは、私たちがバックエンド設計で 20 年かけて捨ててきた「型のない世界」への逆戻りです。

PydanticAI は、この問題に Pydantic の思想で答えるエージェントフレームワークです。開発元は Pydantic そのものを作っているチーム。だから設計の根幹は一貫しています——「システム境界の外から来るデータを、決して信頼しない」。LLM の出力もまた、検証されていない外部入力にすぎない。ならば BaseModel で形を宣言し、境界で検証し、型付きオブジェクトとして内側へ通す。FastAPI が HTTP リクエストに対してやっていることを、LLM の応答に対してやる——それが PydanticAI です。

本記事執筆時点の最新は PydanticAI 2.0(2026 年 6 月 23 日リリース、Python 3.10 以上)。メジャーバージョンで API が刷新されており、ネット上の古いチュートリアルとは重要な点が食い違いますresult_typeoutput_typesystem_promptinstructions 推奨など。後述します)。本記事は 公式ドキュメント に忠実に、2.0 の正しい API でまとめます。

💡 このブログの一貫したテーマ:私はこのポートフォリオで「一人 × 生成 AI で、速く・安く・安全に作る」ことを掲げています。PydanticAI は、まさに**「AI を使いつつ、人間が検証ゲートを握る」**ための道具です。LLM 出力を型で検証し、ツールを決定的なコードに分離し、可観測性で挙動を追う——AI を「賢い当てずっぽう」から「本番に載る部品」へ変える設計を扱います。TypeScript 側の対応物は Vercel AI SDK 本番ガイド、PydanticAI を使わず生 API で構造化出力を作る方法は Pydantic で作る LLM 構造化出力ガイド を参照してください。


1. 最小のエージェント:5 行で動かす

まずインストール。フル版か、プロバイダ別の軽量版(pydantic-ai-slim)を選びます。

# フル版(全プロバイダ同梱)
pip install pydantic-ai

# 軽量版+必要なプロバイダだけ(推奨)
pip install "pydantic-ai-slim[anthropic]"

最小のエージェントはこれだけです。

from pydantic_ai import Agent

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    instructions="簡潔に、1文で答えてください。",
)

result = agent.run_sync('"hello world" の語源は?')
print(result.output)

モデルは "プロバイダ:モデル名" 形式で指定します("anthropic:claude-sonnet-4-6""openai:gpt-5.2""google:gemini-3-flash-preview" など)。実行方法は 3 つ。

メソッド用途
agent.run_sync(...)同期実行(スクリプト・バッチ)
await agent.run(...)非同期実行(FastAPI 等のサーバー)
async with agent.run_stream(...) as result:ストリーミング(第6章)

💡 instructions を使う(system_prompt ではなく):v2 では instructions が推奨です。両者の違いは会話履歴の扱いにあります——message_history を渡したとき、system_prompt は履歴に含まれる過去のプロンプトも一緒に送られますが、instructions現在のエージェントの指示だけが送られます。マルチターンでプロンプトが二重化する事故を避けられるため、特別な理由がなければ instructions を選んでください。


2. 構造化出力:output_type で LLM を「型付き関数」にする

ここが PydanticAI の核心です。output_typeBaseModel を渡すと、LLM の応答が自動でそのモデルとして検証され、result.output型付きオブジェクトになります。

from pydantic import BaseModel
from pydantic_ai import Agent


class CityLocation(BaseModel):
    city: str
    country: str


agent = Agent("anthropic:claude-sonnet-4-6", output_type=CityLocation)

result = agent.run_sync("2012年のオリンピックはどこで開催された?")
print(result.output)        # city='London' country='United Kingdom'
print(result.output.city)   # 'London' ← str として型補完が効く

result.outputCityLocation 型です。エディタの補完が効き、mypy / Pyright が型を検査し、ダウンストリームのコードは「city は必ず str」という保証の上で書ける。LLM が、まるで型付きの純粋関数のように振る舞います

H3: 出力モード——ToolOutput / NativeOutput / PromptedOutput

PydanticAI は既定で、モデルの「ツール呼び出し(function calling)」機能を使って構造化出力を取り出します。これがもっとも移植性が高い方式です。複数の出力型を許したいときはリストで渡します。

from pydantic import BaseModel
from pydantic_ai import Agent, ToolOutput, NativeOutput


class Fruit(BaseModel):
    name: str
    color: str


class Vehicle(BaseModel):
    name: str
    wheels: int


# 既定(ツール経由)。型名でツール名を明示することもできる
agent = Agent(
    "anthropic:claude-sonnet-4-6",
    output_type=[
        ToolOutput(Fruit, name="return_fruit"),
        ToolOutput(Vehicle, name="return_vehicle"),
    ],
)

# モデルがネイティブの構造化出力に対応していれば NativeOutput も使える
native_agent = Agent(
    "anthropic:claude-sonnet-4-6",
    output_type=NativeOutput([Fruit, Vehicle], name="fruit_or_vehicle"),
)
マーカー仕組み使いどころ
ToolOutput(既定)ツール呼び出しで構造化移植性重視。ほぼ全モデルで動く第一選択
NativeOutputモデルのネイティブ構造化出力機能対応モデルで最も確実なスキーマ準拠
PromptedOutputプロンプトで JSON を指示ツールもネイティブも無いモデルの保険

なぜこれが優れているのか? 「自由文を返させて、後で頑張ってパースする」設計は、LLM の気まぐれ(余計な前置き、コードブロックの囲い、微妙に違うキー名)に毎回振り回されます。output_typeスキーマをモデル側に強制し、応答を Pydantic で検証するため、形の崩れは「検証エラー → リトライ」という制御されたフローに収束します。出力の形が契約になり、その契約はそのまま BaseModel のソースコードとして残る——これが文字列地獄からの脱出口です。


3. ツール:LLM に「できること」を型安全に渡す

エージェントが外の世界に作用する手段がツールです。PydanticAI では、普通の Python 関数にデコレータを付けるだけ。関数の型注釈と docstring から、LLM に渡す JSON スキーマが自動生成されます。

import random
from pydantic_ai import Agent, RunContext

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    deps_type=str,  # 第4章で解説。ここではプレイヤー名を注入する
    instructions="サイコロゲームの進行役。出目が予想と一致したら勝ち。",
)


@agent.tool_plain
def roll_dice() -> str:
    """6面ダイスを振り、出目を返す。"""
    return str(random.randint(1, 6))


@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """プレイヤーの名前を取得する。"""
    return ctx.deps

2 種類のデコレータの違いはコンテキストを受け取るかです。

  • @agent.tool:第1引数に ctx: RunContext[...] を取る。依存性(DB 接続、API キー、ユーザー情報など)にアクセスできる。
  • @agent.tool_plain:コンテキスト不要の純粋なツール。

ctx 以外の引数は、そのままツールの入力スキーマになります。docstring の書式(Google / NumPy / Sphinx)を解釈して、各引数の説明まで自動でスキーマに反映されます。

@agent.tool_plain(docstring_format="google", require_parameter_descriptions=True)
def search_products(keyword: str, max_results: int = 10) -> list[str]:
    """商品を検索する。

    Args:
        keyword: 検索キーワード。
        max_results: 返す件数の上限。
    """
    ...

H3: ModelRetry——ツールから LLM に「やり直し」を要求する

ツールが「入力が不正だ」と判断したとき、例外を投げて落とすのではなく、ModelRetry を送って LLM に修正を促せます

from pydantic_ai import Agent, ModelRetry

agent = Agent("anthropic:claude-sonnet-4-6")


@agent.tool_plain
def lookup_user(user_id: str) -> str:
    if not user_id.startswith("usr_"):
        # 例外で落とさず、LLM に「正しい形式で渡し直して」と伝える
        raise ModelRetry("user_id は 'usr_' で始まる必要があります。")
    return f"ユーザー {user_id} の情報"

なぜこれが優れているのか? ツールは「決定的なコード」、LLM は「曖昧な判断」。両者を混ぜないのが堅牢な設計です。在庫照会・決済・DB 書き込みといった確実性が必要な処理はツール(=普通の Python)に閉じ込め、LLM には「いつ・どの引数でそれを呼ぶか」だけを任せる。ModelRetry は、その境界で起きた齟齬を会話の中で自己修復させる仕組みです。これは AI エージェントの tool use 設計 で論じた「決定的コードと確率的判断の分離」を、PydanticAI が言語機能として提供しているということです。


4. 依存性注入:副作用を注入し、テスト可能にする

ツールが DB や外部 API に触れるなら、その依存をハードコードしてはいけません。PydanticAI は **deps_type による依存性注入(DI)**を備えています。FastAPI の Depends を知っているなら、思想は同じです。

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext


@dataclass
class Deps:
    """エージェントが必要とする外部依存。dataclass で束ねるのが定石。"""
    api_key: str
    http_client: httpx.AsyncClient


agent = Agent("anthropic:claude-sonnet-4-6", deps_type=Deps)


@agent.tool
async def fetch_weather(ctx: RunContext[Deps], city: str) -> str:
    """指定都市の天気を取得する。"""
    resp = await ctx.deps.http_client.get(
        "https://api.example.com/weather",
        params={"city": city, "key": ctx.deps.api_key},
    )
    return resp.text


async def main(client: httpx.AsyncClient) -> None:
    deps = Deps(api_key="...", http_client=client)
    result = await agent.run("東京の天気は?", deps=deps)
    print(result.output)

deps_type=Depsを宣言し、実行時に agent.run(..., deps=deps)インスタンスを渡す。ツール内では ctx.deps で型付きアクセスできます。

なぜこれが優れているのか? DI の真価はテスト容易性です。本番では実際の httpx.AsyncClient を、テストではモックを注入できる。さらに agent.override(deps=...) を使えば、テスト中だけ依存を差し替えられます。LLM を呼ばずにツールのロジックだけを検証する、あるいは決定的なテストモデル("test")でフロー全体を回す——AI を含むコードを、AI なしでテストする道が開けます。これは CLAUDE.md の「検証パスを先に作る」原則そのものです。


5. 出力バリデータと自己修復:検証を会話のループに組み込む

output_type の Pydantic 検証だけでは足りない、意味的な検証があります。「LLM が生成した SQL が、実際に実行可能か」「提案された日付が営業日か」——こうした検証は @agent.output_validator で行い、失敗時は ModelRetryLLM に作り直させます

from pydantic import BaseModel
from pydantic_ai import Agent, ModelRetry, RunContext


class SqlQuery(BaseModel):
    sql_query: str


agent = Agent(
    "anthropic:claude-sonnet-4-6",
    output_type=SqlQuery,
    deps_type=DatabaseConn,  # 例:DB 接続
)


@agent.output_validator
async def validate_sql(ctx: RunContext[DatabaseConn], output: SqlQuery) -> SqlQuery:
    try:
        # EXPLAIN で「実行可能か」だけを安全に検証する(実行はしない)
        await ctx.deps.execute(f"EXPLAIN {output.sql_query}")
    except QueryError as e:
        # 失敗をエラーメッセージごと LLM に返し、修正版を生成させる
        raise ModelRetry(f"無効なクエリです: {e}") from e
    return output

このループは強力です。Pydantic の構文的検証(型・必須・制約)と、output_validator意味的検証(業務上正しいか)が二段構えで効き、どちらで失敗しても ModelRetry で会話に差し戻される。**「正しい出力が得られるまで、自動でやり直す」**自己修復エージェントが、宣言的に組めます。

⚠️ リトライには上限をAgent(..., retries=2) でリトライ回数を制限してください。無制限のリトライはコストと遅延を青天井にします。リトライが頻発するなら、それはプロンプトかスキーマの設計不良のサイン——Field(description=...) で各フィールドの意図を LLM に伝えるなど、根本の改善を優先します(Pydantic で作る LLM 構造化出力ガイド で詳述)。


6. ストリーミング:検証しながら部分結果を流す

チャット UI では「全部できてから返す」では遅すぎます。run_stream は、構造化出力を検証しながら部分結果をストリームできます。

async def stream_profile(agent: Agent, user_input: str) -> None:
    async with agent.run_stream(user_input) as result:
        # 検証済みの「途中までのオブジェクト」が順次流れてくる
        async for profile in result.stream_output():
            print(profile)
            # {'name': 'Ben'} → {'name': 'Ben', 'dob': date(1990, 1, 28)} → ...

テキストだけ流すなら result.stream_text()、生の ModelResponse を間引いて受けるなら result.stream_response(debounce_by=0.01) を使います。

⚠️ 部分結果での副作用に注意:ストリーミング中、output_validator には「まだ未完成のオブジェクト」も渡ってきます。DB 書き込みのような副作用は完成版にだけ行いたいはず。ctx.partial_output フラグを見て、部分結果のときは検証・副作用をスキップしてください。これを怠ると、未確定データで外部システムを汚す事故になります。


7. 可観測性:Logfire で「AI の挙動」を一望する

LLM アプリの最大の難所はデバッグです。「なぜこのツールが呼ばれた?」「どのリトライで直った?」「トークンをどこで浪費している?」——これらは print デバッグでは追えません。PydanticAI は、同じ Pydantic チームの Logfire と統合し、OpenTelemetry ベースで全挙動を計装します。

import logfire
from pydantic_ai import Agent

logfire.configure()
logfire.instrument_pydantic_ai()  # これだけで全エージェントが計装される

agent = Agent("anthropic:claude-sonnet-4-6")
result = agent.run_sync("...")
print(result.usage)  # RunUsage(input_tokens=62, output_tokens=1, requests=1)

たった 2 行(configureinstrument_pydantic_ai)で、各エージェント実行・ツール呼び出し・リトライ・トークン使用量がトレースとして可視化されます。result.usage 属性からはトークン数とリクエスト数を直接取得でき、コスト監視の基盤になります。HTTP の生リクエストまで見たいなら logfire.instrument_httpx(capture_all=True) を足します。

なぜこれが優れているのか? PydanticAI の計装は OpenTelemetry の GenAI セマンティック規約に従うため、Logfire 以外の OTel バックエンド(Grafana、Datadog など)にも流せます。これは私が OpenTelemetry 可観測性ガイド で論じた「三本柱を相関させる」設計に、AI の実行が自然に乗るということ。「止まった処理を一目で追える」——AI エージェントを本番運用するうえで、この観測性は機能ではなく前提条件です。どのエージェント実行・ツール呼び出しで時間とトークンが消費されているかをトレースで追えることが、デバッグの短縮とコスト異常の早期検知の土台になります。


8. 本番の回復性:durable execution で「落ちても続きから」

エージェントが外部 API を何度も叩き、ツールを連鎖させ、ときに人間の承認(human-in-the-loop)を待つ——こうした長時間ワークフローは、途中で必ず落ちます。API のレート制限、ネットワーク断、プロセスの再起動。そのたびに最初からやり直すのは、コスト的にも UX 的にも許容できません

PydanticAI は **durable execution(耐久実行)**を、Temporal / DBOS / Prefect / Restate との統合で提供します。既存の Agent を専用クラスで包むだけで、進捗が永続化され、障害をまたいで続きから再開できます。

from pydantic_ai import Agent
from pydantic_ai.durable_exec.temporal import TemporalAgent

# name は必須(ワークフロー/アクティビティの識別に使われる)
agent = Agent(
    "anthropic:claude-sonnet-4-6",
    instructions="...",
    name="geography",
)

temporal_agent = TemporalAgent(agent)  # Temporal ワークフロー内で実行する

DBOS なら状態を DB にチェックポイントします。

from pydantic_ai.durable_exec.dbos import DBOSAgent

dbos_agent = DBOSAgent(agent)
result = await dbos_agent.run("メキシコの首都は?")
バックエンド性質向いている場面
Temporalワークフローエンジン。強力な再試行・タイマー複雑な長時間オーケストレーション
DBOSDB チェックポイント方式。軽量既存 DB に状態を寄せたい
Prefect / Restateデータパイプライン/耐久 RPC 指向それぞれの基盤に合わせて

⚠️ name= を必ず付ける:耐久実行で包むエージェントには name= が必須です(ワークフロー識別子になる)。また TemporalAgent はモジュールのトップレベルで定義するなど、各バックエンド固有の制約があります。導入時は対象バックエンドのドキュメントを必ず参照してください。

なぜこれが効くのか? 私が放送事業者向けに構築した社内 AI プラットフォーム(番組制作支援)では、長時間の AI ジョブを Cloud Workflows / Cloud Run Jobs に切り離して回復性を担保しました。PydanticAI の durable execution は、その「長時間ジョブを落ちない形で回す」という要求を、エージェントのレイヤーで解決します。FastAPI の BackgroundTasks で抱え込むには重すぎる処理を、ジョブ基盤へ逃がす設計(FastAPI 本番運用ガイド 参照)の、AI エージェント版だと考えてください。


結論:AI を「本番に載る部品」に変える

PydanticAI は、LLM アプリを「賢い当てずっぽう」から「型安全で・観測でき・落ちても復元する本番システム」へ引き上げるフレームワークです。本記事の要点を再掲します。

  1. Agent + instructions で最小のエージェントを作り、output_type=BaseModel出力を型付きオブジェクトとして検証する(ToolOutput / NativeOutput / PromptedOutput を使い分け)。
  2. @agent.tool / @agent.tool_plain でツールを定義し、型注釈と docstring からスキーマを自動生成ModelRetry で齟齬を自己修復させる。
  3. deps_type で副作用を注入し、テスト可能にする(override でモック差し替え)。
  4. @agent.output_validator + ModelRetry で意味的検証を会話ループに組み込み、リトライには上限を設ける。
  5. run_stream + stream_output で検証しながらストリーミング(partial_output で副作用を制御)。
  6. Logfire(instrument_pydantic_ai で OpenTelemetry に統合し、全挙動とコストを一望する。
  7. durable execution(Temporal / DBOS 等) で長時間ワークフローを障害耐性のあるものにする。

PydanticAI の根底にあるのは、結局のところ Pydantic 本体と同じ規律です——「外から来るデータ(LLM の出力を含む)を、境界で検証してから内側へ通す」。この一貫性こそが、AI を本番の信頼性へ橋渡しします。

公式の一次情報として、以下を本記事の観点で再読することをお勧めします。


型安全な AI エージェント開発のご相談

筆者は、国内大手放送事業者向けの社内 AI プラットフォームをはじめ、生成 AI を本番品質で組み込むバックエンドを設計・運用してきました。LLM の出力を型で検証し、ツールを決定的コードに分離し、可観測性で挙動を追い、durable execution で落ちないワークフローを組む——「AI を動かす」のではなく「AI を事業の信頼性に載せる」ための設計を、生成 AI を活用して高速かつ高品質に実装します。PydanticAI / FastAPI を用いた AI エージェント・RAG・構造化抽出パイプラインの構築について、お気軽にご相談ください。

友田

友田 陽大

Người phát triển sản phẩm đoạt Giải thưởng Bộ trưởng (METI). Với TypeScript + Python + AWS, tôi một mình triển khai trọn gói SaaS, chuyển đổi số công nghiệp và AI tạo sinh (RAG) sẵn sàng cho production — từ phân tích yêu cầu đến hạ tầng và vận hành.

Got a challenge?

From design to implementation and operations — solo × generative AI

Implementation like this article's, end to end from requirements to production. Start with a free 30-minute technical consult and tell me about your situation.

Nhận cả dự án trọn gói lẫn vai trò cố vấn kỹ thuật. Hãy bắt đầu bằng buổi tư vấn miễn phí 30 phút.

最短ルート:カレンダーから直接予約

相談内容が固まっている方は、フォーム送信よりその場で日程を確定する方がスムーズです。下記から空き時間をお選びください。

  • 30分のオンライン無料相談
  • Google Meet / Zoom / Microsoft Teams
  • NDA 商談前締結可・無理な営業はいたしません
無料相談の空き枠を予約する

Cũng đáng đọc