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`	流式響應結果，支援逐token輸出	即時互動體驗

問題分析：為什麼傳統AI開發不夠好？

類型缺失：LLM回傳原始字串，需要手動解析和驗證，執行時錯誤頻發
依賴混亂：資料庫連線、API金鑰等依賴在函式間層層傳遞，耦合嚴重
工具註冊繁瑣：Function Calling的JSON Schema手寫易錯，與程式碼邏輯脫節
輸出不可控：LLM可能回傳不符合預期的格式，缺乏編譯時保障
流式處理複雜：SSE/WebSocket流式響應的解析和錯誤處理程式碼冗長

模式一：Agent基礎定義與類型安全模型

Pydantic AI的Agent定義將LLM互動抽象為類型安全的函式呼叫。透過Python類型註解，你在撰寫程式碼時就定義了Agent的輸入輸出契約。

"""模式一：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確保結構化輸出穩定

模式二：依賴注入與系統提示

生產級Agent需要存取資料庫、快取、設定等外部依賴。Pydantic AI的依賴注入系統讓這些依賴透過類型註解自動注入，徹底消除全域變數和參數傳遞的混亂。

"""模式二：依賴注入與系統提示"""
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": "zh-TW"},
        ),
    )

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


# 運行範例
if __name__ == "__main__":
    import asyncio

    report = asyncio.run(run_analysis("分析最近一個月的銷售資料趨勢"))
    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補全

模式三：工具註冊與Function Calling

Function Calling是AI Agent連接外部世界的橋樑。Pydantic AI將工具註冊簡化為裝飾器+類型註解，自動生成JSON Schema，讓LLM能精確呼叫你的Python函式。

"""模式三：工具註冊與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": "CA1234", "price": 1280, "departure": "08:00"},
        {"flight": "MU5678", "price": 980, "departure": "14:30"},
        {"flight": "CZ9012", "price": 1580, "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": 580, "rating": 4.5, "location": "市中心"},
        {"name": "湖景度假飯店", "price": 880, "rating": 4.8, "location": "湖畔"},
        {"name": "經濟型連鎖旅館", "price": 280, "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 = {"CNY": 1.0, "USD": 0.14, "EUR": 0.13, "JPY": 21.5}
    if from_currency not in rates or to_currency not in rates:
        return f"不支援的貨幣: {from_currency} -> {to_currency}"
    cny_amount = amount / rates[from_currency]
    result = cny_amount * rates[to_currency]
    return f"{amount} {from_currency} = {result:.2f} {to_currency}"


exchange_tool = Tool(
    name="calculate_exchange_rate",
    description="貨幣換算工具，支援CNY、USD、EUR、JPY",
    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": "8折優惠 + 免費升等機會",
        "silver": "9折優惠",
        "bronze": "95折優惠",
        "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("我想下個月從台北去東京玩5天，預算1萬元"))
    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自動處理

模式四：結構化輸出與結果驗證

結構化輸出是Pydantic AI的核心價值。當LLM回傳的JSON不符合模型定義時，Pydantic會自動驗證並報錯，配合重試機制確保最終輸出符合預期。

"""模式四：結構化輸出與結果驗證"""
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):
    province: 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. 聯合類型輸出 - 讓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 = """
    味千拉麵位於台北市信義區松壽路19號，日式拉麵，評分4.2，人均80元。
    海底撈火鍋在台北市大安區敦化南路1段1號，火鍋，評分4.8，人均150元。
    """
    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包裝器統一錯誤處理

模式五：流式響應與多輪對話

流式響應讓使用者即時看到AI的生成過程，大幅提升互動體驗。Pydantic AI的流式API與結構化輸出完美相容，同時支援多輪對話的上下文管理。

"""模式五：流式響應與多輪對話"""
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. 流式文本輸出（逐token）
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:
        """進行一輪對話"""
        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"找到3個從{origin}到{destination}的航班: CA1234(¥1280), MU5678(¥980), CZ9012(¥1580)"

        @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()兩種模式
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回應超時	增加timeout設定，最佳化prompt減少token消耗
`AuthenticationError`	API金鑰無效或過期	檢查環境變數中的API金鑰設定
`JSONDecodeError in tool args`	LLM生成的工具參數不是有效JSON	在工具描述中明確參數格式，增加範例

進階最佳化

模型降級策略：為Agent設定fallback模型，當主模型不可用時自動切換

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)
    ]

訊息歷史裁剪：對過長的對話歷史進行摘要裁剪，控制token消耗

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金鑰或資料的雜湊值，用於快取鍵生成