依赖项

PydanticAI 使用依赖注入系统为您的代理的 系统提示、工具 和 结果验证器 提供数据和服务。

与 PydanticAI 的设计理念相符，我们的依赖系统尝试使用 Python 开发中现有的最佳实践，而不是发明深奥的“魔法”，这应该使依赖项类型安全、易于理解、更易于测试，并最终更易于在生产环境中部署。

定义依赖项

依赖项可以是任何 Python 类型。虽然在简单情况下，您可能能够将单个对象作为依赖项传递（例如，HTTP 连接），但当您的依赖项包含多个对象时，数据类 通常是一个方便的容器。

这是一个定义需要依赖项的代理的示例。

（注意：此示例中实际上未使用依赖项，请参阅下面的 访问依赖项）

unused_dependencies.py

from dataclasses import dataclass

import httpx

from pydantic_ai import Agent


@dataclass
class MyDeps:  # (1)!
    api_key: str
    http_client: httpx.AsyncClient


agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,  # (2)!
)


async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run(
            'Tell me a joke.',
            deps=deps,  # (3)!
        )
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

1. 定义一个数据类来保存依赖项。
2. 将数据类类型传递给 Agent 构造函数 的 deps_type 参数。注意：我们在此处传递的是类型，而不是实例，此参数实际上在运行时未使用，它在这里是为了我们可以获得代理的完整类型检查。
3. 运行代理时，将数据类的实例传递给 deps 参数。

（此示例是完整的，可以直接“按原样”运行 — 您需要添加 asyncio.run(main()) 来运行 main）

访问依赖项

通过 RunContext 类型访问依赖项，这应该是系统提示函数等的第一个参数。

system_prompt_dependencies.py

from dataclasses import dataclass

import httpx

from pydantic_ai import Agent, RunContext


@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient


agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)


@agent.system_prompt  # (1)!
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:  # (2)!
    response = await ctx.deps.http_client.get(  # (3)!
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},  # (4)!
    )
    response.raise_for_status()
    return f'Prompt: {response.text}'


async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

1. RunContext 可以选择作为唯一参数传递给 system_prompt 函数。
2. RunContext 使用依赖项的类型进行参数化，如果此类型不正确，静态类型检查器将引发错误。
3. 通过 .deps 属性访问依赖项。
4. 通过 .deps 属性访问依赖项。

（此示例是完整的，可以直接“按原样”运行 — 您需要添加 asyncio.run(main()) 来运行 main）

异步与同步依赖项

系统提示函数、函数工具 和 结果验证器 都在代理运行的异步上下文中运行。

如果这些函数不是协程（例如 async def），则会使用 run_in_executor 在线程池中调用它们，因此，如果依赖项执行 IO 操作，则最好使用 async 方法，尽管同步依赖项也可以正常工作。

run 与 run_sync 以及异步与同步依赖项

无论您使用同步还是异步依赖项，都与您是否使用 run 或 run_sync 完全无关 — run_sync 只是 run 的包装器，代理始终在异步上下文中运行。

这是与上面相同的示例，但使用了同步依赖项

sync_dependencies.py

from dataclasses import dataclass

import httpx

from pydantic_ai import Agent, RunContext


@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.Client  # (1)!


agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)


@agent.system_prompt
def get_system_prompt(ctx: RunContext[MyDeps]) -> str:  # (2)!
    response = ctx.deps.http_client.get(
        'https://example.com', headers={'Authorization': f'Bearer {ctx.deps.api_key}'}
    )
    response.raise_for_status()
    return f'Prompt: {response.text}'


async def main():
    deps = MyDeps('foobar', httpx.Client())
    result = await agent.run(
        'Tell me a joke.',
        deps=deps,
    )
    print(result.data)
    #> Did you hear about the toothpaste scandal? They called it Colgate.

1. 这里我们使用同步的 httpx.Client 而不是异步的 httpx.AsyncClient。
2. 为了匹配同步依赖项，系统提示函数现在是一个普通函数，而不是协程。

（此示例是完整的，可以直接“按原样”运行 — 您需要添加 asyncio.run(main()) 来运行 main）

完整示例

除了系统提示之外，依赖项还可以在 工具 和 结果验证器 中使用。

full_example.py

from dataclasses import dataclass

import httpx

from pydantic_ai import Agent, ModelRetry, RunContext


@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient


agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)


@agent.system_prompt
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    response = await ctx.deps.http_client.get('https://example.com')
    response.raise_for_status()
    return f'Prompt: {response.text}'


@agent.tool  # (1)!
async def get_joke_material(ctx: RunContext[MyDeps], subject: str) -> str:
    response = await ctx.deps.http_client.get(
        'https://example.com#jokes',
        params={'subject': subject},
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    response.raise_for_status()
    return response.text


@agent.result_validator  # (2)!
async def validate_result(ctx: RunContext[MyDeps], final_response: str) -> str:
    response = await ctx.deps.http_client.post(
        'https://example.com#validate',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
        params={'query': final_response},
    )
    if response.status_code == 400:
        raise ModelRetry(f'invalid response: {response.text}')
    response.raise_for_status()
    return final_response


async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

1. 要将 RunContext 传递给工具，请使用 tool 装饰器。
2. result_validator 函数可以选择将 RunContext 作为第一个参数传递。

（此示例是完整的，可以直接“按原样”运行 — 您需要添加 asyncio.run(main()) 来运行 main）

覆盖依赖项

在测试代理时，能够自定义依赖项非常有用。

虽然有时可以通过在单元测试中直接调用代理来完成此操作，但我们也可以在调用应用程序代码（反过来调用代理）时覆盖依赖项。

这通过代理上的 override 方法完成。

joke_app.py

from dataclasses import dataclass

import httpx

from pydantic_ai import Agent, RunContext


@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient

    async def system_prompt_factory(self) -> str:  # (1)!
        response = await self.http_client.get('https://example.com')
        response.raise_for_status()
        return f'Prompt: {response.text}'


joke_agent = Agent('openai:gpt-4o', deps_type=MyDeps)


@joke_agent.system_prompt
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    return await ctx.deps.system_prompt_factory()  # (2)!


async def application_code(prompt: str) -> str:  # (3)!
    ...
    ...
    # now deep within application code we call our agent
    async with httpx.AsyncClient() as client:
        app_deps = MyDeps('foobar', client)
        result = await joke_agent.run(prompt, deps=app_deps)  # (4)!
    return result.data

1. 在依赖项上定义一个方法，使系统提示更易于自定义。
2. 从系统提示函数内部调用系统提示工厂。
3. 调用代理的应用程序代码，在实际应用程序中，这可能是 API 端点。
4. 从应用程序代码内部调用代理，在实际应用程序中，此调用可能位于调用堆栈的深处。请注意，当依赖项被覆盖时，此处将不会使用 app_deps。

（此示例是完整的，可以直接“按原样”运行）

test_joke_app.py

from joke_app import MyDeps, application_code, joke_agent


class TestMyDeps(MyDeps):  # (1)!
    async def system_prompt_factory(self) -> str:
        return 'test prompt'


async def test_application_code():
    test_deps = TestMyDeps('test_key', None)  # (2)!
    with joke_agent.override(deps=test_deps):  # (3)!
        joke = await application_code('Tell me a joke.')  # (4)!
    assert joke.startswith('Did you hear about the toothpaste scandal?')

1. 在测试中定义 MyDeps 的子类以自定义系统提示工厂。
2. 创建测试依赖项的实例，我们不需要在此处传递 http_client，因为它未使用。
3. 在 with 代码块的持续时间内覆盖代理的依赖项，运行时将使用 test_deps。
4. 现在我们可以安全地调用我们的应用程序代码，代理将使用覆盖的依赖项。

示例

以下示例演示如何在 PydanticAI 中使用依赖项

• 天气代理
• SQL 生成
• RAG