08-final-article.md - 清書完成版
1. はじめに: なぜPydantic AIなのか
LLMアプリ開発の現状
LLMを使ったアプリケーション開発、ここ1〜2年で一気に広まりました。
OpenAIのGPT-4、AnthropicのClaude、GoogleのGemini...
各社からAPIが提供されていて、誰でもLLMを組み込んだアプリを作れる時代です。
ただ、実際に開発してみると、いくつかの壁にぶつかるんですよね。
1つ目は、型の問題。
LLMの出力はテキストです。JSONで返してほしいと指示しても、
必ずしも期待通りの形式で返ってくるとは限らない。
プログラムで扱うには、パース処理が必要になるし、
エラーハンドリングも複雑になりがちです。
2つ目は、テストの問題。
LLMへのAPIコールは、同じ入力でも毎回違う出力が返ってくる可能性があります。
テストで「期待する出力」と比較するのが難しい。
かといって、毎回APIを叩くテストはコストがかかるし、遅い。
3つ目は、本番運用の問題。
レート制限、タイムアウト、コスト管理...
考えることが多いんですよね。
Pydantic AIが提供する価値
そこで登場したのが Pydantic AI です。
Pydanticというと、Pythonのデータバリデーションライブラリとしておなじみですよね。
FastAPIでも使われているし、知っている人も多いはず。
そのPydanticチームが作ったのが、Pydantic AI。
Pydanticの型安全性を、LLMアプリ開発に持ち込んだフレームワークなんです。
具体的には、こんなことができます。
- 型安全なエージェント定義: 入力・出力の型をPydanticモデルで指定
- 構造化出力: LLMの出力を自動的にPydanticモデルに変換
- 依存性注入（DI）: テストしやすい設計パターン
- 複数LLM対応: OpenAI、Anthropic、Geminiなどに対応
他フレームワークとの違い
LLMアプリフレームワークは他にもあります。
代表的なのはLangChainとLlamaIndex。
簡単に比較してみましょう。
| 特徴 | Pydantic AI | LangChain | LlamaIndex |
|---|---|---|---|
| 学習コスト | 低い | 高い | 中程度 |
| 型安全性 | 高い | 低い | 中程度 |
| 機能の豊富さ | 必要十分 | 非常に豊富 | RAG特化 |
| テスタビリティ | 高い | 中程度 | 中程度 |
| Pydanticとの親和性 | 高い | 低い | 低い |
LangChain は、非常に包括的なフレームワークです。
チェーン、エージェント、ツール、メモリ...
LLMアプリに必要なものが一通り揃っています。
ただ、正直なところ、ちょっと複雑なんですよね。
学習コストが高く、「何がベストプラクティスかわからない」
と感じることも多いです。
LlamaIndex は、RAG（検索拡張生成）に特化しています。
ドキュメントのインデックス作成や検索に強いですが、
汎用的なLLMアプリを作るには物足りない場面もあります。
Pydantic AI は、その中間というか、
「型安全にLLMアプリを作る」ことに特化しています。
シンプルだけど、必要な機能は揃っている。
Pydanticを使い慣れている人なら、すぐに馴染めるはずです。
3. 型安全なエージェントを作る
Agentクラスの基本構造
さて、先ほどのコードはシンプルでしたが、
出力がただのテキストでした。
実務で使うとなると、構造化されたデータが欲しいですよね。
Pydantic AIでは、出力の型をPydanticモデルで指定できます。
from pydantic import BaseModel
from pydantic_ai import Agent
# 出力の型を定義
class GreetingResponse(BaseModel):
greeting: str
language: str
formality: str # formal, casual, etc.
# 型を指定してエージェントを作成
agent = Agent(
"openai:gpt-4o",
output_type=GreetingResponse, # ここがポイント
system_prompt="あなたは多言語の挨拶に詳しいアシスタントです。",
)
Result型で出力を受け取る
エージェントを実行すると、Result オブジェクトが返ってきます。
async def main():
result = await agent.run("日本語で丁寧な挨拶を教えてください。")
# result.data は GreetingResponse 型
response: GreetingResponse = result.data
print(f"挨拶: {response.greeting}")
print(f"言語: {response.language}")
print(f"形式: {response.formality}")
asyncio.run(main())
実行結果:
挨拶: こんにちは、初めまして。
言語: Japanese
形式: formal
ここで注目してほしいのが、response.greeting のように、
属性アクセスでデータを取得できること。
LLMの出力が、自動的にPydanticモデルに変換されているんです。
IDEを使っていると、型ヒントが効くので、
補完も効くし、タイプミスも防げます。
これが「型安全」の恩恵です。
システムプロンプトの設定
システムプロンプトは、エージェントの振る舞いを決める重要な要素です。
静的な文字列だけでなく、関数で動的に生成することもできます。
from pydantic_ai import Agent, RunContext
def system_prompt() -> str:
return """
あなたはPythonの専門家です。
技術的な質問には、コード例を交えて回答してください。
初心者にもわかりやすく説明することを心がけてください。
"""
agent = Agent(
"openai:gpt-4o",
system_prompt=system_prompt,
)
あるいは、コンテキストに応じて動的に変えることも。
def dynamic_prompt(ctx: RunContext[None]) -> str:
# コンテキストに基づいてプロンプトを生成
return f"現在時刻: {ctx.timestamp}。簡潔に回答してください。"
5. ツール（Function Calling）で拡張する
@toolデコレータの使い方
エージェントに「能力」を追加するのが、ツールです。
OpenAIでいうFunction Callingに相当します。
例えば、現在の天気を取得するツールを作ってみましょう。
from pydantic_ai import Agent, tool
@tool
def get_weather(city: str) -> str:
"""指定された都市の天気を返します。"""
# 実際は外部APIを叩くところですが、
# ここではシンプルに固定値を返します
weather_data = {
"東京": "晴れ、気温18度",
"大阪": "曇り、気温16度",
"札幌": "雪、気温-2度",
}
return weather_data.get(city, "データがありません")
agent = Agent(
"openai:gpt-4o",
tools=[get_weather], # ツールを登録
system_prompt="あなたは天気案内アシスタントです。",
)
エージェントは、必要に応じてこのツールを呼び出します。
async def main():
result = await agent.run("東京の天気を教えてください。")
print(result.data)
asyncio.run(main())
実行結果:
東京の天気は晴れで、気温は18度です。良いお天気ですね。
エージェントが自動的に get_weather("東京") を呼び出し、
その結果を踏まえて回答を生成しています。
依存性注入（DI）パターン
ツールの中で外部APIを叩く場合、テスト時にモックしたくなりますよね。
Pydantic AIでは、依存性注入（DI）パターンが使えます。
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
@dataclass
class WeatherDeps:
"""天気APIの依存関係"""
api_key: str
base_url: str
@tool
def get_weather_with_deps(ctx: RunContext[WeatherDeps], city: str) -> str:
"""依存関係を使って天気を取得"""
# ctx.deps に依存関係が入っている
api_key = ctx.deps.api_key
base_url = ctx.deps.base_url
# 実際のAPIコール（省略）
# response = requests.get(f"{base_url}/weather?city={city}&key={api_key}")
return f"{city}の天気: 晴れ（APIから取得）"
agent = Agent(
"openai:gpt-4o",
deps_type=WeatherDeps,
tools=[get_weather_with_deps],
)
実行時に依存関係を渡します。
deps = WeatherDeps(api_key="xxx", base_url="https://api.example.com")
result = await agent.run("東京の天気は？", deps=deps)
テスト時は、モックした依存関係を渡せばいいわけです。
これで、テスト時にAPIを叩かずに済みます。
7. 本番運用のヒント
コスト管理
LLMのAPIは従量課金です。
本番運用では、コスト管理が重要になります。
1. トークン数の監視
result = await agent.run("質問")
usage = result.usage()
print(f"入力トークン: {usage.request_tokens}")
print(f"出力トークン: {usage.response_tokens}")
print(f"合計: {usage.total_tokens}")
2. キャッシュの活用
同じ質問に対しては、キャッシュを返す仕組みを検討しましょう。
Redisなどを使って、応答をキャッシュできます。
3. モデルの選択
GPT-4oは高性能ですが、コストも高い。
用途に応じて、GPT-4o-miniなどの軽量モデルも検討しましょう。
# 軽量モデルを使う例
agent = Agent("openai:gpt-4o-mini")
レート制限対策
APIプロバイダーにはレート制限があります。
大量リクエストを送ると、429エラーが返ってきます。
対策としては:
- リトライ with exponential backoff: 失敗時に待機時間を増やしながらリトライ
- リクエストのキューイング: バックグラウンドで処理
- 複数APIキーのローテーション: 制限を分散
エラーハンドリング戦略
LLM APIは、様々な理由で失敗する可能性があります。
from pydantic_ai import Agent
from pydantic_ai.exceptions import ModelHTTPError
agent = Agent("openai:gpt-4o")
async def safe_run(prompt: str) -> str | None:
try:
result = await agent.run(prompt)
return result.data
except ModelHTTPError as e:
print(f"APIエラー: {e}")
return None
except Exception as e:
print(f"予期しないエラー: {e}")
return None
フォールバック戦略も検討しましょう。
例えば、OpenAIが使えない場合は、Anthropicに切り替えるなど。
async def run_with_fallback(prompt: str) -> str:
try:
agent = Agent("openai:gpt-4o")
result = await agent.run(prompt)
return result.data
except ModelHTTPError:
# フォールバック
agent = Agent("anthropic:claude-3-5-sonnet-latest")
result = await agent.run(prompt)
return result.data
モニタリングとロギング
本番では、エージェントの挙動を監視したいですよね。
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("pydantic_ai")
agent = Agent("openai:gpt-4o")
async def logged_run(prompt: str):
logger.info(f"Input: {prompt}")
result = await agent.run(prompt)
logger.info(f"Output: {result.data}")
logger.info(f"Tokens: {result.usage()}")
return result
より詳細なモニタリングには、LangSmithやLangfuseなどの
LLM observabilityツールとの連携も検討できます。