PydanticAI 实战：用类型安全构建生产级 AI Agent

痛点

当你用 LangChain 或原生 OpenAI SDK 构建 AI Agent 时，大概率踩过这些坑：

输出格式不可控：模型返回的 JSON 经常缺字段、类型错误，需要大量 try-except 兜底
工具调用缺乏类型约束：函数参数靠字符串描述，IDE 无法补全，重构时容易漏改
依赖注入混乱：数据库连接、API client 在 tool 函数间传来传去，代码耦合严重
可观测性差：Agent 链路长，出了问题只能加 print 调试

PydanticAI 是 Pydantic 团队推出的 Agent 框架，核心思路是把 FastAPI 的开发体验带到 AI Agent 领域——类型安全、依赖注入、结构化输出，开箱即用。

方案：为什么选 PydanticAI

特性	PydanticAI	LangChain	原生 SDK
类型安全	✅ 原生 Pydantic 模型	❌ 弱类型 dict	❌ 手动解析
结构化输出	✅ output_type 自动约束	⚠️ 需额外 parser	❌ 手写 schema
依赖注入	✅ RunContext[T]	❌ 无	❌ 无
多模型支持	✅ 20+ provider	✅	❌ 单厂商
可观测性	✅ OpenTelemetry + Logfire	⚠️ 第三方	❌ 手动
学习曲线	低（类 FastAPI）	高（抽象多）	中

实操步骤

Step 1：安装与基础 Agent

# 推荐用 uv 管理依赖
uv add pydantic-ai

# 或 pip
pip install pydantic-ai

最简 Agent——一个运维告警摘要机器人：

from pydantic import BaseModel
from pydantic_ai import Agent


class AlertSummary(BaseModel):
    """结构化告警摘要"""
    severity: str       # critical / warning / info
    service: str        # 受影响服务
    root_cause: str     # 根因分析
    action: str         # 建议操作


agent = Agent(
    'openai:gpt-4o',
    output_type=AlertSummary,
    instructions=(
        '你是一位资深 SRE，负责分析告警信息。'
        '基于告警内容输出结构化摘要，重点是根因和操作建议。'
    ),
)

result = agent.run_sync(
    'CRITICAL: Pod OOMKilled in namespace production, '
    'container api-server memory usage 98%, restarts=5 in 10min'
)

print(result.output)
# severity='critical' service='api-server'
# root_cause='容器内存限制不足或存在内存泄漏，10分钟内OOM重启5次'
# action='立即检查内存 limits 配置，排查最近部署是否引入内存泄漏，临时扩大 memory limit'

关键点：output_type=AlertSummary 让模型必须返回符合 schema 的结构化数据，Pydantic 自动验证，类型不对直接报错——不再需要手动 json.loads + 兜底逻辑。

Step 2：用依赖注入管理外部资源

实际场景中，Agent 需要访问数据库、API、配置。PydanticAI 用 RunContext 做依赖注入：

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext


@dataclass
class OpsDeps:
    """运维 Agent 依赖"""
    prometheus_url: str
    alert_manager_url: str
    team_oncall: str


agent = Agent(
    'anthropic:claude-sonnet-4-20250514',
    deps_type=OpsDeps,
    instructions='你是运维助手，可以查询监控数据辅助故障排查。',
)


@agent.tool
async def query_prometheus(ctx: RunContext[OpsDeps], promql: str) -> str:
    """执行 PromQL 查询获取监控指标"""
    import httpx
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f'{ctx.deps.prometheus_url}/api/v1/query',
            params={'query': promql},
        )
        data = resp.json()
        return str(data['data']['result'][:5])  # 返回前5条


@agent.tool
def get_oncall_engineer(ctx: RunContext[OpsDeps]) -> str:
    """获取当前 oncall 工程师"""
    return ctx.deps.team_oncall


# 运行时注入依赖
deps = OpsDeps(
    prometheus_url='http://prometheus:9090',
    alert_manager_url='http://alertmanager:9093',
    team_oncall='张工 (mobile: 138xxxx)',
)

result = await agent.run(
    'api-server P99 延迟突然飙到 5s，帮我排查',
    deps=deps,
)

优势： - 依赖通过 deps 参数传入，tool 函数通过 ctx.deps 访问，不污染全局状态 - deps_type=OpsDeps 提供完整类型提示，IDE 里 ctx.deps. 自动补全 - 测试时可以注入 mock 依赖，不依赖真实 Prometheus

Step 3：多输出类型 + 流式响应

生产环境中，Agent 可能需要返回不同类型的结果：

from pydantic import BaseModel
from pydantic_ai import Agent


class ScaleUpAction(BaseModel):
    """扩容操作"""
    deployment: str
    replicas: int
    reason: str


class ConfigChangeAction(BaseModel):
    """配置变更"""
    resource: str
    field: str
    old_value: str
    new_value: str
    reason: str


class InvestigateAction(BaseModel):
    """需进一步排查"""
    steps: list[str]
    data_needed: list[str]


# 支持多种输出类型，模型自行选择最合适的
agent = Agent(
    'openai:gpt-4o',
    output_type=[ScaleUpAction, ConfigChangeAction, InvestigateAction],  # type: ignore
    instructions='基于告警信息，决定最合适的响应动作。',
)

result = agent.run_sync('CPU 持续 95%+ 超过 15 分钟，当前 replicas=2')
match result.output:
    case ScaleUpAction() as action:
        print(f'执行扩容: {action.deployment} -> {action.replicas} replicas')
    case ConfigChangeAction() as action:
        print(f'变更配置: {action.resource}.{action.field}')
    case InvestigateAction() as action:
        print(f'需排查: {action.steps}')

Python 3.10+ 的 match 语法 + Pydantic 模型，让分支处理既类型安全又可读。

Step 4：集成 OpenTelemetry 实现可观测性

from pydantic_ai import Agent
from pydantic_ai.integrations.logfire import configure_logfire

# 方式1: 用 Pydantic Logfire（开箱即用）
configure_logfire(token='your-logfire-token')

# 方式2: 对接已有 OTel Collector（Jaeger / Tempo）
import logfire
logfire.configure(
    send_to_logfire=False,
    additional_span_processors=[your_otel_processor],
)

agent = Agent('openai:gpt-4o', instructions='...')
# 所有 agent.run() 调用自动产生 trace span
# 包括：模型调用耗时、token 消耗、tool 执行链路

在 Grafana Tempo 或 Jaeger 中，你能看到完整的 Agent 执行链路：哪个 tool 被调用了、耗时多少、token 消耗多少——这在调试复杂 multi-agent 系统时极其重要。

避坑指南

1. output_type 不要过度嵌套

# ❌ 嵌套太深，模型容易生成错误
class Bad(BaseModel):
    data: dict[str, list[dict[str, Any]]]

# ✅ 扁平化 + 明确字段含义
class Good(BaseModel):
    metrics: list[MetricItem]
    summary: str

模型生成 JSON 的准确率与 schema 复杂度负相关。字段控制在 5-8 个，嵌套不超过 2 层。

2. Tool 函数必须写 docstring

PydanticAI 把 docstring 作为 tool description 发给模型。没有 docstring = 模型不知道什么时候该调用这个工具。

# ❌ 模型不知道何时调用
@agent.tool_plain
def get_pods():
    return kubectl_get_pods()

# ✅ 清晰描述触发条件
@agent.tool_plain
def get_pods() -> str:
    """获取 Kubernetes 集群中所有 Pod 的状态列表，当需要排查 Pod 异常时调用"""
    return kubectl_get_pods()

3. 异步优先，同步兜底

PydanticAI 原生支持 async。生产环境中 Agent 通常嵌入 FastAPI 或事件循环：

# 生产环境：async
result = await agent.run('...')

# 脚本/测试：sync（内部会创建事件循环）
result = agent.run_sync('...')

如果你的 tool 涉及网络 I/O（查 Prometheus、调 API），务必用 async def 避免阻塞。

总结

PydanticAI 的核心价值：把非结构化的 LLM 输出变成类型安全的 Python 对象，把松散的 Agent 代码变成可测试、可观测的工程化系统。

适合的场景： - 运维告警智能分析与自动响应 - 基础设施变更的 AI 审批助手 - 多步骤故障排查 Agent - 任何需要结构化输出 + 工具调用的 LLM 应用

不适合的场景： - 纯聊天机器人（用原生 SDK 更轻量） - 复杂的 multi-agent 图调度（考虑 LangGraph 或 CrewAI）

一句话：如果你用 Python 写 AI Agent，且对代码质量有要求，PydanticAI 是当前最「工程化」的选择。