介绍

代理框架 / 用于与 Pydantic 结合使用的 shim

PydanticAI 是一个 Python 代理框架，旨在使构建生产级应用程序与生成式 AI 的过程变得不那么痛苦。

PydanticAI 是一个 Python 代理框架，旨在减少构建生产级应用程序使用生成性人工智能的痛苦。

FastAPI 通过提供创新和符合人体工程学的设计，彻底改变了网页开发，建立在 Pydantic 的基础上。

同样，几乎所有的代理框架和Python中的LLM库都使用Pydantic，但当我们开始在 Pydantic Logfire 中使用LLM时，我们找不到任何东西能给我们相同的感觉。

我们构建PydanticAI的目的是：将FastAPI的感觉带入GenAI应用开发。

为什么使用 PydanticAI

由Pydantic团队构建：由Pydantic背后的团队构建（OpenAI SDK、Anthropic SDK、LangChain、LlamaIndex、AutoGPT、Transformers、CrewAI、Instructor等的验证层）。
模型无关：支持OpenAI、Anthropic、Gemini、Deepseek、Ollama、Groq、Cohere和Mistral，并且有一个简单的接口来实现对其他模型的支持。
Pydantic Logfire 集成: 无缝集成 Pydantic Logfire，用于实时调试、性能监控和您基于 LLM 的应用程序的行为追踪。
类型安全: 旨在为您提供尽可能强大和信息丰富的类型检查。
以Python为中心的设计：利用Python熟悉的控制流和代理组合来构建您的AI驱动项目，使您能够轻松应用在任何其他（非AI）项目中使用的标准Python最佳实践。
结构化响应：利用Pydantic的强大功能验证和结构化模型输出，确保响应在多次运行中保持一致。
依赖注入系统：提供一个可选的依赖注入系统，以便为您的代理的系统提示、工具和结果验证器提供数据和服务。这对于测试和基于评估的迭代开发非常有用。
流式响应：提供连续流式生成LLM输出的能力，进行即时验证，确保快速和准确的结果。
图形支持: Pydantic Graph 提供了一种强大的方式来使用类型提示定义图形，这在标准控制流可能退化为杂乱代码的复杂应用中非常有用。

测试版

PydanticAI 仍处于早期测试阶段，API 仍可能发生变化，还有很多工作要做。反馈非常欢迎!

你好，世界示例

这是PydanticAI的一个最小示例：

hello_world.py

from pydantic_ai import Agent

agent = Agent(  # (1)!
    'google-gla:gemini-1.5-flash',
    system_prompt='Be concise, reply with one sentence.',  # (2)!
)

result = agent.run_sync('Where does "hello world" come from?')  # (3)!
print(result.data)
"""
The first known use of "hello, world" was in a 1974 textbook about the C programming language.
"""

我们将代理配置为使用 Gemini 1.5's Flash 模型，但您在运行代理时也可以设置模型。
使用关键字参数向代理注册静态 system prompt。
同步运行代理，与LLM进行对话。

(这个例子是完整的，可以“原样”运行)

交换应该非常简短：PydanticAI 将系统提示和用户查询发送给 LLM，模型将返回文本响应。

目前还不是很有趣，但我们可以轻松添加“工具”、动态系统提示和结构化响应，以构建更强大的智能体。

工具与依赖注入示例

这是一个使用PydanticAI构建银行支持代理的简明示例：

bank_support.py

from dataclasses import dataclass

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext

from bank_database import DatabaseConn


@dataclass
class SupportDependencies:  # (3)!
    customer_id: int
    db: DatabaseConn  # (12)!


class SupportResult(BaseModel):  # (13)!
    support_advice: str = Field(description='Advice returned to the customer')
    block_card: bool = Field(description="Whether to block the customer's card")
    risk: int = Field(description='Risk level of query', ge=0, le=10)


support_agent = Agent(  # (1)!
    'openai:gpt-4o',  # (2)!
    deps_type=SupportDependencies,
    result_type=SupportResult,  # (9)!
    system_prompt=(  # (4)!
        'You are a support agent in our bank, give the '
        'customer support and judge the risk level of their query.'
    ),
)


@support_agent.system_prompt  # (5)!
async def add_customer_name(ctx: RunContext[SupportDependencies]) -> str:
    customer_name = await ctx.deps.db.customer_name(id=ctx.deps.customer_id)
    return f"The customer's name is {customer_name!r}"


@support_agent.tool  # (6)!
async def customer_balance(
    ctx: RunContext[SupportDependencies], include_pending: bool
) -> float:
    """Returns the customer's current account balance."""  # (7)!
    return await ctx.deps.db.customer_balance(
        id=ctx.deps.customer_id,
        include_pending=include_pending,
    )


...  # (11)!


async def main():
    deps = SupportDependencies(customer_id=123, db=DatabaseConn())
    result = await support_agent.run('What is my balance?', deps=deps)  # (8)!
    print(result.data)  # (10)!
    """
    support_advice='Hello John, your current account balance, including pending transactions, is $123.45.' block_card=False risk=1
    """

    result = await support_agent.run('I just lost my card!', deps=deps)
    print(result.data)
    """
    support_advice="I'm sorry to hear that, John. We are temporarily blocking your card to prevent unauthorized transactions." block_card=True risk=8
    """

这个 agent 将充当银行的一线支持。代理在接受的依赖类型和返回的结果类型上是通用的。在这种情况下，支持代理的类型是 Agent[SupportDependencies, SupportResult]。
在这里我们配置代理使用 OpenAI的GPT-4o模型，你也可以在运行代理时设置模型。
SupportDependencies 数据类用于传递数据、连接和逻辑到模型中，这些在运行 system prompt 和 tool 函数时是必要的。PydanticAI 的依赖注入系统提供了一种 type-safe 方法来定制您的代理行为，特别是在运行 unit tests 和 evals 时非常有用。
静态系统提示可以通过 system_prompt 关键字参数注册给代理。
动态系统提示可以通过 @agent.system_prompt 装饰器进行注册，并可以利用依赖注入。依赖通过 RunContext 参数进行传递，该参数带有上面的 deps_type。如果这里的类型注解错误，静态类型检查器会捕捉到它。
tool 让您注册LLM在响应用户时可能调用的函数。同样，依赖关系通过 RunContext 传递，任何其他参数成为传递给LLM的工具架构。Pydantic用于验证这些参数，错误会反馈给LLM，以便它可以重试。
工具的文档字符串也作为工具的描述传递给LLM。参数描述从文档字符串中提取并添加到发送给LLM的参数架构中。
异步运行代理，与LLM进行对话，直到达到最终响应。即使在这个相对简单的案例中，代理也会在调用工具以获取结果时与LLM交换多个消息。
代理的响应将保证为一个 SupportResult，如果验证失败 reflection 将意味着代理被提示重新尝试。
结果将通过Pydantic进行验证，以确保它是一个 SupportResult，由于代理是通用的，它也将被类型化为 SupportResult 以帮助静态类型检查。
在实际使用情况下，您会为代理添加更多工具和更长的系统提示，以扩展其具备的上下文和它可以提供的支持。
这是一个数据库连接的简单示例，用于使示例简短易懂。实际上，您将连接到外部数据库（例如 PostgreSQL）以获取有关客户的信息。
这个 Pydantic 模型用于约束代理返回的结构化数据。根据这个简单的定义，Pydantic 构建了 JSON Schema，告诉 LLM 如何返回数据，并执行验证以确保数据在运行结束时是正确的。

完整的 bank_support.py 示例

这里包含的代码因简洁性而不完整（DatabaseConn 的定义缺失）；您可以在这里找到完整的 bank_support.py 示例。

使用Pydantic Logfire进行仪器测量

要了解上述运行的流程，我们可以使用 Pydantic Logfire 观看代理的操作。

为此，我们需要设置logfire，并在代码中添加以下内容：

bank_support_with_logfire.py

...
from bank_database import DatabaseConn

import logfire
logfire.configure()  # (1)!
logfire.instrument_asyncpg()  # (2)!
...

配置logfire，如果项目未设置，将会失败。
在我们的演示中，DatabaseConn 使用 asyncpg 连接到 PostgreSQL 数据库，因此使用 logfire.instrument_asyncpg() 来记录数据库查询。

这足以让你看到你的代理在行动中的以下视图：

查看监控与性能了解更多信息。

接下来的步骤

要自己尝试PydanticAI，请按照示例中的说明进行操作。

阅读文档以了解更多关于使用PydanticAI构建应用程序的信息。

阅读API参考以了解PydanticAI的接口。