Pydantic AI Agent開発実践：型安全でプロダクション級AIアプリを構築する5つのコアパターン

2026年、AI Agentは実験的プロジェクトからプロダクション級デプロイへと移行しました。しかし、PythonエコシステムにおけるAI開発は長らく核心的な矛盾に直面していました：LLMの出力は制御不可能な文字列であり、プロダクションシステムが必要とするのは型安全なデータ構造であるという矛盾です。Pydantic AIはこの問題を解決するために誕生しました——Pydanticの型検証能力とAI Agentの柔軟性を深く組み合わせ、開発者がPython型アノテーションだけでAgentの動作契約を定義できるようにします。

本記事では、実際の開発現場から5つのコアパターンを抽出し、Agent定義からストリーミングレスポンスまでの完全な開発チェーンをカバーします。各パターンには実行可能なコードが付属し、プロジェクトでの迅速な実装を支援します。

コア概念一覧

概念	説明	コアバリュー
`Agent`	Pydantic AIコア抽象化、LLMインタラクションロジックをカプセル化	統一Agent定義インターフェース
`RunContext`	ランタイムコンテキスト、依存性注入オブジェクトを保持	型安全な依存関係管理
`Tool`	関数ツール、Agentが呼び出し可能な外部機能	Function Calling標準化
`ResultType`	構造化出力タイプ、LLM戻り値フォーマットを制約	出力タイプ安全性保証
`StreamedRunResult`	ストリーミングレスポンス結果、トークンごとの出力をサポート	リアルタイムインタラクション体験

問題分析：従来のAI開発ではなぜ不十分なのか？

型欠如：LLMは生の文字列を返し、手動での解析と検証が必要——ランタイムエラーが頻発
依存関係の混乱：データベース接続、APIキーなどの依存関係が関数間で何層にも渡って渡され、密結合が深刻
ツール登録の煩雑さ：Function CallingのJSON Schemaを手書きするとエラーが発生しやすく、コードロジックと乖離
出力の制御不能：LLMが予期しないフォーマットを返す可能性があり、コンパイル時の保証がない
ストリーミング処理の複雑さ：SSE/WebSocketストリーミングレスポンスの解析とエラー処理コードが冗長

パターン1：Agent基礎定義と型安全モデル

Pydantic AIのAgent定義は、LLMインタラクションを型安全な関数呼び出しとして抽象化します。Python型アノテーションを通じて、コード記述時にAgentの入出力契約を定義します。

"""パターン1：Agent基礎定義と型安全モデル"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Optional

from pydantic import BaseModel, Field
from pydantic_ai import Agent


# 1. 構造化出力モデルの定義
class CodeReviewResult(BaseModel):
    """コードレビュー結果 - LLMの出力はこの構造に制約される"""

    quality_score: float = Field(
        ...,
        ge=0.0,
        le=10.0,
        description="コード品質スコア、0-10点",
    )
    issues: list[str] = Field(
        default_factory=list,
        description="発見された問題リスト",
    )
    suggestions: list[str] = Field(
        default_factory=list,
        description="改善提案リスト",
    )
    summary: str = Field(
        ...,
        min_length=10,
        max_length=500,
        description="レビューサマリー",
    )
    language: str = Field(
        ...,
        description="プログラミング言語",
    )


# 2. Agentの定義 - ジェネリックパラメータで出力タイプを指定
code_review_agent = Agent(
    "openai:gpt-4o",
    result_type=CodeReviewResult,
    system_prompt="""あなたはシニアコードレビュー専門家です。
ユーザーが提出したコードを専門的にレビューし、品質スコア、問題リスト、改善提案、レビューサマリーを提供してください。
評価基準：コード可読性、パフォーマンス、セキュリティ、ベストプラクティス。""",
)


# 3. Agentの使用
async def review_code(code: str) -> CodeReviewResult:
    """コードをレビューし構造化結果を返す"""
    result = await code_review_agent.run(code)
    return result.data


# 4. 同期実行方式
def review_code_sync(code: str) -> CodeReviewResult:
    """同期方式でコードレビュー"""
    result = code_review_agent.run_sync(code)
    return result.data


# 5. リトライ付きAgent定義
robust_review_agent = Agent(
    "openai:gpt-4o",
    result_type=CodeReviewResult,
    system_prompt="あなたはシニアコードレビュー専門家です。JSON Schemaに厳密に従って出力してください。",
    retries=3,  # 出力検証失敗時に自動リトライ
    model_settings={
        "temperature": 0.1,  # 低温度で出力の安定性を確保
        "max_tokens": 2000,
    },
)


# 6. 実行と使用量情報の確認
async def review_with_usage(code: str) -> tuple[CodeReviewResult, dict]:
    """コードレビューと使用統計の返却"""
    result = await robust_review_agent.run(code)
    usage = result.usage()
    return result.data, {
        "request_tokens": usage.request_tokens,
        "response_tokens": usage.response_tokens,
        "total_tokens": usage.total_tokens,
    }


# 実行例
if __name__ == "__main__":
    import asyncio

    sample_code = """
def calculate_total(items):
    total = 0
    for i in range(len(items)):
        total = total + items[i]['price'] * items[i]['qty']
    return total
    """

    result = asyncio.run(review_code(sample_code))
    print(f"品質スコア: {result.quality_score}")
    print(f"問題: {result.issues}")
    print(f"提案: {result.suggestions}")
    print(f"サマリー: {result.summary}")
    print(f"言語: {result.language}")

重要ポイント：

result_typeパラメータにより、LLM出力がPydanticモデルに直接バインドされ、検証が自動実行
retriesパラメータは出力検証失敗時に自動リトライし、堅牢性を向上
model_settingsでLLMパラメータを制御、低temperatureで構造化出力の安定性を確保

パターン2：依存性注入とシステムプロンプト

プロダクション級Agentはデータベース、キャッシュ、設定などの外部依存関係にアクセスする必要があります。Pydantic AIの依存性注入システムにより、これらの依存関係が型アノテーションを通じて自動注入され、グローバル変数とパラメータ受け渡しの混乱を完全に排除します。

"""パターン2：依存性注入とシステムプロンプト"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Optional

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext


# 1. 依存タイプの定義
@dataclass
class DatabaseDep:
    """データベース依存"""

    connection_string: str

    async def query(self, sql: str) -> list[dict]:
        """データベースクエリのシミュレーション"""
        print(f"[DB] クエリ実行: {sql}")
        return [{"id": 1, "name": "田中", "score": 95}]


@dataclass
class CacheDep:
    """キャッシュ依存"""

    backend: str = "redis"
    _cache: dict = field(default_factory=dict)

    async def get(self, key: str) -> Optional[str]:
        return self._cache.get(key)

    async def set(self, key: str, value: str, ttl: int = 3600) -> None:
        self._cache[key] = value
        print(f"[Cache] 設定 {key} (TTL={ttl}s)")


@dataclass
class UserContext:
    """ユーザーコンテキスト依存"""

    user_id: str
    username: str
    role: str = "user"
    preferences: dict = field(default_factory=dict)


# 2. 依存関係の合成
@dataclass
class AgentDeps:
    """Agentの全依存関係"""

    db: DatabaseDep
    cache: CacheDep
    user: UserContext


# 3. 出力モデルの定義
class AnalysisReport(BaseModel):
    title: str
    findings: list[str]
    confidence: float
    recommendation: str


# 4. 依存関係付きAgentの作成
analysis_agent = Agent(
    "openai:gpt-4o",
    result_type=AnalysisReport,
    deps_type=AgentDeps,
)


# 5. 動的システムプロンプト - 依存関係に基づいて動的生成
@analysis_agent.system_prompt
async def build_system_prompt(ctx: RunContext[AgentDeps]) -> str:
    """ユーザーロールと設定に基づいてシステムプロンプトを動的生成"""
    role_instructions = {
        "admin": "あなたは管理者権限を持ち、すべてのデータにアクセスし、すべての操作を実行できます。",
        "analyst": "あなたはアナリスト権限を持ち、データを閲覧できますが変更はできません。",
        "user": "あなたは一般ユーザー権限を持ち、自分のデータのみ閲覧できます。",
    }

    base_prompt = f"""あなたはデータ分析アシスタントです。
現在のユーザー: {ctx.deps.user.username} (ロール: {ctx.deps.user.role})
{role_instructions.get(ctx.deps.user.role, role_instructions['user'])}"""

    if ctx.deps.user.preferences.get("detail_level") == "verbose":
        base_prompt += "\n詳細な分析プロセスと推論ステップを提供してください。"
    else:
        base_prompt += "\n簡潔で明確な分析結論を提供してください。"

    return base_prompt


# 6. ツール関数で依存関係を使用
@analysis_agent.tool
async def query_user_data(
    ctx: RunContext[AgentDeps],
    query: str,
) -> str:
    """ユーザー関連データのクエリ"""
    cache_key = f"user:{ctx.deps.user.user_id}:{query}"
    cached = await ctx.deps.cache.get(cache_key)
    if cached:
        return f"[キャッシュヒット] {cached}"

    sql = f"SELECT * FROM user_data WHERE user_id = '{ctx.deps.user.user_id}' AND query LIKE '%{query}%'"
    results = await ctx.deps.db.query(sql)

    await ctx.deps.cache.set(cache_key, str(results))

    return str(results)


@analysis_agent.tool
async def get_user_preferences(
    ctx: RunContext[AgentDeps],
) -> str:
    """ユーザー設定の取得"""
    return str(ctx.deps.user.preferences)


# 7. Agentの実行
async def run_analysis(user_question: str) -> AnalysisReport:
    """分析Agentの実行"""
    deps = AgentDeps(
        db=DatabaseDep(connection_string="postgresql://localhost/mydb"),
        cache=CacheDep(backend="redis"),
        user=UserContext(
            user_id="u_001",
            username="田中",
            role="analyst",
            preferences={"detail_level": "verbose", "language": "ja"},
        ),
    )

    result = await analysis_agent.run(user_question, deps=deps)
    return result.data


# 実行例
if __name__ == "__main__":
    import asyncio

    report = asyncio.run(run_analysis("過去1ヶ月の売上データトレンドを分析して"))
    print(f"タイトル: {report.title}")
    print(f"発見: {report.findings}")
    print(f"信頼度: {report.confidence}")
    print(f"推奨: {report.recommendation}")

重要ポイント：

deps_typeでAgentの依存タイプを宣言、ランタイム時にRunContextに自動注入
@agent.system_promptデコレータは動的システムプロンプト生成をサポート、依存内容に基づいてカスタマイズ
ツール関数はctx: RunContext[AgentDeps]を通じて依存関係にアクセス、型安全でIDE補完あり

パターン3：ツール登録とFunction Calling

Function CallingはAI Agentと外部世界を繋ぐブリッジです。Pydantic AIはツール登録をデコレータ+型アノテーションに簡略化し、JSON Schemaを自動生成してLLMがPython関数を正確に呼び出せるようにします。

"""パターン3：ツール登録とFunction Calling"""
from __future__ import annotations

import json
from datetime import datetime, timezone
from typing import Optional

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext, Tool


# 1. 出力モデルの定義
class TravelPlan(BaseModel):
    """旅行計画結果"""

    destination: str
    start_date: str
    end_date: str
    estimated_cost: float
    itinerary: list[str]
    tips: list[str]


# 2. Agentの作成
travel_agent = Agent(
    "openai:gpt-4o",
    result_type=TravelPlan,
    system_prompt="あなたは旅行計画アシスタントです。ユーザーのニーズに基づいて旅行を計画し、提供されたツールでリアルタイム情報を照会してください。",
)


# 3. 基本ツール登録 - デコレータ方式
@travel_agent.tool
async def search_flights(
    ctx: RunContext[None],
    origin: str = Field(description="出発都市"),
    destination: str = Field(description="目的地都市"),
    date: str = Field(description="出発日、YYYY-MM-DD形式"),
) -> str:
    """フライト情報検索"""
    flights = [
        {"flight": "JL1234", "price": 68000, "departure": "08:00"},
        {"flight": "NH5678", "price": 55000, "departure": "14:30"},
        {"flight": "JL9012", "price": 82000, "departure": "19:00"},
    ]
    return json.dumps(flights, ensure_ascii=False)


@travel_agent.tool
async def get_weather(
    ctx: RunContext[None],
    city: str = Field(description="都市名"),
    date: Optional[str] = Field(default=None, description="日付、未指定の場合は今日"),
) -> str:
    """天気情報照会"""
    weather_data = {
        "city": city,
        "date": date or datetime.now(timezone.utc).strftime("%Y-%m-%d"),
        "temperature": "25°C",
        "condition": "晴れ時々曇り",
        "humidity": "65%",
    }
    return json.dumps(weather_data, ensure_ascii=False)


@travel_agent.tool
async def search_hotels(
    ctx: RunContext[None],
    city: str = Field(description="都市名"),
    check_in: str = Field(description="チェックイン日"),
    check_out: str = Field(description="チェックアウト日"),
    budget: Optional[float] = Field(default=None, description="予算上限（円/泊）"),
) -> str:
    """ホテル情報検索"""
    hotels = [
        {"name": "シティセンター大酒店", "price": 15000, "rating": 4.5, "location": "中心部"},
        {"name": "レイクビューリゾート", "price": 25000, "rating": 4.8, "location": "湖畔"},
        {"name": "ビジネスホテル", "price": 8000, "rating": 4.0, "location": "駅前"},
    ]
    if budget:
        hotels = [h for h in hotels if h["price"] <= budget]
    return json.dumps(hotels, ensure_ascii=False)


# 4. 高度なツール登録 - Toolクラス方式（より詳細な制御）
async def calculate_exchange_rate(
    amount: float,
    from_currency: str,
    to_currency: str,
) -> str:
    """通貨換算"""
    rates = {"JPY": 1.0, "USD": 0.0067, "EUR": 0.0062, "CNY": 0.048}
    if from_currency not in rates or to_currency not in rates:
        return f"未対応の通貨: {from_currency} -> {to_currency}"
    jpy_amount = amount / rates[from_currency]
    result = jpy_amount * rates[to_currency]
    return f"{amount} {from_currency} = {result:.2f} {to_currency}"


exchange_tool = Tool(
    name="calculate_exchange_rate",
    description="通貨換算ツール、JPY、USD、EUR、CNY対応",
    function=calculate_exchange_rate,
)

travel_agent.register_tool(exchange_tool)


# 5. 依存関係付きツール
@dataclass
class TravelDeps:
    user_id: str
    membership_level: str


travel_agent_with_deps = Agent(
    "openai:gpt-4o",
    result_type=TravelPlan,
    deps_type=TravelDeps,
    system_prompt="あなたは旅行計画アシスタントです。",
)


@travel_agent_with_deps.tool
async def get_member_discount(
    ctx: RunContext[TravelDeps],
) -> str:
    """会員割引情報の取得"""
    discounts = {
        "gold": "20%割引 + 無料アップグレードチャンス",
        "silver": "10%割引",
        "bronze": "5%割引",
        "regular": "割引なし",
    }
    level = ctx.deps.membership_level
    return f"会員ランク: {level}, 割引: {discounts.get(level, '割引なし')}"


# 6. ツール実行追跡
class ToolCallLogger:
    """ツール呼び出しロガー"""

    def __init__(self):
        self.calls: list[dict] = []

    def log(self, tool_name: str, args: dict, result: str) -> None:
        self.calls.append(
            {
                "tool": tool_name,
                "args": args,
                "result_preview": result[:100],
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
        )


# 7. Agentの実行
async def plan_travel(user_request: str) -> TravelPlan:
    """旅行計画"""
    result = await travel_agent.run(user_request)

    for msg in result.all_messages():
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            for tc in msg.tool_calls:
                print(f"ツール呼び出し: {tc.function.name}({tc.function.arguments})")

    return result.data


# 実行例
if __name__ == "__main__":
    import asyncio
    from dataclasses import dataclass

    plan = asyncio.run(plan_travel("来月東京から京都へ3日間旅行したい、予算10万円"))
    print(f"目的地: {plan.destination}")
    print(f"日程: {plan.start_date} ~ {plan.end_date}")
    print(f"予算: {plan.estimated_cost}円")
    print(f"旅程: {plan.itinerary}")
    print(f"ヒント: {plan.tips}")

重要ポイント：

@agent.toolデコレータは関数シグネチャと型アノテーションからJSON Schemaを自動生成
Field(description=...)でLLMにパラメータ説明を提供、Function Callingの精度を向上
Toolクラス方式はツール名や説明のカスタマイズなど、より詳細な制御を提供
ツール関数はasyncまたはsyncに対応、Pydantic AIが自動処理

パターン4：構造化出力と結果検証

構造化出力はPydantic AIのコアバリューです。LLMが返すJSONがモデル定義に一致しない場合、Pydanticが自動的に検証してエラーを報告し、リトライメカニズムと組み合わせることで最終出力が期待に合致することを保証します。

"""パターン4：構造化出力と結果検証"""
from __future__ import annotations

import json
from enum import Enum
from typing import Literal, Optional, Union

from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic_ai import Agent


# 1. 基本構造化出力
class SentimentType(str, Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"


class SentimentResult(BaseModel):
    """感情分析結果"""

    text: str = Field(description="分析対象テキスト")
    sentiment: SentimentType = Field(description="感情タイプ")
    confidence: float = Field(ge=0.0, le=1.0, description="信頼度")
    keywords: list[str] = Field(description="キーワードリスト")

    @field_validator("keywords")
    @classmethod
    def keywords_not_empty(cls, v: list[str]) -> list[str]:
        if not v:
            raise ValueError("キーワードリストを空にすることはできません")
        return v


# 2. ネストされた構造化出力
class Address(BaseModel):
    prefecture: str
    city: str
    district: str
    street: Optional[str] = None


class BusinessInfo(BaseModel):
    name: str
    category: str
    address: Address
    rating: float = Field(ge=0.0, le=5.0)
    price_range: str


class BusinessExtraction(BaseModel):
    """店舗情報抽出結果"""

    businesses: list[BusinessInfo] = Field(description="抽出された店舗リスト")
    total_count: int = Field(ge=0, description="店舗総数")
    source_text_summary: str = Field(description="原文サマリー")

    @model_validator(mode="after")
    def validate_count(self) -> "BusinessExtraction":
        if self.total_count != len(self.businesses):
            raise ValueError(
                f"total_count({self.total_count})と実際の店舗数({len(self.businesses)})が一致しません"
            )
        return self


# 3. Union型出力 - LLMに最適な出力タイプを選択させる
class TextSummary(BaseModel):
    type: Literal["summary"] = "summary"
    content: str
    word_count: int


class TextOutline(BaseModel):
    type: Literal["outline"] = "outline"
    sections: list[str]
    total_sections: int


class TextAnalysis(BaseModel):
    type: Literal["analysis"] = "analysis"
    key_points: list[str]
    tone: str
    audience: str


TextOutput = Union[TextSummary, TextOutline, TextAnalysis]


# 4. 複数の専門Agentを作成
sentiment_agent = Agent(
    "openai:gpt-4o",
    result_type=SentimentResult,
    system_prompt="あなたは感情分析専門家です。テキストの感情傾向を分析し、分類、信頼度、キーワードを提供してください。",
    retries=3,
)

extraction_agent = Agent(
    "openai:gpt-4o",
    result_type=BusinessExtraction,
    system_prompt="あなたは情報抽出専門家です。テキストから店舗情報を抽出し、名前、カテゴリ、住所、評価を含めてください。",
    retries=3,
)

flexible_agent = Agent(
    "openai:gpt-4o",
    result_type=TextOutput,
    system_prompt="あなたはテキスト分析アシスタントです。テキストの特徴に基づいて最適な分析方式を選択してください：要約(summary)、アウトライン(outline)、または詳細分析(analysis)。",
    retries=3,
)


# 5. 結果検証と後処理
class ValidatedResult(BaseModel):
    """検証ロジック付き結果ラッパー"""

    raw_result: BaseModel
    is_valid: bool = True
    validation_errors: list[str] = []

    @classmethod
    def from_agent_result(cls, result: BaseModel) -> "ValidatedResult":
        """Agent結果から検証結果を作成"""
        errors = []
        is_valid = True

        if hasattr(result, "confidence") and result.confidence < 0.5:
            errors.append("信頼度が低すぎます。結果が信頼できない可能性があります")
            is_valid = False

        if hasattr(result, "keywords") and len(result.keywords) < 2:
            errors.append("キーワードが少なすぎます。分析が不十分な可能性があります")
            is_valid = False

        return cls(
            raw_result=result,
            is_valid=is_valid,
            validation_errors=errors,
        )


# 6. バッチ処理と結果集約
async def batch_sentiment_analysis(texts: list[str]) -> list[ValidatedResult]:
    """バッチ感情分析"""
    results = []
    for text in texts:
        try:
            result = await sentiment_agent.run(text)
            validated = ValidatedResult.from_agent_result(result.data)
            results.append(validated)
        except Exception as e:
            error_result = ValidatedResult(
                raw_result=SentimentResult(
                    text=text,
                    sentiment=SentimentType.NEUTRAL,
                    confidence=0.0,
                    keywords=["error"],
                ),
                is_valid=False,
                validation_errors=[str(e)],
            )
            results.append(error_result)

    return results


# 7. 構造化出力とJSON Schemaエクスポート
def export_result_schema() -> dict:
    """結果モデルのJSON Schemaをエクスポート"""
    return SentimentResult.model_json_schema()


# 実行例
if __name__ == "__main__":
    import asyncio

    text = "この製品は本当に素晴らしい！パワフルな性能、美しいデザイン、コストパフォーマンス抜群！"
    result = asyncio.run(sentiment_agent.run(text))
    print(f"感情: {result.data.sentiment.value}")
    print(f"信頼度: {result.data.confidence}")
    print(f"キーワード: {result.data.keywords}")

    business_text = """
    一蘭ラーメン渋谷区神南1-22-7、ラーメン、評価4.5、1人1200円。
    すき家新宿区西新宿1-1-1、牛丼、評価3.8、1人500円。
    """
    extraction = asyncio.run(extraction_agent.run(business_text))
    print(f"店舗数: {extraction.data.total_count}")
    for b in extraction.data.businesses:
        print(f"  {b.name} - {b.category} - {b.rating}点")

    schema = export_result_schema()
    print(f"JSON Schema: {json.dumps(schema, indent=2, ensure_ascii=False)[:200]}...")

重要ポイント：

Pydanticのfield_validatorとmodel_validatorはLLM出力検証で重要な役割を果たす
Union型によりLLMが入力に基づいて最適な出力フォーマットを選択
retriesパラメータはバリデータと組み合わせて、検証失敗時にLLM呼び出しを自動リトライ
バッチ処理ではValidatedResultラッパーでエラー処理を統一

パターン5：ストリーミングレスポンスとマルチターン会話

ストリーミングレスポンスにより、ユーザーはAIの生成プロセスをリアルタイムで確認でき、インタラクション体験が大幅に向上します。Pydantic AIのストリーミングAPIは構造化出力と完全に互換性があり、マルチターン会話のコンテキスト管理もサポートします。

"""パターン5：ストリーミングレスポンスとマルチターン会話"""
from __future__ import annotations

from typing import Optional

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext


# 1. 出力モデルの定義
class ChatResponse(BaseModel):
    """チャットレスポンス"""

    content: str = Field(description="返信内容")
    mood: str = Field(default="neutral", description="ムード状態")
    follow_up_questions: list[str] = Field(
        default_factory=list, description="フォローアップ質問の提案"
    )


class CodeExplanation(BaseModel):
    """コード解説結果"""

    explanation: str
    complexity: str
    key_concepts: list[str]
    related_topics: list[str]


# 2. ストリーミングAgentの作成
stream_agent = Agent(
    "openai:gpt-4o",
    result_type=ChatResponse,
    system_prompt="あなたは親しみやすいAIアシスタントで、複雑な概念をシンプルな言葉で説明するのが得意です。",
)


# 3. ストリーミングテキスト出力（トークンごと）
async def stream_text_response(user_message: str) -> None:
    """ストリーミングテキストレスポンス出力"""
    print("ユーザー:", user_message)
    print("アシスタント: ", end="", flush=True)

    async with stream_agent.run_stream(user_message) as stream:
        async for token in stream.stream_text():
            print(token, end="", flush=True)

    print("\n")


# 4. ストリーミング構造化出力（フィールドごと）
async def stream_structured_response(user_message: str) -> ChatResponse:
    """ストリーミング構造化レスポンス出力"""
    async with stream_agent.run_stream(user_message) as stream:
        async for partial in stream.stream_structured():
            if partial.content:
                print(f"\r内容: {partial.content[:50]}...", end="", flush=True)

        result = await stream.get_result()
        return result.data


# 5. マルチターン会話管理
class ConversationManager:
    """マルチターン会話マネージャー"""

    def __init__(self):
        self.agent = Agent(
            "openai:gpt-4o",
            result_type=ChatResponse,
            system_prompt="あなたは親しみやすいAIアシスタントです。以前の会話内容を記憶し、コンテキストの一貫性を保ってください。",
        )
        self.message_history: list = []

    async def chat(self, user_input: str) -> ChatResponse:
        """1ターンの会話"""
        result = await self.agent.run(
            user_input,
            message_history=self.message_history,
        )
        self.message_history = result.all_messages()
        return result.data

    async def chat_stream(self, user_input: str) -> None:
        """ストリーミングマルチターン会話"""
        print(f"ユーザー: {user_input}")
        print("アシスタント: ", end="", flush=True)

        async with self.agent.run_stream(
            user_input,
            message_history=self.message_history,
        ) as stream:
            async for token in stream.stream_text():
                print(token, end="", flush=True)

        result = await stream.get_result()
        self.message_history = result.all_messages()
        print("\n")

    def reset(self) -> None:
        """会話をリセット"""
        self.message_history = []

    def get_history_summary(self) -> dict:
        """会話履歴サマリーの取得"""
        return {
            "total_messages": len(self.message_history),
            "user_messages": sum(
                1 for m in self.message_history if hasattr(m, "content") and m.role == "user"
            ),
        }


# 6. ツール付きストリーミングAgent
from pydantic_ai import Tool


class StreamingTravelAgent:
    """ストリーミング旅行アシスタント"""

    def __init__(self):
        self.agent = Agent(
            "openai:gpt-4o",
            result_type=ChatResponse,
            system_prompt="あなたは旅行アシスタントです。ツールを使用してリアルタイム情報を照会し、専門的なアドバイスを提供してください。",
        )
        self.message_history: list = []
        self._register_tools()

    def _register_tools(self) -> None:
        @self.agent.tool
        async def search_flights(
            ctx: RunContext[None],
            origin: str,
            destination: str,
        ) -> str:
            """フライト検索"""
            return f"{origin}から{destination}へのフライト3件: JL1234(¥68000), NH5678(¥55000), JL9012(¥82000)"

        @self.agent.tool
        async def get_weather(ctx: RunContext[None], city: str) -> str:
            """天気照会"""
            return f"{city}: 今日は晴れ時々曇り、25°C、旅行日和りです"

    async def chat_stream(self, user_input: str) -> ChatResponse:
        """ストリーミング会話"""
        async with self.agent.run_stream(
            user_input,
            message_history=self.message_history,
        ) as stream:
            async for token in stream.stream_text():
                print(token, end="", flush=True)

            result = await stream.get_result()
            self.message_history = result.all_messages()
            print()
            return result.data


# 7. ストリーミング出力と進捗追跡
class StreamProgress:
    """ストリーミング出力進捗トラッカー"""

    def __init__(self):
        self.token_count = 0
        self.tool_calls = 0
        self.start_time: Optional[float] = None

    def report(self) -> dict:
        """進捗レポートの生成"""
        import time

        elapsed = time.time() - self.start_time if self.start_time else 0
        return {
            "token_count": self.token_count,
            "tool_calls": self.tool_calls,
            "elapsed_seconds": round(elapsed, 2),
            "tokens_per_second": round(self.token_count / elapsed, 1) if elapsed > 0 else 0,
        }


# 実行例
if __name__ == "__main__":
    import asyncio

    # ストリーミングテキスト出力
    print("=== ストリーミングテキスト出力 ===")
    asyncio.run(stream_text_response("量子コンピューティングとは何ですか？"))

    # マルチターン会話
    print("\n=== マルチターン会話 ===")
    manager = ConversationManager()

    async def demo_conversation():
        await manager.chat_stream("Pythonを学びたいのですが、どこから始めればいいですか？")
        await manager.chat_stream("データ構造については？おすすめの学習パスはありますか？")
        await manager.chat_stream("クイックソートの例を教えてもらえますか？")
        print(f"会話統計: {manager.get_history_summary()}")

    asyncio.run(demo_conversation())

    # ストリーミング旅行アシスタント
    print("\n=== ストリーミング旅行アシスタント ===")
    travel = StreamingTravelAgent()
    asyncio.run(travel.chat_stream("東京から京都に行きたいのですが、天気はどうですか？"))

重要ポイント：

run_stream()は非同期コンテキストマネージャーを返し、stream_text()とstream_structured()の2つのモードをサポート
message_historyパラメータで履歴メッセージを渡し、マルチターン会話コンテキストを実現
ストリーミングモードではツール呼び出しが自動実行され、LLMはツール結果取得後に生成を継続
ConversationManagerは会話管理ロジックをカプセル化し、プロダクション環境に適しています

よくある落とし穴

落とし穴1：result_type未指定で出力が文字列になる

# ❌ 間違い：result_typeなし、出力は生の文字列
agent = Agent("openai:gpt-4o")
result = await agent.run("このコードを分析して")
print(type(result.data))  # <class 'str'>、型安全な操作が不可能

# ✅ 正しい：result_typeを指定、出力が自動検証されて構造化オブジェクトに
class AnalysisResult(BaseModel):
    score: float
    issues: list[str]

agent = Agent("openai:gpt-4o", result_type=AnalysisResult)
result = await agent.run("このコードを分析して")
print(type(result.data))  # <class 'AnalysisResult'>、型安全

落とし穴2：依存タイプの不一致でランタイムエラー

# ❌ 間違い：deps_typeと実際に渡す依存タイプが不一致
@dataclass
class DepsA:
    api_key: str

agent = Agent("openai:gpt-4o", deps_type=DepsA)

@dataclass
class DepsB:
    token: str  # 異なるフィールド

result = await agent.run("hello", deps=DepsB(token="xxx"))  # ランタイム型エラー

# ✅ 正しい：deps_typeと渡す依存タイプの一致を確保
result = await agent.run("hello", deps=DepsA(api_key="sk-xxx"))

落とし穴3：ツール関数に型アノテーションがない

# ❌ 間違い：ツール関数に型アノテーションがなく、LLMがパラメータの意味を理解できない
@agent.tool
async def search(query):  # 型アノテーションと説明が不足
    return "results"

# ✅ 正しい：完全な型アノテーションとField説明
@agent.tool
async def search(
    ctx: RunContext[None],
    query: str = Field(description="検索キーワード"),
    limit: int = Field(default=10, description="返却結果数の上限"),
) -> str:
    return "results"

落とし穴4：ストリーミングレスポンスで例外処理を忘れる

# ❌ 間違い：ストリーミングレスポンスに例外処理がなく、ネットワーク切断時にプログラムがクラッシュ
async with agent.run_stream(prompt) as stream:
    async for token in stream.stream_text():
        print(token, end="")

# ✅ 正しい：例外処理とタイムアウト制御を追加
import asyncio

try:
    async with asyncio.timeout(30):
        async with agent.run_stream(prompt) as stream:
            async for token in stream.stream_text():
                print(token, end="")
except asyncio.TimeoutError:
    print("\n[タイムアウト] レスポンスに時間がかかりすぎています。再試行してください")
except Exception as e:
    print(f"\n[エラー] {type(e).__name__}: {e}")

落とし穴5：message_historyの未渡しでコンテキストが失われる

# ❌ 間違い：毎回履歴を渡さないと、Agentは以前の会話を記憶できない
result1 = await agent.run("私の名前は田中です")
result2 = await agent.run("私の名前は何ですか？")  # Agentは田中という名前を知らない

# ✅ 正しい：message_historyを渡してコンテキストを維持
result1 = await agent.run("私の名前は田中です")
history = result1.all_messages()
result2 = await agent.run("私の名前は何ですか？", message_history=history)  # Agentは田中という名前を知っている

エラートラブルシューティング表

エラーメッセージ	原因	解決策
`ValidationError: field required`	LLM出力に必須フィールドが欠落	モデル定義を確認、フィールドにデフォルト値を設定するかretriesを増やす
`UnexpectedModelBehavior`	LLM出力をresult_typeとして解析できない	system_promptにフォーマット指示を追加、temperatureを下げる
`ModelHTTPError: 429`	API呼び出しレート制限超過	リクエスト間隔を追加、指数バックオフリトライを使用
`ToolRunError`	ツール関数実行例外	ツール関数ロジックを確認、try/exceptを追加
`RunContextError: deps not provided`	ランタイムで依存関係が未提供	run()呼び出し時にdepsパラメータを渡すことを確認
`StreamCompleteError`	ストリーミングレスポンスが中断	ネットワーク接続を確認、再接続ロジックを追加
`SchemaGenerationError`	result_typeにサポートされていない型が含まれる	複雑なネストされたUnionの使用を避け、モデル定義を簡素化
`TimeoutError`	LLMレスポンスタイムアウト	タイムアウト設定を増やす、プロンプトを最適化してトークン消費を削減
`AuthenticationError`	APIキーが無効または期限切れ	環境変数のAPIキー設定を確認
`JSONDecodeError in tool args`	LLMが生成したツール引数が有効なJSONではない	ツール説明にパラメータフォーマットを明記、例を追加

高度な最適化

モデルフォールバック戦略：Agentにフォールバックモデルを設定し、プライマリモデルが利用不可時に自動切替

from pydantic_ai.models import Model

agent = Agent(
    "openai:gpt-4o",
    result_type=AnalysisResult,
    model_settings={"fallback": "openai:gpt-4o-mini"},
)

ツール呼び出しキャッシュ：冪等なツールの呼び出し結果をキャッシュし、重複APIリクエストを削減

import hashlib
import json

tool_cache: dict[str, str] = {}

async def cached_tool_call(tool_name: str, args: dict) -> str:
    cache_key = hashlib.md5(
        json.dumps({"tool": tool_name, "args": args}, sort_keys=True).encode()
    ).hexdigest()
    if cache_key in tool_cache:
        return tool_cache[cache_key]
    result = await execute_tool(tool_name, args)
    tool_cache[cache_key] = result
    return result

構造化出力Schema最適化：Schema説明を簡素化し、LLMが誤ったフォーマットを生成する確率を低減

class OptimizedResult(BaseModel):
    score: float = Field(ge=0, le=10, description="0-10点")
    issues: list[str] = Field(max_length=5, description="最大5つの問題")
    category: str = Field(default="general", description="カテゴリ")

並行Agent呼び出し：asyncio.gatherを使用して複数のAgentタスクを並行実行

import asyncio

async def parallel_analysis(texts: list[str]) -> list[SentimentResult]:
    tasks = [sentiment_agent.run(text) for text in texts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [
        r.data if isinstance(r, AgentRunResult) else SentimentResult(text=t, sentiment=SentimentType.NEUTRAL, confidence=0, keywords=[])
        for r, t in zip(results, texts)
    ]

メッセージ履歴のトリミング：長すぎる会話履歴を要約・トリミングし、トークン消費を制御

def trim_message_history(
    messages: list,
    max_messages: int = 20,
) -> list:
    if len(messages) <= max_messages:
        return messages
    system_msgs = [m for m in messages if hasattr(m, "role") and m.role == "system"]
    recent_msgs = messages[-max_messages:]
    return system_msgs + recent_msgs

比較

特徴	生OpenAI SDK	LangChain	Pydantic AI
型安全性	❌ なし	⚠️ 部分	✅ 完全なPydantic検証
依存性注入	❌ 手動管理	⚠️ グローバル変数	✅ 型アノテーションによる自動注入
構造化出力	⚠️ 手動解析が必要	⚠️ OutputParserが必要	✅ 自動検証+リトライ
ツール登録	⚠️ JSON Schema手書き	⚠️ デコレータ+手動Schema	✅ 型アノテーションから自動生成
ストリーミングレスポンス	⚠️ 手動処理が必要	✅ サポート	✅ ネイティブサポート+構造化ストリーミング
学習曲線	低い	高い	中程度
コード量	多い	中程度	少ない
プロダクション対応	⚠️ 大量のラッピングが必要	⚠️ 過剰な抽象化	✅ すぐに使える

まとめ

Pydantic AIはPythonの型システムの力をAI Agent開発にもたらします。result_typeによる出力制約、deps_typeによる依存管理、@agent.toolによるツール登録、run_stream()によるストリーミングレスポンスという5つのコアパターンを通じて、開発者は型安全な方法でプロダクション級AIアプリケーションを構築できます。覚えておいてください：型アノテーションはドキュメントではなく、契約です——Pydantic AIはこの契約をLLMインタラクションにおいて厳格に実行します。

オンラインツール推奨

JSONフォーマッター - Agent出力のJSONデータをフォーマット・検証
cURL→コード変換 - APIデバッグ用cURLコマンドをPythonコードに変換
ハッシュ計算 - APIキーやデータのハッシュ値を計算、キャッシュキー生成に活用