PydanticAI：构建生产级 AI 应用程序的综合指南

# Web 开发 # AI # Python

PydanticAI 是一个**强大的 Python 框架**，旨在简化使用生成式 AI 开发生产级应用程序的过程。它由广泛使用的数据验证库 Pydantic 背后的同一团队构建，旨在将 FastAPI 的创新和符合人体工程学的设计引入 AI 应用程序开发领域。PydanticAI 专注于**类型安全、模块化以及与其他 Python 工具的无缝集成**。

核心概念

PydanticAI 围绕几个关键概念：

代理

代理是与大型语言模型 (LLM) 交互的**主要接口**。代理充当各种组件的容器，包括：

系统提示：针对LLM的说明，定义为静态字符串或者动态函数。

函数工具：LLM 可以调用来获取附加信息或执行操作的函数。

结构化结果类型：LLM 在运行结束时必须返回的数据类型。

依赖类型：系统提示功能、工具和结果验证器可能使用的数据或服务。

LLM 模型：代理将使用的 LLM，可以在代理创建时或运行时设置。

代理专为可重用性而设计，通常实例化一次并在整个应用程序中重复使用。

系统提示

系统提示是开发人员向 LLM 提供的指令。它们可以是：

静态系统提示：在创建agent时定义，使用Agent构造函数的system_prompt参数。

动态系统提示：由使用@agent.system_prompt修饰的函数定义。这些函数可以通过RunContext对象访问运行时信息，例如依赖项。

单个代理可以同时使用静态和动态系统提示，这些提示按照运行时定义的顺序附加。

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="Use the customer's name while replying to them.",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"The user's name is {ctx.deps}."

@agent.system_prompt
def add_the_date() -> str:
    return f'The date is {date.today()}.'

result = agent.run_sync('What is the date?', deps='Frank')
print(result.data)
#> Hello Frank, the date today is 2032-01-02.

功能工具

函数工具使 LLM 能够访问外部信息或执行系统提示本身无法执行的操作。工具可以通过多种方式注册：

@agent.tool 装饰器：适用于需要通过 RunContext 访问代理上下文的工具。

@agent.tool_plain 装饰器：适用于不需要访问代理上下文的工具。

Agent 构造函数中的 tools 关键字参数：可以采用普通函数或 Tool 类的实例，从而更好地控制工具定义。

import random
from pydantic_ai import Agent, RunContext

agent = Agent(
    'gemini-1.5-flash',
    deps_type=str,
    system_prompt=(
        "You're a dice game, you should roll the die and see if the number "
        "you get back matches the user's guess. If so, tell them they're a winner. "
        "Use the player's name in the response."
    ),
)

@agent.tool_plain
def roll_die() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))

@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps

dice_result = agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!

工具参数从函数签名中提取，用于构建工具的 JSON 模式。函数的文档字符串用于生成工具的描述和模式中的参数描述。

依赖项

依赖项通过依赖项注入系统向代理的系统提示、工具和结果验证器提供数据和服务。依赖项可通过 `RunContext` 对象访问。它们可以是任何 Python 类型，但 `dataclasses` 是管理多个依赖项的便捷方式。

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
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    response = await 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():
    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.

结果

结果是代理运行返回的最终值。它们被包装在 `RunResult`（用于同步和异步运行）或 `StreamedRunResult`（用于流式运行）中，提供对使用数据和消息历史记录的访问。结果可以是纯文本或结构化数据，并使用 Pydantic 进行验证。

from pydantic import BaseModel
from pydantic_ai import Agent

class CityLocation(BaseModel):
    city: str
    country: str

agent = Agent('gemini-1.5-flash', result_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.data)
#> city='London' country='United Kingdom'

通过 `@agent.result_validator` 装饰器添加的结果验证器提供了一种添加进一步验证逻辑的方法，特别是当验证需要 IO 并且是异步的时候。

主要特点

PydanticAI 拥有几个关键功能，使其成为 AI 应用程序开发的绝佳选择：

与模型无关：PydanticAI 支持多种 LLM，包括 OpenAI、Anthropic、Gemini、Ollama、Groq 和 Mistral。它还提供了一个简单的界面来实现对其他模型的支持。

类型安全：旨在与 mypy 和 pyright 等静态类型检查器无缝协作。它允许对依赖项和结果类型进行类型检查。

以 Python 为中心的设计：利用熟悉的 Python 控制流和代理组合来构建 AI 项目，从而轻松应用标准 Python 实践。

结构化响应：使用 Pydantic 验证和构建模型输出，确保一致的响应。

依赖注入系统：提供依赖注入系统，为代理的组件提供数据和服务，增强可测试性和迭代开发。

流式响应：支持流式 LLM 输出并立即验证，从而获得快速准确的结果。

与代理商合作

运行代理

代理可以通过多种方式运行：

run_sync()：用于同步执行。

run()：用于异步执行。

run_stream()：用于流式响应。

from pydantic_ai import Agent

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

# Synchronous run
result_sync = agent.run_sync('What is the capital of Italy?')
print(result_sync.data)
#> Rome

# Asynchronous run
async def main():
    result = await agent.run('What is the capital of France?')
    print(result.data)
    #> Paris

    async with agent.run_stream('What is the capital of the UK?') as response:
        print(await response.get_data())
        #> London

对话

代理运行可能代表整个对话，但对话也可以由多个运行组成，尤其是在保持交互之间的状态时。您可以使用“message_history”参数传递先前运行的消息以继续对话。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='Be a helpful assistant.')
result1 = agent.run_sync('Tell me a joke.')
print(result1.data)
#> Did you hear about the toothpaste scandal? They called it Colgate.

result2 = agent.run_sync('Explain?', message_history=result1.new_messages())
print(result2.data)
#> This is an excellent joke invent by Samuel Colvin, it needs no explanation.

使用限制

PydanticAI 提供了一个 `settings.UsageLimits` 结构来限制令牌和请求的数量。您可以通过 `usage_limits` 参数将这些设置应用于 `run` 函数。

from pydantic_ai import Agent
from pydantic_ai.settings import UsageLimits
from pydantic_ai.exceptions import UsageLimitExceeded

agent = Agent('claude-3-5-sonnet-latest')
try:
    result_sync = agent.run_sync(
        'What is the capital of Italy? Answer with a paragraph.',
        usage_limits=UsageLimits(response_tokens_limit=10),
    )
except UsageLimitExceeded as e:
    print(e)
    #> Exceeded the response_tokens_limit of 10 (response_tokens=32)

模型设置

`settings.ModelSettings` 结构允许您通过 `temperature`、`max_tokens` 和 `timeout` 等参数微调模型行为。您可以通过 `run` 函数中的 `model_settings` 参数应用这些参数。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')
result_sync = agent.run_sync(
    'What is the capital of Italy?',
    model_settings={'temperature': 0.0},
)
print(result_sync.data)
#> Rome

功能工具详细信息

工具注册

可以使用 `@agent.tool` 装饰器（用于需要上下文的工具）、`@agent.tool_plain` 装饰器（用于没有上下文的工具）或通过 `Agent` 构造函数中的 `tools` 参数来注册工具。

from pydantic_ai import Agent, RunContext

agent_a = Agent(
    'gemini-1.5-flash',
    deps_type=str,
    tools=[
        lambda: str(random.randint(1, 6)),
        lambda ctx: ctx.deps
    ],
)

工具架构

参数描述从文档字符串中提取并添加到工具的 JSON 架构中。如果工具有一个参数可以在 JSON 架构中表示为对象，则架构将简化为该对象。

from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage, ModelResponse
from pydantic_ai.models.function import AgentInfo, FunctionModel

agent = Agent()

@agent.tool_plain
def foobar(a: int, b: str, c: dict[str, list[float]]) -> str:
    """Get me foobar.
    Args:
        a: apple pie
        b: banana cake
        c: carrot smoothie
    """
    return f'{a} {b} {c}'

动态工具

可以使用“prepare”函数定制工具，该函数在每个步骤中被调用来修改工具定义或从该步骤中省略该工具。

from typing import Union
from pydantic_ai import Agent, RunContext
from pydantic_ai.tools import ToolDefinition

agent = Agent('test')

async def only_if_42(
    ctx: RunContext[int], tool_def: ToolDefinition
) -> Union[ToolDefinition, None]:
    if ctx.deps == 42:
        return tool_def

@agent.tool(prepare=only_if_42)
def hitchhiker(ctx: RunContext[int], answer: str) -> str:
    return f'{ctx.deps} {answer}'

result = agent.run_sync('testing...', deps=41)
print(result.data)
#> success (no tool calls)

result = agent.run_sync('testing...', deps=42)
print(result.data)
#> {"hitchhiker":"42 a"}

信息和聊天记录

访问消息

可以通过 `RunResult` 和 `StreamedRunResult` 对象上的 `all_messages()` 和 `new_messages()` 方法访问代理运行期间交换的消息。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='Be a helpful assistant.')
result = agent.run_sync('Tell me a joke.')
print(result.data)
#> Did you hear about the toothpaste scandal? They called it Colgate.

print(result.all_messages())

消息重用

可以将消息传递给 `message_history` 参数，以继续跨多个代理运行的对话。当 `message_history` 已设置且不为空时，不会生成新的系统提示。

消息格式

消息格式与模型无关，允许在不同的代理或使用不同模型的同一代理中使用消息。

调试和监控

Pydantic Logfire

PydanticAI 与 **Pydantic Logfire** 集成，后者是一个可观察性平台，可让您监控和调试整个应用程序。Logfire 可用于：

实时调试：实时查看应用程序中发生的情况。

监控应用程序性能：使用 SQL 查询和仪表板。

要将 PydanticAI 与 Logfire 一起使用，请使用 `logfire` 可选组进行安装：`pip install 'pydantic-ai[logfire]'`。然后，您需要配置一个 Logfire 项目并验证您的环境。

安装和设置

安装

可以使用 pip 安装 PydanticAI：

pip install pydantic-ai

还可以使用精简安装来使用特定型号，例如：

pip install 'pydantic-ai-slim[openai]'

Logfire 集成

要将 PydanticAI 与 Logfire 一起使用，请使用 `logfire` 可选组进行安装：

pip install 'pydantic-ai[logfire]'

示例

示例可作为单独的包使用：

pip install 'pydantic-ai[examples]'

测试与评估

单元测试

单元测试验证您的应用程序代码是否按预期运行。对于 PydanticAI，请遵循以下策略：

使用 pytest 作为您的测试工具。

使用 TestModel 或 FunctionModel 代替您的实际模型。

使用 Agent.override 替换应用程序逻辑中的模型。

全局设置 ALLOW_MODEL_REQUESTS=False 以防止意外调用非测试模型。

import pytest
import anyio

from pydantic_ai import Agent, RunContext
from pydantic_ai.models.test import TestModel
from pydantic_ai.messages import ModelRequest, UserPromptPart
from pydantic_ai.tools import ToolDefinition

from datetime import datetime


@pytest.mark.anyio
async def test_weather():
    model = TestModel()

    class WeatherResult(dict):
        temperature: int
        location: str
        time: datetime

    weather_agent = Agent(
        model=model,
        result_type=WeatherResult,
        system_prompt='Return a valid WeatherResult object.',
    )

    @weather_agent.tool
    def get_current_time(ctx: RunContext) -> datetime:
        return datetime.now()

    @weather_agent.tool
    def get_location(ctx: RunContext) -> str:
        return "london"


    with weather_agent.override(model=model):
        result = await weather_agent.run('What is the weather?')
        assert result.data["temperature"] == 0
        assert result.data["location"] == 'london'
        assert isinstance(result.data["time"], datetime)
        messages = result.all_messages()
        assert len(messages) == 3
        assert messages.kind == "request"
        assert messages.kind == "response"
        assert messages.kind == "request"
        assert isinstance(messages.parts, UserPromptPart)

        assert messages.parts.tool_name == "get_location"
        assert isinstance(messages.parts.tool_name , str)
        assert messages.parts.tool_name == 'get_current_time'

评估

评估用于衡量 LLM 的性能，它更像是基准测试而非单元测试。评估侧重于衡量 LLM 在特定应用中的表现。这可以通过端到端测试、综合自包含测试、使用 LLM 评估 LLM 或通过衡量生产中的代理性能来实现。

示例用例

PydanticAI 可用于多种用例：

轮盘赌：使用具有整数依赖性和布尔结果的代理模拟轮盘赌。

聊天应用程序：创建一个具有多次运行的聊天应用程序，并使用 message_history 传递以前的消息。

银行支持代理：使用工具、依赖注入和结构化响应为银行构建支持代理。

天气预报：使用功能工具和依赖项创建一个根据位置和日期返回天气预报的应用程序。

SQL 生成：根据用户提示生成 SQL 查询，并使用结果验证器进行验证。

结论

PydanticAI 提供了一个**强大而灵活的框架**，用于开发 AI 应用程序，重点强调类型安全性和模块化。使用 Pydantic 进行数据验证和结构化，再加上其依赖项注入系统，使其成为构建**可靠且可维护的 AI 应用程序**的理想工具。凭借其广泛的 LLM 支持以及与 Pydantic Logfire 等工具的无缝集成，PydanticAI 使开发人员能够高效地构建功能强大、可用于生产的 AI 驱动项目。

CLIS.CC