1. 从“魔法”到“工程”:为什么AI Agent需要类型系统
如果你最近在折腾AI Agent,大概率经历过这样的场景:你精心设计了一个提示词,让大模型去调用一个天气查询API。你满怀期待地输入“北京”,结果它返回给你一个JSON,里面temperature字段的值是“25度”。你写的下游代码正等着一个float类型的数字,于是程序毫无悬念地崩溃了,日志里躺着一个TypeError。你挠挠头,修改提示词,加上“请返回一个数字,不要带单位”。第二次,它返回了25。第三次,你问“纽约”,它可能返回"seventy-two"(字符串形式的七十二),或者更糟,直接告诉你“纽约今天天气晴朗”,完全跳过了你设定的JSON结构。
这就是当前AI Agent开发最典型的“坑”:大模型的输出是非确定性的自由文本。无论你的提示词写得多么详尽,它本质上仍是一种“请求”,而非“约束”。LLM(大语言模型)可能会误解、创造、省略或格式化输出,导致下游代码如同在流沙上建房,脆弱不堪。每一次API调用都像一次冒险,你需要写大量的防御性代码(try...except, 类型检查, 格式清洗)来处理各种边界情况,项目80%的精力可能都花在了和模型输出的“不确定性”作斗争上。
而类型系统,就是我们对抗这种不确定性的最强武器。在传统软件开发中,类型系统(Type System)定义了变量、函数参数和返回值的数据结构(如字符串、整数、对象),并在编译或运行时进行检查,确保数据流动符合预期。它带来的核心价值是契约、验证与自动化。
现在,把这种思想引入AI Agent开发:我们不再用自然语言去“描述”我们希望LLM返回什么,而是用严格的、机器可读的类型定义去“声明”它必须返回什么。这就是PydanticAI在做的事情。它不是一个全新的Agent框架,而是基于鼎鼎大名的数据验证库Pydantic V2构建的,专门用于为LLM的输入输出套上“类型安全”的盔甲。简单说,它让你能用定义Python类一样优雅的方式,去定义你与大模型之间的交互协议,从而把LLM不可靠的文本输出,转化为你代码里可预测、可验证的Python对象。
这带来的改变是根本性的。以前,你需要手动解析、清洗、校验;现在,你只需要定义好“我想要什么”,PydanticAI会帮你生成精准的提示词、解析模型的回复、并确保返回的数据结构完全符合你的定义。如果不符合?它会自动尝试让模型重试,或者清晰地抛出错误告诉你哪里出了问题。这意味着,那些因为格式错误、类型不符、字段缺失导致的bug,其发生概率会直线下降。说“少踩80%的坑”绝非夸张,对于中等复杂度的Agent任务,这甚至是保守估计。
2. PydanticAI核心机制拆解:不只是数据验证
PydanticAI的魔力建立在Pydantic V2坚实的基础上,并针对LLM场景做了关键增强。理解其核心机制,能让你从“会用”到“懂用”。
2.1 基石:Pydantic模型作为交互契约
一切始于一个标准的Pydantic模型。这个模型不仅定义了数据的形状,更定义了与LLM交互的“契约”。
from pydantic import BaseModel, Field from typing import List class RestaurantRecommendation(BaseModel): name: str = Field(description="餐厅的名称") cuisine: str = Field(description="菜系,例如:中餐、意大利菜、日料") price_level: int = Field(ge=1, le=5, description="价格等级,1为最便宜,5为最昂贵") reasons: List[str] = Field(description="推荐这家餐厅的理由,至少列出2条")这个RestaurantRecommendation类就是一个契约。它告诉LLM也告诉你的代码:我们交流的信息必须包含这四个字段,且name和cuisine是字符串,price_level是1到5之间的整数,reasons是一个字符串列表。Field中的description至关重要,它会自动被转换为提示词的一部分,指导LLM生成对应内容。
2.2 引擎:Agent与ModelClient的协作
PydanticAI引入了两个核心概念:Agent和ModelClient。
ModelClient是你的LLM供应商抽象层。通过它,你可以对接OpenAI GPT、Anthropic Claude、Google Gemini,甚至是本地部署的Ollama模型。PydanticAI帮你统一了调用接口。
from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel # 创建一个使用GPT-4的ModelClient model = OpenAIModel('gpt-4-turbo', api_key='your-key')Agent则是大脑和协调中心。你将定义好的Pydantic模型和ModelClient喂给它,它就知道如何工作。
recommendation_agent = Agent( model=model, result_type=RestaurantRecommendation, # 核心:声明输出类型 system_prompt="你是一个资深美食顾问,根据用户需求推荐餐厅。" )最关键的一步是result_type=RestaurantRecommendation。这行代码将之前定义的“契约”赋予了Agent。从此,这个Agent的所有运行,其目标就是产出一个符合RestaurantRecommendation结构的实例。
2.3 魔法发生:提示词注入与结构化输出
当你调用Agent时,魔法开始了。
async def main(): result = await recommendation_agent.run( "我想在上海浦东找一家适合商务宴请的餐厅,预算充足。" ) # result.data 已经是一个RestaurantRecommendation实例! print(f"餐厅:{result.data.name}") print(f"菜系:{result.data.cuisine}") print(f"价格等级:{result.data.price_level}") for reason in result.data.reasons: print(f"- {reason}") # 输出可能类似: # 餐厅:菁禧荟 # 菜系:潮州菜 # 价格等级:5 # - 环境私密典雅,非常适合商务洽谈。 # - 菜品精致,凸显待客的诚意与品味。在这个过程中,PydanticAI自动完成了以下工作:
- 提示词合成:它将你的系统提示、用户输入(“我想在上海浦东...”)以及
RestaurantRecommendation模型中每个字段的description,组合成一个结构化的提示词,明确要求LLM以指定JSON格式回复。 - 输出解析与验证:LLM返回文本后,PydanticAI会尝试将其解析为JSON,并立即用
RestaurantRecommendation模型进行验证。如果price_level返回了“五”,验证器会将其转换为整数5;如果返回了6,验证会失败;如果reasons只给了一条,验证也会失败。 - 自动重试:这是减少“坑”的关键一环。如果验证失败(比如格式错误或字段缺失),PydanticAI默认会将错误信息和修正要求反馈给LLM,让其重试。通常最多重试3次。这相当于一个自动的“格式化校对员”,极大地提高了成功率。
2.4 超越基础:工具调用(Function Calling)的标准化
对于需要执行具体操作(如查询数据库、调用API)的Agent,PydanticAI通过@tool装饰器将普通Python函数转化为Agent可安全调用的工具。其强大之处在于,工具的参数和返回值同样用Pydantic模型定义,实现了端到端的类型安全。
from pydantic_ai import tool class WeatherQuery(BaseModel): city: str = Field(description="城市名称") date: str = Field(description="查询日期,格式YYYY-MM-DD") class WeatherResult(BaseModel): city: str date: str temperature: float = Field(description="日均温度,摄氏度") condition: str = Field(description="天气状况,如:晴、多云、雨") @tool async def get_weather(query: WeatherQuery) -> WeatherResult: """根据城市和日期查询天气。""" # 这里模拟一个API调用 return WeatherResult( city=query.city, date=query.date, temperature=22.5, condition="晴" ) # 将工具绑定到Agent weather_agent = Agent( model=model, result_type=WeatherResult, tools=[get_weather] # 注入工具 )当你问“北京明天天气如何?”时,Agent会先让LLM思考,LLM会“决定”需要调用get_weather工具,并自动生成一个符合WeatherQuery模型的参数对象。PydanticAI执行工具后,再将WeatherResult返回给LLM进行总结。全程,工具的参数输入和结果输出都在类型系统的监控之下,杜绝了参数格式错误导致工具调用失败的问题。
3. 实战避坑指南:从“能用”到“好用”的关键配置
掌握了基础,我们来看看如何在实际项目中避开那些深水区,让PydanticAI真正稳定可靠。
3.1 驯服“幻觉”:强化系统提示与字段描述
LLM的“幻觉”在结构化输出中表现为胡编乱造字段值。虽然类型验证能抓住一部分,但更好的方法是从源头遏制。
首先,系统提示要具体、强硬。不要只说“你是一个助手”。要明确指令:
Agent( model=model, result_type=MyModel, system_prompt="""你是一个严格遵循指令的数据提取助手。你的任务是根据用户输入,精确填充下方定义的JSON结构。 你必须: 1. 只使用用户提供的信息,绝不自行编造任何数据。 2. 如果信息缺失,将对应字段设为null(如果允许)或明确说明。 3. 输出的JSON必须完全符合提供的架构定义。 用户输入如下:""" )其次,字段描述是黄金。Field(description=...)是你与LLM沟通的主要渠道。描述要像给实习生写工作说明一样清晰、无歧义。
- 差描述:
address: str - 好描述:
address: str = Field(description="完整的邮政地址,包括街道门牌号、城市、省份和邮政编码。例如:'上海市浦东新区世纪大道100号 200120'。必须从文本中提取,不得虚构。") - 对于枚举值,使用
Literal类型是更佳选择,它能被直接翻译成提示词中的选项。
from typing import Literal status: Literal['pending', 'processing', 'completed', 'failed'] = Field(description="订单状态,只能是以下选项之一:pending, processing, completed, failed。")3.2 控制成本与延迟:重试策略与模型选择
自动重试是福音,但也可能成为成本和延迟的噩梦。一个复杂的模型在3次重试后仍然失败,消耗的token和时间可能很可观。
精细配置重试逻辑:
from pydantic_ai import Agent, RunContext agent = Agent( model=model, result_type=MyModel, retries=2, # 全局重试次数,默认3,可调低 system_prompt=..., )你还可以在run时动态控制:
result = await agent.run( "用户输入", retries=0 # 这次运行不重试,失败即抛错 )更高级的策略是使用RunContext中的defer。例如,你可以先让一个快速但能力稍弱的模型(如gpt-3.5-turbo)尝试,如果失败,再换用更强但更贵的模型(如gpt-4)。这需要在自定义的Agent逻辑中实现。
模型选择经验:对于简单的信息提取和格式化任务,gpt-3.5-turbo在成本效益上往往优于gpt-4,且响应更快。PydanticAI的严格验证部分弥补了其偶尔的格式错误。但对于需要复杂推理、多步骤工具调用的任务,gpt-4系列更高的指令遵循能力可以减少重试次数,整体成功率更高,反而可能更“经济”。
3.3 处理复杂嵌套与可选字段
现实中的数据很少是扁平简单的。PydanticAI完美支持Pydantic的所有功能。
嵌套模型让结构清晰:
class Address(BaseModel): street: str city: str zip_code: str class Customer(BaseModel): id: int name: str shipping_address: Address # 嵌套 billing_address: Address | None = None # 可选嵌套LLM在生成时,会理解这种嵌套关系,并输出对应的JSON对象。
可选字段与默认值是处理信息缺失的关键。使用Optional[...]或... | None,并合理设置default。
from typing import Optional class ProductReview(BaseModel): product_id: str rating: int = Field(ge=1, le=5) comment: Optional[str] = None # 评论可能没有 helpful_votes: int = 0 # 默认值为0这里有个坑:如果你将comment设为Optional[str],LLM在用户没有提供评论时,可能会在JSON中省略该字段,或者将其设为null。Pydantic都能正确处理。但如果你希望它总是出现(即使是null),可以在字段描述中强调“如果无评论,请将comment字段设为null”。
3.4 调试与监控:看清AI的黑箱
当Agent没有返回预期结果时,你需要知道发生了什么。PydanticAI提供了良好的可观测性。
访问原始消息流:agent.run()返回的Result对象包含.messages属性,这是一个完整的对话历史列表,包含系统提示、用户输入、AI的每次回复(包括重试)以及工具调用信息。这是你的一线调试日志。
result = await agent.run("...") for msg in result.messages: print(f"[{msg.type}] {msg.content}") # 查看所有交互利用result.usage进行成本监控:它包含了本次调用消耗的Prompt Token、Completion Token和总Token数。对于需要控制成本的应用,务必记录和分析这个数据。
print(f"本次调用消耗: {result.usage.total_tokens} tokens")自定义日志记录:你可以传入一个logger到Agent中,或者使用Python的标准logging模块来捕获PydanticAI内部的日志,通常设置在logging.INFO或DEBUG级别,可以看到模型调用、重试等详细信息。
注意:在生产环境中,务必对
result.messages中的内容进行脱敏处理,因为它可能包含用户输入和模型生成的敏感信息。
4. 进阶模式:构建健壮的生产级AI Agent
当单个Agent能稳定工作后,我们需要考虑更复杂的场景:多步骤工作流、流式响应、以及与传统系统的集成。
4.1 多智能体协作与状态管理
复杂的任务通常需要多个Agent分工协作。PydanticAI的Agent本身是相对独立的,但你可以通过共享的“状态”(State)将它们串联起来。
Agent.run()方法可以接受一个state参数,这是一个字典,可以在多个Agent调用间传递和修改信息。
# Agent 1: 信息收集与解析 class UserRequest(BaseModel): topic: str depth: Literal['brief', 'detailed'] parser_agent = Agent(model=model, result_type=UserRequest) # Agent 2: 内容生成 class Report(BaseModel): title: str sections: List[str] summary: str report_agent = Agent(model=model, result_type=Report, system_prompt="你是一个专业的内容撰写者。") async def workflow(user_input: str): # 第一步:解析用户意图 parse_result = await parser_agent.run(user_input) user_req = parse_result.data # 将解析结果放入状态,传递给下一个Agent state = {'topic': user_req.topic, 'depth': user_req.depth} # 第二步:根据意图生成报告 report_prompt = f"请生成一份关于{user_req.topic}的{user_req.depth}报告。" report_result = await report_agent.run(report_prompt, state=state) # report_agent的系统提示和工具可以访问state中的信息 return report_result.data通过state,我们实现了简单的、类型安全的智能体间通信。对于更复杂的工作流,可以考虑结合langgraph等编排框架,用PydanticAI作为每个节点的执行引擎。
4.2 流式输出与实时体验
对于生成较长文本(如报告、文章、代码)的Agent,等待全部生成完毕再返回的体验很差。PydanticAI支持流式响应(Streaming)。
async def stream_report(topic: str): agent = Agent(model=model, result_type=str) # 结果类型可以是简单的str async for chunk in agent.run_stream(f"写一篇关于{topic}的短文:"): # chunk是一个Result对象,但其.data在流式过程中是部分内容 if chunk.data: yield chunk.data # 逐块输出给前端流式输出对于result_type是简单类型(如str)或结构简单且LLM能逐步生成的模型非常有效。对于复杂的嵌套对象,流式支持可能有限,因为模型通常需要思考完整结构后才能输出有效的JSON。
4.3 与传统系统集成:数据库与API
PydanticAI Agent可以无缝融入现有的后端架构。最常见的模式是作为“智能路由”或“增强型API”。
场景:智能客服工单分类
- 用户发送一段文字描述问题。
ClassificationAgent(使用PydanticAI定义)分析文本,输出一个结构化工单对象,包含category(技术问题/账单问题/投诉)、urgency(高/中/低)、summary(问题摘要)。- 后端代码收到这个结构化的
Ticket对象,直接根据category和urgency字段的值,路由到不同的处理队列(如Jira、Zendesk),或存入数据库。 - 由于数据是结构化的,后续的所有自动化处理(如分配工程师、发送确认邮件)都可以可靠地进行。
class Ticket(BaseModel): category: Literal['technical', 'billing', 'complaint', 'other'] urgency: Literal['high', 'medium', 'low'] summary: str customer_id: int | None = None classification_agent = Agent(model=model, result_type=Ticket, system_prompt="...") # 在FastAPI/Django视图中的使用示例 @app.post("/create-ticket") async def create_ticket(user_input: str): result = await classification_agent.run(user_input) ticket_data = result.data.dict() # 转换为字典 # 直接存入数据库或发送到消息队列 db_ticket = await TicketORM.create(**ticket_data) await assign_to_queue(db_ticket) return {"ticket_id": db_ticket.id}这种集成方式干净利落,AI负责理解和结构化非标准输入,传统系统负责可靠的存储、流程和业务逻辑处理,两者边界清晰,极大地降低了系统的整体复杂度。
4.4 性能优化与缓存策略
频繁调用LLM成本高、延迟大。对于相对确定性的任务(如根据固定模板提取信息),可以使用缓存。
PydanticAI可以与langchain的缓存组件或自定义缓存结合。一个简单的策略是基于用户输入的哈希值进行缓存:
from functools import lru_cache import hashlib def get_input_hash(user_input: str, system_prompt: str) -> str: combined = f"{system_prompt}|{user_input}" return hashlib.md5(combined.encode()).hexdigest() @lru_cache(maxsize=100) async def cached_agent_run(user_input: str) -> MyModel: # 这里实际上调用真实的agent.run result = await my_agent.run(user_input) return result.data # 使用时 data = await cached_agent_run("重复的查询内容")注意:缓存仅适用于输入确定、且期望输出也确定的场景。对于创造性任务或实时信息查询,缓存不适用。同时,要警惕缓存可能带来的数据陈旧问题,需要设置合理的过期策略。
经过这几个层次的构建,你的AI Agent已经从一个小脚本,进化为一个拥有严格接口、可观测、可集成、甚至具备一定性能优化能力的生产级组件。PydanticAI提供的类型系统,就是贯穿这一切、保证其内在一致性和可靠性的钢筋骨架。它没有替代你对业务逻辑的思考,而是让你从繁琐的文本解析和错误处理中解放出来,专注于设计更强大的Agent能力本身。