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(ge1, le5, 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_keyyour-key)Agent则是大脑和协调中心。你将定义好的Pydantic模型和ModelClient喂给它它就知道如何工作。recommendation_agent Agent( modelmodel, result_typeRestaurantRecommendation, # 核心声明输出类型 system_prompt你是一个资深美食顾问根据用户需求推荐餐厅。 )最关键的一步是result_typeRestaurantRecommendation。这行代码将之前定义的“契约”赋予了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的AgentPydanticAI通过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( cityquery.city, datequery.date, temperature22.5, condition晴 ) # 将工具绑定到Agent weather_agent Agent( modelmodel, result_typeWeatherResult, tools[get_weather] # 注入工具 )当你问“北京明天天气如何”时Agent会先让LLM思考LLM会“决定”需要调用get_weather工具并自动生成一个符合WeatherQuery模型的参数对象。PydanticAI执行工具后再将WeatherResult返回给LLM进行总结。全程工具的参数输入和结果输出都在类型系统的监控之下杜绝了参数格式错误导致工具调用失败的问题。3. 实战避坑指南从“能用”到“好用”的关键配置掌握了基础我们来看看如何在实际项目中避开那些深水区让PydanticAI真正稳定可靠。3.1 驯服“幻觉”强化系统提示与字段描述LLM的“幻觉”在结构化输出中表现为胡编乱造字段值。虽然类型验证能抓住一部分但更好的方法是从源头遏制。首先系统提示要具体、强硬。不要只说“你是一个助手”。要明确指令Agent( modelmodel, result_typeMyModel, 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( modelmodel, result_typeMyModel, retries2, # 全局重试次数默认3可调低 system_prompt..., )你还可以在run时动态控制result await agent.run( 用户输入, retries0 # 这次运行不重试失败即抛错 )更高级的策略是使用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(ge1, le5) 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(modelmodel, result_typeUserRequest) # Agent 2: 内容生成 class Report(BaseModel): title: str sections: List[str] summary: str report_agent Agent(modelmodel, result_typeReport, 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, statestate) # report_agent的系统提示和工具可以访问state中的信息 return report_result.data通过state我们实现了简单的、类型安全的智能体间通信。对于更复杂的工作流可以考虑结合langgraph等编排框架用PydanticAI作为每个节点的执行引擎。4.2 流式输出与实时体验对于生成较长文本如报告、文章、代码的Agent等待全部生成完毕再返回的体验很差。PydanticAI支持流式响应Streaming。async def stream_report(topic: str): agent Agent(modelmodel, result_typestr) # 结果类型可以是简单的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 与传统系统集成数据库与APIPydanticAI 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(modelmodel, result_typeTicket, 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(maxsize100) 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能力本身。