pydantic-ai-dependency-injection

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

PydanticAI Dependency Injection

PydanticAI 依赖注入

Core Pattern

核心模式

Dependencies flow through

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,  # 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)

依赖项通过

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,  # 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) )

undefined

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

undefined

Defining Dependencies

定义依赖项

Use dataclasses or Pydantic models:

python

from dataclasses import dataclass
from pydantic import BaseModel

使用数据类或Pydantic模型：

python

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

undefined

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

undefined

Accessing Dependencies

访问依赖项

In tools and instructions:

python

@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}"

在工具和指令中：

python

@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}"

Type Safety

类型安全

Full type checking with generics:

python

undefined

通过泛型实现完整的类型检查：

python

undefined

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

undefined

undefined

No Dependencies Pattern

无依赖项模式

When you don't need dependencies:

python

undefined

当你不需要依赖项时：

python

undefined

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

undefined

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

undefined

Complete Example

完整示例

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='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)

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='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)

Override for Testing

测试用重写

python

from pydantic_ai.models.test import TestModel

python

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')

undefined

with agent.override(model=TestModel(), deps=mock_deps): result = agent.run_sync('Test prompt')

undefined

Best Practices

最佳实践

Keep deps immutable: Use frozen dataclasses or Pydantic models
Pass connections, not credentials: Deps should hold initialized clients
Type your agents: Use
```
Agent[DepsType, OutputType]
```
for full type safety
Scope deps appropriately: Create deps at the start of a request, close after

保持依赖项不可变：使用冻结数据类或Pydantic模型
传递连接而非凭证：依赖项应包含已初始化的客户端
为Agent添加类型注解：使用
```
Agent[DepsType, OutputType]
```
实现完整的类型安全
合理划分依赖项作用域：在请求开始时创建依赖项，请求结束后关闭