Pydantic AI 進階工具與延遲執行

當你熟悉了基礎的 Function Tools 之後，你可能會面臨更複雜的實務情境。例如，你的專案可能包含數十個工具，需要有條理地管理；或者，某些工具涉及金流操作與修改資料庫，AI 不能直接自作主張執行，必須經過人類的審核批准。

Pydantic AI 針對這些進階需求，提供了 Toolsets (工具集) 以及 Deferred Tools (延遲執行工具) 來解決這些問題。

Toolsets 工具集：模組化管理工具

當你的系統變得龐大，如果將所有工具都寫在同一個檔案並用 @agent.tool 掛載，程式碼會變得非常難以維護。

Toolsets (工具集) 讓你可以將相關聯的工具分組打包。你可以建立一個獨立的 Toolset 實例，然後在裡面註冊相關的工具，最後再將整個 Toolset 傳入 Agent 的設定中。

from pydantic_ai import Agent
from pydantic_ai.toolsets import Toolset

# 建立一個負責處理電子郵件的工具集
email_tools = Toolset()

@email_tools.tool_plain
def read_latest_email(count: int = 5) -> str:
    """讀取最新幾封電子郵件的摘要"""
    return "您的最新信件..."

@email_tools.tool_plain
def send_email(to_address: str, subject: str, content: str) -> bool:
    """發送一封電子郵件"""
    return True

# 建立另一個處理行事曆的工具集
calendar_tools = Toolset()
# ... 註冊行事曆相關工具

# 在建立 Agent 時，將這些工具集組合進來
# 這樣做讓程式碼的職責分離非常清晰
agent = Agent(
    'openai:gpt-4o',
    toolsets=[email_tools, calendar_tools]
)

Human-in-the-Loop 與 Deferred Tools

在自動化流程中，如果 AI 呼叫的工具會造成不可逆的結果（例如：發送退款、刪除使用者帳號），我們通常不放心讓 AI 完全自主執行。這時候就需要 Human-in-the-Loop (人在迴圈內) 的設計。

Pydantic AI 提供了 Deferred Tools (延遲執行工具) 機制。你可以標記特定的工具需要人類同意後才能執行，或者將工具的執行交由外部系統處理。

處理延遲執行的兩種方式

當模型呼叫了一個延遲工具時，Pydantic AI 支援兩種處理流程：

Inline Resolution (內聯解析)：使用 HandleDeferredToolCalls Capability，在同一個 Python 行程中攔截並處理審核邏輯。
Stop-the-world Flow (中斷執行流)：中斷當前的 Agent 執行，回傳 DeferredToolRequests 給呼叫端，等待外部系統（如前端 UI）取得人類同意後，再發起一次新的 Agent 執行。

方式一：內聯解析 (HandleDeferredToolCalls)

如果你可以在同一個後端程式中處理審核邏輯（例如檢查使用者的權限等級），你可以使用 HandleDeferredToolCalls 來攔截請求：

from pydantic_ai import Agent, RunContext, ToolDenied
from pydantic_ai.tools import DeferredToolRequests, DeferredToolResults
from pydantic_ai.capabilities import HandleDeferredToolCalls

# 定義處理延遲請求的 Handler
async def handle_deferred(
    ctx: RunContext[None], requests: DeferredToolRequests
) -> DeferredToolResults:
    approvals = {}
    
    # 遍歷所有需要審核的工具呼叫
    for call in requests.approvals:
        if call.tool_name == 'delete_file':
            # 拒絕刪除檔案的操作
            approvals[call.tool_call_id] = ToolDenied('不允許刪除檔案')
        else:
            # 同意其他操作
            approvals[call.tool_call_id] = True

    # 回傳審核結果，Agent 會自動接續執行
    return requests.build_results(approvals=approvals)

# 將 Handler 作為 Capability 傳入
agent = Agent(
    'openai:gpt-4o',
    capabilities=[HandleDeferredToolCalls(handler=handle_deferred)],
)

# 標記此工具需要審批 (requires_approval=True)
@agent.tool_plain(requires_approval=True)
def delete_file(path: str) -> str:
    """刪除檔案"""
    # 因為在 handler 中被拒絕，這段程式碼永遠不會被執行
    return f'檔案 {path!r} 已刪除'

方式二：中斷執行流 (Stop-the-world)

如果審核需要等待真實人類在網頁上點擊按鈕，你就必須中斷 Agent 的執行，將請求傳給前端，等待回應後再繼續。

from typing import Union
from pydantic_ai import Agent, ToolApproved, ToolDenied
from pydantic_ai.tools import DeferredToolRequests, DeferredToolResults

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

@agent.tool_plain(requires_approval=True)
def delete_database_record(record_id: str) -> str:
    """從資料庫中刪除一筆關鍵紀錄。"""
    return f"紀錄 {record_id} 已成功刪除。"

# 1. 執行時，將 result_type 設為允許回傳 DeferredToolRequests
result = agent.run_sync(
    "請幫我刪除紀錄 12345", 
    result_type=Union[str, DeferredToolRequests]
)

# 2. 檢查回傳的結果資料是否為「延遲的工具請求」
if isinstance(result.data, DeferredToolRequests):
    # 代表 AI 決定執行 delete_database_record，但被暫停了
    pending_calls = result.data.approvals
    tool_call_id = list(pending_calls.keys())[0]
    
    print("等待人類審核中... (假設前端傳回同意)")
    
    # 3. 使用者同意後，建立 DeferredToolResults
    deferred_results = DeferredToolResults(
        approvals={tool_call_id: ToolApproved()}
    )
    
    # 4. 再次呼叫 Agent，將之前的對話紀錄與審核結果傳進去
    final_result = agent.run_sync(
        "使用者已審核",
        message_history=result.all_messages(),
        deferred_tool_results=deferred_results
    )
    print("最終結果：", final_result.data)
else:
    print("AI 回覆：", result.data)

除了 requires_approval=True 之外，你也可以在工具函數內主動拋出 ApprovalRequired 或 CallDeferred 異常，來動態觸發延遲執行的流程。透過 Toolsets 和 Deferred Tools 機制，Pydantic AI 具備了開發企業級、高安全性、並且融入人類決策流程的複雜系統能力。