依赖项

Pydantic AI 使用依赖注入系统，为您的代理（agent）的系统提示、工具和输出验证器提供数据和服务。

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

定义依赖项

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

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

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

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.output)
        #> 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.output)
        #> 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 在一个线程池中被调用。因此，当依赖项执行 I/O 操作时，稍微更推荐使用 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.output)
    #> 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.output_validator  # (2)!
async def validate_output(ctx: RunContext[MyDeps], output: str) -> str:
    response = await ctx.deps.http_client.post(
        'https://example.com#validate',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
        params={'query': output},
    )
    if response.status_code == 400:
        raise ModelRetry(f'invalid response: {response.text}')
    response.raise_for_status()
    return output


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.output)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

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

（此示例是完整的，可以“按原样”运行——您需要添加 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.output

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. 现在我们可以安全地调用我们的应用程序代码，代理将使用被覆盖的依赖项。

示例

以下示例演示了如何在 Pydantic AI 中使用依赖项。

• 天气代理
• SQL 生成
• RAG