pydantic/pydantic-ai:基于 Pydantic 的强类型 Python Agent 框架¶
先用一句话说清楚它是什么:PydanticAI 是基于 Pydantic 做类型校验的 Python Agent 框架,核心卖点是让 LLM 的输出能被强类型约束住。在智能体开发中,大模型生成的“非结构化文本”是不确定性的源头。Pydantic 团队推出这个框架,就是为了解决一个核心工程痛点:如何将 LLM 的概率性输出强行拉回确定性的后端强类型轨道。
0. 最小示例:不加任何约束的最简单用法¶
在讲“强类型契约”之前,先看 PydanticAI 最基础的调用方式——不指定输出类型、不注入依赖,就是普通的一问一答:
from pydantic_ai import Agent
agent = Agent('openai:gpt-4o-mini')
result = agent.run_sync("What is the capital of France?")
print(result.output) # 此时 result.output 只是一段普通字符串
这时候 result.output 拿到的就是模型原样吐出来的文本,跟直接调用 OpenAI API 没有本质区别。但业务代码往往不想要一段自由文本,而是想要一个能直接拿来用的 Python 对象——这正是 PydanticAI 真正要解决的问题。
1. 核心设计思路:先声明“我要模型返回这个 Python 类型”,剩下的交给框架¶
PydanticAI 的核心思路说白了很简单:你直接声明“我要模型返回这个 Python 类型”,剩下的解析、校验、甚至出错后的重试,都由框架帮你做。它拒绝把大模型当成“自由对话者”,而是将其物理建模为一个类型安全的强契约函数:Agent[Deps, ResultModel](核心实现参见 pydantic/pydantic-ai 仓库 pydantic_ai_slim/pydantic_ai/agent.py 中的 Agent 类定义)。
1.1 结构化约束机制¶
落到具体 API 上,做法是给 Agent 指定 output_type=MyPydanticModel:PydanticAI 会自动将该 Pydantic 模型的 JSON Schema 转换为系统 Prompt 附件发给 LLM。大模型返回的文本必须能够被该 Pydantic 模型成功反序列化(Parse),否则无法进入下行数据流。
版本提示:早期版本 PydanticAI 使用的是
result_type参数与result.data取值方式,这套命名在后续版本中已被正式重命名为output_type/result.output(result_type/.data等旧命名已在 v0.6.0,2025-08-06 发布的清理版本中被彻底移除,而非仅停留在废弃警告阶段)。本文以当前命名为准。
那如果模型返回的内容校验不通过怎么办?这里 PydanticAI 没有简单地直接报错抛给业务层,而是先自己尝试补救——下面这套自纠错闭环是它比“裸调用 + 手写校验”更省心的地方。
1.2 反应式自纠错闭环 (Reactive Self-Correction)¶
┌────────────────────────┐
│ LLM 采样输出原始文本 │
└───────────┬────────────┘
│
v
┌────────────────────────┐
│ Pydantic 强类型校验引擎 │
└───────────┬────────────┘
│
┌──────┴──────┐
│ (校验通过) │ (校验失败: ValidationError)
v v
┌─────────┐ ┌──────────────────────────────────────────────┐
│ 返回强 │ │ 1. 拦截并格式化 ValidationError Traceback │
│ 类型对象│ │ 2. 自动封装为 System Feedback 错误消息 │
│ (Model) │ │ 3. 带错误上下文回灌大模型,发起自动纠偏重试 │
└─────────┘ └──────────────────────────────────────────────┘
- 纠错回灌:若校验失败(触发
ValidationError),运行时会自动拦截异常,将具体字段的校验报错详情(如field_x: expected int, got string)封装为一轮新的系统反馈消息发送给大模型,让大模型在了解“自己哪里写错了”的前提下自动进行下一轮纠偏生成,默认重试预算为 1 次(可通过Agent(..., retries=3)等方式调大),超出预算仍未通过校验则抛出UnexpectedModelBehavior异常,直至输出完全合规的结构方可放行。
2. 另一个设计问题:怎么让工具拿到数据库连接、密钥这类敏感依赖¶
结构化输出解决的是“模型吐什么格式”的问题,但 Agent 的工具函数往往还需要访问数据库连接池、租户 ID、API Client 这类真实的运行时资源。如果把这些东西直接拼进 Prompt 字符串下发给 Agent,是极其危险的反模式(Anti-Pattern),也容易引发提示词注入攻击(Prompt Injection)。PydanticAI 给出的方案是一套依赖注入(Dependency Injection)机制:
RunContext[Deps]机制:通过泛型Deps将可信数据源与句柄注入 Agent 运行上下文。- 物理隔离:所有的 Tools 和中间件都在本地运行时通过
RunContext静态获取依赖,这些可信资源完全不经过大模型的推理空间,大模型只决定“调不调工具”以及“传入什么逻辑参数”,而无法触碰或污染底层的系统物理句柄。
3. 基于 PydanticAI 的强契约智能体构建(Python 实践)¶
把 output_type 的结构化约束和 RunContext 的依赖注入放在一起,一段接近生产环境的骨架代码大致如下:
from dataclasses import dataclass
from typing import List
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIChatModel # 旧名 OpenAIModel 已重命名为 OpenAIChatModel(旧名仍保留作为向后兼容的废弃别名)
# 1. 声明强契约输出数据结构
class AuditResult(BaseModel):
user_id: int = Field(description="The validated user identifier.")
has_violation: bool = Field(description="True if system policies are violated.")
violations: List[str] = Field(default=[], description="List of specific violations.")
# 2. 定义安全的运行时物理依赖 (例如数据库连接句柄)
@dataclass
class DatabaseDeps:
db_conn_pool: str # 模拟数据库连接池
operator_role: str
# 3. 初始化强契约 Agent 实例
# 绑定依赖泛型与期望输出的模型结构
model = OpenAIChatModel('gpt-4o-mini')
agent = Agent(
model,
deps_type=DatabaseDeps,
output_type=AuditResult,
system_prompt="Analyze the query logs and output the structured audit result."
)
# 4. 注册工具,通过 RunContext 安全获取外部物理依赖句柄
@agent.tool
def check_user_db_logs(ctx: RunContext[DatabaseDeps], user_name: str) -> str:
"""
Fetch raw database logs for a specific user from internal storage.
:param user_name: The raw username to query.
"""
# 依赖项物理隔离,大模型无法直接读取或伪造 db_conn_pool
pool = ctx.deps.db_conn_pool
role = ctx.deps.operator_role
print(f"[SYSTEM DI] Fetching from pool '{pool}' under operator role '{role}' for user '{user_name}'...")
return f"LOGS FOR {user_name}: SELECT * FROM credit WHERE amount > 10000; (Operator: {role})"
# 5. 生产级运行示例
if __name__ == "__main__":
# 初始化外部可信依赖
my_deps = DatabaseDeps(
db_conn_pool="postgresql://prod_db:5432/audit",
operator_role="SecOps_Lead"
)
# 触发运行,系统将自动进行 Pydantic 格式校验与自动纠偏循环
result = agent.run_sync(
"Audit the activity of user 'zhangxi' and check for unauthorized credit access.",
deps=my_deps
)
# 6. 获取强类型输出实体,类型自动转换为 AuditResult
audit_data: AuditResult = result.output
print("--- Struct Validation Passed Successfully ---")
print(f"User ID: {audit_data.user_id}")
print(f"Has Violation: {audit_data.has_violation}")
print(f"Violation List: {audit_data.violations}")
4. 生产级故障演进与运维排查¶
| 故障现象 | 底层诱因 | 系统级表现 | 防御与排查手段 |
|---|---|---|---|
重试次数耗尽异常 (UnexpectedModelBehavior) |
模型持续输出无法通过 Pydantic 校验的格式,耗尽了自动纠偏重试次数(对应源码 pydantic_ai/_tool_execution.py 中的 Exceeded maximum output retries (...) 报错)。 |
业务请求中断,抛出 UnexpectedModelBehavior 异常并返回 HTTP 500。 |
1. 简化 output_type 对应的模型结构,避免深层嵌套 Pydantic Model。2. 在 Pydantic 字段 Field(description=...) 中增加极其详尽的描述引导模型。 |
| 依赖注入丢失 (Dependency Missing) | 在执行 run() 时忘记传入 deps 参数,或传入了 None。 |
工具 Handler 试图访问 ctx.deps 时触发 AttributeError 崩溃。 |
1. 严格对 Deps 引入强类型静态检查器(如 MyPy)。2. 在 Handler 入口增加首行 assert ctx.deps is not None。 |
| 数据范围越界 (Validation Fail) | 模型生成的参数通过了 Python 基本类型校验,但违反了 Pydantic 的值域限制(如 gt=100)。 |
触发自纠错逻辑,循环重试,P99 延迟显著抬升。 | 1. 在 Prompt 中明确指出取值约束界限。 2. 换用上下文理解力更强的中大参数模型。 |
5. 资深系统架构师面试表达方案¶
面试提问:你们是如何解决大模型生成数据不确定、格式易错的问题,来保证后端业务线消费到的是 100% 格式安全的数据?
回答模版:
在我们的架构里,大模型的输出被当作一份不可信的外部输入来处理,所以我们选了 PydanticAI 来把这层校验做扎实。
具体是两件事:
第一,用 output_type 把输出锁定在结构化模型上:
我们不接受模型直接吐 Markdown 或自由文本,而是给 Agent 指定一个 Pydantic 模型作为 output_type。一旦返回内容触发 ValidationError,框架会把具体报错字段封装成一轮反馈消息发回给模型,让它照着报错自己纠正,重试几次仍失败才会真正抛错给上层——这样业务代码拿到的 result.output 基本都是已经校验过的结构化数据。
第二,用 RunContext 把物理依赖和模型推理隔开:
数据库连接池、密钥这类东西我们不会拼进 Prompt,而是包在 Deps 结构体里,通过 RunContext 在工具函数里按需取用。模型只负责决定“调用哪个工具、传什么逻辑参数”,真正碰数据库连接的代码始终留在我们自己的进程里,不经过模型的上下文。