跳转至

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.outputresult_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 在工具函数里按需取用。模型只负责决定“调用哪个工具、传什么逻辑参数”,真正碰数据库连接的代码始终留在我们自己的进程里,不经过模型的上下文。