如何使用Pydantic AI构建具备生产级功能的智能代理
使用原始的LLM SDK来构建AI代理，在开发原型阶段确实可行，但一旦你需要结构化输出、可测试的代码以及具备生产环境可靠性的系统，这些问题就会显现出来。 这些问题的出现具有很强的规律性。你的笔记本代码可以正常运行，于是你将其应用到生产环境中，并开始添加各种补丁：比如为`json.loads`添加异常处理逻辑，编写辅助函数来去除Markdown格式的标记，使用`if`语句检查字段类型，设置重试机制，以及创建一个将工具名称与对应的可调用函数关联起来的映射函数。这些代码单独来看并不复杂，但当它们汇集在一起时，就会占据你代码库的大部分内容，而真正的代理逻辑反而被这些辅助代码所掩盖。 本文将按照这些问题
使用原始的LLM SDK来构建AI代理，在开发原型阶段确实可行，但一旦你需要结构化输出、可测试的代码以及具备生产环境可靠性的系统，这些问题就会显现出来。
这些问题的出现具有很强的规律性。你的笔记本代码可以正常运行，于是你将其应用到生产环境中，并开始添加各种补丁：比如为`json.loads`添加异常处理逻辑，编写辅助函数来去除Markdown格式的标记，使用`if`语句检查字段类型，设置重试机制，以及创建一个将工具名称与对应的可调用函数关联起来的映射函数。这些代码单独来看并不复杂，但当它们汇集在一起时，就会占据你代码库的大部分内容，而真正的代理逻辑反而被这些辅助代码所掩盖。
本文将按照这些问题出现的顺序，逐一分析它们，并通过具体代码示例展示Pydantic AI是如何解决这些问题的：
非结构化输出需要复杂的解析流程——你的输出数据格式仅以字符串的形式存在，与代码期望接收的字典结构完全脱节。
工具定义中充斥着大量重复代码——例如为三个工具编写了大约70行的JSON格式定义和调用逻辑，但却没有任何机制来确保这些定义与函数签名保持一致。
没有合适的方法来传递运行时上下文——一旦框架调用了这些工具，你就必须通过全局变量或闭包才能将数据库连接信息或用户身份等信息传递给它们。
进行测试需要实际调用大型语言模型——每次测试都会产生费用，耗时数秒，还需要网络支持，而且结果还可能不稳定。
重试和验证逻辑都是手动实现的——你在构建的每一个代理系统中都要重复编写相同的验证、提示重新输入以及重试的处理流程。
更换模型意味着需要重新编写集成代码——不同的提供商提供的SDK在格式、工具接口和响应结构上都有所不同。
在整个讲解过程中，我们将以一个具体的运行示例作为贯穿始终的案例：这个示例是一个用于分析收据信息的代理系统。它接收原始的收据文本（比如通过图像识别软件转换得到的文本），调用相应的工具来查询商家类别和汇率，然后生成一份结构化输出结果——包括商家名称、消费类别、商品明细以及置信度评分——这些信息可以直接被预算管理工具或费用统计软件所使用。这种结构化的输入方式、用于数据查询的工具接口，以及为下游系统提供的格式化输出结果，都是非常常见的设计模式，因此由此产生的问题也会让人感到十分熟悉。
阅读完本文后，你将获得一个功能完备的代理系统：它能够生成结构化输出结果，使用依赖注入机制来管理工具组件，具备自动重试的功能来进行业务规则验证，并且其测试套件可以在不需要API密钥的情况下在几毫秒内完成运行。
目录
关于Pydantic AI的简要介绍
Pydantic是一个基于Python的数据验证库。你可以通过定义带有类型提示的普通Python类来构建数据结构，Pydantic会在运行时对这些结构进行验证——在合适的情况下强制转换数据类型，拒绝不符合要求的输入，并生成能明确指出错误所在的具体错误信息。
from pydantic import BaseModel, Field
class Item(BaseModel):
name: str
amount: float = Field(gt=0)
Item(name="Espresso", amount="2.50") # → amount被强制转换为2.5
Item(name="Espresso", amount=-1) # → 输出错误：amount必须大于0
Pydantic AI则是一个用于处理大语言模型相关任务的代理框架。它能够将你的模型转换成适合相应工具使用的结构化请求，解析并验证返回的数据，根据函数签名生成相应的工具定义，自动处理依赖关系，并在验证失败时重新尝试执行操作。你的代理逻辑仍然可以使用Python编写，而这个框架会负责完成双向的转换工作。
先决条件
本文假设你已经掌握了以下内容：
Python 3.10+ — 包含类型提示、数据类以及异步编程功能
大语言模型API的基础知识 — 至少曾经使用过OpenAI、Anthropic或类似的SDK进行过调用
代理的概念 — 了解什么是AI代理（即大语言模型加上相关工具与推理机制）。如果还不熟悉，可以先阅读《AI代理——构建指南》
你不需要事先具备使用Pydantic AI的经验，我们可以从零开始学习。
问题：在没有框架的情况下构建代理程序
为了更好地说明这些问题，我们以一个具体的例子来进行讲解：一个用于分析收据信息的代理程序。这种程序会接收原始的收据文本（比如通过照片识别软件转换得到的文本），对支出项目进行分类，查询商家信息，然后生成结构化的摘要结果。这个摘要会包含商家名称、支出类别、详细的项目列表以及置信度评分。下游系统（如预算管理工具或费用报告软件）可以直接使用这种结构化输出数据。
这种模式在现实世界中非常常见：输入数据需要是结构化的，需要通过特定的工具进行查询操作，而输出结果也需要满足下游系统的格式要求。接下来我们就来看看，如果直接使用OpenAI SDK的原始接口来构建这样的代理程序，会遇到哪些问题。
问题1：非结构化输出数据需要复杂的解析流程
从根本上来说，任何与大语言模型的交互都只是文本形式的输入和输出。你与模型之间的所有约定——包括输入数据、目标以及期望的输出格式——都被压缩成了一条简单的文本指令。没有专门的框架、类型系统或编译器来确保这些约定的正确性。你只能用英语描述自己的需求，然后希望模型能够按照这些要求进行操作。
下面是使用OpenAI SDK实现的简单示例。请注意，系统的指令中必须同时包含输入数据的上下文信息、任务要求，以及输出数据的结构格式，所有这些内容都被合并成了一条文本字符串：
import json
from openai import OpenAI
client = OpenAI()
def analyzereceipt(receipt_text: str) -> dict:
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": """分析这张收据，并返回以下格式的JSON数据：
{
"merchant": "字符串",
"category": "食品、交通、公共事业、娱乐、购物或其他之一",
"total": 浮点数,
"currency": "字符串",
"items": [{"name": "字符串", "amount": 浮点数}],
"is_business_expense": 布尔值,
"confidence": 0到1之间的浮点数
}""},
{"role": "user", "content": receipt_text}
]
)
raw = response.choices[0].message.content
# 解析响应数据
try:
if raw.startswith("```"):
raw = raw.split("\n", 1)[1].rsplit("```", 1)[0]
result = json.loads(raw)
except json.JSONDecodeError:
raise ValueError(f"大语言模型返回了无效的JSON数据：{raw[:200]}")
# 手动验证字段内容
allowed_categories = {"food", "transport", "utilities", "entertainment", "shopping", "other"}
if result.get("category") not in allowed_categories:
result["category"] = "other"
return result
这段代码结构清晰、易于阅读，而且在你的笔记本环境中可以正常运行。但根本的问题在于：你与大语言模型之间的交互协议（包括输入格式、输出格式等）仅仅存储在一个非结构化的字符串中，没有任何机制来确保双方都能严格遵守这一协议。
如果将这种代码应用到生产环境中，就会遇到各种问题：
提示信息本身就代表了输出格式，但这个描述只是用英语写的：系统使用的提示语是用自然语言描述输出格式的，而你的代码却期望接收某种特定的数据结构。如果在提示语中添加了某个字段，但却在后续处理环节忽略了它，那么在生产环境中才会出现问题。
大语言模型并不总是会返回格式规范的JSON数据：它可能会用````json ```这样的标签将输出数据括起来，或者在前后添加解释性文字，还可能在数据末尾添加逗号，或者在中断时只返回部分数据。你的解析代码目前只能处理其中一种情况，而其他问题却无法得到解决。
缺乏有效的验证机制：例如，`total`这个字段到底是不是一个数字？大语言模型会不会返回像`"$45.99"`这样的字符串？`confidence`这个值应该在0到1之间，但它会不会返回95这样百分比形式的数值？`items`列表里是否真的包含键值对格式的数据？这些都需要你手动去检查。
错误处理机制存在问题：当出现错误时，系统要么什么反应都没有，要么会引发严重的后果。例如，当分类信息无法确定时，代码只是简单地将`result["category"]`设置为“other”，而实际上这种情况下应该重新尝试获取数据。另外，`json.loads`方法失败时会抛出异常，但并没有提供任何恢复机制。
我们真正需要的是一种方式：能够通过代码明确地定义输出数据的格式，使用结构化的数据类型来表示这些信息，而不是用英语写的描述性文字。这种格式规范应该成为大语言模型和调用它的程序双方都必须遵守的统一标准。
我们还需要这样的框架，以便能够自动执行模式验证，并根据类型定义来检查大语言模型的响应结果，当出现不匹配的情况时，能够正确地生成错误提示。
最后，如果验证失败，系统应该能够自动进行重试，而无需人工干预。如果输出结果不符合预定的模式，就应该向大语言模型重新发送包含验证错误的请求，使其能够自我纠正。
简而言之：输出格式应该是一种用类型系统表达的“规范”，而不是用英语写成的“建议”。
Pydantic AI是如何解决这个问题的
Pydantic AI允许你将输出结果定义为一个Pydantic模型。该框架会负责生成模式结构、插入提示信息、解析JSON数据、进行验证以及处理重试逻辑，而所有这些功能都是基于同一个模型定义来实现的：
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from enum import Enum
class SpendingCategory(str, Enum):
FOOD = "food"
TRANSPORT = "transport"
UTILITIES = "utilities"
ENTERTAINMENT = "entertainment"
SHOPPING = "shopping"
OTHER = "other"
class LineItem(BaseModel):
name: str
amount: float
class ReceiptAnalysis(BaseModel):
merchant: str
category: SpendingCategory
total: float = Field(gt=0)
currency: str = Field(min_length=3, max_length=3)
items: list[LineItem]
is_business_expense: bool
confidence: float = Field(ge=0, le=1)
receipt_agent = Agent(
"openai:gpt-4o",
output_type=ReceiptAnalysis,
system_prompt="Analyze the provided receipt and extract structured details.",
)
result = receipt_agent.run_sync("CAFE PARIS\n€12.50\nCroissant x2 €5.00\nEspresso €2.50\nCroque Monsieur €5.00")
print(result.output)
# merchant='CAFE PARIS' category= total=12.5 ...
那么，这里有什么不同之处呢？
首先，模式本身就构成了这个模型。ReceiptAnalysis类定义了各个字段的类型及约束条件，Pydantic AI会将这些信息转换成大语言模型能够理解的JSON格式，并据此验证模型的输出结果。同一个模型定义可以被到处使用。
其次，根本不需要手动解析代码。无需去除Markdown标记，也不需要调用json.loads或处理JSONDecodeError异常，因为整个框架会自动完成这些工作。
另外，验证是严格且实时的。confidence字段上设置的ge=0, le=1约束意味着值为95的结果会被直接拒绝，而不会被默默接受。而SpendingCategory作为枚举类型，只允许使用有效的类别值，不存在任何默认值或兜底处理机制。
最后，重试是自动进行的。如果大语言模型返回的结果无法通过验证，Pydantic AI会将验证错误信息传回模型，要求其进行自我修正，完全不需要人工编写重试逻辑。
最终得到的结果是一个ReceiptAnalysis对象——这个对象的类型是明确的，已经过验证，而且非常适合在集成开发环境中使用自动补全功能。它并不是一个你希望其中包含正确键值的dict对象。
幕后发生了什么
当您指定 `output_type=ReceiptAnalysis` 时，Pydantic AI会在每次运行代理程序时执行一些关键操作。
在数据输入阶段，它会根据您的 Pydantic 模型生成一个 JSON 结构规范，并将其嵌入到 LLM 的请求中。根据不同的模型提供者，这种方式会利用相应的原生结构化输出机制（例如 OpenAI 的 `response_format`、Anthropic 的工具调用接口等），从而确保 LLM 能够“准确”地知道应该生成什么样的数据结构。
在数据返回阶段，Pydantic AI 会解析 LLM 提供的原始响应数据，然后根据之前生成的 JSON 结构规范对其进行验证——包括类型检查、字段约束以及枚举成员资格的验证。如果验证失败，它会将错误信息反馈给 LLM，并要求其重新生成正确的输出结果（系统会自动尝试多次重试，直到达到可配置的重试次数上限为止）。
┌──────────────────────────────────────────────────────────────────────┐
│ Pydantic AI — 结构化数据输出流程 │
│ │
│ ┌────────────────┐ ┌──────────────────────────────────┐ │
│ │ 您的代码 │ │ Pydantic AI 框架 │ │
│ │ │ │ │ │
│ │ output_type = │────────>│ 1. 根据 ReceiptAnalysis 模型生成 JSON 结构规范 │ │
│ │ │ │ │ │
│ └────────────────┘ │ 2. 将结构规范嵌入到 LLM 的请求中 │ │
│ │ （使用提供者支持的原生格式，如 response_format、tool_call 等） │ │
│ │ │ │ │
│ └──────────────┼───────────────────┘ │
│ ▼ │
│ ┌──────────────────────────────────┐ │
│ │ LLM │ │
│ │ 接收到结构规范后，会生成对应的 JSON 数据 │ │
│ └──────────────┬───────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────┐ │
│ │ Pydantic AI 框架会继续对 JSON 数据进行验证 │ │
│ │ │ │
│ │ 3. 解析 LLM 返回的原始响应数据 │ │
│ │ 4. 根据模型规范进行验证： │ │
│ │ - 类型检查 │ │
│ │ - 字段范围检查 │ │
│ │ - 枚举成员资格验证 │ │
│ │ │ │ │
│ │ ┌────┴────┐ │ │
│ │ │ │ │ │
│ │ 通过 ✓ 失败 ✗ │ │
│ │ │ │ │ │
│ │ ▼ ▼ │ │
│ │ 将验证结果以对象形式返回给 LLM │ │
│ │ 以便 LLM 自动进行修正 │ │
│ └──────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────┐ │
│ │ 您的代码最终会收到： │ │
│ │ 已经过验证、结构清晰的结果数据 │ │
│ └──────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────────┘你只需将相关功能定义为一个Python类，该框架就会自动处理大语言模型相关的所有事务——它会告诉模型应该生成什么结果，并验证模型是否确实生成了预期的结果。
问题2：工具定义过于冗余、缺乏灵活性
你的收单系统需要一些工具，以便能够查询商家类别、查看汇率以及检索消费记录。
如果使用原始的函数调用方式来实现这些功能，代码会看起来像这样：
tools = [
{
"type": "function",
"function": {
"name": "lookup_merchant_category",
"description": "根据商家名称查询对应的消费类别"
"parameters": {
"type": "object",
"properties": {
"merchant_name": {
"type": "string",
"description": "收单记录中提到的商家名称"
}
},
"required": ["merchant_name"]
}
}
},
{
"type": "function",
"function": {
"name": "get_exchange_rate",
"description": "获取两种货币之间的当前汇率"
"parameters": {
"type": "object",
"properties": {
"from_currency": {
"type": "string",
"description": "源货币代码（例如：EUR）"
},
"to_currency": {
"type": "string",
"description": "目标货币代码（例如：USD）"
}
},
"required": ["from_currency", "to_currency"]
}
}
},
{
"type": "function",
"function": {
"name": "get_spending_history",
"description": "查询指定日期范围内的消费明细"
"parameters": {
"type": "object",
"properties": {
"category": {
"type": "string",
"description": "消费类别"
},
"days": {
"type": "integer",
"description": "需要查询的过去天数"
}
},
"required": ["category", "days"]
}
}
}
]
# 此外，你还需要编写相应的逻辑来处理这些工具调用请求：
def handle_tool_call-tool_call):
name = tool_call.function.name
args = json.loads(tool_call.functionarguments)
if name == "lookup_merchant_category":
return lookup_merchant_category(args["merchant_name"])
elif name == "get_exchange_rate":
return get_exchange_rate(args["from_currency"], args["to_currency"])
elif name == "get_spending_history":
return get_spending_history(args["category"], args["days"])
else:
raise ValueError(f"未知的工具：{name}")
对于这三个工具来说，你总共编写了大约70行的JSON模式代码以及相应的调度逻辑。然而这种设计存在一个问题：模式与函数的实际参数签名是脱节的——如果你修改了函数中的参数名称却忘记更新对应的JSON模式，那么在运行时程序就会出错。
解决方案应该具备哪些特点
其实，工具的定义应该直接从函数本身中得出。函数的名称、文档字符串以及类型提示已经足以说明该工具的功能及其接受的参数类型，这些信息完全足够了。
此外，这些信息还应该能够自动保持同步。如果你更改了某个参数的名称或类型，那么发送给大语言模型的JSON模式也应该会自动更新，而你根本不需要去修改任何额外的文件。
最后，这种工具的设计不应该需要额外的调度逻辑。框架应该直接调用相应的函数，而不需要通过复杂的if/elif语句来手动将字符串名称与可调用对象进行匹配。
Pydantic AI是如何解决这个问题的
在Pydantic AI中，工具其实就是带有装饰器的函数。框架会根据函数的参数签名和文档字符串自动生成JSON模式，并且会自动处理调用的逻辑：
from pydantic_ai import Agent, RunContext
receipt_agent = Agent(
"openai:gpt-4o",
output_type=ReceiptAnalysis,
systemprompt="Analyze the provided receipt and extract structured details.",
)
@receipt_agent.toolplain
def lookup_merchant_category(merchant_name: str) -> str:
"""根据商家名称查找对应的消费类别。"""
# 实际的实现代码
categories_db = {"CAFE PARIS": "food", "UBER": "transport", "NETFLIX": "entertainment"}
return categories_db.get(merchant_name.upper(), "other")
@receipt_agent.toolplain
def get_exchange_rate(from_currency: str, to_currency: str) -> float:
"""获取两种货币之间的当前汇率。"""
# 实际的实现代码——可能是调用API或查询缓存等
rates = {"EUR_USD": 1.08, "GBP_USD": 1.27}
return rates.get(f"{from_currency}_{to_currency}", 1.0)
@receipt_agent.toolplain
def get_spending_history(category: str, days: int) -> dict:
"""获取指定日期范围内的按类别划分的支出总额。"""
# 实际的实现代码
return {"category": category, "total": 142.50, "transaction_count": 12}
就是这样，根本不需要编写JSON模式字典或额外的调度函数。这三个工具总共只需要大约25行代码，而不是70行。
框架为你做了以下几件事：
根据类型提示自动生成JSON模式：例如`merchant_name: str`在JSON模式中会变成`{"type": "string"}`，文档字符串则会成为工具的`description`字段，参数名称也会自动转换为相应的属性名。所有这些信息都直接从你已经编写好的代码中提取而来。
自动进行函数调用：当大语言模型调用`get_exchange_rate`时，框架会直接跳转到被装饰过的函数进行执行，根本不需要进行任何字符串匹配或手动映射操作。
确保信息同步：如果你在函数签名中将`from_currency`改为`source_currency`，那么下次运行时JSON模式会自动更新。这样就完全避免了遗漏修改的情况。
问题3：无法以简洁的方式传递运行时上下文
在自动调度机制下（参见问题2），框架会调用你的工具函数，而不是你直接控制这些调用过程。因此你无法简单地将`db`或`user_id`作为额外参数传递给这些函数。
而这些本来也不应该是大型语言模型应该提供的功能。你需要一种机制，以便将运行时所需的依赖项传递给框架代为调用的工具函数。
如果没有这样的机制，最终就会出现如下这种情况：
# 方案A：使用全局变量（不可测试且不安全）
db = get_database_connection()
current_user = None # 应该在工具函数运行之前设置这个值…
def get_spending_history(category: str, days: int) -> dict:
# 这个函数使用了全局变量`db`和`current_user`，如何对其进行测试呢？
# 如何让两个用户同时使用这个函数呢？
return db.query(
"SELECT sum(amount) FROM transactions WHERE user_id = ? AND category = ? AND date > ?",
current_user.id, category, days_ago(days)
)
# 方案B：使用闭包（虽然可行，但结构复杂）
def make_tools(db, user):
def get_spending_history(category: str, days: int) -> dict:
return db.query(...) # 这个函数会捕获外部作用域中的`db`和`user`变量
def lookup_merchant_category(merchant_name: str) -> str:
return db.query(...) # 同样使用闭包机制
return [get_spending_history, lookup_merchant_category]
# 每次添加新的依赖项，都需要重新调整闭包的结构这两种方法都会让测试变得非常麻烦。你无法轻松地替换为模拟数据库或测试用户，而不需要修改代码结构。
理想的解决方案应该是什么样的
为了获得更好的解决方案，你应该明确说明你的工具函数需要哪些依赖项。将这些依赖项（如数据库、HTTP客户端、用户会话信息）以类型化的方式定义出来，使其与大型语言模型提供的参数区分开来。
此外，这些依赖项应该在运行时注入，而不是在代码定义阶段就确定下来。在运行代理程序时传递具体的实例，而在定义工具函数时则不需要指定这些依赖项。这样可以让工具函数的定义更加简洁、易于复用。
对于测试来说，也可以灵活地替换这些依赖项。例如，可以用内存中的模拟数据库代替真实的数据库，用测试用例代替真实用户，而无需修改工具函数的代码。
Pydantic AI是如何解决这个问题的
Pydantic AI拥有一个功能强大的依赖项注入系统。你可以在代理程序中定义所需的依赖项类型，然后工具函数会通过类型化的`RunContext`对象来获取这些依赖项，从而避免使用全局变量或闭包：
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
@dataclass
class ReceiptDeps:
db: DatabaseClient
user_id: str
http_client: HttpClient
receipt_agent = Agent(
"openai:gpt-4o",
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
system_prompt="分析提供的收据并提取结构化信息。",
)
@receipt_agent.tool
def get_spending_history(ctx: RunContext[ReceiptDeps], category: str, days: int) -> dict:
"""按类别统计指定日期范围内的消费总额。"""
return ctx.deps.db.query(
"SELECT sum(amount), count(*) FROM transactions WHERE user_id = ? AND category = ? AND date > ?",
ctx.deps.user_id, category, days_ago(days)
)
@receipt_agent.tool
def get_exchange_rate(ctx: RunContext[ReceiptDeps], from_currency: str, to_currency: str) -> float:
"""获取两种货币之间的当前汇率。"""
response = ctx.deps.http_client.get(f"/rates/{from_currency}/{to_currency}")
return response.json()["rate"]
# 在运行时传递真实的依赖项
result = receipt_agent.run_sync(
"CAFE PARIS\n€12.50\nCroissant x2",
deps=ReceiptDeps(
db=get_database_connection(),
user_id="user_123",
http_client=HttpClient(base_url="https://api.exchangerate.host"),
),
)
# 在测试时使用模拟依赖项，无需修改工具函数的代码
result = receipt_agent.run_sync(
"CAFE PARIS\n€12.50\nCroissant x2",
deps=ReceiptDeps(
db=InMemoryDb(fake_transactions),
user_id="test_user",
http_client=MockHttpClient(fixed_rate=1.08),
),
)
这能为您带来以下好处：
工具会明确说明自己需要什么，而不会指定如何获取这些资源： `ctx.deps.db` 的类型是明确的。您的集成开发环境会自动完成与之相关的代码补全功能，类型检查器也能检测到不当的使用情况。该工具并不关心所使用的连接到底是真实的 Postgres 数据库还是测试用模拟对象。
不存在全局变量或闭包： 依赖关系会在 `run_sync()` 被调用时被明确地处理。两个同时运行的用户会分别获得两个独立的 `ReceiptDeps` 实例，这些实例之间没有共享的可变状态。
测试变得非常简单： 只需要将 `DatabaseClient` 替换为 `InMemoryDb`，将 `HttpClient` 替换为 `MockHttpClient` 即可。工具代码本身无需做任何修改。既不需要进行任何临时性的修改，也不需要依赖依赖注入框架或测试 fixture 来访问模块级别的状态。
大语言模型根本看不到这些依赖关系： `RunContext` 并不会作为参数被传递给大语言模型。大语言模型只能看到 `category` 和 `days` 这些信息，因为框架会自动从数据结构中剔除这些依赖关系。
问题 4：测试需要真正的大语言模型调用
您希望验证自己的代理程序能否正确处理一些边缘情况，比如外币收据、缺失的商家名称或模糊的分类信息。但所有的测试实际上都会直接访问真实的 API：
def test_foreign_currencyreceipt():
# 这个测试：
# - 需要消耗网络资源（会调用 API）
# - 执行时间约为 2-5 秒
#> 结果具有不确定性（今天可能通过，明天可能会失败）
#> 需要网络连接；如果没有配置相应的访问凭据，在持续集成环境中就会出现问题
result = analyzereceipt("CAFÉ PARIS\n€12.50\nCroissant x2")
assert result["currency"] == "EUR"
assert result["category"] == "food" # 但实际上可能会返回 "dining" —— 这种情况很不稳定！您无法在持续集成环境中可靠地运行这类测试。如果进行 50 次这样的边缘情况测试，您的 API 资源很快就会被耗尽。最终，要么根本无法进行任何测试，要么进行的集成测试结果会反复出现不稳定现象。
解决方案应该是什么样的
首先，可以用一个可预测的替代品来取代大语言模型——这个替代品能够返回可控的结果，这样测试就能快速、高效且可重复地进行。
其次，代理程序的逻辑本身不需要改变。测试应该仍然能够调用真实的工具来进行数据分发、验证以及结果解析，只有模型部分才是被模拟的。
最后，测试的重点应该是检查程序的实际行为，而不是大语言模型生成的具体文本。要确认是否使用了正确的工具，并且这些工具是否被传入了正确的参数；同时还要验证输出结果是否符合预期的结构。
Pydantic AI 是如何解决这个问题的
Pydantic AI 提供了 `TestModel` 和 `FunctionModel` 这两种工具。使用这些工具，您可以完全控制“大语言模型”返回的结果，而无需进行任何网络调用。
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai.models.function import FunctionModel
# TestModel — 会自动返回符合规范且可预测的响应结果
def testreceipt_analysis_structure():
"""测试该代理是否能够返回一个有效的ReceiptAnalysis对象。」
with receipt_agent.override(model=TestModel()):
result = receipt_agent.run_sync(
"CAFE PARIS\n€12.50\nCroissant x2",
deps=ReceiptDeps(
db=InMemoryDb(fake_transactions),
user_id="test_user",
http_client=MockHttpClient(fixed_rate=1.08),
),
)
# TestModel会用符合规范的有效虚拟数据填充各种字段
assert isinstance(result.output, ReceiptAnalysis)
assert 0 <= result.output.confidence <= 1
# FunctionModel — 允许你为特定场景定制响应内容
def test_foreign_currency_triggers_exchange_rate_tool():
"""测试当收到以欧元计价的订单时，该代理是否会调用getexchange_rate函数。」
def mock_model(messages, info):
# 模拟大语言模型决定调用汇率查询工具的行为
return ModelResponse(
tool_calls=[ToolCall(name="get_exchange_rate", args={"from_currency": "EUR", "to_currency": "USD"})]
)
with receipt_agent.override(model=FunctionModel(mock_model)):
result = receipt_agent.run_sync(
"CAFE PARIS\n€12.50\nCroissant x2",
deps=ReceiptDeps(
db=InMemoryDb(fake_transactions),
user_id="test_user",
http_client=MockHttpClient(fixed_rate=1.08),
),
)
# 确认汇率查询工具确实被调用了
tool_calls = [msg for msg in result.all_messages() if hasattr(msg, "tool_name")]
assert any(tc.tool_name == "get_exchange_rate" for tc in tool_calls)
这种方法非常实用，因为它既快速又免费：无需进行API调用，也不会消耗网络资源或令牌，测试耗时仅几毫秒而已。
它的结果也是可预测的——对于相同的输入，每次都会得到相同的输出。由于大语言模型的状态变化或表述上的差异，不会出现测试结果不一致的情况。
真正的智能体逻辑依然会正常运行。工具调度、依赖注入以及输出验证等流程都不会受到影响，只是使用的模型被替换成了其他模型而已。
此外，这种方法也非常适合集成到持续集成环境中。你在持续集成环境中不需要使用任何API密钥，也无需管理任何敏感信息或遵守任何速率限制。
你还可以拥有两种级别的控制方式：可以使用`TestModel`来进行“基础功能是否正常”的测试；而使用`FunctionModel`则可以测试“智能体是否能做出正确的决策”，因为你可以通过脚本来指定大语言模型的行为。
问题5：重试与验证逻辑是手工实现的
当大语言模型返回错误的响应时，就需要进行重试。但这种重试逻辑很快就会变得非常复杂：
def analyzereceipt_with_retry(receipt_text: str, max_retries: int = 3) -> dict:
for attempt in range(max_retries):
try:
response = client.chat.completions.create(...)
raw = response.choices[0].message.content
result = json.loads(strip_markdown(raw))
# 验证结果
if not isinstance(result.get("total"), (int, float)):
raise ValueError("总金额必须是数字")
if result.get("confidence", 0) > 1 or result.get("confidence", 0) < 0:
raise ValueError("置信度必须在0到1之间")
if result.get("category") not in ALLOWED_CATEGORIES:
raise ValueError(f"无效的分类：{result.get('category')}")
return result
except (json.JSONDecodeError, ValueError, KeyError) as e:
if attempt == max_retries - 1:
raise
# 是否需要将错误信息反馈给大语言模型？或者需要修改提示语？
# 如何记录哪些尝试失败了以及原因是什么？
continue
raise RuntimeError("不应该走到这一步")
你开发的每一个智能体都需要这种重试/验证/重新提示的机制，而每次你都得重新编写这些代码。将验证错误信息反馈给大语言模型以便其自我纠正，这一机制又增加了额外的复杂性。
解决方案应该具备哪些特点
首先，验证过程应该是声明式的——也就是说，应该根据预定义的输出格式来进行验证，而不是通过在代码中编写大量的if语句来手动完成验证工作。
其次，重试机制应该是自动化的。如果输出结果无法通过验证，框架应该自动向大语言模型重新发送请求，并附上错误信息，以便它能够自我纠正。
最后，自定义的验证规则应该能够方便地被添加进来。对于那些超出类型检查范围的业务规则（例如“如果分类为‘其他’，置信度必须低于0.8”），你应该能够直接添加相应的验证逻辑，而无需重新修改重试机制的相关代码。
Pydantic AI是如何解决这个问题的
对于模式级别的验证，Pydantic模型已经能够处理这类任务（如问题1中所示）。但对于业务逻辑方面的验证，Pydantic AI提供了`result_validator`这一功能。这是一个在数据解析完成后执行的装饰器，它可以在验证失败时触发自动重试机制：
from pydantic_ai import Agent, RunContext, ModelRetry
receipt_agent = Agent(
"openai:gpt-4o",
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
system_prompt="分析提供的收据并提取其中的结构化信息。",
retries=3, # 验证失败时最多重试3次
)
@receipt_agent.result_validator
def validate Receipt_analysis(ctx: RunContext[ReceiptDeps], result: ReceiptAnalysis) -> ReceiptAnalysis:
"""业务逻辑验证——在模式验证通过后执行。"""
# 规则：如果总金额与各项金额之和不一致，让LLM进行修正
items_sum = sum(item.amount for item in result.items)
if abs(result.total - items_sum) > 0.01:
raise ModelRetry(
f"总金额({result.total})与各项金额之和({items_sum})不一致。"
"请重新核对收据，确保总金额或各项金额无误。"
)
# 规则：如果分类为“其他”，且置信度较低，很可能意味着LLM无法给出准确结果——需要重试
if result.category == SpendingCategory.OTHER and result.confidence < 0.5:
raise ModelRetry(
"分类为‘其他’，且置信度很低。请仔细核对商家名称和商品信息，以确定更准确的分类。"
)
return result
当验证失败时，系统会按照以下流程进行操作：
┌─────────────────────────────────────────────────────────────┐
│ 自动重试流程 │
│ │
│ LLM的响应 │
│ │ │
│ ▼ │
│ 模式验证（Pydantic模型） │
│ │ │
│ ├── 失败 → 向LLM发送错误信息 → 重试 │
│ │ │
│ ▼ │
│ business逻辑验证 │
│ │ │
│ └─ 触发ModelRetry → 向LLM发送提示信息 → 重试 │
│ │ │
│ ▼ │
│ 验证通过 → 返回最终结果 │
└─────────────────────────────────────────────────────────────┘
Pydantic会自动检测各种规范违规情况，例如数据类型错误、字段缺失或枚举值不匹配等。这些验证错误会被反馈给大语言模型，以便它知道应该修复哪些问题。
业务规则违规情况则由你的`result_validator`代码来检测。
ModelRetry会将你自定义生成的提示信息发送给大语言模型，从而引导它给出正确的响应。你的代码中并没有设置重试机制。参数`retries=3`用于指定最大尝试次数。该框架会负责处理重试逻辑、重新提示用户以及错误信息的格式化工作。
问题6：更换模型意味着需要重新编写集成代码
你的智能体目前是与OpenAI合作的。现在你想要尝试使用Anthropic（对于你的使用场景来说，它的成本更低），或者为了保护数据隐私而选择在本地使用Ollama。不过，不同的提供商提供的SDK、工具调用格式以及响应结构都是不一样的：
# OpenAI
response = openai_client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools # OpenAI的工具调用格式
)
tool_calls = response.choices[0].message.toolCalls
# Anthropic——其API的格式完全不同
response = anthropic_client.messages.create(
model="claude-sonnet-4-20250514",
messages=messages,
tools=anthropic_tools # 与OpenAI的格式截然不同！
)
tool_use_blocks = [b for b in response.content if b.type == "tool_use"]
# Google——又是另一种不同的格式
response = genai_client.generate_content(
contents=messages,
tools=google.tools # 又一种完全不同的格式！
)
function_calls = response.candidates[0].content.parts
最终，你会得到一些与特定提供商相关的代码路径和适配层，而智能体的核心逻辑则被埋藏在这些集成代码之中。
解决方案应该具备什么特点
首先，你应该一次性定义好智能体的核心逻辑。你的工具、输出类型、系统提示语以及验证机制都应该与具体的模型无关。
你应该能够通过简单地更改一个字符串来切换不同的模型，而无需重新编写SDK调用代码、修改工具接口格式或调整响应解析方式。
此外，你应该将那些与特定提供商相关的细节隐藏起来。框架应该能够将你定义的通用智能体逻辑转换成各个提供商所要求的格式。
Pydantic AI是如何解决这个问题的
Pydantic AI对模型类型是完全不敏感的。模型仅仅被视作一个字符串标识符而已：只要更改这个字符串，其他所有部分都不会发生变化：
# 你的智能体定义——工具、输出类型、依赖关系、验证机制——全部保持不变
receipt_agent = Agent(
"openai:gpt-4o", # ← 这是唯一需要更改的行
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
system_prompt="分析提供的收据并提取结构化信息。",
)
# 切换到Anthropic——相同的智能体，相同的工具，相同的输出类型
receipt_agent = Agent(
"anthropic:claude-sonnet-4-20250514",
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
systemprompt="分析提供的收据并提取结构化信息。",
)
# 通过Ollama使用本地模型
receipt_agent = Agent(
"ollama:llama3.1",
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
system_prompt="分析提供的收据并提取结构化信息。",
)
# 或者在运行时进行配置
import os
receipt_agent = Agent(
os.getenv("RECEIPT_AGENT_MODEL", "openai:gpt-4o"),
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
systemprompt="分析提供的收据并提取结构化信息。",
)
该框架在幕后所完成的工作包括：
工具规范的转换： 用户定义的 `
@receipt_agent.tool` 函数会被转换为 OpenAI 的 `tools` 格式、Anthropic 的 `tools` 格式，或 Google 的 `function_declarations` 格式——具体取决于所选择的服务提供商的要求。用户根本不会察觉到这种转换带来的差异。响应数据的标准化： 无论模型返回的是 `choices[0].message.tool_calls`（OpenAI）、`content[].type == "tool_use"`（Anthropic），还是 `candidates[0].content_parts`（Google），该框架都会将其统一转换为一种标准的内部表示形式。
针对不同服务提供商的特殊功能的透明处理： 不同的服务提供商在实现结构化输出、数据流处理以及令牌计数等功能时存在差异，但该框架能够自动适应这些差异，而不会让用户代码感受到这些区别的存在。
只需定义一次代理对象，就可以使用任何模型。这种配置切换可以通过配置文件或环境变量来实现。
总结
本文中提到的所有问题其实都源于同一个根本原因：当调用大型语言模型时，输入的是文本形式的数据，输出也是文本形式的结果；而处理这些请求的代码却是以编程语言编写的。传统的原始 SDK 方法通过一些手工编写的语句来填补这种差距——比如使用条件判断语句、调度表或者重试机制等。虽然每一部分代码单独来看都很简单，但当它们组合在一起时，就会变得复杂不堪，而且用户还需要自己负责管理所有这些细节。
Pydantic AI 则通过将相关规范定义为明确的契约来消除这种问题。下面我们逐一看看这种方式带来了哪些好处：
| 问题 | 原始 SDK 的处理方式 | Pydantic AI 的处理方式 |
|---|---|---|
| 结构化输出 | 需要用英语描述规范，然后手工进行解析 | 使用 `output_type=ReceiptAnalysis`，规范由框架自动生成，响应也会经过验证 |
| 工具定义 | 大约需要 70 行 JSON 代码来定义规范并实现调度逻辑 | 只需在函数上添加 `@agent.toolPLAIN` 即可 |
| 运行时上下文 | 依赖项信息存储在全局变量中或通过嵌套闭包来管理 | 使用 `deps_type` 和 `RunContext`，每次运行时都会重新生成这些信息 |
| 测试 | 需要调用真实的 API，速度慢、费用高且稳定性不可靠 | 可以使用 `TestModel` 或 `FunctionModel` 进行测试，无需连接网络 |
| 重试与错误处理 | 需要手动编写循环代码来处理错误 | 通过字段约束条件和 `ModelRetry` 类来处理错误，并将错误信息反馈给模型 |
| 模型切换 | 需要针对不同的服务提供商分别编写 SDK 代码和解析逻辑 | 只需使用一个字符串即可完成切换，例如 `"openai:gpt-4o"` → `"anthropic:claude-sonnet-4-20250514"` |
最终我们得到的这个“收据代理”其实就是一个 Pydantic 模型，再加上一些类型化的函数、一个用于存储依赖项信息的数据类以及一个验证器而已。整个过程中不需要进行任何解析操作、调度处理或编写重试循环代码。
访问 Pydantic AI 文档 可以了解到更多详细信息，这些文档非常简明易懂，而且其中的内容都与他们的 API 参考文档相对应。
相关文章
如何在没有服务器的情况下为静态网站添加动态功能
静态网站目前正受到人们的青睐，这是有原因的。一个包含HTML、CSS和JavaScript文件的文件夹，加载速度很快，托管成本也很低，而且几乎不可能出现故障。 像 Astro 、 Eleventy 和 Hugo 这样的工具，能够利用Markdown文件和模板帮您生成这样的网站结构。而Netlify、Vercel以及Cloudflare Pages等托管服务，则可以通过内容分发网络来提供这些生成的网站内容，而且通常还是免费的。 不过，您的网站还需要具备实际的功能。读者可能想要留下评论，或者有人想通过电子邮件与您联系。也许您还想实时显示价格信息，需要用户登录后才能查看某些页面，或者在网站发布之前收
阅读全文
如何利用人工智能对传统应用程序进行现代化改造，同时又避免对其进行彻底的重写？
我见过一些旧系统的迁移项目被认为取得了成功，因为那些旧的框架已经从代码库中消失了。 但六个月后，团队仍然在面对同样的耦合问题、同样不清晰的业务规则，以及几乎相同的部署难题。 虽然技术已经发生了变化，但整个系统本身并没有发生太大的改变。 人工智能让这个问题变得更加复杂了。 它能够比人类团队更快地翻译代码，能够解释那些不熟悉的类结构，生成测试用例，创建适配器，更新API接口，从而大大减少重复性工作。 但是，如果你让一个人工智能编码工具去处理一个旧应用程序，并简单地要求它将所有内容都迁移到现代的技术架构中，那么很可能会得到你想要的结果： 还是那个系统，只不过被更快地重新编写了一遍而已。 这并不一定算
阅读全文
如何使用针对用户的OAuth访问机制来构建人工智能代理程序【完整手册】
当你的AI代理同时为多个人提供服务时，每一次工具调用都必须明确：该代理究竟是在代表哪位用户行事。让我们通过构建一个能够与Slack和GitHub连接的AI代理来学习如何解决这个问题。 当使用Slack时，系统会使用 해당用户的 workspace；而在GitHub上创建问题时，也会以该用户的身份在其有权访问的仓库中操作。虽然代理可能会犯错，但它绝对不能使用错误用户的权限来进行操作。 解决这个问题的方法分为两个部分，而这两个部分都在本教程的前半部分进行了讲解： 每位用户都需要单独授权。 Alice为自己授权Slack，Bob也为自己授权Slack。 代理传递的是标识符，而不是令牌。 像 alic
阅读全文
Flutter前端系统设计：在人工智能时代，如何像资深工程师一样思考
系统设计长期以来一直被视为后端领域的问题。 如果你问一群Flutter工程师“系统设计到底意味着什么”，他们中的大多数人会提到服务器架构：负载均衡器、数据库以及微服务。 但如果你让他们设计一个分布式缓存系统或画出一个消息队列的示意图，他们会犹豫不决。而当你要求他们为社交Feed应用开发Flutter客户端时，他们就会立刻打开新文件开始编写组件代码。 这种差距确实存在，不过正在迅速缩小。 随着Flutter应用程序变得越来越复杂——它们具备了实时功能、离线支持、多平台兼容性，同时还包含需要维护的人工智能生成代码——在编写任何一个组件之前所做出的架构决策，其重要性已经与后端架构相当了。 在那些以产
阅读全文