1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@softbase(FullStack ITエンジニア)

Pydantic AIの紹介

1
Posted at

Pydantic AIの紹介

Pydantic AIは、Pydanticチームによって開発されたPythonのエージェントフレームワークです。Generative AI（生成AI）を使った生産グレードのアプリケーションやワークフローを、迅速で信頼性が高く、簡単に構築することを目的としています。PydanticのデータバリデーションとPythonの型ヒントを活用し、FastAPIのようなエレガントなデザインをGenAI開発に持ち込んでいます。モデル非依存で、さまざまなLLM（Large Language Models）プロバイダーをサポートし、観測可能性、型安全性、評価機能などを備えています。

以下では、Pydantic AIの基本的な使い方から応用例まで、たくさんの実例を交えてわかりやすく説明します。インストールはpip install pydantic-aiで行い、必要に応じてLLMのAPIキーを設定してください（例: OpenAIのOPENAI_API_KEY環境変数）。

1. Pydantic AIの基本

Pydantic AIのコアはAgentクラスです。エージェントを定義し、指示、ツール、出力型などを指定してAIを制御します。

例1: Hello World（シンプルなエージェントの作成）

基本的なエージェントを作成し、クエリを実行する例です。AnthropicのClaudeモデルを使い、簡潔な応答を指示します。

from pydantic_ai import Agent

# エージェントの定義
agent = Agent(
    'anthropic:claude-sonnet-4-0',  # 使用するモデル（AnthropicのClaude Sonnet 4.0）
    instructions='Be concise, reply with one sentence.'  # 指示: 簡潔に1文で応答
)

# 同期実行
result = agent.run_sync('Where does "hello world" come from?')
print(result.output)

出力例:

The phrase "hello world" originates from a 1974 Bell Laboratories internal memorandum by Brian Kernighan, and was popularized in the 1978 book "The C Programming Language" by Kernighan and Dennis Ritchie.

ポイント:

  • Agentをインスタンス化し、モデルと指示を指定。
  • run_syncで同期的にクエリを実行。結果はRunResultオブジェクトで、outputにテキスト応答が入る。
  • モデルはopenai:gpt-4oやgemini:gemini-1.5-proなど、さまざまなものを指定可能。

例2: ルーレットホイール（ツールを使ったシンプルなエージェント）

エージェントにツールを追加し、依存性を注入する例。ルーレットゲームをシミュレートします。

from pydantic_ai import Agent, RunContext
import random

# エージェントの定義（依存型: int, 出力型: bool）
roulette_agent = Agent(
    'openai:gpt-4o',
    deps_type=int,
    output_type=bool,
    system_prompt='Use the `roulette_wheel` function to determine if the user wins or loses based on their chosen square.'
)

# ツールの定義
@roulette_agent.tool
async def roulette_wheel(ctx: RunContext[int], square: int) -> str:
    """Simulates a roulette wheel spin and checks if the user wins."""
    spin_result = random.randint(0, 36)  # ルーレットのスピン結果
    win = spin_result == ctx.deps  # 依存性（ユーザーの選択した数字）と比較
    return f'The wheel landed on {spin_result}. You {"win" if win else "lose"}!'

# 実行
result = roulette_agent.run_sync('Put my money on square eighteen', deps=18)
print(result.output)  # True または False（勝敗）

出力例 (スピン結果次第):

False

ポイント:

  • deps_type=intで依存性を指定。run_sync時にdeps=18を渡す。
  • @agent.toolでツールを定義。RunContextから依存性にアクセス。
  • 出力型boolで、応答を構造化（Pydanticで検証）。

2. 高度な機能: ツールと依存性の注入

Pydantic AIは、ツールの追加や依存性の注入が強み。データベース接続などの外部リソースを安全に扱えます。

例3: 銀行サポートエージェント（ツール、依存性、構造化出力）

銀行のカスタマーサポートエージェントの例。データベースを模したクラスを使い、顧客情報を扱います。

from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

# 偽のデータベースクラス
class DatabaseConn:
    async def customer_name(self, id: int) -> str | None:
        if id == 123: return 'John'
    async def customer_balance(self, id: int, include_pending: bool) -> float:
        if id == 123: return 123.45 if include_pending else 100.00
        raise ValueError('Customer not found')

# 依存性データクラス
@dataclass
class SupportDependencies:
    customer_id: int
    db: DatabaseConn

# 出力モデル
class SupportOutput(BaseModel):
    support_advice: str
    block_card: bool
    risk: int

# エージェント定義
support_agent = Agent(
    'openai:gpt-4o',
    deps_type=SupportDependencies,
    output_type=SupportOutput,
    instructions='You are a support agent in our bank, give the customer support and judge the risk level of their query. Reply using the customer\'s name.'
)

# 動的指示の追加
@support_agent.instructions
async def add_customer_name(ctx: RunContext[SupportDependencies]) -> str:
    name = await ctx.deps.db.customer_name(id=ctx.deps.customer_id)
    return f"The customer's name is {name!r}"

# ツールの追加
@support_agent.tool
async def customer_balance(ctx: RunContext[SupportDependencies], include_pending: bool) -> str:
    """Returns the customer's current account balance."""
    balance = await ctx.deps.db.customer_balance(id=ctx.deps.customer_id, include_pending=include_pending)
    return f'${balance:.2f}'

# 実行例
deps = SupportDependencies(customer_id=123, db=DatabaseConn())
result = support_agent.run_sync('What is my balance?', deps=deps)
print(result.output)

result = support_agent.run_sync('I just lost my card!', deps=deps)
print(result.output)

出力例:

support_advice='Hello John, your current account balance, including pending transactions, is $123.45.' block_card=False risk=1
support_advice="I'm sorry to hear that, John. We are temporarily blocking your card to prevent unauthorized transactions." block_card=True risk=8

ポイント:

  • deps_typeで依存性を注入。RunContextからアクセス。
  • @agent.instructionsで動的指示を追加（顧客名を自動挿入）。
  • output_typeで構造化出力（Pydanticモデルで検証）。
  • ツールで外部操作（残高取得）を実行。

3. ストリーミングと会話の管理

Pydantic AIは、ストリーミング出力や会話履歴をサポートします。

例4: ストリーミング出力

クエリの応答をリアルタイムでストリーミングする例。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

async def main():
    async with agent.run_stream('What is the capital of the UK?') as response:
        async for text in response.stream_text():
            print(text, end='')  # ストリーミング出力

# 実行（asyncioでラップ）
import asyncio
asyncio.run(main())

出力例 (ストリーミング):

The capital of the UK is London.

ポイント:

  • run_streamでコンテキストマネージャを使い、テキストをストリーミング。
  • イベントハンドラでツール呼び出しなどのイベントを監視可能。

例5: 会話履歴の使用

複数回のやり取りを管理する例。

from pydantic_ai import Agent, Message

agent = Agent('openai:gpt-4o')

# 最初のメッセージ
history = [Message(role='user', content='Hello, what is Python?')]

result = agent.run_sync(history=history)
print(result.output)

# 続きの会話
history.append(Message(role='assistant', content=result.output))
history.append(Message(role='user', content='Tell me more about Pydantic.'))

result = agent.run_sync(history=history)
print(result.output)

ポイント:

  • message_historyで会話コンテキストを渡す。
  • Messageオブジェクトで役割（user/assistant）と内容を指定。

4. 観測可能性と評価

Pydantic AIはPydantic Logfireと統合し、リアルタイム監視が可能。評価機能でエージェントのパフォーマンスをテスト。

例6: Logfireを使った監視

from pydantic_ai import Agent
import logfire

logfire.configure()  # Logfireの設定

agent = Agent('openai:gpt-4o')
result = agent.run_sync('Explain quantum computing.')
# Logfireダッシュボードでトレース、コスト、性能を確認

ポイント:

  • Logfireでデバッグ、トレース、コスト追跡。
  • OpenTelemetry互換で他のツールも使用可能。

5. 高度な応用: グラフと耐久性

複雑なワークフローはグラフで定義。耐久実行で長時間タスクを扱う。

例7: グラフのイテレーション

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

with agent.iter('What is the weather today?') as it:
    for node in it:
        print(node)  # 各ノード（UserPromptNode, ModelRequestNodeなど）を処理
        it.advance()  # 次へ進む

ポイント:

  • iter()でグラフをステップバイステップで制御。
  • 複雑なアプリケーションに有用。

まとめ

Pydantic AIの主な特徴と利点：

  1. モデル非依存: OpenAI, Anthropic, Geminiなど多様なLLMをサポート。
  2. 型安全性: 型ヒントでIDE支援とエラー削減。
  3. ツール統合: 依存注入とPydantic検証で安全なツール作成。
  4. ストリーミングと耐久性: リアルタイム出力と長時間ワークフロー対応。
  5. 観測可能性: Logfireで監視・評価。
  6. 標準対応: Agent2Agent, AG-UIなどで相互運用性。

実例のポイント

  • 基本エージェント（例1）
  • ツール付き（例2）
  • 銀行サポート（例3）
  • ストリーミング（例4）
  • 会話管理（例5）
  • 監視（例6）
  • グラフ（例7）

Pydantic AIは、AIエージェント開発を効率化します。詳細は公式ドキュメント（https://ai.pydantic.dev/）を参照。

1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?