利用 Pydantic AI Agent 框架构建可靠的生成式 AI 应用

当前的LLM（大型语言模型）正在向更加复杂、多功能的AI代理体系的转变。在这一趋势的推动下，各大科技巨头如AWS、OpenAI、Microsoft等纷纷推出了各自的AI Agent框架（探索LangGraph：开启AI Agent构建的新路径）。其中，PydanticAI代理框架以其对Pydantic与LLM的卓越支持，在众多框架中脱颖而出，成为构建健壮、生产级AI应用的重要工具。

一、Pydantic 基础

（一）定义与功能

Pydantic 是一个用于轻松验证和解析数据的 Python 库，其核心功能在于确保数据的准确性以及遵循预期的结构，这在处理诸如 JSON 文件、用户数据或 API 响应等外部输入时显得尤为关键。它摒弃了手动编写每个字段检查（如 “这是一个整数吗？这个字符串是否过长？”）的繁琐方式，而是借助模型实现自动化检查。

（二）示例说明

假设我们正在开发一个应用程序，用户需要在其中提交姓名、年龄和电子邮件。使用 Pydantic，我们可以轻松定义一个数据模型来确保输入数据的有效性。

1. 定义模型

from pydantic import BaseModel, EmailStrclass User(BaseModel):    name: str    age: int    email: EmailStr

2、验证输入数据

user_data = {    "name": "Alice",    "age": 25,    "email": "alice@example.com"}user = User(**user_data)print(user.name)  print(user.age)  print(user.email)  

在上述示例中，当用户提交的数据符合模型定义时，Pydantic 能够顺利解析并验证数据。然而，如果用户提交了无效数据，例如将年龄设置为字符串 “twenty-five”，Pydantic 会自动抛出错误，从而确保数据的质量。

（三）在部署中的关键作用

1. 大规模数据验证

   自动验证大规模输入数据，保证其与预期结构匹配，有效减少错误。

2. 错误处理与调试

   针对无效数据提供清晰的错误消息，极大地加快和简化生产环境中的调试过程。

3. 自动解析与序列化

   自动将原始数据转换为可用格式，简化数据处理流程。

4. 与 FastAPI 无缝集成

   与 FastAPI 协同工作，定义输入 / 输出模型，以便在生产环境中处理 HTTP 数据。

5. 确保微服务的健壮性

   保证微服务之间的数据交换一致且有效，避免集成问题。

6. 预防安全问题

   在入口点验证数据，阻止恶意输入，增强应用程序的安全性。

二、PydanticAI 框架详解

（一）关键特性

1. 结构化响应处理

   利用 Pydantic 验证静态和流式响应，确保数据处理的可靠性。无论是一次性获取全部响应，还是处理实时流式传输的数据，PydanticAI 都能够依据预定义的模型对数据进行严格验证，确保数据的准确性和完整性。

2. 广泛的模型支持

   兼容 OpenAI、Gemini 和 Groq 等多种模型，并提供简洁的接口以便集成其他模型。这意味着开发者可以根据项目需求灵活选择最适合的模型，而无需担心框架的兼容性问题。

3. 基于 Pydantic 专业知识构建

   由 Pydantic 的创造者开发，是 LangChain、OpenAI SDK 等流行框架的基础。凭借其深厚的专业积累，PydanticAI 继承了 Pydantic 的优势，并在此基础上进行了针对性的优化和扩展，为开发者提供了更加稳定和强大的功能。

4. 简化的依赖管理

   引入类型安全的依赖注入系统，简化测试和迭代开发过程。通过清晰地管理依赖关系，开发者可以更轻松地维护代码，提高开发效率，同时降低因依赖问题导致的错误风险。

5. Pythonic 设计理念

   采用标准的 Python 编程实践进行智能体的组合和控制流，对开发者来说更加直观。熟悉 Python 的开发者可以迅速上手，减少学习成本，更快地构建出高质量的 AI 智能体应用。

6. 集成 Logfire 监控功能

   与 Logfire 集成，实现对 AI 驱动应用程序的性能跟踪和调试。开发者可以实时监控应用程序的运行状态，及时发现并解决性能瓶颈或其他问题，确保应用程序的稳定运行。

7. 类型安全操作

   确保强大的类型检查工作流程，最大程度减少运行时错误。在代码编写阶段，通过严格的类型检查，提前发现潜在的类型不匹配等问题，提高代码的质量和可靠性。

8. 处于积极的测试阶段

   目前处于测试阶段，为持续改进和反馈驱动的更新提供了空间。这意味着开发者可以积极参与到框架的完善过程中，提出宝贵的意见和建议，同时也能够及时享受到框架不断优化带来的好处。

（二）、创建一个 PydanticAI 代理​​​​​​​

from dataclasses import dataclassfrom pydantic import BaseModel, Fieldfrom pydantic_ai import Agent, RunContextfrom bank_database import DatabaseConn@dataclassclass SupportDependencies:      customer_id: int    db: DatabaseConn  class SupportResult(BaseModel):      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(      'openai:gpt-4o',      deps_type=SupportDependencies,    result_type=SupportResult,      system_prompt=(          'You are a support agent in our bank, give the '        'customer support and judge the risk level of their query.'    ),)@support_agent.system_prompt  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  async def customer_balance(    ctx: RunContext[SupportDependencies], include_pending: bool) -> float:    """Returns the customer's current account balance."""      return await ctx.deps.db.customer_balance(        id=ctx.deps.customer_id,        include_pending=include_pending,    )async def main():    deps = SupportDependencies(customer_id=123, db=DatabaseConn())    result = await support_agent.run('What is my balance?', deps=deps)      print(result.data)      """    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    """

（三）代码解析

1. 数据类装饰器（支持依赖项）
   • 目的

     定义智能体所需的依赖项，例如客户 ID 和数据库连接（db）。

   • 用途

     在查询过程中传递给智能体，以便访问特定于客户的数据。

2. 支持结果（SupportResult）
   • 目的

     指定智能体输出的结构化格式。包括给客户的文本建议（support_advice）、是否应阻止客户的卡（block_card）以及查询情况的风险级别（risk，范围为 0 - 10）。

   • 验证

     确保数据的完整性，例如风险级别必须在 0 到 10 之间。

3. 智能体设置

   • 使用 openai:gpt - 4o 模型生成响应。

   • 指定依赖项（deps_type）和预期输出（result_type）。

   • 包含一个系统提示，为智能体设置上下文，指示其提供支持并评估查询风险。

4. 系统提示函数
   • 目的

     通过查询数据库动态丰富系统提示中的客户姓名。

   • 工作方式

     使用客户 ID（ctx.deps.customer_id）访问数据库（ctx.deps.db），并返回包含客户姓名的字符串，以实现个性化交互。

5. 自定义工具
   • 目的

     添加一个工具来获取客户的账户余额，可选择是否包括未完成交易。

   • 工作方式

     使用客户 ID 和 include_pending 标志查询数据库，并返回余额作为浮点数。

6. 运行智能体
   • 目的

     演示如何通过传递查询和依赖项与智能体进行交互。

   • 步骤

     创建一个包含客户 ID 和数据库连接的 SupportDependencies 对象，使用查询（如 'What is my balance?'）调用智能体的 run 方法，Agent（哪款Multi-Agent框架更胜一筹？深入剖析五大热门选项）根据查询、依赖项和工具生成响应。

PydanticAI框架为构建健壮、生产级AI应用提供了一个强大的工具。通过结合Pydantic的数据验证能力和LLM的生成能力，PydanticAI确保了结构化、类型安全和安全的交互。其易用性、对多种AI模型的支持以及持续演进的特性使其成为开发人员不可或缺的工具。