5.2 读 PydanticAI：run 循环与重试调用链
第 2 章用 PydanticAI 重写了 ResearchCopilot 的问答路径，第 4 章比过它的分数。但「agent.run_sync() 之后到底发生了什么」一直被当黑盒。本节把它拆开：你会发现 PydanticAI 的 run 根本不是一个 while 循环，而是一张用 GraphBuilder 拼出来的状态机图；而结构化输出的「校验失败就重试」也不是 try/except，而是图里长出来的一个新节点。
本节把「PydanticAI 怎么跑一轮、校验失败怎么重试」推进到源码级：读出
build_agent_graph的节点与边、ModelRetry的消费路径、_build_retry_node如何把异常变成新请求节点，并用一次真跑的失败→重试把调用链与提示文本对出来。
5.2.1 先定位源码
/tmp/agentbook/venv3/bin/python -c "import importlib.metadata as m; print(m.version('pydantic-ai'))"
/tmp/agentbook/venv3/bin/python -c "import pydantic_ai; print(pydantic_ai.__file__)"
2.55.0
/tmp/agentbook/venv3/lib/python3.14/site-packages/pydantic_ai/__init__.py
run 循环的主体在 pydantic_ai/_agent_graph.py（158 KB，是包内最大的模块之一）；输出校验与重试提示在 pydantic_ai/_output.py 与 pydantic_ai/messages.py；传输层的重试在 pydantic_ai/retries.py——这个文件和「模型重试」是两回事，后面会专门澄清。
5.2.2 run 循环不是 while：Agent 是一张图
Agent 构造时会构建一张执行图。入口在 pydantic_ai/_agent_graph.py:3002：
# pydantic_ai/_agent_graph.py:3002
def build_agent_graph(name, deps_type, output_type) -> Graph[...]:
"""Build the execution Graph for a given agent."""
return _build_agent_graph(name)
真正的装配在 _build_agent_graph（第 3020 行，带 @lru_cache，同名 Agent 只建一次图）：
# pydantic_ai/_agent_graph.py:3029
g = GraphBuilder(
name=name or 'Agent',
state_type=GraphAgentState,
deps_type=GraphAgentDeps[Any, Any],
input_type=UserPromptNode[Any, Any],
output_type=result.FinalResult[Any],
auto_instrument=False,
)
g.add(
g.edge_from(g.start_node).to(UserPromptNode[Any, Any]),
g.node(UserPromptNode[Any, Any]),
g.node(ModelRequestNode[Any, Any]),
g.node(CallToolsNode[Any, Any]),
g.node(SetFinalResult[Any, Any]),
)
return g.build(validate_graph_structure=False)
四个节点、一套状态。GraphAgentState（第 382 行）里有个字段是重试的核心：
# pydantic_ai/_agent_graph.py:387
output_retries_used: int = 0
图的「循环」由节点返回下一个节点来驱动：节点 run() 返回谁，执行器就接着跑谁。所以这不是 while True，是图上的转移。
5.2.3 节点各干什么、往哪转
UserPromptNode：把用户输入与历史拼成第一条ModelRequest，返回一个ModelRequestNode。ModelRequestNode：真正发起模型调用。非流式路径在_make_request（第 1692 行）；拿到响应后转CallToolsNode。CallToolsNode：处理响应——要么执行工具、要么产出最终结果。它的_next_node决定去向（_agent_graph.py:2596）：
# pydantic_ai/_agent_graph.py:2596
if output_final_result:
final_result = output_final_result[0]
self._next_node = self._handle_final_result(ctx, final_result, output_parts)
else:
if self.user_prompt is not None:
output_parts.append(_messages.UserPromptPart(self.user_prompt))
self._next_node = ModelRequestNode[DepsT, NodeRunEndT](_messages.ModelRequest(parts=output_parts))
「还有工具调用 → 造一个新的 ModelRequestNode 带上工具结果」，这就是 agent loop 的迭代点；「有最终输出 → _handle_final_result 返回 End(...)」，图终止。SetFinalResult（第 2724 行）只在流式路径下用来立刻结束。
5.2.4 结构化输出校验失败：ModelRetry 的消费
当 output_type 是一个 Pydantic 模型时，PydanticAI 会把它变成一个「输出工具」，模型必须调它并给出符合 schema 的参数。参数不合规时，校验层抛出 ModelRetry。真正把它接住并转成重试的地方，是 ModelRequestNode 的两条路径。非流式那条在 _make_request 末尾（_agent_graph.py:1762）：
# pydantic_ai/_agent_graph.py:1762
except exceptions.ModelRetry as e:
self._enforce_usage_limits(ctx, accounted_responses)
# `ModelRetry` from any model lifecycle hook retries the model request.
# If the handler was called, preserve the response in history for context.
if _handler_response is not None:
self._append_response(ctx, _handler_response, record_usage=not _handler_usage_recorded)
return await self._build_retry_node(ctx, e)
注意「先把这次响应 append 进历史」——这样模型下次能看到自己上次错在哪。然后交给 _build_retry_node。
5.2.5 重试节点怎么造出来
_build_retry_node（pydantic_ai/_agent_graph.py:2194）短小但信息量足：
# pydantic_ai/_agent_graph.py:2194
async def _build_retry_node(self, ctx, error: exceptions.ModelRetry):
"""Build a retry ModelRequestNode from a ModelRetry exception."""
ctx.state.consume_output_retry(ctx.deps.max_output_retries, error=error)
m = _messages.RetryPromptPart(content=error.message)
retry_node = ModelRequestNode[DepsT, NodeRunEndT](_messages.ModelRequest(parts=[m]))
self._result = retry_node
return retry_node
三步：扣重试预算 → 造一个 RetryPromptPart → 把它包进新的 ModelRequestNode 返回。于是「重试」在图里就是一次普通的节点转移——上一个 ModelRequestNode 直接返回了另一个 ModelRequestNode，执行器继续跑，模型被再问一次。
预算的扣减在 consume_output_retry（_agent_graph.py:447）：
# pydantic_ai/_agent_graph.py:447
def consume_output_retry(self, max_output_retries: int, error: BaseException | None = None) -> None:
"""Record one unit of output-retry budget consumption."""
self.output_retries_used += 1
if self.output_retries_used > max_output_retries:
self.check_incomplete_tool_call()
message = f'Exceeded maximum output retries ({max_output_retries})'
raise exceptions.UnexpectedModelBehavior(message) from error
max_output_retries 来自 GraphAgentDeps（第 487 行），就是 Agent(...) 的 output_retries。超了抛 UnexpectedModelBehavior——重试是有界的，不会无限打模型。
5.2.6 重试提示长什么样
RetryPromptPart.from_error（pydantic_ai/messages.py:1829）决定模型看到的文本。校验错误（ValidationError）取 error.errors(...)，ModelRetry 取 error.message：
# pydantic_ai/messages.py:1842
content = (
error.errors(include_url=False, include_context=False)
if isinstance(error, pydantic_core.ValidationError)
else error.message
)
part = cls(content=content, tool_name=tool_name)
而 model_response()（第 1852 行）把它渲染成自然语言：
# pydantic_ai/messages.py:1871
plural = isinstance(self.content, list) and len(self.content) != 1
description = (
f'{len(self.content)} validation error{"s" if plural else ""}:\n```json\n{json_errors.decode()}\n```'
)
return f'{description}\n\nFix the errors and try again.'
5.2.7 真跑：失败 → 重试的完整调用链
用一个 FunctionModel 第一次返回非法值、第二次返回合法值，output_type 是一个带校验的模型：
import os
os.environ["PYDANTIC_AI_NO_BANNER"] = "1"
from pydantic import BaseModel, field_validator
from pydantic_ai import Agent
from pydantic_ai.models.function import FunctionModel, AgentInfo
from pydantic_ai.messages import ModelResponse, ToolCallPart
class Answer(BaseModel):
value: int
@field_validator("value")
@classmethod
def positive(cls, v):
if v <= 0:
raise ValueError("value must be positive")
return v
seen = []
def fn(messages, info: AgentInfo) -> ModelResponse:
seen.append(len(messages))
name = info.output_tools[0].name
if len(seen) == 1:
return ModelResponse(parts=[ToolCallPart(name, {"value": -5})])
return ModelResponse(parts=[ToolCallPart(name, {"value": 42})])
agent = Agent(FunctionModel(fn), output_type=Answer)
res = agent.run_sync("give me an answer")
print("OUTPUT", res.output)
print("MSG_COUNTS_AT_EACH_CALL", seen)
真跑输出：
OUTPUT value=42
MSG_COUNTS_AT_EACH_CALL [1, 3]
[1, 3] 是本节最有说服力的一行：第一次调用模型时历史里只有 1 条消息（用户提问）；重试时历史变成 3 条——多了「模型上次的非法工具调用」和「RetryPromptPart」两条，正是 _append_response + _build_retry_node 两步的产物。模型第二次看到校验反馈，才改成了 42。
模型在重试里到底看到什么，把那条 RetryPromptPart 打印出来：
for m in res.all_messages():
for p in getattr(m, "parts", []):
if type(p).__name__ == "RetryPromptPart":
print(p.model_response())
1 validation error:
[
{
"type": "value_error",
"loc": [ "value" ],
"msg": "Value error, value must be positive",
"input": -5
}
]
Fix the errors and try again.
（模型实际收到的文本里，那段 JSON 由一层 json 围栏包住；这里为免代码块嵌套去掉了那对围栏标记。）这段文本就是 model_response() 拼出来的——loc、msg、input 全在，模型据此知道「value 得是正数」。
5.2.8 别把「模型重试」和「传输重试」搞混
pydantic_ai/retries.py 里也有一个 RetryConfig（第 72 行），但它是基于 tenacity 的 HTTP 传输层重试，配在 httpx transport 上，处理的是 5xx / 超时 / 连接错误：
# pydantic_ai/retries.py:72
class RetryConfig(TypedDict, total=False):
"""The configuration for tenacity-based retrying."""
sleep: Callable[[int | float], None | Awaitable[None]]
stop: StopBaseT
wait: WaitBaseT
retry: SyncRetryBaseT | RetryBaseT
两条重试的层次完全不同：
| 维度 | 传输重试（retries.py） | 模型重试（_agent_graph.py） |
|---|---|---|
| 触发 | HTTP 5xx / 超时 / 连接失败 | 输出校验失败 / 显式 ModelRetry |
| 机制 | tenacity 装饰器，透明重发 | 图里新造一个 ModelRequestNode |
| 模型可见 | 否，模型完全不知道 | 是，收到 RetryPromptPart 反馈 |
| 预算 | RetryConfig 的 stop | output_retries_used vs max_output_retries |
| 计数影响 | 不计入 usage.requests | 计一次真实模型调用 |
传输重试是「网络抖动再试一次」，模型重试是「模型答错了，把错误告诉它再问一次」——后者才需要把失败信息喂回上下文，也才会消耗 token 和额度。卷一 4.2 参数校验与执行隔离
里自研引擎用「把错误 JSON 回灌」实现的重试，与这里的 RetryPromptPart 是同一思路，只是 PydanticAI 把它做成了图上的节点。
5.2.9 这套框架替你做了什么、你又让渡了什么
| 维度 | PydanticAI 替你做了 | 你让渡了 |
|---|---|---|
| 循环 | 状态机图的转移调度全自动 | 「循环」的形态被节点/边固定 |
| 重试 | 校验失败自动转 RetryPromptPart 并重发 | 重试提示的文案由框架渲染，改文案得改 model_response |
| 预算 | output_retries_used 计数与上限 | 超限直接 UnexpectedModelBehavior，行为不可插话 |
| 历史 | 失败响应自动 append，保证上下文连续 | 历史形状由框架决定，自研时得自己拼 |
一句话：PydanticAI 把 agent loop 写成了一张可枚举节点的状态机，重试是这张图上一条普通的边；代价是「循环」不再是一段你能随手改的代码。下一节读 Claude Agent SDK——它更极端：根本没有 Python 里的 loop，循环跑在另一个进程（CLI）里，Python 侧只是个客户端。
阅读导航：上一节：5.1 读 LangGraph：图执行器与检查点机制 · 下一节：5.3 读 Claude Agent SDK：subagent 与 hook 调度（含 TS SDK 对照） 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。