Google Gemini-2.0-Flashの無料枠でpydantic_aiを動かす

概要

PydanticAIは大規模言語モデル(以後LLM)を使ったフレームワークツール。今回もGoogle Gemini APIの無料枠を使用する。この記事ではGoogle Gemini APIの無料枠を使用することにより、無料でPydanticAIを試せる事を示す。Google Gemini APIではgemini-2.0-flashを使用する。今回はGoogle Colaboratoryを使用する。

手順

Google Colaboratoryの準備

Colaboratoryを起動し、GeminiのAPIキーをGOOGLE_API_KEYとしてSecretを入力

!pip install -qq pydantic_ai nest_asyncio
import nest_asyncio
nest_asyncio.apply() # Google Colab自体がasyncio配下で動いているのでネストさせる。
from google.colab import userdata
GOOGLE_API_KEY=userdata.get("GOOGLE_API_KEY") # Secrets（🔑）からGOOGLE_API_KEYをNotebook access可能にする
import os
os.environ["GEMINI_API_KEY"] = GOOGLE_API_KEY

PydanticAIはデフォルトがasyncによる非同期関数だらけなので、nest_asyncio.apply()しないとうまいことJupyterNotebookで動かない。

Agent

from pydantic_ai import Agent
from pydantic_ai.models.gemini import GeminiModel
from pydantic import BaseModel

model = GeminiModel("gemini-2.0-flash")

# デフォルトでは文字列が返ってくる上に、最後に改行が入ってくる
agent = Agent(model=model)
result = await agent.run("1+1=?")
print(result.data)

# === 応答の型の変更 ===

# 強制的に数字で返させることも可能
agent = Agent(model=model, result_type=int)
result = await agent.run("1+1=?")
print(result.data) # 2 が数字で返ってくる
agent = Agent(model=model, result_type=int)
result = await agent.run("水曜日の次は何曜日？")
print(result.data) # これもなんとかして数字で返そうとしてくる。日曜日は0、月曜日は1、といった塩梅である。

# run実行時にさらに強制的に文字列で返させることも可能
agent = Agent(model=model, result_type=int)
result = await agent.run("土曜日の次は何曜日？", result_type=str)
print(result.data) # 文字列で返される。最後に改行が入ってくる

# 構造体も可能。この場合、変数名が意味を持つので注意が必要。
class Answer(BaseModel):
    曜日: str
    日付: int
agent = Agent(model=model, result_type=Answer)
result = await agent.run("土曜日が12日だとして、その4日後は何曜日の何日？")
print(result.data.曜日, result.data.日付)

• Python3ではUnicodeの変数名が使えるが、PydanticAIではBaseModelの変数名が激しく意味を持つので、あえて日本語で変数名を記述している点がポイント。プロンプトが日本語である場合に有効。

from pydantic_ai import Agent, ModelRetry
from pydantic_ai.models.gemini import GeminiModel

from pydantic import BaseModel, StringConstraints
from typing import Annotated

model = GeminiModel("gemini-2.0-flash")

# === 応答の型の変更 (制限付き応答) ===

class Answer(BaseModel):
    曜日: Annotated[str, StringConstraints(max_length=2)] # 曜日を2文字以内で表現するように制限
    日付: int

agent = Agent(model=model, result_type=Answer)
result = await agent.run("土曜日が12日だとして、その4日後は何曜日の何日？")
print(result.data.曜日, result.data.日付)

class Answer(BaseModel):
    曜日: Annotated[str, StringConstraints(pattern="^.曜日$")] # "水曜日" という表現に制限
    日付: int

agent = Agent(model=model, result_type=Answer)
result = await agent.run("土曜日が12日だとして、その4日後は何曜日の何日？")
print(result.data.曜日, result.data.日付)

• Pydanticのリッチなバリデーション機構も使える。

SystemPrompt

from pydantic_ai import Agent, RunContext
from pydantic_ai.models.gemini import GeminiModel

model = GeminiModel("gemini-2.0-flash")

# デコレータを使い、動的にシステムプロンプトを付与する場合
agent = Agent(model=model)
@agent.system_prompt
async def 入れ知恵(ctx: RunContext[None]) -> str:
    return "あなたはカラスの専門家です。"
result = await agent.run("多くのカラスは黒いですか？")
print(result.data)

# 静的にエージェントにシステムプロンプトを付与する場合
agent = Agent(model=model, system_prompt="あなたは逆のことを伝えるAIエージェントです。")
result = await agent.run("多くのカラスは黒いですか？")
print(result.data)

• 一般的なシステムプロンプトも動的に記述できる。

Deps

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.gemini import GeminiModel

model = GeminiModel("gemini-2.0-flash")

class Species(BaseModel):
    名前: str
    色: str

class Answer(BaseModel):
    回答: bool
    補足事項: str

agent = Agent(model=model, deps_type=Species, result_type=Answer)
@agent.system_prompt
async def 入れ知恵(ctx: RunContext[Species]) -> str:
    return f"あなたは{ctx.deps.名前}の専門家です。" # 変数化することで、この関数の単体テストがしやすくなる。

species = Species(名前="カラス", 色="黒い")
result = await agent.run(f"多くの{species.名前}は{species.色}ですか？", deps=species) #
print(result.data.回答)
print(result.data.補足事項)

species = Species(名前="サイコロ", 色="白い") # 変数化することで、プロンプトの再利用もしやすくなる。
result = await agent.run(f"多くの{species.名前}は{species.色}ですか？", deps=species) #
print(result.data.回答)
print(result.data.補足事項)

• Depsを使えば動的にプロンプトも作成できる。

Tools

import datetime
import zoneinfo
from typing import Annotated
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.gemini import GeminiModel
from pydantic import BaseModel, StringConstraints

model = GeminiModel("gemini-2.0-flash")

agent = Agent(model=model)
@agent.tool_plain
def get_today_date() -> str:
    """get today date as format "%04Y-%02m-%02d" in UTC"""
    now = datetime.datetime.now(tz=datetime.timezone.utc)
    return now.strftime("%04Y-%02m-%02d")

result = await agent.run("今日はUTCで何日？")
print(result.data)

# === toolとdepsの組み合わせ ===

class Timezone(BaseModel):
    tzname: Annotated[str, StringConstraints(pattern="|".join(zoneinfo.available_timezones()))]

agent = Agent(model=model, deps_type=Timezone)

@agent.tool
def get_now_time(ctx: RunContext[Timezone]) -> str:
    """get now time as format "%02H-%02M-%02S" """
    now = datetime.datetime.now(tz=zoneinfo.ZoneInfo(ctx.deps.tzname))
    return now.strftime("%02H-%02M-%02S")

tz = Timezone(tzname="Asia/Tokyo")
result = await agent.run("今何時ですか？", deps=tz)
print(result.data) # JSTで帰って来る

tz = Timezone(tzname="UTC")
result = await agent.run("今何時ですか？", deps=tz)
print(result.data) # UTCで帰って来る

• ToolはDepsと共に使う事で真価を発揮する。

Message History

from pydantic_ai import Agent
from pydantic_ai.models.gemini import GeminiModel
from pydantic_ai.messages import (
    ModelRequest,
    ModelResponse,
    UserPromptPart,
    SystemPromptPart,
    TextPart
)

model = GeminiModel("gemini-2.0-flash")

# === 連続的な会話のプロンプト ===

agent = Agent(model=model, system_prompt="あなたは数学者です。")
result = await agent.run("1+1=?")
messages = result.all_messages()
result = await agent.run("それに2を掛けると？", message_history=messages,)
print(result.data)

# Langchainのように途中の会話を改ざんすることも可能 詳細は https://ai.pydantic.dev/api/messages/
forged_messages = [
    ModelRequest(parts=[SystemPromptPart(content="あなたはしりとりのエキスパートです。"), UserPromptPart(content="しりとり")]),
    ModelResponse(parts=[TextPart(content="りんり"),]),
    ModelRequest(parts=[UserPromptPart(content="りか")]),
    ModelResponse(parts=[TextPart(content="かりん"),]),
]
result = await agent.run("「かりん」は「ん」で終わります。あなたの負けです", message_history=forged_messages)
print(result.data)

• AIとの会話をシリアライズして注入することが可能なので、会話の保存も容易。

Multi Agent Systems

PydanticAIは汎用性が高いのでマルチエージェントシステムは自分で作る必要がある。Agenticなワークフローを提供しているControlFlowのページによれば、様々な種類の形がある為、自分で選択し実装する必要がある。基本的にToolからAgentを呼び出す形である。

• マルチエージェントシステムは「どのエージェントがタスクの完了フラグを立てる権限を持つか？」が鍵となる。基本的にLLM達の思考は無限に連鎖する為、どこかで思考停止を行わなければ、映画「War Games」のようにシステムが壊れるだろう。

SingleAgent

• Full-DirectedなMASである。一から全部お膳立てしてあげる必要がある為、コーディング量が多くなってしまうが、確実である。

import random
from typing import Annotated, Optional
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.gemini import GeminiModel
from pydantic import BaseModel, StringConstraints
from dataclasses import dataclass

model = GeminiModel("gemini-2.0-flash")

# 呼び出すAgentが固定的である場合、オーケストレーターは不要

@dataclass
class JankenResult:
    result: Annotated[str, StringConstraints(pattern="^(グー|チョキ|パー)$")] | None

@dataclass
class JankenResults:
    mario: JankenResult
    luigi: JankenResult

mario = Agent(
    model=model, name="Mario",
    system_prompt="あなたはジャンケンをするAIエージェントです。"
                  "'get_janken' toolを使用し、ジャンケンで「グー」、「チョキ」、「パー」のどれかをランダムに出します。",
    deps_type=None, result_type=JankenResult
)
luigi = Agent(
    model=model, name="Luigi",
    system_prompt="あなたはジャンケンをするAIエージェントです。"
                  "'get_janken' toolを使用し、ジャンケンで「グー」、「チョキ」、「パー」のどれかをランダムに出します。",
    deps_type=None, result_type=JankenResult
)
peach = Agent(
    model=model, name="Peach",
    system_prompt="あなたはジャンケン勝負の結果を出力するAIエージェントです。"
                  "'prompt_janken' toolを使用し、ジャンケン勝負の結果と勝敗を出力してください。",
    deps_type=JankenResults, result_type=str
)

yoshi = Agent(
    model=model, name="Yoshi",
    system_prompt="あなたは親切なAIエージェントです。"
                  "get_janken_results' toolを使用し結果を教えてください。",
    deps_type=None, result_type=str
)

@mario.tool
def get_janken(ctx: RunContext[None]) -> str:
    return random.choice(["グー", "チョキ", "パー"])

@luigi.tool
def get_janken(ctx: RunContext[None]) -> str:
    return random.choice(["グー", "チョキ", "パー"])

@peach.tool
async def prompt_janken(ctx: RunContext[None]) -> str:
    mario_response = await mario.run(
        f'ジャンケン、',
        usage=ctx.usage,
    )
    luigi_response = await luigi.run(
        f'ジャンケン、',
        usage=ctx.usage,
    )
    jankenResults = JankenResults(mario=mario_response.data, luigi=luigi_response.data)
    print(jankenResults)
    if jankenResults.mario.result == "グー" and jankenResults.luigi.result == "グー":
        return "双方グーで引き分けです。"
    elif jankenResults.mario.result == "グー" and jankenResults.luigi.result == "チョキ":
        return "Marioがグーで勝ちです。"
    elif jankenResults.mario.result == "グー" and jankenResults.luigi.result == "パー":
        return "Luigiがパーで勝ちです。"
    elif jankenResults.mario.result == "チョキ" and jankenResults.luigi.result == "グー":
        return "Luigiがグーで勝ちです。"
    elif jankenResults.mario.result == "チョキ" and jankenResults.luigi.result == "チョキ":
        return "双方チョキで引き分けです。"
    elif jankenResults.mario.result == "チョキ" and jankenResults.luigi.result == "パー":
        return "Marioがチョキで勝ちです。"
    elif jankenResults.mario.result == "パー" and jankenResults.luigi.result == "グー":
        return "Marioがパーで勝ちです。"
    elif jankenResults.mario.result == "パー" and jankenResults.luigi.result == "チョキ":
        return "Luigiがチョキで勝ちです。"
    elif jankenResults.mario.result == "パー" and jankenResults.luigi.result == "パー":
        return "双方パーで引き分けです。"
    else:
        return "不明な結果です。"

@yoshi.tool
async def get_janken_results(ctx: RunContext[None]) -> str:
    response = await peach.run("ジャンケン勝負の結果と勝敗を出力してください", usage=ctx.usage)
    return response.data

result = await yoshi.run("どちらがどの手を出したかと、勝敗を教えて下さい。", deps=None, result_type=str)
print(result.data)

Popcorn

TBD

Moderated

TBD

RoundRobin

RBD

MostBusy

TBD

Random

TBD

Graphs

Pythonの条件式が使えるのにわざわざDAGを書くのはやめよう。

Image and Audio Input

PydanticAIは画像と音声の入力をサポートしている。 多分バイナリで渡せば動画もいける。

# 画像認識
import httpx
from pydantic_ai import Agent, BinaryContent
from pydantic_ai.models.gemini import GeminiModel

image_response = httpx.get('http://www.sakado-jigenji.jp/images/k_logo.png')
model = GeminiModel("gemini-2.0-flash")
agent = Agent(model=model)
result = await agent.run(
    [
        'これは何？',
        BinaryContent(data=image_response.content, media_type='image/png'),  # だいたいの画像に対応している
    ]
)
print(result.data)

# 音声認識

import httpx
from pydantic_ai import Agent, BinaryContent
from pydantic_ai.models.gemini import GeminiModel

audio_response = httpx.get('http://www.sakado-jigenji.jp/dl/pannyashingyou16.mp3')
model = GeminiModel("gemini-2.0-flash")
agent = Agent(model=model)
result = await agent.run(
    [
        '文字起こししてください',
        BinaryContent(data=audio_response.content, media_type='audio/mpeg'), # mp3, wavに対応しているみたい。
    ]
)
print(result.data)

Tracing

動作が基本的に不安定なLLMはトレースが必須である。 LangchainではLangsmithやLangfuseを使うがPydanticAIではLogFireを使うケースが多い。 しかし、LogFireは有償サービスであり、オンプレ動作は難しいのでLangfuseを使うことをオススメする。 Langfuseはオンプレもクラウドも可能である。トレースに関しては別の記事で紹介する。

Langfuseの無料枠でpydantic_aiのトレーサビリティを確保する