如何用Pydantic AI构建生产级智能体
使用原始LLM SDK构建AI代理在原型阶段没问题，直到你需要结构化输出、可测试的代码和生产级可靠性为止。
这种差距会以可预期的方式暴露出来。你的笔记本代码能运行，于是你把它推向生产环境并开始打补丁：在json.loads周围加try/except，写一个去除Markdown围栏的辅助函数，几个检查字段类型的if语句，一个重试循环，一个将工具名称映射到可调用对象的调度函数。这些单独来看都不难，但它们合在一起会成为代码库的主体，真正的代理逻辑湮没在胶水代码之中。
本文按你实际会遇到这些问题的顺序逐一讨论六个问题，并展示Pydantic AI如何用代码解决每一个问题：
非结构化输出需要脆弱的解析——你的输出模式存在于一段英文提示字符串中，与代码所期望的字典脱节。
工具定义充满样板代码——三个工具就需要约70行手写的JSON Schema和调度代码，而且没有任何机制让Schema与函数签名保持同步。
没有干净的方式传递运行时上下文——一旦框架调用你的工具，你无法在不借助全局变量或闭包的情况下向它们传递数据库连接或用户ID。
测试需要真实的LLM调用——每次测试都要花钱、耗时数秒、依赖网络，还可能不稳定。
重试和验证逻辑需要手工编写——你在构建的每个代理中都要重写相同的验证/重新提示/重试模式。
切换模型意味着重写集成代码——每个提供商的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, coerced
Item(name="Espresso", amount=-1) # → ValidationError: amount must be > 0
Pydantic AI 是一个智能体框架，负责处理 LLM 边界——将你的模型转换为提供商原生的模式请求，解析并验证返回内容，从函数签名生成工具定义，注入依赖项，并在验证失败时重试。你的智能体逻辑仍然是 Python 代码；框架负责双向转换。
先决条件
本文假定你熟悉以下内容：
Python 3.10+ — 类型提示、数据类、async/await
LLM API 基础 — 你至少调用过几次 OpenAI、Anthropic 或类似 SDK
智能体概念 — 你理解什么是 AI 智能体（LLM + 工具 + 推理循环）。如果不理解，从 AI 智能体——构建者指南 开始阅读
你不需要有 Pydantic AI 的使用经验。我们将从零开始构建。
问题：在没有框架的情况下构建智能体
为了更好地揭示和解释这些问题，我们将使用一个贯穿全文的示例：一个收据分析智能体。它接收原始收据文本（即通过照片转文字扫描得到的内容），对支出进行分类，查找商家信息，并返回结构化摘要。这样你就能获得商家名称、支出类别、逐项明细和置信度分数。下游系统（预算仪表板、费用报告工具）直接消费这种结构化输出。
这是一种常见的现实世界模式：结构化输入、用于查找的工具调用，以及面向下游系统的类型化输出。让我们看看使用原始 OpenAI SDK 调用来构建它会是什么样子。
问题 1：非结构化输出需要脆弱的解析
从核心来看，与 LLM 的每一次交互都只是文本进、文本出。你与模型的整个约定（输入数据、目标和期望的输出格式）都被塞进一个单一的文本提示中。没有任何模式、类型系统或编译器来强制正确性。你用英语描述你想要的内容，然后希望模型遵守。
以下是使用 OpenAI SDK 的直接实现。注意系统提示如何将输入上下文、任务指令和输出模式全部编码在一大段文本中：
import json
from openai import OpenAI
client = OpenAI()
def analyze_receipt(receipt_text: str) -> dict:
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": """Analyze this receipt and return JSON:
{
"merchant": "string",
"category": "one of: food, transport, utilities, entertainment, shopping, other",
"total": float,
"currency": "string",
"items": [{"name": "string", "amount": float}],
"is_business_expense": bool,
"confidence": float between 0 and 1
}"""},
{"role": "user", "content": receipt_text}
]
)
raw = response.choices[0].message.content
# Parse the response
try:
if raw.startswith("```"):
raw = raw.split("\n", 1)[1].rsplit("```", 1)[0]
result = json.loads(raw)
except json.JSONDecodeError:
raise ValueError(f"LLM returned invalid JSON: {raw[:200]}")
# Validate fields manually
allowed_categories = {"food", "transport", "utilities", "entertainment", "shopping", "other"}
if result.get("category") not in allowed_categories:
result["category"] = "other"
return result
这段代码清晰易读，并且在你的notebook中可以运行。但根本问题在于，你与LLM之间的整个契约（输入、目标和输出格式）存在于一个非结构化的字符串中。双方都没有任何机制来强制执行这个契约。
如果你将其推向生产环境，问题就会开始显现：
提示词就是schema，而且只是英文：系统提示词用自然语言描述了输出格式。没有任何东西将该描述与你的代码实际期望的
dict关联起来。在提示词中添加一个字段却忘记在下游处理它，直到生产环境你才会遇到错误。LLM并不总是返回干净的JSON：它会把输出包裹在
```json ```代码围栏中，在前后添加解释性文本，包含尾随逗号，或者在超时时返回部分响应。你的解析代码只处理了一种情况（代码围栏），但无法处理其他情况。没有真正的验证：
total真的是数字吗，还是LLM返回了字符串"$45.99"？confidence是否在0到1之间，还是返回了95（百分比）？items列表是否包含具有正确键的字典？你不得不手动检查所有这些。失败要么是静默的，要么是灾难性的：类别回退（
result["category"] = "other"）隐藏了一个本应触发重试的问题。而json.loads失败会引发异常，且没有恢复路径。
我们真正需要的是一种方式，能够在代码中一次性定义输出schema，作为类型化的数据结构，而不是提示词字符串中的英文描述。该schema应该是LLM和使用方代码的单一事实来源。
我们还需要框架自动强制实施该schema，并根据类型定义验证LLM的响应，在出现不匹配时给出适当的错误。
最后，我们需要在验证失败时自动重试，无需手动逻辑。如果输出不符合schema，则使用验证错误重新提示LLM，以便其自我纠正。
简而言之：输出格式应该是类型系统中表达的契约，而不是用英文表达的建议。
Pydantic AI 如何解决这个问题
Pydantic AI 允许你将输出定义为 Pydantic 模型。该框架处理 schema 生成、提示词注入、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 ...
So what's different here?
First, the schema is the model. ReceiptAnalysis defines the fields, types, and constraints. Pydantic AI converts this into the appropriate JSON schema for the LLM and validates the response against it. One definition, used everywhere.
Second, there's no parsing of code. You don't strip markdown fences, call json.loads, or catch JSONDecodeError. The framework handles all of that.
Also, validation is real. Field(ge=0, le=1) on confidence means a value of 95 is rejected, not silently accepted. SpendingCategory as an Enum means only valid categories are allowed with no fallback masking.
Finally, retry is automatic. If the LLM returns output that fails validation, Pydantic AI sends the validation error back to the model and asks it to correct itself. There's no hand-rolled retry loop.
The function returns a ReceiptAnalysis object: typed, validated, and IDE-autocomplete-friendly. Not a dict you hope has the right keys.
What happens beneath the surface
When you define output_type=ReceiptAnalysis, Pydantic AI does a few key things on each agent run.
On the way in, it generates a JSON schema from your Pydantic model and injects it into the LLM request. Depending on the model provider, this uses the native structured output / tool-call mechanism (OpenAI's response_format, Anthropic's tool-use, and so on) so the LLM knows exactly what structure to produce.
On the way back, it takes the LLM's raw response, parses it against the Pydantic model, and runs full validation (type coercion, field constraints, and enum membership). If validation fails, it feeds the error message back into the conversation and asks the LLM to correct its output (automatically, up to a configurable retry limit).
┌──────────────────────────────────────────────────────────────────────┐
│ Pydantic AI — Structured Output Flow │
│ │
│ ┌────────────────┐ ┌──────────────────────────────────┐ │
│ │ Your Code │ │ Pydantic AI Framework │ │
│ │ │ │ │ │
│ │ output_type = │────────>│ 1. Generate JSON schema from │ │
│ │ ReceiptAnalysis │ ReceiptAnalysis model │ │
│ │ │ │ │ │
│ └────────────────┘ │ 2. Inject schema into LLM │ │
│ │ request (provider-native │ │
│ │ format: response_format, │ │
│ │ tool_call, etc.) │ │
│ │ │ │ │
│ └──────────────┼───────────────────┘ │
│ ▼ │
│ ┌──────────────────────────────────┐ │
│ │ LLM │ │
│ │ Sees schema → produces JSON │ │
│ └──────────────┬───────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────┐ │
│ │ Pydantic AI Framework │ │
│ │ │ │
│ │ 3. Parse raw LLM response │ │
│ │ 4. Validate against model: │ │
│ │ - Type checks │ │
│ │ - Field constraints (ge, le) │ │
│ │ - Enum membership │ │
│ │ │ │ │
│ │ ┌────┴────┐ │ │
│ │ │ │ │ │
│ │ PASS ✓ FAIL ✗ │ │
│ │ │ │ │ │
│ │ ▼ ▼ │ │
│ │ Return typed Send validation │ │
│ │ object error back to LLM │ │
│ │ for self-correct │ │
│ │ (auto-retry) │ │
│ └──────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────┐ │
│ │ Your Code receives: │ │
│ │ result.output → ReceiptAnalysis │ │
│ │ (typed, validated, ready to use)│ │
│ └──────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────────┘
你只需用 Python 类定义一次契约。框架会处理 LLM 边界的两端，既告诉模型要生成什么，又验证它是否真的做到了。
问题 2：工具定义充满样板代码
你的收据代理需要一些工具，用来查找商家类别、检查汇率以及查询消费历史。
以下是使用原始函数调用时的样子：
tools = [
{
"type": "function",
"function": {
"name": "lookup_merchant_category",
"description": "Look up the spending category for a merchant name",
"parameters": {
"type": "object",
"properties": {
"merchant_name": {
"type": "string",
"description": "The merchant name from the receipt"
}
},
"required": ["merchant_name"]
}
}
},
{
"type": "function",
"function": {
"name": "get_exchange_rate",
"description": "Get current exchange rate between two currencies",
"parameters": {
"type": "object",
"properties": {
"from_currency": {
"type": "string",
"description": "Source currency code (e.g., EUR)"
},
"to_currency": {
"type": "string",
"description": "Target currency code (e.g., USD)"
}
},
"required": ["from_currency", "to_currency"]
}
}
},
{
"type": "function",
"function": {
"name": "get_spending_history",
"description": "Get spending totals by category for a date range",
"parameters": {
"type": "object",
"properties": {
"category": {
"type": "string",
"description": "Spending category"
},
"days": {
"type": "integer",
"description": "Number of past days to query"
}
},
"required": ["category", "days"]
}
}
}
]
# Then you ALSO need to write the dispatch logic:
def handle_tool_call(tool_call):
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
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"Unknown tool: {name}")
对于三个工具，你已经编写了大约70行JSON schema和分派代码。schema与实际函数签名脱节：更改函数中的参数名而忘记更新schema，它会在运行时悄悄出错。
解决方案应有的样子
相反，工具定义应该从函数本身派生。函数的名称、文档字符串和类型提示已经描述了该工具的功能及其接受的参数。这应该足够了。
它还应该自动保持同步。如果你重命名参数或更改其类型，发送给LLM的schema应该自动更新，而无需你触碰第二个文件。
而且它应该是免分派的。框架应该直接调用正确的函数。不需要手动编写if/elif链来将字符串名称映射到可调用对象。
Pydantic AI如何解决这个问题
在Pydantic AI中，工具只是一个带有装饰器的函数。框架根据函数的签名和文档字符串生成JSON schema，并自动处理分派：
from pydantic_ai import Agent, RunContext
receipt_agent = Agent(
"openai:gpt-4o",
output_type=ReceiptAnalysis,
system_prompt="Analyze the provided receipt and extract structured details.",
)
@receipt_agent.tool_plain
def lookup_merchant_category(merchant_name: str) -> str:
"""Look up the spending category for a merchant name."""
# Your actual implementation
categories_db = {"CAFE PARIS": "food", "UBER": "transport", "NETFLIX": "entertainment"}
return categories_db.get(merchant_name.upper(), "other")
@receipt_agent.tool_plain
def get_exchange_rate(from_currency: str, to_currency: str) -> float:
"""Get current exchange rate between two currencies."""
# Your actual implementation — call an API, hit a cache, etc.
rates = {"EUR_USD": 1.08, "GBP_USD": 1.27}
return rates.get(f"{from_currency}_{to_currency}", 1.0)
@receipt_agent.tool_plain
def get_spending_history(category: str, days: int) -> dict:
"""Get spending totals by category for a date range."""
# Your actual implementation
return {"category": category, "total": 142.50, "transaction_count": 12}
就是这样。没有 JSON schema 字典，没有分发函数。同样的三个工具，约25行代码，而不是约70行。
框架为你做的事情：
从类型提示生成 Schema：
merchant_name: str在 JSON schema 中变成{"type": "string"}。文档字符串变成工具的description。参数名变成属性名。这一切都来自你已经写好的内容。自动分发： 当 LLM 调用
get_exchange_rate时，框架直接路由到被装饰的函数。无需字符串匹配，也无需手动映射。同步有保证： 在函数签名中把
from_currency重命名为source_currency，schema 会在下次运行时自动更新。没有第二个需要记住的地方。
问题3：没有干净的方式传递运行时上下文
有了自动分发（问题2），框架调用你的工具函数，而不是你调用。你不再控制调用点，所以你不能简单地传递 db 或 user_id 作为额外参数。
而且这些也不应该由 LLM 提供。你需要一个侧信道，将运行时依赖传递给框架代表你调用的工具。
如果没有这种机制，你最终会得到类似这样的代码：
# Option A: Global state (untestable, unsafe)
db = get_database_connection()
current_user = None # Set somewhere else... hopefully before tools run
def get_spending_history(category: str, days: int) -> dict:
# Uses global `db` and `current_user` — how do you test this?
# How do you run two users concurrently?
return db.query(
"SELECT sum(amount) FROM transactions WHERE user_id = ? AND category = ? AND date > ?",
current_user.id, category, days_ago(days)
)
# Option B: Closure-based (awkward, deeply nested)
def make_tools(db, user):
def get_spending_history(category: str, days: int) -> dict:
return db.query(...) # Captures db and user from enclosing scope
def lookup_merchant_category(merchant_name: str) -> str:
return db.query(...) # Same closure trick
return [get_spending_history, lookup_merchant_category]
# Every time you add a dependency, you restructure the closure nesting
两种方法都让测试变得痛苦。如果不重构代码，你无法轻松地换用模拟数据库或测试用户。
解决方案应该是什么样子
为了更好的解决方案，你应该声明你的工具需要什么。将依赖（数据库、HTTP客户端、用户会话）表达为类型化需求，与LLM提供的工具参数分离。
你还应该在运行时注入，而不是在定义时注入。在运行智能体时传递具体实例，而不是在定义工具时。这使工具定义保持纯净且可复用。
并且为了测试替换依赖。用内存模拟替代真实数据库，或用测试夹具替代真实用户，而无需更改工具代码。
Pydantic AI如何解决这个问题
Pydantic AI拥有一流的依赖注入系统。你在智能体上定义deps_type，工具通过类型化的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="Analyze the provided receipt and extract structured details.",
)
@receipt_agent.tool
def get_spending_history(ctx: RunContext[ReceiptDeps], category: str, days: int) -> dict:
"""Get spending totals by category for a date range."""
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:
"""Get current exchange rate between two currencies."""
response = ctx.deps.http_client.get(f"/rates/{from_currency}/{to_currency}")
return response.json()["rate"]
# At runtime — pass real dependencies
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"),
),
)
# In tests — swap with mocks, no code changes to tools
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是类型化的。你的IDE会自动补全其方法，类型检查器会捕获误用。该工具不知道也不关心它是真实的Postgres连接还是测试模拟。没有全局变量，没有闭包： 依赖在
run_sync()时显式流入。两个并发用户获得两个独立的ReceiptDeps实例，没有共享的可变状态。测试变得非常简单： 将
DatabaseClient替换为InMemoryDb，将HttpClient替换为MockHttpClient。工具代码不变。无需猴子补丁、依赖注入框架，或触及模块级状态的测试夹具。LLM永远不会看到依赖：
RunContext不会作为工具参数暴露。LLM只能看到category和days。框架会自动将其从schema中剥离。
问题4：测试需要真实的LLM调用
你想验证你的代理能处理边缘情况：外币收据、缺失的商户名称或模糊的类别。但每个测试都会命中真实的API：
def test_foreign_currency_receipt():
# This test:
# - Costs money (API call)
# - Takes 2-5 seconds
# - Is non-deterministic (might pass today, fail tomorrow)
# - Requires network access (breaks in CI without secrets)
result = analyze_receipt("CAFÉ PARIS\n€12.50\nCroissant x2")
assert result["currency"] == "EUR"
assert result["category"] == "food" # Might return "dining" instead — flaky!
你无法在CI中可靠地运行它。你无法在不消耗完API预算的情况下运行50个边缘案例测试。你最终要么没有测试，要么拥有不稳定的集成测试。
解决方案应该是什么样
首先，将LLM替换为确定性的替身——一个返回可预测、可控响应的东西，从而使测试快速、免费且可重复。
接下来，保持智能体逻辑完整。测试应该演练真实的工具调度、验证和输出解析。只有模型是伪造的。
最后，对行为进行断言，而不是对LLM的措辞。验证正确的工具是否以正确的参数被调用，以及输出是否符合预期的结构。
Pydantic AI 如何解决这个问题
Pydantic AI 提供了TestModel和FunctionModel。它们是即插即用的模型替代品，让你能够精确控制“LLM”返回的内容，无需网络调用：
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai.models.function import FunctionModel
# TestModel — returns a predictable, schema-valid response automatically
def test_receipt_analysis_structure():
"""Test that the agent returns a valid ReceiptAnalysis object."""
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 fills fields with valid dummy data matching the schema
assert isinstance(result.output, ReceiptAnalysis)
assert 0 <= result.output.confidence <= 1
# FunctionModel — you control the exact response for specific scenarios
def test_foreign_currency_triggers_exchange_rate_tool():
"""Test that a EUR receipt causes the agent to call get_exchange_rate."""
def mock_model(messages, info):
# Simulate the LLM deciding to call the exchange rate tool
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),
),
)
# Assert the exchange rate tool was actually invoked
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调用、网络或令牌消耗。测试在毫秒内运行。
它也是确定性的，意味着对于相同的输入，每次都能得到相同的输出。不会因LLM温度或措辞变化而产生不稳定的测试。
真实的代理逻辑仍然运行。工具调度、依赖注入和输出验证都得到执行。只有模型被替换。
它也对CI友好。你的CI环境中不需要任何API密钥，也没有需要管理的机密或需要达到的速率限制。
你还可以获得两个级别的控制。你有用于“管道是否工作？”测试的TestModel。还有用于“代理是否做出正确决策？”测试的FunctionModel，你可以在其中编写特定的LLM行为脚本。
问题5：重试和验证逻辑是手写的
当LLM返回错误响应时，你需要重试。但重试逻辑很快变得复杂：
def analyze_receipt_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))
# Validate
if not isinstance(result.get("total"), (int, float)):
raise ValueError("total must be numeric")
if result.get("confidence", 0) > 1 or result.get("confidence", 0) < 0:
raise ValueError("confidence must be 0-1")
if result.get("category") not in ALLOWED_CATEGORIES:
raise ValueError(f"invalid category: {result.get('category')}")
return result
except (json.JSONDecodeError, ValueError, KeyError) as e:
if attempt == max_retries - 1:
raise
# Should we feed the error back to the LLM? Modify the prompt?
# How do we track which attempts failed and why?
continue
raise RuntimeError("Should not reach here")
你构建的每个智能体都需要相同的重试/验证/重新提示模式，而你每次都要重写它。将验证错误反馈给LLM（以便其自我纠正）的逻辑又增加了一层复杂性。
解决方案应具备的样子
首先，验证应该是声明式的：也就是说，由输出模式定义，而不是通过散落在代码中的手写if语句。
其次，重试应该是自动的。如果输出未通过验证，框架应使用错误信息重新提示LLM，使其能够自我纠正。
而且自定义验证应该能够干净地插入。对于超出类型检查的业务规则（例如，"如果类别是'其他'，置信度必须低于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="Analyze the provided receipt and extract structured details.",
retries=3, # Max retry attempts on validation failure
)
@receipt_agent.result_validator
def validate_receipt_analysis(ctx: RunContext[ReceiptDeps], result: ReceiptAnalysis) -> ReceiptAnalysis:
"""Business logic validation — runs after schema validation passes."""
# Rule: if total doesn't match sum of items, ask LLM to fix it
items_sum = sum(item.amount for item in result.items)
if abs(result.total - items_sum) > 0.01:
raise ModelRetry(
f"Total ({result.total}) doesn't match sum of items ({items_sum}). "
f"Please recheck the receipt and correct either the total or the item amounts."
)
# Rule: low confidence + "other" category likely means the LLM gave up — retry
if result.category == SpendingCategory.OTHER and result.confidence < 0.5:
raise ModelRetry(
"Category is 'other' with low confidence. Look more carefully at the "
"merchant name and items to determine a more specific category."
)
return result
以下是验证失败时会发生的情况：
┌─────────────────────────────────────────────────────────────┐
│ Automatic Retry Flow │
│ │
│ LLM response │
│ │ │
│ ▼ │
│ Schema validation (Pydantic model) │
│ │ │
│ ├── FAIL → error message sent back to LLM → retry │
│ │ │
│ ▼ │
│ result_validator (your business rules) │
│ │ │
│ ├── ModelRetry raised → message sent to LLM → retry │
│ │ │
│ ▼ │
│ PASS → return typed result │
└─────────────────────────────────────────────────────────────┘
Schema违规（类型错误、字段缺失、枚举不匹配）会被Pydantic自动捕获。验证错误会作为上下文发送回LLM，让它知道需要修复什么。
业务规则违规由你的
result_validator捕获。ModelRetry将你的自定义消息发送给LLM，引导其做出正确响应。你的代码中不需要重试循环。
retries=3参数控制最大尝试次数。框架负责处理循环、重新提示和错误消息格式化。
问题6：切换模型意味着重写集成代码
你的智能体目前使用OpenAI。现在你想尝试Anthropic（对你的用例更便宜）或使用Ollama在本地运行（出于数据隐私考虑）。每个提供商都有不同的SDK、工具调用格式和响应结构：
# OpenAI
response = openai_client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools # OpenAI tool format
)
tool_calls = response.choices[0].message.tool_calls
# Anthropic — completely different API shape
response = anthropic_client.messages.create(
model="claude-sonnet-4-20250514",
messages=messages,
tools=anthropic_tools # Different format than OpenAI!
)
tool_use_blocks = [b for b in response.content if b.type == "tool_use"]
# Google — yet another shape
response = genai_client.generate_content(
contents=messages,
tools=google_tools # Yet another format!
)
function_calls = response.candidates[0].content.parts
你最终会得到供应商特定的代码路径、适配器层，而智能体逻辑被淹没在集成胶水代码之下。
解决方案应有的样子
首先，将智能体逻辑定义一次。你的工具、输出类型、系统提示和验证都应与模型无关。
你还应该能够通过更改一个字符串来切换模型，而不是重写SDK调用、工具模式或响应解析。
并且你应该将供应商特定的细节隐藏起来。框架应该将你的通用智能体定义转换为每个供应商所期望的任何格式。
Pydantic AI 如何解决这个问题
智能体定义完全与模型无关。模型只是一个字符串标识符：更改它，其他一切都保持不变：
# Your agent definition — tools, output type, deps, validators — all unchanged
receipt_agent = Agent(
"openai:gpt-4o", # ← this is the only line that changes
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
system_prompt="Analyze the provided receipt and extract structured details.",
)
# Switch to Anthropic — same agent, same tools, same output type
receipt_agent = Agent(
"anthropic:claude-sonnet-4-20250514",
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
system_prompt="Analyze the provided receipt and extract structured details.",
)
# Switch to a local model via Ollama
receipt_agent = Agent(
"ollama:llama3.1",
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
system_prompt="Analyze the provided receipt and extract structured details.",
)
# Or make it configurable at runtime
import os
receipt_agent = Agent(
os.getenv("RECEIPT_AGENT_MODEL", "openai:gpt-4o"),
output_type=ReceiptAnalysis,
deps_type=ReceiptDeps,
system_prompt="Analyze the provided receipt and extract structured details.",
)
框架在幕后处理的内容：
工具模式转换：你的
@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），框架都会将其规范化为一致的内部表示。透明处理提供商特有功能：每个提供商以不同方式实现结构化输出模式、流式传输和令牌计数。框架会适应这些差异，而不会将其暴露给你的代码。
一个代理定义，任意模型。通过配置或环境变量即可切换。
总结
本文中的每个问题都源自同一个地方：LLM调用是文本进、文本出，而其两侧的代码是有类型的。原生SDK方法通过手写的胶水代码弥合这一差距——一个分隔符剥离器、一连串的if检查、一个调度表、一个重试循环。每个部分都很简单。但组合在一起，它们就超过了所包围的代理逻辑，而且这些代码全部归你所有。
Pydantic AI通过将边界变为声明式契约来弥合这一差距。下面逐项说明这为我们带来了什么：
| 问题 | 原生SDK | Pydantic AI |
|---|---|---|
| 结构化输出 | 用英文描述模式，手动解析 | output_type=ReceiptAnalysis — 自动生成模式，校验响应 |
| 工具定义 | 约70行JSON模式+调度 | 在类型化函数上使用@agent.tool_plain |
| 运行时上下文 | 全局变量或嵌套闭包 | deps_type + RunContext，每次运行注入 |
| 测试 | 真实API调用：缓慢、付费、不稳定 | TestModel / FunctionModel，无需网络 |
| 重试与验证 | 手写循环，错误丢弃 | 字段约束 + ModelRetry，错误反馈给模型 |
| 模型切换 | 每个提供商各自的SDK和解析代码 | 一个字符串："openai:gpt-4o" → "anthropic:claude-sonnet-4-20250514" |
我们最终得到的收据代理是一个Pydantic模型、几个类型化函数、一个deps数据类和一个验证器。没有解析、没有调度、没有重试循环。
Pydantic AI 文档很简短，值得从头到尾阅读；本文中的一切都对应其API参考。
——
zhirenhun
一个热爱技术的程序员，喜欢分享前沿AI知识和开发经验。
