pydantic-ai-dependency-injection

PydanticAI 依赖注入

核心模式

依赖通过 RunContext 传递：

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,  # Type for static analysis
)

@agent.tool
async def get_user_balance(ctx: RunContext[Deps]) -> float:
    """Get the current user's account balance."""
    return await ctx.deps.db.get_balance(ctx.deps.user_id)

# At runtime, provide deps
result = await agent.run(
    'What is my balance?',
    deps=Deps(db=db_conn, api_client=client, user_id=123)
)

定义依赖

使用数据类或 Pydantic 模型：

from dataclasses import dataclass
from pydantic import BaseModel

# Dataclass (recommended for simplicity)
@dataclass
class Deps:
    db: DatabaseConnection
    cache: CacheClient
    user_context: UserContext

# Pydantic model (if you need validation)
class Deps(BaseModel):
    api_key: str
    endpoint: str
    timeout: int = 30

访问依赖

在工具和指令中：

@agent.tool
async def query_database(ctx: RunContext[Deps], query: str) -> list[dict]:
    """Run a database query."""
    return await ctx.deps.db.execute(query)

@agent.instructions
async def add_user_context(ctx: RunContext[Deps]) -> str:
    user = await ctx.deps.db.get_user(ctx.deps.user_id)
    return f"User name: {user.name}, Role: {user.role}"

@agent.system_prompt
def add_permissions(ctx: RunContext[Deps]) -> str:
    return f"User has permissions: {ctx.deps.permissions}"

类型安全

使用泛型进行完整的类型检查：

# Explicit agent type annotation
agent: Agent[Deps, OutputModel] = Agent(
    'openai:gpt-4o',
    deps_type=Deps,
    output_type=OutputModel,
)

# Now these are type-checked:
# - ctx.deps in tools is typed as Deps
# - result.output is typed as OutputModel
# - agent.run() requires deps: Deps

无依赖模式

不需要依赖时：

# Option 1: No deps_type (defaults to NoneType)
agent = Agent('openai:gpt-4o')
result = agent.run_sync('Hello')  # No deps needed

# Option 2: Explicit None for type checker
agent: Agent[None, str] = Agent('openai:gpt-4o')
result = agent.run_sync('Hello', deps=None)

# In tool_plain, no context access
@agent.tool_plain
def simple_calc(a: int, b: int) -> int:
    return a + b

完整示例

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='You are a weather assistant.',
)

@agent.tool
async def get_weather(
    ctx: RunContext[WeatherDeps],
    city: str
) -> dict:
    """Fetch weather data for a city."""
    response = await ctx.deps.client.get(
        f'https://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('Weather in London?', deps=deps)
        print(result.output.temperature)

测试替换

from pydantic_ai.models.test import TestModel

# Create mock dependencies
mock_deps = Deps(
    db=MockDatabase(),
    api_client=MockClient(),
    user_id=999
)

# Override model and deps for testing
with agent.override(model=TestModel(), deps=mock_deps):
    result = agent.run_sync('Test prompt')

检查关卡

在认定 Agent 正确之前，按顺序执行以下检查；每一步都有明确的通过条件。

1. 依赖涵盖所有访问 — 收集工具、@agent.instructions 和 @agent.system_prompt 中的所有 ctx.deps.<attr>（及嵌套用法）。通过条件： 每个 <attr> 都存在于 deps_type 上（如果使用 mypy/pyright 检查 Agent[DepsType, …]，静态检查也须通过）。
2. 每次需要依赖的运行都传入依赖 — 通过条件： 每条会执行这些工具的 agent.run / run_sync 路径都传入类型与 deps_type 匹配的 deps=（除非 Agent 确实没有依赖，否则不得为 None）。
3. 测试固定依赖结构 — 通过条件： 使用 agent.override 的测试传入的 deps= 值，其字段和类型与生产环境中的 Deps 相同（除非被测工具不会访问缺失字段，否则不得使用不完整的模拟对象）。

最佳实践

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