2.2 结构化输出与工具：输出模型 + 校验重试
2.1 节的 Agent 默认输出一段纯文本。对演示够用，对卷二验收的 UC1 不够：UC1 要求「每条回答至少带 1 条 chunk 级引用」，而一段自由文本里，「引用」既不能被程序解析、也不能被断言。这一节把 output_type 从 str 换成一个 Pydantic 模型，让回答变成带类型、带校验的结构化对象；再演示当模型给出非法结构时，框架如何自动把校验错误回灌给模型并重试——这正是「自己造轮子」时最容易漏掉、也最难写对的一环。
本节把 PydanticAI 推进到「能产出可断言的回答」：用
output_type=<PydanticModel>约束输出形状，用嵌套Citation模型表达引用，用FunctionModel注入一次非法输出真跑出校验重试，并实测默认重试上限与失败行为。
2.2.1 为什么把回答写成模型
先看卷二 UC1 对输出的要求：回答里每条论断都要能回链到引用，引用要能回溯到真实 chunk。用纯文本表达时，这两条只能靠人眼看；换成结构化对象后，它们变成类型系统能检查的事实：
| 要求 | 纯文本 | Pydantic 模型 |
|---|---|---|
| 必须有引用 | 靠人读 | citations: list[Citation]，min_length=1 |
| 引用有结构 | 自由字符串 | 嵌套 Citation(source, quote) |
| 置信度有范围 | 无所谓 | confidence: float = Field(ge=0, le=1) |
| 能写断言 | 要解析文本 | result.output.citations[0].source |
一句话：结构化输出把「内容质量」从「人工评审」搬到「程序断言」。这也是本卷反复出现的主题——凡是能被断言的东西，才谈得上被评测（第 12 章）。
2.2.2 output_type：用 Pydantic 模型约束回答
定义一个回答模型，再定义它的引用模型：
from pydantic import BaseModel, Field
class Citation(BaseModel):
source: str = Field(description="来源标记，如 pg / cache")
quote: str = Field(description="被引用的原文片段")
class Answer(BaseModel):
answer: str = Field(description="对问题的直接回答")
citations: list[Citation] = Field(min_length=1, description="至少一条可回溯引用")
confidence: float = Field(ge=0, le=1, description="0~1 的置信度")
把它交给 Agent：
from pydantic_ai import Agent
agent = Agent("test", deps_type=Deps, output_type=Answer,
system_prompt="先检索再回答，回答必须带可回溯引用。")
这里有两个关键点：
Field(description=...)不是给文档看的。PydanticAI 会把每个字段的 description 连同类型一起，生成成给模型看的 JSON Schema。description 写得越清楚，模型填得越准——它和卷一工具 description 的定位完全一样，是提示词的一部分。- 约束写在类型里。
min_length=1、ge=0、le=1这些不是「建议」，是模型输出必须满足的条件；不满足就走重试。
2.2.3 输出是怎么变成工具调用的
一个自然的疑问：模型 API 本身只会返回文本或工具调用，output_type=Answer 是怎么落地的？答案是——框架把「产出结构化输出」也实现成一次工具调用。这个隐藏工具的名字实测是 final_result：
from pydantic_ai.models.function import FunctionModel, AgentInfo
from pydantic_ai.messages import ModelMessage, ModelResponse, ToolCallPart
def probe(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
print("业务工具:", [t.name for t in info.function_tools])
print("输出工具:", [t.name for t in info.output_tools])
return ModelResponse(parts=[ToolCallPart(
tool_name=info.output_tools[0].name,
args={"answer": "x", "citations": ["[pg]"]})])
agent.run_sync("q", model=FunctionModel(probe))
本机实测输出：
业务工具: []
输出工具: ['final_result']
info.output_tools 是框架在每次请求里附加的「输出工具」列表；info.function_tools 才是我们自己 @agent.tool 注册的业务工具。所以一次带结构化输出的 run，模型看到的工具集合是「业务工具 + final_result」，它要么继续调业务工具，要么调 final_result 交卷。交卷时 final_result 的参数就是 Answer 的字段，框架拿到后做一次 Pydantic 校验，通过则 result.output 是一个 Answer 实例。
这个设计的好处是复用同一条工具调用链路：结构化输出不需要供应商 API 支持任何特殊参数，任何能调工具的模型都能用它。代价是输出多占一次工具调用（多一轮往返），成本核算时要算进去。
2.2.4 校验失败触发重试
现在做本节的核心演示：让假模型第一次交卷交错了——citations 为空，违反 min_length=1；框架应当拦下、把校验错误回灌，模型第二次交出合法结构。用一个三步脚本：第 1 步调业务工具 search，第 2 步交非法卷，第 3 步交合法卷。
import os
os.environ["PYDANTIC_AI_NO_BANNER"] = "1"
from dataclasses import dataclass
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.function import FunctionModel, AgentInfo
from pydantic_ai.messages import ModelMessage, ModelResponse, ToolCallPart
class Citation(BaseModel):
source: str = Field(description="来源标记，如 pg / cache")
quote: str = Field(description="被引用的原文片段")
class Answer(BaseModel):
answer: str = Field(description="对问题的直接回答")
citations: list[Citation] = Field(min_length=1, description="至少一条可回溯引用")
confidence: float = Field(ge=0, le=1, description="0~1 的置信度")
class FakeRetriever:
def __init__(self) -> None:
self.calls: list[str] = []
def search(self, query: str, k: int = 2) -> list[str]:
self.calls.append(query)
return ["[pg] ResearchCopilot 使用 PostgreSQL 17.11。"]
@dataclass
class Deps:
retriever: FakeRetriever
agent = Agent("test", deps_type=Deps, output_type=Answer,
system_prompt="先检索再回答，回答必须带可回溯引用。")
@agent.tool
def search(ctx: RunContext[Deps], query: str) -> list[str]:
"""按关键词检索本地语料。"""
return ctx.deps.retriever.search(query)
state = {"step": 0}
def scripted(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
state["step"] += 1
out = info.output_tools[0].name
if state["step"] == 1:
return ModelResponse(parts=[ToolCallPart(tool_name="search", args={"query": "postgres"})])
if state["step"] == 2: # 非法：citations 为空，触发校验重试
return ModelResponse(parts=[ToolCallPart(
tool_name=out, args={"answer": "17.11", "citations": [], "confidence": 0.9})])
return ModelResponse(parts=[ToolCallPart(
tool_name=out,
args={"answer": "PostgreSQL 17.11",
"citations": [{"source": "pg", "quote": "ResearchCopilot 使用 PostgreSQL 17.11。"}],
"confidence": 0.8})])
retriever = FakeRetriever()
res = agent.run_sync("用的哪个 Postgres？", deps=Deps(retriever), model=FunctionModel(scripted))
print("输出类型:", type(res.output).__name__)
print("output:", res.output)
print("模型调用次数:", state["step"])
for m in res.all_messages():
if any(p.part_kind == "retry-prompt" for p in m.parts):
for p in m.parts:
if p.part_kind == "retry-prompt":
print("重试反馈:", str(p.content))
本机实测（pydantic-ai 2.55.0）：
输出类型: Answer
output: answer='PostgreSQL 17.11' citations=[Citation(source='pg', quote='ResearchCopilot 使用 PostgreSQL 17.11。')] confidence=0.8
模型调用次数: 3
重试反馈: [{'type': 'too_short', 'loc': ('citations',), 'msg': 'List should have at least 1 item after validation, not 0', 'input': []}]
三个信号，一个比一个关键：
输出类型: Answer：result.output是真的Answer实例，不是 dict、不是 JSON 字符串。可以直接res.output.citations[0].source。模型调用次数: 3：第 1 次调工具，第 2 次交非法卷，第 3 次交合法卷。中间那次非法输出没有让整个 run 崩掉——框架拦下、重试、成功。这就是「校验重试」。重试反馈：框架回灌给模型的是一段结构化的 Pydantic 校验错误，含loc（出错字段）、msg（人话解释）、input（模型给的原值）。模型下一轮看到这段反馈，就知道「citations不能为空」。
对比自研版：卷一的引擎里，模型输出没有校验——工具参数不对由 ToolRegistry.call 包成 {"error": ...} 回灌，但「最终回答的结构」完全没人管。要在自研版里加这一层，你得自己写「校验 → 拼错误消息 → 塞回历史 → 重跑」的循环；PydanticAI 把它做成了 output_type 的一个属性。
2.2.5 重试上限与失败行为
重试不是无限的。默认只重试 1 次。实测：让假模型一直交非法卷，run 会抛 UnexpectedModelBehavior：
from pydantic_ai.exceptions import UnexpectedModelBehavior
calls = {"n": 0}
def always_bad(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
calls["n"] += 1
return ModelResponse(parts=[ToolCallPart(
tool_name=info.output_tools[0].name,
args={"answer": "x", "citations": []})]) # 一直非法
try:
agent.run_sync("q", deps=Deps(FakeRetriever()), model=FunctionModel(always_bad))
except UnexpectedModelBehavior as e:
print("异常:", type(e).__name__)
print("消息:", str(e))
print("模型被调用次数:", calls["n"])
本机实测：
异常: UnexpectedModelBehavior
消息: Exceeded maximum output retries (1)
模型被调用次数: 2
要点：
- 默认重试上限是 1（模型总共被调 2 次：原始 + 1 次重试）。要放宽就在
Agent(...)里传retries=（构造参数实测存在）或每次run_sync(retries=...)覆盖。 - 耗尽后抛
UnexpectedModelBehavior，而不是静默返回一个不合法对象。这个选择是对的：一个结构不合法的「回答」如果被放行，下游（写库、渲染、引用回链）会在更远的地方以更难查的方式炸掉。在边界处响亮失败，是护栏的基本素养（第 15 章会把这个思路扩展到输入 / 输出两类护栏）。 - 重试是有成本的：每次重试都是一次完整的模型往返（真模型下按 token 计费）。所以「把约束写严」和「重试上限放宽」要权衡——约束太严会让合法但边缘的输出反复重试，上限太高会放大成本。
2.2.6 工具与结构化输出并存
注意 2.2.4 的例子同时用了业务工具 search 和结构化输出 Answer。框架把两类工具拼在一起发给模型：
| 类别 | 来源 | 出现在 | 作用 |
|---|---|---|---|
| 业务工具 | @agent.tool | info.function_tools | 检索、查库等真实动作 |
| 输出工具 | output_type=Answer | info.output_tools | 交结构化答卷（final_result） |
这个「并存」正是 Agent 与「一次性结构化抽取」的区别：抽取只要输出工具，Agent 要在多个业务工具之间循环、最后才交卷。PydanticAI 把两者统一在同一套工具机制里，所以我们既不用为「结构化」单独写一条路径，也不用担心业务工具和输出工具打架——模型自己按 system_prompt 与工具描述决定先检索还是先交卷。
一个实践提醒：final_result 的 description 与各字段 description 会显著影响模型「什么时候交卷」。写得太松，模型可能一次工具都不调就硬编一个答案交卷（退化成卷二 UC1 的幻觉失败模式）；写得太严，模型可能反复检索不交卷。调这两处 description，是 PydanticAI 里最像「调提示」的活。
2.2.7 框架替我做了什么、我让渡了什么
| 维度 | 自研 minagent | PydanticAI |
|---|---|---|
| 输出结构 | 无（纯文本） | output_type=<Model> |
| 输出校验 | 需自写 | Pydantic 自动 |
| 校验失败 | 无机制 | 自动回灌 + 重试 |
| 重试上限 | 无 | retries=，默认 1 |
| 失败信号 | 无 | UnexpectedModelBehavior |
| 结构化落地的实现 | — | 隐藏的 final_result 工具 |
框架替你做的是**「类型即契约 + 校验即护栏 + 重试即自愈」三件套**；你让渡的是输出形状被 Pydantic 的约束能力绑定（想要 schema 里表达不了的结构就得改设计），以及多一次工具调用往返的成本。
小结
output_type=<PydanticModel>把回答变成带类型、带校验的对象；字段description会进 JSON Schema，是提示词的一部分。- 结构化输出在底层被实现成一次隐藏工具调用，实测工具名是
final_result，因此任何能调工具的模型都能用。 - 校验失败会自动回灌 Pydantic 报错并重试；实测默认重试上限 1，耗尽后抛
UnexpectedModelBehavior。 - 业务工具（
function_tools）与输出工具（output_tools）并存于同一次请求，这是 Agent 与一次性抽取的分界。 - 框架替你做了「类型契约 + 校验护栏 + 自动重试」；你让渡了输出形状的表达自由与一轮往返成本。
结构化输出立住了，但它也带来了新问题：这套「自动重试 + 隐藏工具」的实现，比自研版多了多少代码、又让测试变简单还是变复杂了？下一节 2.3 与 minagent 对照：代码量与可测性对比 用同一任务的两版实现把这个问题量化。想回看卷一为什么把「结构化输出」和「参数校验」分开讲，翻 4.2 参数校验与执行隔离 。
阅读导航：上一节：2.1 核心模型：Agent/RunContext/依赖注入重写问答路径 · 下一节：2.3 与 minagent 对照：代码量与可测性对比 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。