pydanticai

PydanticAI & PydanticGraph エキスパートスキル

PydanticAI は、Pydantic を開発したチームが構築した、本番環境向けの GenAI アプリケーションを構築するための Python エージェントフレームワークです。PydanticGraph は、これに付随するグラフ／ステートマシンライブラリです。

インストール：

pip install pydantic-ai               # Full install (all providers)
pip install "pydantic-ai-slim[openai]" # Minimal install + your provider

クイックリファレンス

from pydantic_ai import Agent

# Basic agent — one line
agent = Agent('openai:gpt-5.2', instructions='Be concise.')

# Run it
result = agent.run_sync('What is the capital of France?')
print(result.output)

読み込むリファレンスの選び方

トピック	読み込むタイミング	ファイル
エージェントの作成とライフサイクル	エージェントの作成、設定、実行が必要なとき — ツール、依存関係、出力型、実行メソッド、ストリーミングの定義	references/core-agents.md
機能とフック	組み込み機能（Thinking、WebSearch、MCP など）、オンデマンド読み込み、ライフサイクルフック、カスタム機能が必要なとき	references/capabilities-hooks.md
PydanticGraph	ステートマシン、グラフベースの制御フロー、並列実行、BaseNode サブクラス、結合や判断を伴う GraphBuilder が必要なとき	references/graph.md
モデル、出力、ストリーミング	複数モデルの設定、FallbackModel、ストリーミング出力、出力関数、検証付きの構造化出力が必要なとき	references/models-output.md
マルチエージェントのパターンと統合	エージェントへの委譲、プログラムによる引き継ぎ、MCP サーバー、耐久性のある実行、UI アダプターが必要なとき	references/patterns.md
テストと評価	TestModel、FunctionModel、pytest のパターン、オーバーライド、体系的な評価のための Pydantic Evals が必要なとき	references/testing-evals.md
実例の全体	実行可能な完全な例（銀行サポートエージェント、メールフィードバックグラフ、マルチエージェントによる航空券予約）が必要なとき	references/examples.md
フレームワークの境界	プロジェクトで PydanticAI と LangGraph を比較する必要があるとき、または両者を組み合わせたいとき	references/hybrid-pydanticai-langgraph.md — skill_view(name='langgraph') も読み込んでください
エージェント内での型付き System One 判断	型付きの判断サービスを注入する、応答の形状を検証する、または判断を生成出力から分離する必要があるとき	references/core-agents.md と System One。配置、権限、復旧、タスク全体のエビデンスについては harness-engineering を使用してください
API サーフェスのリファレンス	正しいインポートパス、クラス名、メソッドシグネチャをすばやく見つける必要があるとき	references/api-reference.md

よく使うパターンの概要

ツールと構造化出力を備えたエージェント

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

class WeatherResult(BaseModel):
    temperature: float
    conditions: str

agent = Agent('openai:gpt-5.2', output_type=WeatherResult)

@agent.tool
async def get_weather(ctx: RunContext, city: str) -> str:
    """Get current weather for a city."""
    return f"24°C and sunny in {city}"

result = agent.run_sync('Weather in London?')
print(result.output.temperature)

→ エージェントのライフサイクル全体、実行メソッド、ツールのパターンについては、references/core-agents.md を参照してください。

依存性注入を使うエージェント

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class MyDeps:
    api_key: str
    db_conn: str

agent = Agent('openai:gpt-5.2', deps_type=MyDeps)

@agent.tool
async def query_db(ctx: RunContext[MyDeps], sql: str) -> str:
    return f"Query results using {ctx.deps.db_conn}"

→ 依存性注入のパターンとテスト用オーバーライドについては、references/core-agents.md を参照してください。

複数ノードを含むグラフ

from dataclasses import dataclass
from pydantic_graph import BaseNode, End, GraphRunContext, GraphBuilder

@dataclass
class MyState:
    value: int = 0

@dataclass
class ProcessNode(BaseNode[MyState]):
    async def run(self, ctx: GraphRunContext[MyState]) -> End | NextNode:
        ctx.state.value += 1
        if ctx.state.value >= 5:
            return End(ctx.state.value)
        return NextNode()

→ BaseNode と GraphBuilder の両方の API、並列実行、結合／リデューサーのパターンについては、references/graph.md を参照してください。

使用する実行メソッドの選び方

必要なこと…	使用するもの	主な動作
同期コードで単一の回答を得る	run_sync()	完了までブロックし、RunResult を返します
非同期コードで単一の回答を得る	run()	非同期で実行し、RunResult を返します
生成中のテキストをストリーミングする	run_stream()	非同期コンテキストマネージャー。stream_text() / stream_output() を返します
詳細なイベント（ツール呼び出し、パートの開始、差分）を確認する	run_stream_events()	AgentStreamEvent の型（FunctionToolCallEvent、PartStartEvent、FinalResultEvent）を返します
グラフの各ステップを手動で制御する	iter()	エージェント内部のグラフノード（UserPromptNode → ModelRequestNode → CallToolsNode）を反復処理します
ストリーミング中に実行するツール呼び出しを扱う	run_stream_events() または run(event_stream_handler=...)	run_stream() は output_type に一致する最初の出力で停止し、その後のツール呼び出しは実行しません

各実行メソッドの詳細は references/core-agents.md を参照してください。

Graph API：BaseNode と GraphBuilder

要素	BaseNode（クラスベース）	GraphBuilder（関数ベース）
スタイル	BaseNode[StateT] をサブクラス化し、async run() を実装します	非同期関数を @g.step でデコレートします
状態の変更	run() メソッド内で ctx.state を使用します	ステップ関数内で ctx.state を使用します
並列処理	フォーク／ジョインのロジックを手動で実装します	要素ごとのファンアウトには .map()、同じ入力を複数に渡す場合は .broadcast() を組み込みで利用できます
ジョイン／集約	戻り値の型で集約を手動で行います	組み込みのリデューサー（reduce_list_append、reduce_sum、reduce_dict_update など）を利用できます
エッジの宣言	run() の戻り値の型注釈から推論されます	g.edge_from(source).to(target) で明示的に指定します
使用する場面	複雑なノードロジック、オブジェクト指向のパターン、条件付きエッジロジック	シンプルな線形フロー、並列データ処理、簡潔な構文

両方の API の詳細は references/graph.md を参照してください。

フレームワークの境界：PydanticAI と LangGraph

どちらのフレームワークも、グラフとツールを使ってエージェント型システムを構築しますが、設計思想は大きく異なります。適切な選択は、何を最適化するかによって決まります。

要素	PydanticAI + PydanticGraph	LangGraph	両方を併用する場合
設計思想	型安全で、データスキーマを重視します。FastAPI のような使い心地です。	低レベルのグラフプリミティブ（Pregel／Beam に着想を得ています）。NetworkX のような使い心地です。	エージェント層には PydanticAI、複雑なオーケストレーションには LangGraph
エージェントの定義	Agent(model, tools, deps, output_type) — 宣言的で、1 行で記述できます	メッセージパッシングを使った StateGraph ノードを手動で定義します	LangGraph StateGraph 内のノード関数として PydanticAI の Agent を使用します
ツール呼び出し	@agent.tool デコレーターを使い、型ヒントからスキーマを自動生成し、RunContext DI を使用します	ツールを手動で登録し、tool_node = ToolNode(tools) とします	LangGraph ノード内で PydanticAI の型付きツール定義を使用します
状態管理	GraphRunContext.state — 可変データクラスで、メモリー内に保持されます	型付きリデューサーとチェックポインター（SQLite／Postgres）を使った State	外側のフローには LangGraph のチェックポインター、サブグラフの状態には PydanticGraph
マルチエージェントのパターン	エージェントへの委譲（ツール呼び出し）、プログラムによる引き継ぎ、グラフベース	スーパーバイザー（中央ルーター）、スウォーム（直接引き継ぎ）、階層型（サブグラフ）	LangGraph のスーパーバイザーノード内で PydanticAI が委譲します
永続化	Temporal、Inngest、Prefect、DBOS による耐久実行	組み込みのチェックポインター（MemorySaver、SqliteSaver、PostgresSaver）	グラフレベルで LangGraph のチェックポインターを使用します
ストリーミング	5 つのメソッド：run、run_sync、run_stream、run_stream_events、iter	コンパイル済みグラフで .stream()／.astream_events() を使用します	PydanticAI のイベントハンドラーを LangGraph の .astream_events() でラップします
学習曲線	低め — 型ヒントがあらゆる場面で役立ちます	高め — 手動で配線する部分がより多くあります	最も高い — 2 つのメンタルモデルを習得する必要があります
最適な用途	単一エージェント、ツールを使うワークフロー、型安全な構造化出力、エージェントを初めて扱うチーム	複雑な状態機械、分岐／循環を伴うマルチエージェント、HITL、既存の LangChain ユーザー	型安全なエージェントと高度なオーケストレーションの両方が必要な大規模システム

境界条件 — 次の場合は LangGraph を検討してください：

• 長時間実行される会話に、組み込みのチェックポイント機能／永続化が必要な場合（SQLite、Postgres バックエンドが組み込まれています）
• マルチエージェントシステムで、分離された状態名前空間を持つサブグラフの合成が必要な場合
• Human-in-the-loop パターン（割り込み／再開、状態の編集、承認ワークフロー）が必要な場合
• すでに LangChain を使用しており、一貫性を保ちたい場合
• グラフで循環や Send() による動的なファンアウトが必要な場合

次の場合は PydanticAI を検討してください：

• 型安全性と IDE のオートコンプリートを優先する場合
• ボイラープレートを最小限に抑えた宣言的なエージェントが必要な場合
• 自動検証と再試行を備えた構造化出力が必要な場合
• マルチエージェントの要件が、単純な委譲または逐次的な引き継ぎで済む場合
• 合成可能な機能システム（Thinking、WebSearch、プラグインとしての MCP）を重視する場合

次の場合は両方の併用を検討してください：

• 外側のループには LangGraph のオーケストレーション（チェックポイント機能、サブグラフ、HITL）が必要で、内側のエージェントロジックには PydanticAI の型安全なエージェント定義とツールスキーマを使いたい場合
• 混成チームの場合：一部のエージェントには PydanticAI の型付けが適しており、ほかのエージェントには LangGraph の低レベル制御が必要な場合
• 完全な実例については、references/hybrid-pydanticai-langgraph.md を参照してください。

LangGraph スキル： skill_view(name='langgraph') — スーパーバイザー／スウォーム／階層型のパターン、永続化、本番環境へのデプロイ、評価について説明しています。

エラー処理のクイックリファレンス

from pydantic_ai import UnexpectedModelBehavior, capture_run_messages

with capture_run_messages() as messages:
    try:
        result = agent.run_sync('Query')
    except UnexpectedModelBehavior as e:
        cause = e.__cause__  # Often ModelRetry('reason')
        print(f"Root cause: {cause}")
        print("Full conversation:", messages)  # Inspect every message
        # Common recovery: raise ModelRetry from tools with clear instructions

例外	意味	対処方法
UnexpectedModelBehavior	再試行の上限に達したか、モデルが予期しない応答を返しました	e.__cause__を確認し、メッセージを確認して、指示またはツールの再試行回数を調整します
ModelRetry（ツールから発生）	ツールが異なる引数でモデルに再試行を求めています	そのまま伝播させます。PydanticAIがretriesの上限まで自動的に処理します
ModelAPIError	プロバイダーが4xx/5xxを返しました	APIキー、レート制限、モデルの利用可否を確認します
UsageLimitExceeded	トークンまたはリクエストの予算を使い切りました	UsageLimitsを増やすか、プロンプトを最適化します
HookTimeoutError	ライフサイクルフックがタイムアウトしました	フックのタイムアウトを延長するか、フックのロジックを最適化します

主なCLIコマンド

pip install pydantic-ai                    # Everything
pip install "pydantic-ai-slim[openai]"      # Minimal
pip install "pydantic-ai-slim[openai,google,anthropic]"  # Multi-provider

ディレクトリ構成

pydanticai/
├── SKILL.md
├── references/
│   ├── core-agents.md         # Agent lifecycle, tools, deps, output
│   ├── capabilities-hooks.md  # Capabilities system & lifecycle hooks
│   ├── graph.md               # PydanticGraph (BaseNode + GraphBuilder)
│   ├── models-output.md       # Models, streaming, structured output
│   ├── patterns.md            # Multi-agent patterns & integrations
│   ├── testing-evals.md       # Testing & evaluation framework
│   ├── examples.md            # Complete worked examples
│   ├── hybrid-pydanticai-langgraph.md  # PydanticAI as LangGraph node (hybrid pattern)
│   └── api-reference.md       # Quick API surface reference

注意点

• 出力タイプ = final のみ： output_type は最終応答を制約します。モデルは実行中にツール（関数ツール）を呼び出すこともできます。一方、出力関数は強制的に呼び出され、実行を終了します。

• pydantic-graph は pydantic-ai にまったく依存しません： 単独で使えるライブラリです。GenAI 以外のステートマシンにも使用できます。pip install pydantic-graph でインストールしてください。

• pydantic-ai-slim と pydantic-ai の違い： slim パッケージにはコア依存関係と OpenTelemetry のみが含まれます。完全版の pydantic-ai はメタパッケージで、openai、anthropic、google、cli、mcp、evals、web、retries、logfire の追加機能が含まれます。

• ストリーミング中のツール呼び出しは、デフォルトでは実行されません： run_stream() は出力タイプに一致する最初の出力で停止します。ツール呼び出しの実行を続けるには、run_stream_events() を使用するか、event_stream_handler を指定して run() を使用してください。

• システムプロンプト ≠ 指示： システムプロンプトはメッセージ履歴の一部であり、往復してやり取りされます。指示はサーバー側にあり、クライアントに送信されるメッセージには含まれません。message_history を再利用する場合、ReinjectSystemPrompt を追加しない限り、エージェントの新しいシステムプロンプトは自動的に送信されません。

• フォークする場合、conversation_id は手動で指定します： conversation_id='new' を渡すと、既存の履歴から新しい会話チェーンが開始されます。自動ではありません。

• モデル名は provider:model_name の形式です： PydanticAI は文字列のプレフィックスからモデルクラスを自動的に解決します。カスタムエンドポイントでは、OpenAIChatModel(model_name, provider=OpenAIProvider(base_url=...)) を使用してください。

• TestModel ではネイティブツールをエミュレートできません： エージェントが WebSearch などを使用する場合、テストでは agent.override(model=TestModel(), native_tools=[]) で上書きしてください。

• テスト可能なモジュールレベルのエージェントには defer_model_check=True を指定します： Agent をモジュールレベル（関数の外）で宣言し、テストで agent.override(model=TestModel()) を使って TestModel を使用する場合は、コンストラクターに defer_model_check=True を設定してください。設定しないと、実際のモデルがテスト実行前に上書きされていても、インポート時にエージェントがモデル文字列を解決しようとするため、API 認証情報がないと失敗します。

• メッセージ履歴では対応関係を保つ必要があります： 履歴をスライスするときは、ツール呼び出しとその戻り値を対応させたままにしてください。対応していないと、LLM でエラーが発生します。

• stream_text() は BaseModel の出力タイプでは失敗します： output_type が BaseModel（構造化出力）の場合、result.stream_text() を呼び出すと UserError('stream_text() can only be used with text responses') が発生します。ストリーミング中に部分的に検証済みのオブジェクトを取得するには、代わりに result.stream_output() を使用してください。構造化出力をテキストレベルでストリーミングする必要がある場合は、run_stream_events() を使用し、TextPartDelta の差分を含む PartDeltaEvent を調べてください。この2つのメソッドは異なる出力モードを対象としています。テキスト出力には stream_text()、構造化出力には stream_output() を使用します。

• graph.run() が返すのは状態オブジェクトではなく OutputT です： graph.run() に state=MyState() を渡しても、戻り値はグラフの output_type（例：list[int]）であり、状態ではありません。state オブジェクトは実行中にその場で変更されるため（変更可能なデータクラスであるため）、別途参照を保持してください：

  state = MyState(items_processed=0)
  result = await graph.run(state=state, inputs=[1, 2, 3])
  # result -> [2, 4, 6]       (OutputT = list[int])
  # state.items_processed -> 3 (state mutated in-place)
  

  この落とし穴は、並列の .map() パターンで、読み手が result.items_processed を使えると思い込む場合によく起こります。これは使えません。items_processed の件数は戻り値ではなく、渡した state オブジェクトに格納されています。

非チャットアプリケーションの提案

アプリケーションがエージェントを呼び出して編集可能なアクション案を生成する場合は、型付きアプリケーション提案を参照してください。型付きデータをアプリケーション所有の状態に返し、コマンドの実行権限はモデルの外部に保持してください。インタラクションの動作についてはproduct-design-and-ux、アーキテクチャについてはsoftware-architecture、配置についてはharness-engineering、評価についてはagent-evals-and-observability、本番環境での権限と復旧についてはagent-production-operationsに案内してください。

非チャットの提案では、候補 ID をアプリケーションから提供されたスコープと照合して検証してください。承認者の識別情報とコマンドは、モデル出力の外部に保持してください。実行時に、アプリケーションは現在の権限、リビジョン、ポリシー、および承認されたアクションの正確な内容を再確認する必要があります。承認時のロール確認だけでは不十分です。コマンドのステータスが不確かな場合は、再試行する前に永続化された ID を使って状態を照合してください。