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

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

- 公開日: 2026-06-26
- 著者: 友田 陽大
- タグ: Python, Pydantic, PydanticAI, AIエージェント, LLM, 型安全, 可観測性
- URL: https://tomodahinata.com/blog/pydantic-ai-agent-framework-production-guide

## 要点

- PydanticAIは「境界の外を信頼しない」Pydanticの規律をLLM出力に持ち込むエージェントフレームワーク。2026年6月にv2.0がリリースされた
- 最大の武器はoutput_type=BaseModel。LLMの自由文を型付きオブジェクトとして検証し、失敗時はModelRetryで自己修復させる
- ツールは@agent.tool（RunContext付き）/@agent.tool_plainで定義し、docstringと型注釈が自動でJSONスキーマになる。副作用はdeps_typeで注入してテスト可能にする
- 可観測性はLogfire（logfire.instrument_pydantic_ai）でOpenTelemetryに統合。全ツール呼び出し・リトライ・トークン使用量を一望できる
- 本番の回復性はTemporal/DBOS等のdurable execution（TemporalAgent/DBOSAgent）で担保し、API障害やクラッシュをまたいで進捗を復元する

---

## **導入：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_type` → `output_type`、`system_prompt` → `instructions` 推奨など。後述します）。本記事は [公式ドキュメント](https://pydantic.dev/docs/ai/overview/) に忠実に、2.0 の正しい API でまとめます。

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

---

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

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

```bash
# フル版（全プロバイダ同梱）
pip install pydantic-ai

# 軽量版＋必要なプロバイダだけ（推奨）
pip install "pydantic-ai-slim[anthropic]"
```

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

```python
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_type` に `BaseModel` を渡すと、**LLM の応答が自動でそのモデルとして検証**され、`result.output` が**型付きオブジェクト**になります。

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

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

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

```python
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 スキーマが**自動生成**されます。

```python
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）を解釈して、各引数の説明まで自動でスキーマに反映されます。

```python
@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 に修正を促せます**。

```python
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 設計](/blog/ai-agent-tool-use-function-calling-production-design) で論じた「決定的コードと確率的判断の分離」を、PydanticAI が言語機能として提供しているということです。

---

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

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

```python
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` で行い、失敗時は `ModelRetry` で**LLM に作り直させます**。

```python
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 構造化出力ガイド](/blog/pydantic-llm-structured-output-json-schema-validation-guide) で詳述）。

---

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

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

```python
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 ベース**で全挙動を計装します。

```python
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 行（`configure` ＋ `instrument_pydantic_ai`）で、**各エージェント実行・ツール呼び出し・リトライ・トークン使用量**がトレースとして可視化されます。`result.usage` 属性からはトークン数とリクエスト数を直接取得でき、コスト監視の基盤になります。HTTP の生リクエストまで見たいなら `logfire.instrument_httpx(capture_all=True)` を足します。

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

---

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

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

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

```python
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 にチェックポイントします。

```python
from pydantic_ai.durable_exec.dbos import DBOSAgent

dbos_agent = DBOSAgent(agent)
result = await dbos_agent.run("メキシコの首都は？")
```

| バックエンド | 性質 | 向いている場面 |
| --- | --- | --- |
| Temporal | ワークフローエンジン。強力な再試行・タイマー | 複雑な長時間オーケストレーション |
| DBOS | DB チェックポイント方式。軽量 | 既存 DB に状態を寄せたい |
| Prefect / Restate | データパイプライン／耐久 RPC 指向 | それぞれの基盤に合わせて |

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

**なぜこれが効くのか？**
私が放送事業者向けに構築した社内 AI プラットフォーム（[番組制作支援](/case-studies/broadcaster-ai-content-platform)）では、長時間の AI ジョブを **Cloud Workflows / Cloud Run Jobs** に切り離して回復性を担保しました。PydanticAI の durable execution は、その「長時間ジョブを落ちない形で回す」という要求を、**エージェントのレイヤーで**解決します。FastAPI の `BackgroundTasks` で抱え込むには重すぎる処理を、ジョブ基盤へ逃がす設計（[FastAPI 本番運用ガイド](/blog/fastapi-production-async-pydantic-observability-guide) 参照）の、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 を本番の信頼性へ橋渡しします。

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

- [PydanticAI Overview](https://pydantic.dev/docs/ai/overview/)
- [Agents](https://pydantic.dev/docs/ai/core-concepts/agent/)
- [Output（構造化出力・ストリーミング）](https://pydantic.dev/docs/ai/core-concepts/output/)
- [Tools](https://pydantic.dev/docs/ai/tools-toolsets/tools/)
- [Dependencies](https://pydantic.dev/docs/ai/core-concepts/dependencies/)
- [Durable Execution](https://pydantic.dev/docs/ai/integrations/durable_execution/overview/)
- [Logfire 連携](https://pydantic.dev/docs/ai/integrations/logfire/)

---

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

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