Pydantic AI 依賴注入 (Dependency Injection)
在開發實際可用的 AI 代理人時,代理人往往需要與外界系統互動。例如:查詢資料庫中的使用者設定、呼叫外部 API 獲取即時天氣、或是讀取當下使用者的認證資訊 (Token)。
為了讓代理人能在執行時獲取這些上下文資料,Pydantic AI 提供了一個強大且型別安全的設計模式——依賴注入 (Dependencies Injection)。
什麼是 Dependencies?
簡單來說,Dependencies (依賴項) 就是 Agent 執行任務所需要的「外部資料」或「工具物件」。
在 Pydantic AI 中,你可以定義一個自訂的資料結構或類別來作為依賴項,並在 Agent 執行 (run, run_sync) 時將其傳入。接著,Agent 內部的動態系統提示 (System Prompts) 或自訂工具 (Tools) 就可以透過 RunContext 提取並使用這些依賴項。
型別安全的設計
Pydantic AI 是透過 Python 的泛型 (Generics) 機制來確保依賴項的型別安全。當你在宣告 Agent 時,可以指定依賴項的型別(即 deps_type)。
這樣做的好處是:IDE 會知道 RunContext.deps 到底是什麼型別,進而提供完整的自動完成,並且靜態檢查工具(如 mypy 或 Pyright)也能提早抓出型別不符的錯誤。
基本範例:注入資料結構
讓我們看一個具體的範例。假設我們有一個代理人,職責是提供個人的財務建議。這個代理人需要知道使用者的帳戶餘額與名字。
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
# 1. 定義一個資料類別,用來裝載我們要注入的依賴項
@dataclass
class UserContext:
name: str
account_balance: float
# 2. 建立 Agent,並透過 deps_type 參數宣告預期的依賴項型別
agent = Agent(
'gemini-1.5-flash',
deps_type=UserContext, # 告訴框架:這個代理人需要 UserContext 作為依賴
)
# 3. 透過動態系統提示,在執行時提取依賴資料並加入提示中
@agent.system_prompt
def inject_user_context(ctx: RunContext[UserContext]) -> str:
# IDE 能明確知道 ctx.deps 是一個 UserContext 物件,提供屬性提示
return (
f"這是一位名為 {ctx.deps.name} 的客戶。"
f"他目前的帳戶餘額為 ${ctx.deps.account_balance} 萬元。"
f"請根據他的餘額提供適當且客氣的理財建議。"
)
# 4. 在執行 run_sync 時,透過 deps 參數將實際的資料實例傳入
# 如果此處傳入錯誤型別,型別檢查工具會報錯
current_user = UserContext(name='王大明', account_balance=150.5)
result = agent.run_sync('請問我該投資什麼比較好?', deps=current_user)
print(result.data)
在工具 (Tools) 中使用 Dependencies
依賴注入更強大的應用場景在於與 Function Tools(函數工具) 結合。
當你為 Agent 賦予工具,讓它能夠主動查詢外部資訊時,這些工具函數通常需要依賴資料庫連線或是 HTTP Client 物件。你可以將這些物件定義為依賴項,讓工具函數透過 RunContext 獲取它們。
import httpx
from pydantic_ai import Agent, RunContext, tool
# 建立 Agent,指定其依賴為 httpx 的異步客戶端
agent = Agent('openai:gpt-4o', deps_type=httpx.AsyncClient)
@agent.system_prompt
def base_prompt() -> str:
return "你是一個可以查詢即時天氣的助手。如果使用者詢問天氣,請使用提供的工具。"
# 定義一個工具供 Agent 使用,這工具需要發送網路請求
@agent.tool
async def get_weather(ctx: RunContext[httpx.AsyncClient], city_name: str) -> str:
# 透過 ctx.deps 提取注入的 AsyncClient 實例來發送 API 請求
client = ctx.deps
response = await client.get(f'https://wttr.in/{city_name}?format=3')
# 這裡將取得的天氣字串回傳給 Agent
return response.text
async def main():
# 在應用程式最外層建立 AsyncClient,並透過 deps 注入給 Agent
async with httpx.AsyncClient() as client:
result = await agent.run('請查一下台北現在的天氣', deps=client)
print(result.data)
import asyncio
asyncio.run(main())
在這個範例中,代理人(以及它的工具)本身並不負責管理 HTTP Client 的生命週期。HTTP Client 的建立與關閉是在最外層的主程式完成,並作為 Dependency 被傳遞進去的。這正是依賴反轉原則(Dependency Inversion Principle)的最佳實踐,讓你的程式碼更容易測試與維護。
透過 Pydantic AI 的 RunContext 與依賴注入機制,你能以乾淨、安全的方式將外部世界與 AI 代理人連結起來。
如何傳遞多個 Dependencies?
有時候你的 Agent 會同時需要多種外部依賴。例如:一個代理人可能同時需要存取資料庫連線、HTTP Client 以及當下的使用者資訊。
在 Pydantic AI 中,deps_type 與傳入的 deps 參數 只能接受單一個物件。但你不用擔心,我們可以透過將不同的依賴項「打包」到一個單一的資料結構中,來達成傳遞多個依賴項的目的。
常見的最佳實踐是使用 dataclass、pydantic.BaseModel 或是 typing.TypedDict 來組合這些依賴項:
from dataclasses import dataclass
import httpx
# 假設使用 asyncpg 作為資料庫連線套件
import asyncpg
from pydantic_ai import Agent, RunContext
# 1. 建立一個包含所有需要的依賴項的資料類別 (Data Class)
@dataclass
class AppDependencies:
http_client: httpx.AsyncClient
db_pool: asyncpg.Pool
user_id: int
# 2. 將這個組合後的類別指定為 deps_type
agent = Agent(
'gemini-1.5-flash',
deps_type=AppDependencies,
)
@agent.tool
async def get_user_data(ctx: RunContext[AppDependencies]) -> str:
# 3. 在工具中,透過 ctx.deps 分別提取並使用對應的依賴項
db = ctx.deps.db_pool
user_id = ctx.deps.user_id
# 實際應用中會使用 db 查詢資料庫...
return f"為使用者 {user_id} 獲取的資料"
@agent.tool
async def fetch_external_api(ctx: RunContext[AppDependencies], query: str) -> str:
# 同樣地,提取 HTTP Client 來發送請求
client = ctx.deps.http_client
# response = await client.get(...)
return "API 回應內容"
透過這種打包的方式,你不僅能依據需求傳入任意數量的依賴項,同時也完美保留了框架主打的靜態型別檢查與編輯器 (IDE) 的自動完成提示功能。