1. 项目概述当Pydantic遇上AI数据验证的范式革命如果你在过去几年里写过Python尤其是做过Web开发或者数据处理那你大概率听说过或用过Pydantic。它凭借基于Python类型注解的、运行时强制的数据验证和序列化能力几乎成了FastAPI的“官方CP”也让无数开发者从手动写一堆if-else判断参数有效性的繁琐工作中解放出来。但今天聊的不是Pydantic本身而是一个可能改变我们构建AI应用方式的新玩意儿——pydantic-ai。简单说它试图用Pydantic那套优雅、声明式的范式来“管束”和“结构化”大语言模型LLM那看似自由不羁的输出。这听起来有点“给野马套上缰绳”的意思。我们调用OpenAI、Anthropic的API时最头疼的问题之一就是输出的不可控性。你让模型生成一段JSON它可能给你来个残缺的括号或者多出几个注释你让它严格按某个格式回答它一高兴就自由发挥了。于是我们不得不写大量的后处理代码用正则表达式去抠、用json.loads配合try...except去解析、写复杂的逻辑去校验和修正。pydantic-ai的核心想法就是为什么不让模型直接输出一个Pydantic模型实例呢这样一来从模型那里拿到结果的那一刻它就已经是一个经过验证的、类型安全的、可以直接在你的业务逻辑里使用的Python对象了。这不仅仅是方便更是一种思维模式的转变将AI从“黑盒文本生成器”变成了“可靠的结构化数据生成器”。这个项目适合任何正在或将要在生产环境中集成LLM的开发者、工程师无论是构建智能客服、数据分析助手还是复杂的多智能体工作流都能从中获得巨大的确定性和开发效率提升。2. 核心设计理念声明式AI交互的四大支柱pydantic-ai不是简单地在openai库外面包一层。它的设计渗透着Pydantic哲学旨在为AI交互提供一个坚固、可预测的框架。我们可以从四个核心支柱来理解它的设计思路。2.1 支柱一结果即对象验证前置传统流程是发送Prompt - 接收文本/JSON字符串 - 解析 - 验证 - 转换为对象。pydantic-ai将其压缩为发送Prompt附带输出模型定义 - 接收并自动验证 - 得到Pydantic对象。验证被前置到了模型调用环节由框架替你完成。这意味着如果模型输出不符合你定义的Pydantic模型比如缺少必需字段、字段类型不对这次调用在框架层面就会失败或进行重试而不会让一个半成品数据流入你的核心业务代码。这极大地增强了系统的健壮性。实操心得这种设计特别适合对数据质量要求高的场景。比如你让AI从一段客户邮件中提取“订单号”、“产品SKU”和“问题描述”。订单号必须是特定格式的数字SKU必须是存在的产品代码。用传统方法你需要在解析后手动校验这些规则。用pydantic-ai你只需要在Pydantic模型里用Field和自定义验证器定义好这些规则框架会确保你拿到的对象一定是合规的。2.2 支柱二提示词Prompt的模块化与结构化在pydantic-ai中Prompt不再是简单的字符串模板拼接。它被抽象成更结构化的组件。你可以定义“系统消息”、“用户消息”并且可以动态地注入上下文context。更重要的是这个上下文本身也可以是Pydantic模型对象框架会自动将其序列化成适合模型理解的格式通常是JSON。这使得构建复杂的、多轮次的对话或工作流变得更加清晰避免了字符串格式化带来的混乱和潜在错误。一个典型场景你有一个客户支持机器人需要根据用户的历史工单一个列表、当前产品信息一个对象和本次问题描述来生成回复。传统方式下你需要手动把这些数据拼进一个巨大的Prompt字符串里。在pydantic-ai中你可以为“对话上下文”定义一个Pydantic模型包含history_tickets,product_info,current_query等字段。在调用时你只需创建这个上下文对象并传入框架负责将其整合到Prompt中。2.3 支柱三依赖注入Dependency Injection式的工具调用OpenAI的Function Calling后更名为Tools是让LLM与现实世界交互的关键。pydantic-ai对此提供了原生支持并且设计得非常巧妙。你定义的工具函数可以像FastAPI的路由处理函数一样声明参数而这些参数的类型提示会被框架用来自动生成工具的JSON Schema。更强大的是它支持依赖注入。这意味着你的工具函数可以声明需要“数据库会话”或“当前用户”这样的依赖项框架会在调用工具时自动解析并提供这些依赖而不是让你在全局状态中到处传递。这解决了什么痛点想象一下你的AI助手有一个工具叫get_user_order需要user_id和db_session。没有依赖注入你或许需要在一个全局变量中保存数据库连接或者以某种方式传递。有了依赖注入你只需在工具函数中声明db: DBSession参数并在框架中配置好如何获取DBSession比如从连接池获取剩下的框架自动搞定。这让工具函数的编写更纯粹、更易于测试。2.4 支柱四多模型支持与统一抽象虽然例子常用OpenAI但pydantic-ai在设计上就考虑了对多种模型后端的支持。它提供了一个统一的抽象层让你可以用几乎相同的代码与OpenAI、Anthropic Claude、甚至是本地部署的模型通过兼容OpenAI API的服务器进行交互。你只需要更换配置中的base_url和api_key。这为未来切换模型供应商或进行多模型降级提供了便利。注意事项尽管抽象是统一的但不同模型在性能、成本、对特定Prompt格式的偏好上仍有差异。在pydantic-ai中切换模型后端后可能需要对Prompt的措辞或温度temperature等参数进行微调以达到最佳效果。框架负责通信协议但无法完全抹平模型之间的特性差异。3. 从零开始构建你的第一个结构化AI代理理论说了不少现在让我们动手从一个最简单的例子开始感受pydantic-ai的工作流。我们将构建一个“天气信息提取器”从一段用户文本中提取结构化信息。3.1 环境准备与安装首先确保你的Python环境在3.8以上。然后安装核心库pip install pydantic-ai openai这里我们同时安装了openai因为我们将使用OpenAI的模型作为后端。pydantic-ai本身不绑定特定模型客户端它使用httpx进行HTTP通信但需要对应的SDK来提供API密钥等配置管理或者你直接配置端点。重要配置你需要设置你的OpenAI API密钥。最安全的方式是使用环境变量export OPENAI_API_KEYyour-api-key-here或者在代码中通过os.environ设置仅用于测试生产环境切勿硬编码。3.2 定义输出模型告诉AI你想要什么这是最核心的一步。我们想要从“上海明天天气怎么样”或“纽约下周一的温度是多少”这样的句子中提取出location地点、date日期和information_type信息类型是温度、天气状况还是风速。from pydantic import BaseModel, Field from datetime import date from enum import Enum class WeatherInfoType(str, Enum): TEMPERATURE “temperature” CONDITION “condition” # 如 sunny, rainy WIND “wind” ALL “all” # 全部信息 class WeatherQuery(BaseModel): 从用户查询中提取的天气查询信息 location: str Field(description“城市或地区名如‘上海’、‘New York’) query_date: date Field(description“查询的日期默认为今天”, default_factorydate.today) info_type: WeatherInfoType Field(description“用户想查询的天气信息类型”, defaultWeatherInfoType.ALL) # 一个自定义验证器示例确保location不是空字符串 field_validator(‘location’) classmethod def location_not_empty(cls, v: str) - str: if not v or not v.strip(): raise ValueError(‘location cannot be empty’) return v.strip()我们定义了两个模型。WeatherInfoType是一个枚举限定了信息类型的可选值这能极大减少模型的“胡思乱想”。WeatherQuery是我们的主输出模型使用了Field来为每个字段提供更详细的描述这些描述会作为上下文送给大模型指导它如何理解并填充这些字段。default_factory用于提供默认值今天的日期。我们还添加了一个简单的验证器确保地点非空。3.3 创建AI代理Agent并运行在pydantic-ai中与模型交互的核心单元是Agent。我们需要创建一个Agent并指定它使用哪个模型以及输出格式。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel # 1. 选择模型。这里使用gpt-3.5-turbo成本较低适合实验。 # 你也可以用‘gpt-4’, ‘gpt-4-turbo’等。 model OpenAIModel(‘gpt-3.5-turbo’) # 2. 创建Agent。关键参数result_type指定了我们期望的输出模型。 weather_agent Agent( modelmodel, result_typeWeatherQuery, system_prompt“你是一个精准的信息提取助手。请从用户的自然语言查询中严格提取出与天气相关的结构化信息。” ) # 3. 运行Agent传入用户消息。 async def main(): user_message “请问旧金山下周五的天气状况” result await weather_agent.run(user_message) # result.data 就是我们想要的WeatherQuery对象 extracted_info: WeatherQuery result.data print(f“地点: {extracted_info.location}”) print(f“日期: {extracted_info.query_date}”) print(f“查询类型: {extracted_info.info_type}”) # 你可以像使用任何Pydantic对象一样使用它 if extracted_info.location “San Francisco”: # 调用你的天气API... pass # 使用asyncio运行 import asyncio asyncio.run(main())运行这段代码你会看到控制台打印出结构化的信息。result.data直接就是一个WeatherQuery实例你可以直接访问它的属性调用它的方法或者将它序列化成JSONextracted_info.model_dump_json()发给其他服务。整个过程你没有写一行解析JSON或校验数据的代码。踩坑点在初期测试时你可能会遇到模型输出不符合格式导致验证失败的情况。pydantic-ai的Agent.run()默认会重试可配置次数。如果一直失败你需要检查1. 字段描述是否足够清晰2. 枚举值是否覆盖了所有可能情况3. Prompt特别是系统提示是否明确要求了输出格式通常在系统提示中强调“请严格按指定JSON格式输出”会有帮助。4. 进阶实战构建带工具调用的客服助手仅仅提取信息还不够真正的AI应用需要“行动”。我们构建一个更复杂的客服助手它不仅能理解用户问题还能通过调用工具查询数据库、创建工单。4.1 定义工具Tools假设我们有两个工具一个根据订单号查询订单状态另一个为用户创建新的支持工单。from pydantic import BaseModel, Field from typing import Optional # 首先定义工具调用所需的输入模型 class OrderQueryInput(BaseModel): order_id: str Field(description“订单编号格式为‘ORD-XXXXXX’”) class CreateTicketInput(BaseModel): user_id: int Field(description“用户ID”) problem: str Field(description“用户描述的问题”) priority: str Field(description“优先级可选‘low’, ‘medium’, ‘high’”, default“medium”) # 然后实现工具函数本身。这里用模拟函数代替真实数据库操作。 async def query_order_status(order: OrderQueryInput) - str: 根据订单号查询订单状态模拟函数 # 这里应该是数据库查询逻辑 print(f“[工具调用] 正在查询订单 {order.order_id}...”) # 模拟一个响应 fake_statuses [“已发货” “配送中” “待付款” “已完成”] import random return f“订单 {order.order_id} 状态为{random.choice(fake_statuses)}” async def create_support_ticket(ticket: CreateTicketInput) - str: 为用户创建支持工单模拟函数 print(f“[工具调用] 正在为用户 {ticket.user_id} 创建工单问题{ticket.problem[:50]}... 优先级{ticket.priority}”) # 模拟创建成功返回工单号 import uuid ticket_id str(uuid.uuid4())[:8] return f“工单创建成功工单号TICKET-{ticket_id}。我们的客服将在24小时内联系您。”注意工具函数的输入参数是一个Pydantic模型OrderQueryInput/CreateTicketInput。pydantic-ai会利用这个模型的字段定义自动生成供大模型理解的Tool Schema包括名称、描述、参数JSON Schema。4.2 创建带工具的Agent并运行现在我们将工具装配到Agent上。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel model OpenAIModel(‘gpt-4’) # 复杂任务建议使用理解能力更强的GPT-4 # 创建Agent时通过tools参数注册工具 customer_agent Agent( modelmodel, system_prompt“””你是一个专业的电商客服助手。你的任务是 1. 理解用户关于订单或账户问题的查询。 2. 如果需要查询订单请使用query_order_status工具。 3. 如果用户需要人工帮助或报告新问题请使用create_support_ticket工具。 4. 根据工具返回的结果用友好、清晰的语气回复用户。 请严格判断只在必要时调用工具。不要对工具能力之外的问题进行猜测。“””, tools[query_order_status, create_support_ticket] # 注册工具 ) async def handle_customer_query(query: str, current_user_id: int): 处理用户查询的入口函数 print(f“用户提问{query}”) # 运行Agent。注意我们还可以传入‘context’这里把当前用户ID作为上下文。 result await customer_agent.run( query, context{‘user_id’: current_user_id} # 上下文可以传递给工具 ) # result.data 是模型的最终文本回复 final_answer result.data print(f“助手回复{final_answer}”) return final_answer # 模拟对话 async def simulate_conversation(): # 用户1查询订单 await handle_customer_query(“我的订单ORD-123456到哪里了”, current_user_id1001) print(“---”) # 用户2报告问题 await handle_customer_query(“我昨天买的手机屏幕不亮了怎么办”, current_user_id1002) asyncio.run(simulate_conversation())当你运行这段代码时观察控制台输出。你会看到类似[工具调用] 正在查询订单 ORD-123456...的日志。模型会根据你的系统提示和用户问题自主决定调用哪个工具甚至可能不调用并将工具返回的结果整合到它对用户的最终文本回复中。result.data就是这段最终回复。核心机制解析这个过程背后是OpenAI的Tool Calling功能。pydantic-ai将你的工具函数列表转换成标准的OpenAI Tool Schema发送给模型。模型在思考后如果决定调用工具会在响应中返回一个特殊的“tool call”请求其中包含要调用的函数名和参数一个JSON字符串。pydantic-ai框架会捕获这个请求找到对应的本地函数用Pydantic模型解析参数执行函数再将执行结果作为新一轮消息发送给模型让模型生成面向用户的总结性回答。4.3 依赖注入实战为工具提供数据库会话上面的工具函数是“无状态”的。真实场景中query_order_status需要访问数据库。我们通过依赖注入来实现。首先定义一个“依赖提供者”函数from typing import Annotated from pydantic_ai import Depends # 模拟一个数据库会话类 class DBSession: async def execute(self, query): print(f“[DB] 执行查询{query}”) return “模拟的数据库结果” # 依赖提供者这个函数负责创建或获取一个DB会话 async def get_db_session() - DBSession: # 这里可以从连接池获取一个会话 print(“[依赖注入] 创建新的DB会话。”) return DBSession() # 重写工具函数声明它对db_session的依赖 async def query_order_status_with_db( order: OrderQueryInput, db_session: Annotated[DBSession, Depends(get_db_session)] # 声明依赖 ) - str: 带数据库依赖的订单查询工具 # 现在我们可以使用db_session了 result await db_session.execute(f“SELECT status FROM orders WHERE id‘{order.order_id}’”) return f“订单 {order.order_id} 状态为{result}”然后在创建Agent时需要启用依赖注入支持并注册新的工具customer_agent_advanced Agent( modelmodel, system_prompt“...”, tools[query_order_status_with_db, create_support_ticket], deps[get_db_session] # 注册依赖提供者 )现在当模型决定调用query_order_status_with_db时pydantic-ai会自动调用get_db_session()来获取一个数据库会话并将其注入到工具函数中。这使得工具函数的测试变得极其简单你可以注入一个mock的session也完美解耦了业务逻辑和资源管理。5. 性能调优与生产级部署考量当你的AI代理从demo走向生产性能、成本和可靠性就成为关键。pydantic-ai提供了一些机制来应对这些挑战。5.1 控制成本与延迟缓存、重试与限流1. 结果缓存Caching对于具有确定性的查询例如从固定文本中提取信息可以启用缓存避免为相同输入重复调用昂贵的模型API。from pydantic_ai import Agent from pydantic_ai.caching import InMemoryCache agent Agent( modelmodel, result_typeWeatherQuery, cacheInMemoryCache() # 使用内存缓存 )pydantic-ai支持自定义缓存后端你可以轻松集成Redis等分布式缓存实现跨进程/服务的缓存共享。2. 智能重试与退避网络波动或模型服务暂时不可用可能导致失败。Agent的run方法内置了重试逻辑可配置次数和退避策略。更重要的是由于输出是结构化的你可以针对“验证失败”设计特定的重试策略。例如第一次调用模型输出格式错误第二次重试时可以在系统提示中加强格式要求。3. 限流Rate Limiting如果你直接使用OpenAI等云服务其本身有速率限制。在客户端你也可以通过像asyncio.Semaphore这样的机制或者在调用Agent的上一层实现一个简单的令牌桶来控制并发请求数避免触发服务端的429错误。5.2 可观测性Observability与调试生产系统必须可监控、可调试。pydantic-ai通过结构化日志和回调Callbacks提供了支持。1. 结构化日志框架内部记录了关键事件如模型调用开始、工具调用、结果验证成功/失败等。你可以配置Python的logging模块来捕获这些日志。import logging logging.basicConfig(levellogging.INFO) # 你会看到类似 INFO:pydantic_ai.agent:Running agent... 的日志2. 回调函数这是更强大的监控和调试工具。你可以注册回调函数在Agent运行的各个生命周期节点如on_run_start,on_model_call,on_tool_call,on_result_validated执行自定义逻辑。from pydantic_ai import Agent, RunContext from pydantic_ai.messages import ModelMessage async def log_model_call(ctx: RunContext, messages: list[ModelMessage]): 记录每次发给模型的Prompt和收到的响应 print(f“### 模型调用 ###”) print(f“Prompt消息数{len(messages)}”) # 可以记录到文件或监控系统 # 注意messages可能包含长文本生产环境建议采样或记录摘要 agent Agent( modelmodel, result_typeWeatherQuery, system_prompt“...”, ) # 注册回调 agent.register_callback(‘on_model_call’ log_model_call)通过回调你可以实现记录每次AI调用的耗时和Token使用量估算成本、将对话历史存入数据库供分析、在特定错误发生时触发告警等。5.3 测试策略Mock与隔离测试AI应用一直是个难题因为模型输出非确定。pydantic-ai的架构让单元测试变得可行。1. Mock模型响应你可以创建一个“MockModel”直接返回你预设的响应从而在不调用真实API的情况下测试你的Agent逻辑、工具调用链和结果验证。from pydantic_ai import Agent from pydantic_ai.models.mock import MockModel from pydantic_ai.messages import ModelMessage, ModelResponse class MyMockModel(MockModel): async def run(self, messages: list[ModelMessage]) - ModelResponse: # 根据messages的内容返回一个预设的、符合你输出模型格式的响应 # 例如如果用户消息包含“天气”就返回一个填充好的WeatherQuery的JSON fake_response ModelResponse( messageModelMessage.assistant(‘{“location”: “MockCity” “query_date”: “2023-10-01” “info_type”: “all”}’) ) return fake_response mock_agent Agent(modelMyMockModel(), result_typeWeatherQuery) # 现在对mock_agent.run的调用将是完全确定性的适合单元测试。2. 测试工具函数由于工具函数是普通的异步Python函数且依赖可以被注入你可以像测试任何业务函数一样测试它们使用pytest和asyncio轻松模拟各种输入和依赖状态。6. 常见陷阱、排查技巧与最佳实践在实际使用pydantic-ai构建复杂应用的过程中我踩过不少坑也总结出一些让项目更稳健的经验。6.1 陷阱一模型不按格式输出这是最常见的问题。你定义了一个精致的Pydantic模型但模型返回的JSON缺胳膊少腿或者多出一些未知字段。排查与解决强化系统提示在system_prompt中明确、反复强调输出格式。例如“你必须且只能返回一个合法的JSON对象该对象必须严格遵循以下JSON Schema定义不要添加任何额外的解释、注释或Markdown代码块标记。”简化输出模型初期尽量使用简单字段字符串、数字、枚举。避免复杂的嵌套或联合类型Union直到模型能稳定处理简单结构。使用strict模式在Pydantic模型配置中设置model_config ConfigDict(strictTrue)。这会让模型在遇到额外字段时直接报错迫使你检查是模型输出问题还是Schema定义过宽。启用调试查看原始响应使用前面提到的on_model_call回调打印出模型返回的原始消息。很多时候问题在于模型在JSON外面包裹了json ...这样的Markdown标记你需要调整Prompt或在后处理中剥离它们pydantic-ai有尝试自动处理一些常见情况。6.2 陷阱二工具调用不准确或泛滥模型可能错误地调用工具或者在不需要时调用工具。排查与解决精炼工具描述工具函数及其输入模型的description字段至关重要。用清晰、无歧义的语言描述工具的功能和适用场景。例如query_order_status的描述应该是“根据一个有效的订单编号查询该订单的当前物流状态”而不是简单的“查询订单”。在系统提示中设定规则明确告诉模型什么情况下该调用什么工具什么情况下不应该调用。例如“用户只有提供明确的订单号时才使用query_order_status工具。如果用户只是泛泛地问‘我的订单怎么样了’你应该引导用户提供订单号而不是直接调用工具。”控制工具暴露范围不是每个Agent都需要所有工具。根据Agent的职责只给它注册必要的工具集减少模型的困惑。6.3 陷阱三依赖注入的循环依赖或初始化开销如果依赖提供者get_db_session本身又依赖其他复杂服务可能会形成循环依赖或导致每次调用都初始化昂贵资源。最佳实践使用单例或缓存依赖对于数据库连接池、配置对象等确保依赖提供者返回的是共享实例而不是每次都新建。你可以在get_db_session内部从全局连接池获取会话。分层依赖复杂的依赖链可以拆解。例如一个工具需要UserService和DB你可以定义两个独立的依赖提供者而不是让一个依赖另一个。惰性初始化在依赖提供者内部直到真正需要时才初始化资源。6.4 性能最佳实践批处理请求如果你有大量独立的结构化数据提取任务不要用for循环一个个调用agent.run()。考虑将这些任务组合成一个列表让模型一次处理多个如果模型上下文窗口允许或者使用异步并发asyncio.gather来同时发起多个请求但要注意云服务的速率限制。精简上下文传递给Agent的context和对话历史message_history会占用Token增加成本和延迟。定期清理过长的历史只保留必要的上下文。模型选型不是所有任务都需要GPT-4。对于格式固定、逻辑简单的提取任务gpt-3.5-turbo甚至更小的模型可能表现相当且成本大幅降低。做好A/B测试。验证开销对于超大型或深度嵌套的Pydantic模型验证本身可能成为性能瓶颈。在极高吞吐场景下评估验证开销必要时可考虑对已验证过的缓存结果跳过二次验证。pydantic-ai代表了一种构建AI应用的更工程化、更可靠的思路。它将Python生态中久经考验的数据验证模式与新兴的LLM能力相结合在享受AI灵活性的同时为程序注入了确定性。从简单的数据提取到复杂的、带有多步工具调用的智能体它提供了一个可扩展、可测试、可观测的坚实基础。当然它也不是银弹Prompt设计、工具定义、错误处理依然需要开发者的精心打磨。但有了这个框架你至少可以告别那些脆弱的字符串解析和冗长的数据校验代码更专注于构建真正有价值的AI功能逻辑。
Pydantic-AI：用声明式数据验证范式构建可靠的大语言模型应用
发布时间：2026/5/16 2:34:05
1. 项目概述当Pydantic遇上AI数据验证的范式革命如果你在过去几年里写过Python尤其是做过Web开发或者数据处理那你大概率听说过或用过Pydantic。它凭借基于Python类型注解的、运行时强制的数据验证和序列化能力几乎成了FastAPI的“官方CP”也让无数开发者从手动写一堆if-else判断参数有效性的繁琐工作中解放出来。但今天聊的不是Pydantic本身而是一个可能改变我们构建AI应用方式的新玩意儿——pydantic-ai。简单说它试图用Pydantic那套优雅、声明式的范式来“管束”和“结构化”大语言模型LLM那看似自由不羁的输出。这听起来有点“给野马套上缰绳”的意思。我们调用OpenAI、Anthropic的API时最头疼的问题之一就是输出的不可控性。你让模型生成一段JSON它可能给你来个残缺的括号或者多出几个注释你让它严格按某个格式回答它一高兴就自由发挥了。于是我们不得不写大量的后处理代码用正则表达式去抠、用json.loads配合try...except去解析、写复杂的逻辑去校验和修正。pydantic-ai的核心想法就是为什么不让模型直接输出一个Pydantic模型实例呢这样一来从模型那里拿到结果的那一刻它就已经是一个经过验证的、类型安全的、可以直接在你的业务逻辑里使用的Python对象了。这不仅仅是方便更是一种思维模式的转变将AI从“黑盒文本生成器”变成了“可靠的结构化数据生成器”。这个项目适合任何正在或将要在生产环境中集成LLM的开发者、工程师无论是构建智能客服、数据分析助手还是复杂的多智能体工作流都能从中获得巨大的确定性和开发效率提升。2. 核心设计理念声明式AI交互的四大支柱pydantic-ai不是简单地在openai库外面包一层。它的设计渗透着Pydantic哲学旨在为AI交互提供一个坚固、可预测的框架。我们可以从四个核心支柱来理解它的设计思路。2.1 支柱一结果即对象验证前置传统流程是发送Prompt - 接收文本/JSON字符串 - 解析 - 验证 - 转换为对象。pydantic-ai将其压缩为发送Prompt附带输出模型定义 - 接收并自动验证 - 得到Pydantic对象。验证被前置到了模型调用环节由框架替你完成。这意味着如果模型输出不符合你定义的Pydantic模型比如缺少必需字段、字段类型不对这次调用在框架层面就会失败或进行重试而不会让一个半成品数据流入你的核心业务代码。这极大地增强了系统的健壮性。实操心得这种设计特别适合对数据质量要求高的场景。比如你让AI从一段客户邮件中提取“订单号”、“产品SKU”和“问题描述”。订单号必须是特定格式的数字SKU必须是存在的产品代码。用传统方法你需要在解析后手动校验这些规则。用pydantic-ai你只需要在Pydantic模型里用Field和自定义验证器定义好这些规则框架会确保你拿到的对象一定是合规的。2.2 支柱二提示词Prompt的模块化与结构化在pydantic-ai中Prompt不再是简单的字符串模板拼接。它被抽象成更结构化的组件。你可以定义“系统消息”、“用户消息”并且可以动态地注入上下文context。更重要的是这个上下文本身也可以是Pydantic模型对象框架会自动将其序列化成适合模型理解的格式通常是JSON。这使得构建复杂的、多轮次的对话或工作流变得更加清晰避免了字符串格式化带来的混乱和潜在错误。一个典型场景你有一个客户支持机器人需要根据用户的历史工单一个列表、当前产品信息一个对象和本次问题描述来生成回复。传统方式下你需要手动把这些数据拼进一个巨大的Prompt字符串里。在pydantic-ai中你可以为“对话上下文”定义一个Pydantic模型包含history_tickets,product_info,current_query等字段。在调用时你只需创建这个上下文对象并传入框架负责将其整合到Prompt中。2.3 支柱三依赖注入Dependency Injection式的工具调用OpenAI的Function Calling后更名为Tools是让LLM与现实世界交互的关键。pydantic-ai对此提供了原生支持并且设计得非常巧妙。你定义的工具函数可以像FastAPI的路由处理函数一样声明参数而这些参数的类型提示会被框架用来自动生成工具的JSON Schema。更强大的是它支持依赖注入。这意味着你的工具函数可以声明需要“数据库会话”或“当前用户”这样的依赖项框架会在调用工具时自动解析并提供这些依赖而不是让你在全局状态中到处传递。这解决了什么痛点想象一下你的AI助手有一个工具叫get_user_order需要user_id和db_session。没有依赖注入你或许需要在一个全局变量中保存数据库连接或者以某种方式传递。有了依赖注入你只需在工具函数中声明db: DBSession参数并在框架中配置好如何获取DBSession比如从连接池获取剩下的框架自动搞定。这让工具函数的编写更纯粹、更易于测试。2.4 支柱四多模型支持与统一抽象虽然例子常用OpenAI但pydantic-ai在设计上就考虑了对多种模型后端的支持。它提供了一个统一的抽象层让你可以用几乎相同的代码与OpenAI、Anthropic Claude、甚至是本地部署的模型通过兼容OpenAI API的服务器进行交互。你只需要更换配置中的base_url和api_key。这为未来切换模型供应商或进行多模型降级提供了便利。注意事项尽管抽象是统一的但不同模型在性能、成本、对特定Prompt格式的偏好上仍有差异。在pydantic-ai中切换模型后端后可能需要对Prompt的措辞或温度temperature等参数进行微调以达到最佳效果。框架负责通信协议但无法完全抹平模型之间的特性差异。3. 从零开始构建你的第一个结构化AI代理理论说了不少现在让我们动手从一个最简单的例子开始感受pydantic-ai的工作流。我们将构建一个“天气信息提取器”从一段用户文本中提取结构化信息。3.1 环境准备与安装首先确保你的Python环境在3.8以上。然后安装核心库pip install pydantic-ai openai这里我们同时安装了openai因为我们将使用OpenAI的模型作为后端。pydantic-ai本身不绑定特定模型客户端它使用httpx进行HTTP通信但需要对应的SDK来提供API密钥等配置管理或者你直接配置端点。重要配置你需要设置你的OpenAI API密钥。最安全的方式是使用环境变量export OPENAI_API_KEYyour-api-key-here或者在代码中通过os.environ设置仅用于测试生产环境切勿硬编码。3.2 定义输出模型告诉AI你想要什么这是最核心的一步。我们想要从“上海明天天气怎么样”或“纽约下周一的温度是多少”这样的句子中提取出location地点、date日期和information_type信息类型是温度、天气状况还是风速。from pydantic import BaseModel, Field from datetime import date from enum import Enum class WeatherInfoType(str, Enum): TEMPERATURE “temperature” CONDITION “condition” # 如 sunny, rainy WIND “wind” ALL “all” # 全部信息 class WeatherQuery(BaseModel): 从用户查询中提取的天气查询信息 location: str Field(description“城市或地区名如‘上海’、‘New York’) query_date: date Field(description“查询的日期默认为今天”, default_factorydate.today) info_type: WeatherInfoType Field(description“用户想查询的天气信息类型”, defaultWeatherInfoType.ALL) # 一个自定义验证器示例确保location不是空字符串 field_validator(‘location’) classmethod def location_not_empty(cls, v: str) - str: if not v or not v.strip(): raise ValueError(‘location cannot be empty’) return v.strip()我们定义了两个模型。WeatherInfoType是一个枚举限定了信息类型的可选值这能极大减少模型的“胡思乱想”。WeatherQuery是我们的主输出模型使用了Field来为每个字段提供更详细的描述这些描述会作为上下文送给大模型指导它如何理解并填充这些字段。default_factory用于提供默认值今天的日期。我们还添加了一个简单的验证器确保地点非空。3.3 创建AI代理Agent并运行在pydantic-ai中与模型交互的核心单元是Agent。我们需要创建一个Agent并指定它使用哪个模型以及输出格式。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel # 1. 选择模型。这里使用gpt-3.5-turbo成本较低适合实验。 # 你也可以用‘gpt-4’, ‘gpt-4-turbo’等。 model OpenAIModel(‘gpt-3.5-turbo’) # 2. 创建Agent。关键参数result_type指定了我们期望的输出模型。 weather_agent Agent( modelmodel, result_typeWeatherQuery, system_prompt“你是一个精准的信息提取助手。请从用户的自然语言查询中严格提取出与天气相关的结构化信息。” ) # 3. 运行Agent传入用户消息。 async def main(): user_message “请问旧金山下周五的天气状况” result await weather_agent.run(user_message) # result.data 就是我们想要的WeatherQuery对象 extracted_info: WeatherQuery result.data print(f“地点: {extracted_info.location}”) print(f“日期: {extracted_info.query_date}”) print(f“查询类型: {extracted_info.info_type}”) # 你可以像使用任何Pydantic对象一样使用它 if extracted_info.location “San Francisco”: # 调用你的天气API... pass # 使用asyncio运行 import asyncio asyncio.run(main())运行这段代码你会看到控制台打印出结构化的信息。result.data直接就是一个WeatherQuery实例你可以直接访问它的属性调用它的方法或者将它序列化成JSONextracted_info.model_dump_json()发给其他服务。整个过程你没有写一行解析JSON或校验数据的代码。踩坑点在初期测试时你可能会遇到模型输出不符合格式导致验证失败的情况。pydantic-ai的Agent.run()默认会重试可配置次数。如果一直失败你需要检查1. 字段描述是否足够清晰2. 枚举值是否覆盖了所有可能情况3. Prompt特别是系统提示是否明确要求了输出格式通常在系统提示中强调“请严格按指定JSON格式输出”会有帮助。4. 进阶实战构建带工具调用的客服助手仅仅提取信息还不够真正的AI应用需要“行动”。我们构建一个更复杂的客服助手它不仅能理解用户问题还能通过调用工具查询数据库、创建工单。4.1 定义工具Tools假设我们有两个工具一个根据订单号查询订单状态另一个为用户创建新的支持工单。from pydantic import BaseModel, Field from typing import Optional # 首先定义工具调用所需的输入模型 class OrderQueryInput(BaseModel): order_id: str Field(description“订单编号格式为‘ORD-XXXXXX’”) class CreateTicketInput(BaseModel): user_id: int Field(description“用户ID”) problem: str Field(description“用户描述的问题”) priority: str Field(description“优先级可选‘low’, ‘medium’, ‘high’”, default“medium”) # 然后实现工具函数本身。这里用模拟函数代替真实数据库操作。 async def query_order_status(order: OrderQueryInput) - str: 根据订单号查询订单状态模拟函数 # 这里应该是数据库查询逻辑 print(f“[工具调用] 正在查询订单 {order.order_id}...”) # 模拟一个响应 fake_statuses [“已发货” “配送中” “待付款” “已完成”] import random return f“订单 {order.order_id} 状态为{random.choice(fake_statuses)}” async def create_support_ticket(ticket: CreateTicketInput) - str: 为用户创建支持工单模拟函数 print(f“[工具调用] 正在为用户 {ticket.user_id} 创建工单问题{ticket.problem[:50]}... 优先级{ticket.priority}”) # 模拟创建成功返回工单号 import uuid ticket_id str(uuid.uuid4())[:8] return f“工单创建成功工单号TICKET-{ticket_id}。我们的客服将在24小时内联系您。”注意工具函数的输入参数是一个Pydantic模型OrderQueryInput/CreateTicketInput。pydantic-ai会利用这个模型的字段定义自动生成供大模型理解的Tool Schema包括名称、描述、参数JSON Schema。4.2 创建带工具的Agent并运行现在我们将工具装配到Agent上。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel model OpenAIModel(‘gpt-4’) # 复杂任务建议使用理解能力更强的GPT-4 # 创建Agent时通过tools参数注册工具 customer_agent Agent( modelmodel, system_prompt“””你是一个专业的电商客服助手。你的任务是 1. 理解用户关于订单或账户问题的查询。 2. 如果需要查询订单请使用query_order_status工具。 3. 如果用户需要人工帮助或报告新问题请使用create_support_ticket工具。 4. 根据工具返回的结果用友好、清晰的语气回复用户。 请严格判断只在必要时调用工具。不要对工具能力之外的问题进行猜测。“””, tools[query_order_status, create_support_ticket] # 注册工具 ) async def handle_customer_query(query: str, current_user_id: int): 处理用户查询的入口函数 print(f“用户提问{query}”) # 运行Agent。注意我们还可以传入‘context’这里把当前用户ID作为上下文。 result await customer_agent.run( query, context{‘user_id’: current_user_id} # 上下文可以传递给工具 ) # result.data 是模型的最终文本回复 final_answer result.data print(f“助手回复{final_answer}”) return final_answer # 模拟对话 async def simulate_conversation(): # 用户1查询订单 await handle_customer_query(“我的订单ORD-123456到哪里了”, current_user_id1001) print(“---”) # 用户2报告问题 await handle_customer_query(“我昨天买的手机屏幕不亮了怎么办”, current_user_id1002) asyncio.run(simulate_conversation())当你运行这段代码时观察控制台输出。你会看到类似[工具调用] 正在查询订单 ORD-123456...的日志。模型会根据你的系统提示和用户问题自主决定调用哪个工具甚至可能不调用并将工具返回的结果整合到它对用户的最终文本回复中。result.data就是这段最终回复。核心机制解析这个过程背后是OpenAI的Tool Calling功能。pydantic-ai将你的工具函数列表转换成标准的OpenAI Tool Schema发送给模型。模型在思考后如果决定调用工具会在响应中返回一个特殊的“tool call”请求其中包含要调用的函数名和参数一个JSON字符串。pydantic-ai框架会捕获这个请求找到对应的本地函数用Pydantic模型解析参数执行函数再将执行结果作为新一轮消息发送给模型让模型生成面向用户的总结性回答。4.3 依赖注入实战为工具提供数据库会话上面的工具函数是“无状态”的。真实场景中query_order_status需要访问数据库。我们通过依赖注入来实现。首先定义一个“依赖提供者”函数from typing import Annotated from pydantic_ai import Depends # 模拟一个数据库会话类 class DBSession: async def execute(self, query): print(f“[DB] 执行查询{query}”) return “模拟的数据库结果” # 依赖提供者这个函数负责创建或获取一个DB会话 async def get_db_session() - DBSession: # 这里可以从连接池获取一个会话 print(“[依赖注入] 创建新的DB会话。”) return DBSession() # 重写工具函数声明它对db_session的依赖 async def query_order_status_with_db( order: OrderQueryInput, db_session: Annotated[DBSession, Depends(get_db_session)] # 声明依赖 ) - str: 带数据库依赖的订单查询工具 # 现在我们可以使用db_session了 result await db_session.execute(f“SELECT status FROM orders WHERE id‘{order.order_id}’”) return f“订单 {order.order_id} 状态为{result}”然后在创建Agent时需要启用依赖注入支持并注册新的工具customer_agent_advanced Agent( modelmodel, system_prompt“...”, tools[query_order_status_with_db, create_support_ticket], deps[get_db_session] # 注册依赖提供者 )现在当模型决定调用query_order_status_with_db时pydantic-ai会自动调用get_db_session()来获取一个数据库会话并将其注入到工具函数中。这使得工具函数的测试变得极其简单你可以注入一个mock的session也完美解耦了业务逻辑和资源管理。5. 性能调优与生产级部署考量当你的AI代理从demo走向生产性能、成本和可靠性就成为关键。pydantic-ai提供了一些机制来应对这些挑战。5.1 控制成本与延迟缓存、重试与限流1. 结果缓存Caching对于具有确定性的查询例如从固定文本中提取信息可以启用缓存避免为相同输入重复调用昂贵的模型API。from pydantic_ai import Agent from pydantic_ai.caching import InMemoryCache agent Agent( modelmodel, result_typeWeatherQuery, cacheInMemoryCache() # 使用内存缓存 )pydantic-ai支持自定义缓存后端你可以轻松集成Redis等分布式缓存实现跨进程/服务的缓存共享。2. 智能重试与退避网络波动或模型服务暂时不可用可能导致失败。Agent的run方法内置了重试逻辑可配置次数和退避策略。更重要的是由于输出是结构化的你可以针对“验证失败”设计特定的重试策略。例如第一次调用模型输出格式错误第二次重试时可以在系统提示中加强格式要求。3. 限流Rate Limiting如果你直接使用OpenAI等云服务其本身有速率限制。在客户端你也可以通过像asyncio.Semaphore这样的机制或者在调用Agent的上一层实现一个简单的令牌桶来控制并发请求数避免触发服务端的429错误。5.2 可观测性Observability与调试生产系统必须可监控、可调试。pydantic-ai通过结构化日志和回调Callbacks提供了支持。1. 结构化日志框架内部记录了关键事件如模型调用开始、工具调用、结果验证成功/失败等。你可以配置Python的logging模块来捕获这些日志。import logging logging.basicConfig(levellogging.INFO) # 你会看到类似 INFO:pydantic_ai.agent:Running agent... 的日志2. 回调函数这是更强大的监控和调试工具。你可以注册回调函数在Agent运行的各个生命周期节点如on_run_start,on_model_call,on_tool_call,on_result_validated执行自定义逻辑。from pydantic_ai import Agent, RunContext from pydantic_ai.messages import ModelMessage async def log_model_call(ctx: RunContext, messages: list[ModelMessage]): 记录每次发给模型的Prompt和收到的响应 print(f“### 模型调用 ###”) print(f“Prompt消息数{len(messages)}”) # 可以记录到文件或监控系统 # 注意messages可能包含长文本生产环境建议采样或记录摘要 agent Agent( modelmodel, result_typeWeatherQuery, system_prompt“...”, ) # 注册回调 agent.register_callback(‘on_model_call’ log_model_call)通过回调你可以实现记录每次AI调用的耗时和Token使用量估算成本、将对话历史存入数据库供分析、在特定错误发生时触发告警等。5.3 测试策略Mock与隔离测试AI应用一直是个难题因为模型输出非确定。pydantic-ai的架构让单元测试变得可行。1. Mock模型响应你可以创建一个“MockModel”直接返回你预设的响应从而在不调用真实API的情况下测试你的Agent逻辑、工具调用链和结果验证。from pydantic_ai import Agent from pydantic_ai.models.mock import MockModel from pydantic_ai.messages import ModelMessage, ModelResponse class MyMockModel(MockModel): async def run(self, messages: list[ModelMessage]) - ModelResponse: # 根据messages的内容返回一个预设的、符合你输出模型格式的响应 # 例如如果用户消息包含“天气”就返回一个填充好的WeatherQuery的JSON fake_response ModelResponse( messageModelMessage.assistant(‘{“location”: “MockCity” “query_date”: “2023-10-01” “info_type”: “all”}’) ) return fake_response mock_agent Agent(modelMyMockModel(), result_typeWeatherQuery) # 现在对mock_agent.run的调用将是完全确定性的适合单元测试。2. 测试工具函数由于工具函数是普通的异步Python函数且依赖可以被注入你可以像测试任何业务函数一样测试它们使用pytest和asyncio轻松模拟各种输入和依赖状态。6. 常见陷阱、排查技巧与最佳实践在实际使用pydantic-ai构建复杂应用的过程中我踩过不少坑也总结出一些让项目更稳健的经验。6.1 陷阱一模型不按格式输出这是最常见的问题。你定义了一个精致的Pydantic模型但模型返回的JSON缺胳膊少腿或者多出一些未知字段。排查与解决强化系统提示在system_prompt中明确、反复强调输出格式。例如“你必须且只能返回一个合法的JSON对象该对象必须严格遵循以下JSON Schema定义不要添加任何额外的解释、注释或Markdown代码块标记。”简化输出模型初期尽量使用简单字段字符串、数字、枚举。避免复杂的嵌套或联合类型Union直到模型能稳定处理简单结构。使用strict模式在Pydantic模型配置中设置model_config ConfigDict(strictTrue)。这会让模型在遇到额外字段时直接报错迫使你检查是模型输出问题还是Schema定义过宽。启用调试查看原始响应使用前面提到的on_model_call回调打印出模型返回的原始消息。很多时候问题在于模型在JSON外面包裹了json ...这样的Markdown标记你需要调整Prompt或在后处理中剥离它们pydantic-ai有尝试自动处理一些常见情况。6.2 陷阱二工具调用不准确或泛滥模型可能错误地调用工具或者在不需要时调用工具。排查与解决精炼工具描述工具函数及其输入模型的description字段至关重要。用清晰、无歧义的语言描述工具的功能和适用场景。例如query_order_status的描述应该是“根据一个有效的订单编号查询该订单的当前物流状态”而不是简单的“查询订单”。在系统提示中设定规则明确告诉模型什么情况下该调用什么工具什么情况下不应该调用。例如“用户只有提供明确的订单号时才使用query_order_status工具。如果用户只是泛泛地问‘我的订单怎么样了’你应该引导用户提供订单号而不是直接调用工具。”控制工具暴露范围不是每个Agent都需要所有工具。根据Agent的职责只给它注册必要的工具集减少模型的困惑。6.3 陷阱三依赖注入的循环依赖或初始化开销如果依赖提供者get_db_session本身又依赖其他复杂服务可能会形成循环依赖或导致每次调用都初始化昂贵资源。最佳实践使用单例或缓存依赖对于数据库连接池、配置对象等确保依赖提供者返回的是共享实例而不是每次都新建。你可以在get_db_session内部从全局连接池获取会话。分层依赖复杂的依赖链可以拆解。例如一个工具需要UserService和DB你可以定义两个独立的依赖提供者而不是让一个依赖另一个。惰性初始化在依赖提供者内部直到真正需要时才初始化资源。6.4 性能最佳实践批处理请求如果你有大量独立的结构化数据提取任务不要用for循环一个个调用agent.run()。考虑将这些任务组合成一个列表让模型一次处理多个如果模型上下文窗口允许或者使用异步并发asyncio.gather来同时发起多个请求但要注意云服务的速率限制。精简上下文传递给Agent的context和对话历史message_history会占用Token增加成本和延迟。定期清理过长的历史只保留必要的上下文。模型选型不是所有任务都需要GPT-4。对于格式固定、逻辑简单的提取任务gpt-3.5-turbo甚至更小的模型可能表现相当且成本大幅降低。做好A/B测试。验证开销对于超大型或深度嵌套的Pydantic模型验证本身可能成为性能瓶颈。在极高吞吐场景下评估验证开销必要时可考虑对已验证过的缓存结果跳过二次验证。pydantic-ai代表了一种构建AI应用的更工程化、更可靠的思路。它将Python生态中久经考验的数据验证模式与新兴的LLM能力相结合在享受AI灵活性的同时为程序注入了确定性。从简单的数据提取到复杂的、带有多步工具调用的智能体它提供了一个可扩展、可测试、可观测的坚实基础。当然它也不是银弹Prompt设计、工具定义、错误处理依然需要开发者的精心打磨。但有了这个框架你至少可以告别那些脆弱的字符串解析和冗长的数据校验代码更专注于构建真正有价值的AI功能逻辑。
