Pydantic AI Capabilities 與 Hooks

在開發大型且複雜的 AI 應用程式時，我們常常需要跨越多個不同的代理人 (Agent) 來重複使用相同的工具、系統指令或設定檔，或者需要在程式執行的特定階段插入自訂邏輯。為了提升程式碼的可維護性與重用性，Pydantic AI 提供了 Capabilities（能力套件） 與 Hooks（生命週期掛鉤） 的強大機制。

Capabilities：模組化的能力套件

在 Pydantic AI 中，Capability (能力套件) 是一個可重複使用、可組合的代理人行為單元。與其將指令 (instructions)、模型設定 (model settings)、工具 (tools) 散落在各處，你可以將相關的行為打包成一個 Capability，並透過 Agent 的 capabilities 參數傳入。

Capabilities 可以包含：

工具 (Tools)
生命週期掛鉤 (Lifecycle hooks)
動態或靜態的系統指令 (Instructions)
模型設定 (Model settings)

內建的 Capabilities

Pydantic AI 提供了許多實用的內建 Capabilities，涵蓋了常見的應用需求：

Thinking：啟用模型的思考與推理能力（例如 Claude 3.7 Sonnet 的 thinking mode）。
WebSearch / WebFetch：提供網頁搜尋與抓取內容的能力（支援內建工具或本地降級處理）。
ImageGeneration：提供圖片生成的能力。
MCP：連接 MCP (Model Context Protocol) 伺服器，擴充外部資源存取。
Hooks：提供基於裝飾器 (decorator) 的生命週期掛鉤註冊。

使用內建 Capabilities 範例

你可以像堆積木一樣，把不同的能力組合在一起，賦予給 Agent。以下範例展示了如何同時賦予 Agent「深度思考」與「網路搜尋」的能力：

from pydantic_ai import Agent
from pydantic_ai.capabilities import Thinking, WebSearch

# 建立一個具備「深度思考」與「網路搜尋」能力的 Agent
agent = Agent(
    'anthropic:claude-3-7-sonnet-latest',
    instructions='你是一個專業的研究助理。請詳細思考並附上資料來源。',
    capabilities=[
        # 啟用高強度的思考模式
        Thinking(effort='high'),
        # 賦予網路搜尋能力
        WebSearch(),
    ],
)

# 執行 Agent 進行查詢
result = agent.run_sync('請分析 2026 年最新的 AI 發展趨勢？')
print(result.output)

透過這種設計，你可以將與核心業務邏輯無關的監控、日誌記錄或是特定領域的工具，抽離成獨立的套件供團隊內部共用。

Hooks：生命週期攔截器

在軟體工程中，Hooks 是一種常見的模式，允許開發者在程式執行的特定階段介入。Pydantic AI 提供了專門的 Hooks Capability，讓你可以攔截並修改 Agent 在各個階段的行為（例如：發送請求給模型前、執行工具前、處理串流事件時等），而不需要繼承複雜的類別。

攔截模型請求

你可以使用 @hooks.on.* 裝飾器來註冊生命週期事件。例如，我們可以在發送請求給模型之前，攔截請求並印出日誌，這對於除錯與監控非常有用：

from pydantic_ai import Agent, ModelRequestContext, RunContext
from pydantic_ai.capabilities import Hooks

# 建立 Hooks 實例
hooks = Hooks()

# 註冊 Hook：在發送請求給模型之前觸發
@hooks.on.before_model_request
async def log_request(ctx: RunContext[None], request_context: ModelRequestContext) -> ModelRequestContext:
    # 印出即將發送的訊息數量
    print(f'準備發送 {len(request_context.messages)} 則訊息給模型...')
    
    # 必須回傳 request_context，你也可以在這裡修改它
    return request_context

# 將 hooks 作為 capability 傳入 Agent
agent = Agent('openai:gpt-4o', capabilities=[hooks])
result = agent.run_sync('你好！')

結果驗證與重試機制

Hooks 也可以用來驗證模型的輸出。如果不符合商業邏輯，可以拋出 ModelRetry 異常，Pydantic AI 會自動帶著錯誤訊息要求模型重新思考並修正錯誤：

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import Hooks
from pydantic_ai.exceptions import ModelRetry
from pydantic_ai.messages import ModelResponse
from pydantic_ai.models import ModelRequestContext

hooks = Hooks()

# 註冊 Hook：在模型回傳結果之後觸發
@hooks.on.after_model_request
async def check_response(
    ctx: RunContext[None],
    *,
    request_context: ModelRequestContext,
    response: ModelResponse,
) -> ModelResponse:
    # 檢查模型的回覆內容是否包含敏感字眼
    if "密碼" in str(response.parts) or "password" in str(response.parts).lower():
        # 拋出 ModelRetry，要求模型重試
        raise ModelRetry('請注意安全規範，絕對不要在回覆中包含任何密碼資訊。')
        
    # 驗證通過，回傳原始 response
    return response

agent = Agent('gemini-1.5-flash', capabilities=[hooks])

Agent 層級的快捷裝飾器

除了使用 Hooks Capability 來進行深度的生命週期攔截外，Pydantic AI 也直接在 Agent 實例上提供了幾個常用的快捷裝飾器，適合用於簡單的動態提示或最終結果驗證。

動態系統提示

我們可以在模型開始生成回答「之前」，根據當下的執行環境或時間，動態地調整系統提示。這透過 @agent.system_prompt 裝飾器來達成：

from pydantic_ai import Agent, RunContext
import datetime

agent = Agent('gemini-1.5-flash')

# 每次執行前都會觸發這個 Hook，將回傳的字串加入到系統提示中
@agent.system_prompt
def inject_current_time(ctx: RunContext) -> str:
    # 取得當前日期
    now = datetime.datetime.now().strftime("%Y-%m-%d")
    return f"今天的日期是 {now}。在回答結尾請加上一句鼓勵的話。"

最終結果驗證

如果你只需要針對 Agent 的「最終輸出結果」進行驗證（特別是當你有定義結構化輸出 result_type 時），使用 @agent.result_validator 會比 Hooks Capability 更簡單直覺：

from pydantic_ai import Agent, RunContext, ModelRetry

agent = Agent('gemini-1.5-flash')

# 模型生成最終結果後會觸發
@agent.result_validator
def validate_no_secrets(ctx: RunContext, result: str) -> str:
    # 檢查最終結果字串
    if "密碼" in result:
        # 觸發重試機制
        raise ModelRetry('請注意安全規範，不要在回覆中包含任何密碼資訊。')
    return result

透過巧妙運用 Capabilities 與 Hooks，你能大幅提升 Agent 程式碼的可維護性，並建立出具備強大防錯機制的企業級 AI 應用。