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自动完成了以下工作：

1. 提示词合成 ：它将你的系统提示、用户输入（“我想在上海浦东...”）以及 RestaurantRecommendation 模型中每个字段的 description ，组合成一个结构化的提示词，明确要求LLM以指定JSON格式回复。
2. 输出解析与验证 ：LLM返回文本后，PydanticAI会尝试将其解析为JSON，并立即用 RestaurantRecommendation 模型进行验证。如果 price_level 返回了“五”，验证器会将其转换为整数 5 ；如果返回了 6 ，验证会失败；如果 reasons 只给了一条，验证也会失败。
3. 自动重试 ：这是减少“坑”的关键一环。如果验证失败（比如格式错误或字段缺失），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”。

场景：智能客服工单分类

1. 用户发送一段文字描述问题。
2. ClassificationAgent （使用PydanticAI定义）分析文本，输出一个结构化工单对象，包含 category （技术问题/账单问题/投诉）、 urgency （高/中/低）、 summary （问题摘要）。
3. 后端代码收到这个结构化的 Ticket 对象，直接根据 category 和 urgency 字段的值，路由到不同的处理队列（如Jira、Zendesk），或存入数据库。
4. 由于数据是结构化的，后续的所有自动化处理（如分配工程师、发送确认邮件）都可以可靠地进行。

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能力本身。