PydanticAI：构建生产就绪型人工智能应用程序的综合指南

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

核心概念

PydanticAI 围绕几个关键概念：

代理商

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

• 系统提示：LLM说明，定义为静态字符串或动态函数。
• 函数工具：LLM 可以调用以获取其他信息或执行操作的函数。
• 结构化结果类型：LLM 在运行结束时必须返回的数据类型。
• 依赖类型：系统提示函数、工具和结果验证器可能使用的数据或服务。
• LLM 模型：代理将使用的 LLM，可以在代理创建时或运行时设置。

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

系统提示

系统提示是开发者向LLM提供的说明。他们可以是：

• 静态系统提示：在创建代理时定义，使用 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.

功能工具

功能工具使法学硕士能够访问外部信息或执行系统提示本身不可用的操作。工具可以通过多种方式注册：

• @agent.tool 装饰器：适用于需要通过 RunContext 访问代理上下文的工具。
• @agent.tool_plain 装饰器：适用于不需要访问代理上下文的工具。
Agent 构造函数中的
• tools 关键字参数：可以采用 Tool 类的普通函数或实例，从而更好地控制工具定义。

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.

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

依赖关系

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

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!

结果

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

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.

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

主要特点

PydanticAI 拥有多项关键功能，使其成为人工智能应用程序开发的绝佳选择：

• 模型不可知论：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, 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.

对话

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

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!

使用限制

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

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.

模型设置

settings.ModelSettings 结构允许您通过温度、max_tokens 和超时等参数微调模型行为。您可以通过运行函数中的 model_settings 参数应用这些。

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.tool装饰器（对于需要上下文的工具）、@agent.tool_plain装饰器（对于没有上下文的工具）或通过Agent构造函数中的tools参数来注册工具。

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

工具架构

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

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.

动态工具

可以使用准备函数来自定义工具，该函数在每个步骤中调用以修改工具定义或从该步骤中省略工具。

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)

消息和聊天记录

访问消息

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

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

消息重用

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

消息格式

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

调试与监控

派丹提克原火

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

• 实时调试：实时查看应用程序中发生的情况。
• 监控应用程序性能：使用 SQL 查询和仪表板。

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

安装和设置

安装

PydanticAI 可以使用 pip 安装：

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.

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

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!

日志火集成

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

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.

示例

示例可作为单独的包提供：

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'

测试与评估

单元测试

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

• 使用 pytest 作为您的测试工具。
• 使用 TestModel 或 FunctionModel 代替您的实际模型。
• 使用 Agent.override 替换应用程序逻辑中的模型。
• 全局设置 ALLOW_MODEL_REQUESTS=False 以防止意外调用非测试模型。

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

埃瓦尔斯

评估用于衡量 LLM 的表现，更像是基准测试而不是单元测试。评估的重点是衡量法学硕士在特定申请中的表现。这可以通过端到端测试、综合独立测试、使用 LLM 评估 LLM 或通过测量生产中的代理性能来完成。

示例用例

PydanticAI 可用于多种用例：

• 轮盘赌轮：使用具有整数依赖项和布尔结果的代理模拟轮盘赌轮。
• 聊天应用程序：创建多次运行的聊天应用程序，使用 message_history 传递以前的消息。
• 银行支持代理：使用工具、依赖项注入和结构化响应为银行构建支持代理。
• 天气预报：创建一个应用程序，使用函数工具和依赖项根据位置和日期返回天气预报。
• SQL 生成：根据用户提示生成 SQL 查询，并使用结果验证器进行验证。

结论

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

以上是PydanticAI：构建生产就绪型人工智能应用程序的综合指南的详细内容。更多信息请关注PHP中文网其他相关文章！