依赖项

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

与PydanticAI的设计理念相匹配，我们的依赖系统尝试使用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.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

定义一个数据类来保存依赖项。
将数据类类型传递给Agent构造函数的deps_type参数。注意：我们在这里传递的是类型，而不是实例，这个参数在运行时实际上并没有使用，它在这里的目的是为了让我们能够对代理进行完整的类型检查。
在运行代理时，将数据类的实例传递给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.

RunContext 可以作为唯一参数可选地传递给 system_prompt 函数。
RunContext 使用依赖项的类型进行参数化，如果该类型不正确，静态类型检查器将会引发错误。
通过 .deps 属性访问依赖项。
通过 .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.

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

(这个例子是完整的，可以“直接运行”——你需要添加 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.

要将 RunContext 传递给一个工具，请使用 tool 装饰器。
RunContext 可以选择性地作为第一个参数传递给result_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.data

定义一个依赖的方法，以使系统提示更容易定制。
从系统提示函数中调用系统提示工厂。
调用代理的应用程序代码，在实际应用中，这可能是一个API端点。
在应用程序代码中调用代理，在真实应用中，此调用可能深嵌在调用栈中。注意这里的 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?')

在测试中定义一个MyDeps的子类，以自定义系统提示工厂。
创建测试依赖的实例，我们在这里不需要传递一个 http_client，因为它没有被使用。
在with块的持续时间内重写代理的依赖项，test_deps将在运行代理时使用。
现在我们可以安全地调用我们的应用程序代码，代理将使用重写的依赖项。

示例

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