`pydantic/pydantic-ai`：大模型结构化输出的类型契约与依赖注入内核¶

在智能体开发中，大模型生成的“非结构化文本”是不确定性的源头。Pydantic 团队推出的 PydanticAI 框架，旨在解决核心工程痛点：如何将 LLM 的概率性输出强行拉回确定性的后端强类型轨道。它通过 Pydantic 核心数据校验层，确立了模型与系统之间的执行契约（Execution Contract），并引入了优雅的运行时依赖注入（Dependency Injection）设计。

1. 结构化输出契约与反应式自纠错（Self-Correction Loop）¶

PydanticAI 拒绝将大模型视为“自由对话者”，而是将其物理建模为一个类型安全的强契约函数：Agent[Deps, ResultModel]。

1.1 结构化约束机制¶

当为 Agent 指定了 result_type=MyPydanticModel 时，PydanticAI 自动将该 Pydantic 模型的 JSON Schema 转换为系统 Prompt 附件发给 LLM。大模型返回的文本必须能够被该 Pydantic 模型成功反序列化（Parse），否则无法进入下行数据流。

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）封装为一轮新的系统反馈消息发送给大模型，让大模型在了解“自己哪里写错了”的前提下自动进行下一轮纠偏生成，默认重试 3 次，直至输出完全合规的结构。

2. 强类型依赖注入 (Dependency Injection) 与 RunContext 架构¶

将可信的运行时参数（如数据库连接池、租户 ID、API Client）通过 Prompt 字符串拼接下发给 Agent 是极其危险的反模式（Anti-Pattern），这极易引发提示词注入攻击（Prompt Injection）。

RunContext[Deps] 机制：PydanticAI 引入了依赖注入架构。通过泛型 Deps 将可信数据源与句柄注入 Agent 运行上下文。
物理隔离：所有的 Tools 和中间件都在本地运行时通过 RunContext 静态获取依赖，这些可信资源完全不经过大模型的推理空间，大模型只决定“调不调工具”以及“传入什么逻辑参数”，而无法触碰或污染底层的系统物理句柄。

3. 基于 PydanticAI 的强契约智能体构建（Python 实践）¶

以下是使用 Python 编写的 PydanticAI 生产级智能体服务骨架，展示了依赖注入与强契约自纠偏的融合：

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 OpenAIModel

# 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 = OpenAIModel('gpt-4o-mini')
agent = Agent(
    model,
    deps_type=DatabaseDeps,
    result_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.data
    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. 生产级故障演进与运维排查¶

故障现象	底层诱因	系统级表现	防御与排查手段
`ModelRetryLimitExceeded` 异常	模型持续输出无法通过 Pydantic 校验的格式，耗尽了自动纠偏重试次数。	业务请求中断，抛出崩溃错误并返回 HTTP 500。	1. 简化 ResultType 结构，避免深层嵌套 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 框架来建立硬契约。

我们实施了两层硬性隔离与治理：第一，结构化类型契约与自纠错环（Structured Contract Loop）：我们绝不让大模型输出 Markdown 或 Free Text，而是将输出强制约束在自定义的 Pydantic 模型泛型中。当模型输出发生偏离引发 ValidationError 时，框架会自动捕获详细的字段堆栈报错，将其封装为一轮 System Feedback 反馈给模型，由底层引擎在毫秒级内自动发起就地纠错重试（Self-Correction Retry），直到 100% 通过格式校验才放行下传给业务线消费。

第二，可信资源依赖注入隔离（DI Isolation）：我们坚信“可信物理句柄绝对不进模型推理域”。所有的数据库连接池、加密 Token，全部封装在強类型的 Deps 结构体中，通过 RunContext 进行运行时依赖注入。大模型仅仅做出调用哪个工具的逻辑决策，而物理执行句柄在本地运行时直接被 DI 捕获，实现了模型决策与系统物理副作用的强力解耦，从而保证了线上智能体系统的高可用、高安全性。

pydantic/pydantic-ai：大模型结构化输出的类型契约与依赖注入内核¶