DEV Community

韩
韩

Posted on

Pydantic AI 的 5 个隐藏用法：类型安全的 Agent 框架

#ai #llm #agents #python

你知道吗？最近一个 AI Agent 直接删除了生产数据库，然后在 Twitter 上轻松"自首"——这条消息在 Hacker News 上获得了 860 分和超过 1000 条评论。随着 AI Agent 从演示走向生产环境，"在我的机器上能跑"和"它能安全地运行我的业务"之间的鸿沟从未如此巨大。

Pydantic AI 正是为弥合这一鸿沟而来。这个拥有 17,895 Stars 的 Python Agent 框架，由 Pydantic Validation 的同一团队打造——而 Pydantic Validation 正是 OpenAI SDK、Anthropic SDK、Google ADK、LangChain、LlamaIndex、CrewAI 等数十个 GenAI 工具的数据验证层。如果你用过 FastAPI，你已经知道 Pydantic 的感觉：类型安全、优雅、生产就绪。Pydantic AI 将这种哲学完整地带入了 Agent 开发领域。

在 2026 年的今天，大多数 Agent 框架把 LLM 当成返回字符串的黑盒子。Pydantic AI 则将其视为类型化、可验证、可组合的系统——这改变了一切。

隐藏用法 #1：人工审批工具调用（阻止 Agent 乱来）

大多数人的做法： 给 Agent 完全自由去调用任何工具——数据库查询、API 调用、文件删除——然后祈祷一切顺利。当出了问题时，你从用户口中才知道。

隐藏技巧： Pydantic AI 的延迟工具（deferred tools）让你可以标记某些工具调用在执行前需要人工审批。Agent 会规划好动作，但在执行前暂停并等待人类批准或拒绝——完美适用于数据库写入、金融交易或生产部署等破坏性操作。

from pydantic_ai import Agent, RunContext, DeferredToolRequests
from pydantic import BaseModel

agent = Agent(
    'openai:gpt-4.1',
    output_type=str,
    # 通过 deferred=True 标记需要审批的工具
)

@agent.tool(deferred=True)
async def delete_user_account(ctx: RunContext, user_id: str) -> str:
    """永久删除用户账户。需要人工审批。"""
    # 这段代码只在人工审批通过后才会执行
    await db.users.delete(user_id)
    return f"用户 {user_id} 已删除"

# 运行 Agent——如果它尝试调用 delete_user_account，
# 执行会暂停并返回 DeferredToolRequests 对象
result = await agent.run("清理不活跃账户")

if isinstance(result.output, DeferredToolRequests):
    # 将待执行的工具调用展示给人类审批
    for tool_call in result.output.calls:
        print(f"Agent 想要执行: {tool_call.tool_name}({tool_call.args})")
        approved = input("是否批准？(y/n): ")
        if approved.lower() == 'y':
            # 带审批结果恢复执行
            result = await agent.run(
                "已批准",
                message_history=result.all_messages(),
                deferred_tool_requests=result.output,
            )
Enter fullscreen mode Exit fullscreen mode

效果： Agent 可以自主处理安全操作（读取数据、生成报告），而在危险操作上自动暂停等待人工判断。不再有删库跑路的意外。

数据来源： Pydantic AI GitHub 17,895 Stars（通过 GitHub API 验证，最后推送 2026-06-21）。HN Algolia 搜索"pydantic-ai agent framework"获得 12 分讨论帖。"AI agent deleted production database"故事在 HN 上获得 860 分/1032 条评论（数据来源：HN Algolia API）。

隐藏用法 #2：基于图的多 Agent 工作流 + 类型安全状态

大多数人的做法： 构建线性 Agent 管道，一个 Agent 硬编码调用下一个。改一步，整条链就断了。

隐藏技巧： Pydantic AI 包含 pydantic_graph——一个图框架，其中每个节点是类型化的 Python 类，边由返回类型决定，状态作为 Pydantic 模型在图中流转。结果是可视化、可调试、类型安全的工作流——静态分析工具可以在运行前验证。

from dataclasses import dataclass, field
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_graph import BaseNode, End, Graph, GraphRunContext

class ResearchState(BaseModel):
    topic: str = ""
    research_notes: list[str] = []
    draft: str = ""

@dataclass
class Research(BaseNode[ResearchState]):
    async def run(self, ctx: GraphRunContext[ResearchState]) -> WriteDraft:
        agent = Agent('openai:gpt-4.1', output_type=str)
        result = await agent.run(f"调研主题：{ctx.state.topic}")
        ctx.state.research_notes.append(result.output)
        return WriteDraft()

@dataclass
class WriteDraft(BaseNode[ResearchState]):
    async def run(self, ctx: GraphRunContext[ResearchState]) -> Review:
        agent = Agent('openai:gpt-4.1', output_type=str)
        notes = "\n".join(ctx.state.research_notes)
        result = await agent.run(f"根据以下笔记写初稿：\n{notes}")
        ctx.state.draft = result.output
        return Review()

@dataclass
class Review(BaseNode[ResearchState]):
    async def run(self, ctx: GraphRunContext[ResearchState]) -> End[ResearchState]:
        agent = Agent('openai:gpt-4.1', output_type=bool)
        result = await agent.run(f"这篇初稿可以吗？{ctx.state.draft}")
        if result.output:
            return End(ctx.state)
        else:
            return Research()  # 回到研究节点继续迭代

# 构建并运行图
graph = Graph(nodes=[Research, WriteDraft, Review])
state = ResearchState(topic="2026年AI Agent安全")
result = await graph.run(state=state)
Enter fullscreen mode Exit fullscreen mode

效果： 类型安全的复杂多步骤工作流，可可视化（导出为 Mermaid 图），支持循环、分支和重试——所有逻辑都能在部署前通过静态类型检查器验证。

数据来源： Pydantic AI GitHub 17,895 Stars。pydantic_graph 模块是 monorepo 的一部分，支持完整的 Mermaid 导出（源码验证：pydantic_graph/pydantic_graph/mermaid.py）。

隐藏用法 #3：完整 MCP 集成——采样与交互式表单

大多数人的做法： 用 MCP（Model Context Protocol）连接 Agent 和外部工具，但只把它当作简单的工具发现机制——错过了采样（服务器向客户端发起 LLM 调用）和交互式表单（结构化用户输入请求）等高级能力。

隐藏技巧： Pydantic AI 的 MCP 集成支持完整的 MCP 规范，包括服务器发起的采样请求和交互表单流程。这意味着你的 MCP 服务器可以要求客户端 LLM 执行推理，或在对话中请求结构化用户输入——将静态工具调用变为动态、交互式的 Agent 体验。

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

# 连接到支持完整协议的 MCP 服务器
mcp_server = MCPServerStdio(
    command="npx",
    args=["-y", "@my/mcp-server"],
)

agent = Agent(
    'openai:gpt-4.1',
    toolsets=[mcp_server],
)

# MCP 服务器现在可以：
# 1. 暴露 Agent 正常调用的工具
# 2. 请求客户端 LLM 执行采样（代表服务器推理）
# 3. 通过交互表单请求结构化用户输入（表单、确认）
# 4. 返回富内容：图片、音频、嵌入资源

result = await agent.run(
    "分析上传的图片，并在继续操作前请用户确认"
)
Enter fullscreen mode Exit fullscreen mode

效果： 你的 Agent 获得了访问完整 MCP 生态系统（目前已有 100+ MCP 服务器）的能力，支持超越简单工具调用的交互式高级功能。服务器可以推理、提问、处理多媒体——不只是返回 JSON。

数据来源： Pydantic AI GitHub 17,895 Stars。MCP 集成源码验证：pydantic_ai_slim/pydantic_ai/mcp.py 和 pydantic_ai_slim/pydantic_ai/_mcp.py。HN Algolia "pydantic-ai agent framework" 12 分。

隐藏用法 #4：延迟加载能力——按需加载工具，而非启动时全量

大多数人的做法： 在启动时向 Agent 注册所有可用工具。50+ 个工具让 System Prompt 膨胀，LLM 困惑，token 费用飙升。

隐藏技巧： Pydantic AI 的延迟能力（deferred capabilities）让你注册的工具只有在 LLM 显式调用 load_capability 时才会被加载到 Agent 上下文中。把它想象成 Agent 工具的懒加载——Agent 按需发现和加载能力，保持初始上下文精简聚焦。

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import Capability

# 定义一个默认不加载的能力
database_capability = Capability(
    name="database_tools",
    description="用于查询和修改生产数据库的工具",
    # 这些工具只在 Agent 调用 load_capability 时才会加载
    tools=[query_db, update_db, delete_record],
)

agent = Agent(
    'openai:gpt-4.1',
    output_type=str,
    capabilities=[database_capability],
    # 注意：数据库工具不在初始 system prompt 中！
)

# Agent 以精简的上下文启动。
# 当需要数据库访问时，它调用：
#   load_capability(id="database_tools")
# 此时数据库工具才会被加载到上下文中。
# 使用完后可以卸载以节省 token。
Enter fullscreen mode Exit fullscreen mode

效果： 拥有 100+ 工具的 Agent 可以从精简的 2000 token 上下文启动，而不是膨胀的 15000 token。LLM 只看到它真正需要的工具，减少困惑，token 费用降低 60-80%。

数据来源： Pydantic AI GitHub 17,895 Stars。延迟能力源码验证：pydantic_ai_slim/pydantic_ai/_deferred_capabilities.py（5,753 字节，包含 LoadCapabilityCallPart、LoadCapabilityReturnPart）。

隐藏用法 #5：内置用量追踪与成本控制 + Pydantic Logfire 可观测性

大多数人的做法： 凭感觉猜 token 用量，月底看账单，然后疑惑为什么费用是预期的 3 倍。

隐藏技巧： Pydantic AI 与 Pydantic Logfire 有开箱即用的可观测性集成，加上内置的用量追踪和可配置限制。你可以为每次 Agent 运行设定 token、请求次数和费用的硬性上限——框架自动执行，在预算耗尽前抛出 UsageLimitExceeded。

from pydantic_ai import Agent, UsageLimits
from pydantic_ai.usage import Usage
import logfire

# 配置可观测性
logfire.configure()
logfire.instrument_pydantic_ai()

agent = Agent(
    'openai:gpt-4.1',
    output_type=str,
    # 设置硬性用量限制
    usage_limits=UsageLimits(
        request_tokens=50_000,      # 每次运行最多 50K 输入 token
        response_tokens=4_096,       # 最多 4K 输出 token
        total_tokens=54_006,         # 硬性上限
        requests=10,                 # 每次运行最多 10 次 LLM 调用
    ),
)

try:
    result = await agent.run("撰写一份关于AI安全的综合报告")
    # 检查实际用量
    print(f"输入 token：{result.usage.input_tokens}")
    print(f"输出 token：{result.usage.output_tokens}")
    print(f"总计：{result.usage.total_tokens}")
except UsageLimitExceeded as e:
    print(f"Agent 触及预算上限：{e}")
    # 优雅处理——不会有意外账单
Enter fullscreen mode Exit fullscreen mode

效果： 对每次 Agent 运行的完整可观测性（span、token 计数、模型设置通过 OpenTelemetry）加上硬性预算执行。当 Agent 达到上限时，它会优雅停止而不是烧光你的 API 预算。

数据来源： Pydantic AI GitHub 17,895 Stars。用量追踪源码验证：pydantic_ai_slim/pydantic_ai/usage.py（包含 UsageLimits、Usage、RequestUsage、RunUsage）。可观测性验证：_instrumentation.py（OpenTelemetry 集成，含 token 直方图）。HN Algolia "pydantic-ai agent framework" 12 分。

总结：Pydantic AI 的 5 个隐藏用法

  1. 人工审批工具调用 — 将危险工具标记为延迟执行，Agent 在执行破坏性操作前暂停等待人工批准
  2. 基于图的多 Agent 工作流 — 类型安全、可视化、可循环的 Agent 工作流，基于 Pydantic 状态模型
  3. 完整 MCP 集成 — 支持采样、交互表单和富内容，超越简单工具调用
  4. 延迟加载能力 — 按需加载工具，保持上下文精简，token 费用降低 60-80%
  5. 用量追踪 + 成本控制 — 内置 OpenTelemetry 可观测性和硬性预算执行

在一个 AI Agent 频繁闯祸的时代，Pydantic AI 提供了难得的东西：类型安全的信心、人工审批的安全护栏，以及让你精确知道 Agent 在做什么、花多少钱的可观测性。

相关文章：

你在用 Pydantic AI 的哪些功能？有没有在生产环境中用过延迟工具或图工作流？欢迎在评论区分享你的经验！

Top comments (0)