PydanticAI 依赖注入

核心模式

依赖通过 RunContext 流转：

python
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class Deps:
db: DatabaseConn
api_client: HttpClient
user_id: int

agent = Agent(
openai:gpt-4o,
deps_type=Deps, # 用于静态分析的类型
)

@agent.tool
async def getuserbalance(ctx: RunContext[Deps]) -> float:
获取当前用户的账户余额。
return await ctx.deps.db.getbalance(ctx.deps.userid)

运行时提供依赖

result = await agent.run( 我的余额是多少？, deps=Deps(db=dbconn, apiclient=client, user_id=123) )

定义依赖

使用数据类或 Pydantic 模型：

python
from dataclasses import dataclass
from pydantic import BaseModel

数据类（为简单起见推荐使用）

@dataclass class Deps: db: DatabaseConnection cache: CacheClient user_context: UserContext

Pydantic 模型（如果需要验证）

class Deps(BaseModel): api_key: str endpoint: str timeout: int = 30

访问依赖

在工具和指令中：

python
@agent.tool
async def query_database(ctx: RunContext[Deps], query: str) -> list[dict]:
运行数据库查询。
return await ctx.deps.db.execute(query)

@agent.instructions
async def addusercontext(ctx: RunContext[Deps]) -> str:
user = await ctx.deps.db.getuser(ctx.deps.userid)
return f用户名：{user.name}，角色：{user.role}

@agent.system_prompt
def add_permissions(ctx: RunContext[Deps]) -> str:
return f用户拥有权限：{ctx.deps.permissions}

类型安全

使用泛型实现完整类型检查：

python

显式代理类型注解

agent: Agent[Deps, OutputModel] = Agent(
openai:gpt-4o,
deps_type=Deps,
output_type=OutputModel,
)

现在这些都会进行类型检查：

- 工具中的 ctx.deps 类型为 Deps

- result.output 类型为 OutputModel

- agent.run() 需要 deps: Deps 参数

无依赖模式

当不需要依赖时：

python

选项1：不设置 deps_type（默认为 NoneType）

agent = Agent(openai:gpt-4o)
result = agent.run_sync(你好) # 无需依赖

选项2：为类型检查器显式设置为 None

agent: Agent[None, str] = Agent(openai:gpt-4o) result = agent.run_sync(你好, deps=None)

在 tool_plain 中，无法访问上下文

@agent.tool_plain def simple_calc(a: int, b: int) -> int: return a + b

完整示例

python
from dataclasses import dataclass
from httpx import AsyncClient
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

@dataclass
class WeatherDeps:
client: AsyncClient
api_key: str

class WeatherReport(BaseModel):
location: str
temperature: float
conditions: str

agent: Agent[WeatherDeps, WeatherReport] = Agent(
openai:gpt-4o,
deps_type=WeatherDeps,
output_type=WeatherReport,
instructions=你是一个天气助手。,
)

@agent.tool
async def get_weather(
ctx: RunContext[WeatherDeps],
city: str
) -> dict:
获取指定城市的天气数据。
response = await ctx.deps.client.get(
fhttps://api.weather.com/{city},
headers={Authorization: ctx.deps.api_key}
)
return response.json()

async def main():
async with AsyncClient() as client:
deps = WeatherDeps(client=client, api_key=secret)
result = await agent.run(伦敦的天气怎么样？, deps=deps)
print(result.output.temperature)

测试覆盖

python
from pydantic_ai.models.test import TestModel

创建模拟依赖

mock_deps = Deps( db=MockDatabase(), api_client=MockClient(), user_id=999 )

覆盖模型和依赖以进行测试

with agent.override(model=TestModel(), deps=mock_deps): result = agent.run_sync(测试提示)

最佳实践

1. 1. 保持依赖不可变：使用冻结数据类或 Pydantic 模型
2. 传递连接，而非凭据：依赖应持有已初始化的客户端
3. 为代理添加类型注解：使用 Agent[DepsType, OutputType] 实现完整类型安全
4. 合理限定依赖范围：在请求开始时创建依赖，使用后关闭