Skip to content

Pydantic AI

Pydantic AI は、 Pydantic を作ってるチームが開発してる AI エージェント用のフレームワーク。 LLM に送った結果を Pydantic のモデルで受け取れるのが特徴で、出力の型を BaseModel で宣言しておくと、バリデーション済みのインスタンスが返ってくる。

LLM に「JSON 形式で返して」ってプロンプトで頼んで、返ってきた文字列を自分で json.loads して……というのをやらなくていい。出力の形を型で定義するだけで、 Pydantic AI が構造化して返してくれる。

バリデーション(Pydantic) で見た BaseModel の書き方がそのまま使えるし、同じ開発チームの Logfire とも相性がいい。 OpenAI・Anthropic・Google など主要なプロバイダに対応してるけど、ここでは Google の Gemini を使う。


インストールと API キーの準備

Pydantic AI 本体を入れる。 pydantic-ai は全プロバイダの依存をまとめて含むので、これだけで Google(Gemini)も使える。

uv add pydantic-ai

pydantic-ai と pydantic-ai-slim

pydantic-ai は全プロバイダの依存が入るぶん重い。使うプロバイダが決まってるなら pydantic-ai-slim[google] のように必要なエクストラだけ入れると軽い。複数使うなら pydantic-ai-slim[google,openai] のようにカンマで並べる。

サンプルでは .env から API キーを読むので、 python-dotenv も入れる。

uv add python-dotenv

Gemini の API キーは Google AI Studio で発行できる。取得したキーを .envGOOGLE_API_KEY として置く。

.env
GOOGLE_API_KEY=your-api-key

モデルの指定と環境変数

Agent("google:gemini-2.5-flash-lite")google: がプロバイダの接頭辞。この指定だと Pydantic AI は環境変数 GOOGLE_API_KEY を見に行く。だから .env のキー名は GOOGLE_API_KEY にしておく。


サンプル:問い合わせのトリアージ

サポートに届いた問い合わせ文(自然文)を Gemini に読ませて、カテゴリ・優先度・要約・返信要否に構造化するスクリプト。

main.py
from typing import Literal

from dotenv import load_dotenv
from pydantic import BaseModel, Field
from pydantic_ai import Agent

# .env の GOOGLE_API_KEY を環境変数に読み込む
load_dotenv()


class Inquiry(BaseModel):
    category: Literal["bug", "request", "question", "other"] = Field(
        description="問い合わせの種類"
    )
    priority: Literal["low", "mid", "high"] = Field(description="対応の優先度")
    summary: str = Field(description="問い合わせ内容の要約(日本語)")
    needs_reply: bool = Field(description="こちらからの返信が必要かどうか")


agent = Agent(
    "google:gemini-2.5-flash-lite",
    output_type=Inquiry,
    system_prompt="サポート窓口の担当者として、届いた問い合わせ文を分類・要約してください。",
)


def main():
    text = input("問い合わせ内容を入力してください: ").strip()
    if not text:
        print("問い合わせ内容が入力されていません。終了します。")
        return

    result = agent.run_sync(text)

    # result.output はバリデーション済みの Inquiry インスタンス
    print(result.output.model_dump_json(indent=2))


if __name__ == "__main__":
    main()
  • Inquiry が出力の型。 categorypriorityLiteral[...] にしてるので、モデルは決められた選択肢の中からしか選べない。自由な文字列じゃなくて必ず "bug""high" に収まるので、返ってきた値をそのまま if の分岐や DB のカラムに使える。 needs_replybool にしておけば True / False で受け取れる。

  • 各フィールドの Field(description=...) は LLM へのヒントになる。「このフィールドに何を入れてほしいか」を description で伝えると、その通りに埋めてくれる。 system_prompt はエージェントの役割(サポート窓口の担当者)を伝える固定の指示。

  • Agent(...) の最初の引数がモデル(google:gemini-2.5-flash-lite)。 output_type=Inquiry で「この型で返して」と指定する。
  • agent.run_sync(text) に問い合わせ文を渡すと、同期的に実行されて結果が返る。 result.outputInquiry のインスタンス。バリデーション済みなので、そのまま result.output.categoryresult.output.priority のように属性でアクセスできるし、 priority"high" 以外の想定外の値になることはない。

  • result.output.model_dump_json(indent=2) で JSON 文字列にして表示してる。 Pydantic の model_dump_json は日本語をエスケープせずそのまま出す(標準ライブラリの json.dumps と違って ensure_ascii=False 相当がデフォルト)ので、これだけで読める JSON になる。

モデルは flash-lite にした

最初は gemini-2.5-flash を使ってたけど、無料枠だと混雑して 503 UNAVAILABLE("This model is currently experiencing high demand")が返ることがある。 gemini-2.5-flash-lite は軽量で「分類・単純な抽出」向けに作られてるモデルなので、今回のトリアージ用途に合ってるし、混雑にも当たりにくい。


実行してみる

uv run python main.py

問い合わせ例1: アプリにログインできない

% uv run python main.py
問い合わせ内容を入力してください: ログインしようとするとエラー画面が出てアプリが落ちます。至急なんとかしてほしいです。
{
  "category": "bug",
  "priority": "high",
  "summary": "ログイン時にエラーが発生し、アプリがクラッシュする。",
  "needs_reply": true
}

問い合わせ例2: ダークモードのリクエスト

% uv run python main.py
問い合わせ内容を入力してください: もし可能であれば、将来的にダークモードに対応してもらえると嬉しいです。急ぎではありません。
{
  "category": "request",
  "priority": "low",
  "summary": "ダークモードへの対応希望",
  "needs_reply": false
}

自由文だった問い合わせが、 categoryprioritysummaryneeds_reply という決まった形に落ちた。 categorypriorityLiteral で選択肢を固定してるので、このあと「優先度でソートする」「カテゴリで担当を振り分ける」みたいな処理がそのまま書ける。プロンプトで「JSON で返して」と指示しなくても、型を定義するだけでこの構造とバリデーションが効く。


FastAPI で使う

result.output は Pydantic のモデルなので、 FastAPI のエンドポイントの response_model にそのまま渡せる。問い合わせ文を受け取って、トリアージ結果を返すエンドポイントはこれだけ。スクリプトでは run_sync を使ったけど、 async def のエンドポイントの中では await agent.run(...) を使う。

main.py
from typing import Literal

from dotenv import load_dotenv
from fastapi import FastAPI
from pydantic import BaseModel, Field
from pydantic_ai import Agent

# .env の GOOGLE_API_KEY を環境変数に読み込む
load_dotenv()


class Inquiry(BaseModel):
    category: Literal["bug", "request", "question", "other"] = Field(
        description="問い合わせの種類"
    )
    priority: Literal["low", "mid", "high"] = Field(description="対応の優先度")
    summary: str = Field(description="問い合わせ内容の要約(日本語)")
    needs_reply: bool = Field(description="こちらからの返信が必要かどうか")


agent = Agent(
    "google:gemini-2.5-flash-lite",
    output_type=Inquiry,
    system_prompt="サポート窓口の担当者として、届いた問い合わせ文を分類・要約してください。",
)

app = FastAPI(title="inquiry-triage")


class InquiryInput(BaseModel):
    text: str


@app.post("/inquiries/triage", response_model=Inquiry)
async def triage(body: InquiryInput):
    result = await agent.run(body.text)
    return result.output

curl -X POST http://127.0.0.1:8000/inquiries/triage \ -H "Content-Type: application/json" \ -d '{"text": "ログインしようとするとエラー画面が出てアプリが落ちます。至急なんとかしてほしいです。"}'{"category":"bug","priority":"high","summary":"ユーザーがログインしようとするとエラーが発生し、アプリがクラッシュするとのこと。緊急の対応が必要。","needs_reply":true}

Logfire で計測する

Logfire と組み合わせると、エージェントの実行をトレースとして可視化できる。 Logfire も Pydantic と同じチームなので連携が用意されてて、 logfire.instrument_pydantic_ai() を足すだけで、どんなプロンプトを送って・モデルがどう応答して・何トークン使ったかが span になる。

上のコードに Logfire の設定を足したのが下のコード。 logfire.md でやった instrument_fastapi(app) も一緒に入れてるので、 HTTP リクエストからモデル呼び出しまでが 1 本のトレースにつながる。

main.py
from typing import Literal

import logfire
from dotenv import load_dotenv
from fastapi import FastAPI
from pydantic import BaseModel, Field
from pydantic_ai import Agent

# .env の GOOGLE_API_KEY を環境変数に読み込む
load_dotenv()

logfire.configure(
    service_name="inquiry-triage",
    send_to_logfire="if-token-present",
)
logfire.instrument_pydantic_ai()


class Inquiry(BaseModel):
    category: Literal["bug", "request", "question", "other"] = Field(
        description="問い合わせの種類"
    )
    priority: Literal["low", "mid", "high"] = Field(description="対応の優先度")
    summary: str = Field(description="問い合わせ内容の要約(日本語)")
    needs_reply: bool = Field(description="こちらからの返信が必要かどうか")


agent = Agent(
    "google:gemini-2.5-flash-lite",
    output_type=Inquiry,
    system_prompt="サポート窓口の担当者として、届いた問い合わせ文を分類・要約してください。",
)

app = FastAPI(title="inquiry-triage")
logfire.instrument_fastapi(app)


class InquiryInput(BaseModel):
    text: str


@app.post("/inquiries/triage", response_model=Inquiry)
async def triage(body: InquiryInput):
    result = await agent.run(body.text)
    return result.output

FatAPI を起動して curl コマンドでリクエストを送る。

uv run fastapi dev

curl -X POST http://127.0.0.1:8000/inquiries/triage \ -H "Content-Type: application/json" \ -d '{"text": "ログインしようとするとエラー画面が出てアプリが落ちます。至急なんとかしてほしいです。"}'{"category":"bug","priority":"high","summary":"ログイン試行時にエラーが発生し、アプリがクラッシュする。","needs_reply":true

トークン未設定でもコンソールに span が出る。クラウドのダッシュボードで見たいときは、 Logfire の手順どおりに認証してプロジェクトを作る。

curl コマンドを実行した後の Logfire のダッシュボード。

Logfire の Live 画面。POST /inquiries/triage の下に agent run、さらに chat gemini-2.5-flash-lite が入れ子で並んでいる

POST /inquiries/triage → 200(scope は fastapi)を親に、その下に agent run、さらに下に chat gemini-2.5-flash-lite(Gemini への実際のリクエスト)が入れ子で並ぶ。 agent runchat の scope は pydantic-ai になってて、 instrument_pydantic_ai() が作った span だとわかる。

右側の ↗154 ↙44 は入力(プロンプト)154 トークン・出力 44 トークンの意味で、 Σ が付いた行は子孫の span を合算した値。 1 リクエストで何トークン使ったか、モデル呼び出しにどれだけかかったかが、コードを足さずに見える。