Pydantic AI 实战：类型安全的 AI 应用开发，比 LangChain 更优雅

LangChain 用多了会发现：返回格式不固定、类型错误运行时才暴露、调试靠 print——Pydantic AI 用 Pydantic 模型定义 AI 输入输出，让 AI 应用的类型安全达到传统后端的水准。

为什么需要类型安全的 AI 框架？

# LangChain 常见问题：返回格式不确定
from langchain.output_parsers import JsonOutputParser

parser = JsonOutputParser()
result = chain.invoke({"query": "分析这篇文章"})
# result 是什么类型？字段有哪些？IDE 完全不知道 ❌
# 运行时才可能发现字段缺失

Pydantic AI 的解法：用 Pydantic Model 明确定义 AI 输出结构，IDE 有完整补全，运行时自动校验。

核心优势对比

对比维度	LangChain	Pydantic AI
输出类型安全	❌ 运行时才知道	✅ 编译时类型检查
IDE 补全	弱	完整（Pydantic模型驱动）
依赖重量	重（50+ 依赖包）	轻量（核心仅 Pydantic + httpx）
学习曲线	陡峭（抽象层次多）	平缓（类似 FastAPI）
结构化输出	靠 Prompt 约束	原生支持，强制校验
流式输出	支持	原生支持，类型保持

实战1：5 行代码跑通第一个 Agent

# pip install pydantic-ai==0.0.26
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel

# 1. 定义模型
model = OpenAIModel(
    model_name="gpt-4o",
    api_key="sk-xxxxxxxx"
)

# 2. 创建 Agent（指定返回类型）
agent = Agent(
    model=model,
    result_type=str,  # 返回类型：str / Pydantic Model / int / bool
    system_prompt="你是一个技术分析专家，回答简洁准确。"
)

# 3. 运行
result = agent.run_sync("RAG 和 Fine-tuning 应该怎么选？")
print(result.data)
# 输出：RAG适合... Fine-tuning适合...（类型为 str，IDE有补全）

实战2：用 Pydantic Model 强制结构化输出

这是 Pydantic AI 的杀手锏功能——让 AI 输出严格符合你的数据模型：

from pydantic import BaseModel, Field
from typing import List, Optional
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel

# 定义结构化输出模型
class CodeReviewResult(BaseModel):
    """代码审查结果——AI输出必须完全符合这个结构"""
    overall_score: int = Field(ge=0, le=100, description="代码质量评分0-100")
    issues: List[str] = Field(description="发现的问题列表")
    suggestions: List[str] = Field(description="改进建议列表")
    severity: str = Field(description="严重程度：low/medium/high/critical")
    needs_rewrite: bool = Field(description="是否需要重写")

# 创建 Agent
reviewer = Agent(
    model=OpenAIModel(model_name="gpt-4o", api_key="sk-xxx"),
    result_type=CodeReviewResult,
    system_prompt="你是一个资深代码审查专家。审查代码后，严格按照指定JSON结构输出结果。"
)

# 运行
code_snippet = """
def login(password):
    if password == "admin123":
        return True
    return False
"""

result = reviewer.run_sync(f"请审查以下代码：\n{code_snippet}")
review = result.data  # 类型为 CodeReviewResult，IDE 完整补全！

print(f"评分：{review.overall_score}/100")
print(f"严重程度：{review.severity}")
print(f"问题：{review.issues}")
print(f"建议：{review.suggestions}")
print(f"需要重写：{review.needs_rewrite}")

运行输出示例（自动校验，字段一个不少）：

评分：30/100
严重程度：critical
问题：['硬编码密码，存在严重安全风险', '未使用参数化验证', '无密码复杂度检查']
建议：['使用bcrypt/scrypt加密存储密码', '实现密码复杂度策略', '添加登录失败锁定机制']
需要重写：True

实战3：集成工具调用（Function Tools）

Pydantic AI 的工具调用基于 Python 函数签名自动生成工具定义，比 LangChain 简洁太多：

import httpx
from pydantic_ai import Agent, RunContext
from pydantic_ai.tools import Tool

# 定义工具函数（类型注解自动生成工具schema）
def get_weather(city: str) -> str:
    """获取指定城市的当前天气"""
    # 实际对接天气 API
    weather_data = {
        "北京": "晴，28°C",
        "上海": "多云，25°C",
        "深圳": "阵雨，30°C"
    }
    return weather_data.get(city, "天气数据不可用")

def search_documentation(query: str, ctx: RunContext) -> str:
    """搜索内部技术文档（使用上下文依赖）"""
    es_url = ctx.deps["elasticsearch_url"]  # 从依赖注入获取配置
    # 实际调用 ES 搜索
    return f"关于'{query}'的文档：..."

# 创建带工具的 Agent
agent = Agent(
    model=OpenAIModel(model_name="gpt-4o", api_key="sk-xxx"),
    result_type=str,
    system_prompt="你是技术助手，可以查天气和搜索文档。"
)

# 注册工具（自动生成 OpenAI Function Calling schema）
agent.register_tool(Tool(get_weather))
agent.register_tool(Tool(search_documentation))

# 运行（AI自动决定调用哪个工具）
result = agent.run_sync("北京今天天气怎么样？适合户外部署服务器吗？")
print(result.data)

实战4：依赖注入（Deps）—— 复用连接和配置

from pydantic_ai import Agent, RunContext
from typing import Dict
import redis

# 定义依赖类型
class AppDeps(BaseModel):
    redis_client: Dict[str, str] = {}  # 实际用 Redis 连接池
    db_url: str
    api_keys: Dict[str, str]

# 创建带依赖的 Agent
agent = Agent(
    model=OpenAIModel(model_name="gpt-4o", api_key="sk-xxx"),
    deps_type=AppDeps,  # 声明依赖类型
    result_type=str,
)

# 工具函数可以通过 ctx.deps 访问依赖
@agent.tool
def get_cache(ctx: RunContext[AppDeps], key: str) -> str:
    """从 Redis 缓存获取数据"""
    # 实际：return ctx.deps.redis_client.get(key)
    return f"cache:{key} = xxx"

# 运行前注入依赖
deps = AppDeps(
    redis_client={"host": "localhost", "port": "6379"},
    db_url="postgresql://user:pass@localhost/appdb",
    api_keys={"openai": "sk-xxx"}
)

result = agent.run_sync("从缓存获取用户配置", deps=deps)

实战5：流式输出（Streaming）

from pydantic_ai import Agent
import asyncio

agent = Agent(
    model=OpenAIModel(model_name="gpt-4o", api_key="sk-xxx"),
    result_type=str,
)

# 异步流式输出
async def stream_demo():
    async with agent.run_stream("详细讲解 Kubernetes 的 Service 和 Ingress 区别") as response:
        async for chunk in response.stream_text():
            print(chunk, end="", flush=True)  # 逐字输出，类似打字效果

# 结构化流式输出（高级用法）
async def stream_structured():
    async with agent.run_stream(
        "分析这段代码的问题",
        result_type=CodeReviewResult  # 流式输出也能保持类型安全！
    ) as response:
        async for partial in response.stream():
            # partial 是部分构建的 Pydantic 模型
            if hasattr(partial, 'overall_score'):
                print(f"当前评分：{partial.overall_score}")

asyncio.run(stream_demo())

Pydantic AI vs LangChain 代码量对比

场景：构建一个”代码审查 + 自动修复”的 AI 工具

# ===== LangChain 版本（约 80 行）=====
from langchain.chat_models import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import ChatPromptTemplate

parser = PydanticOutputParser(pydantic_object=CodeReviewResult)
prompt = ChatPromptTemplate.from_messages([
    ("system", "审查代码\n{format_instructions}"),
    ("human", "{code}")
])
chain = prompt | ChatOpenAI() | parser
result = chain.invoke({
    "code": code_snippet,
    "format_instructions": parser.get_format_instructions()
})

# ===== Pydantic AI 版本（约 30 行）=====
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel

agent = Agent(
    model=OpenAIModel(model_name="gpt-4o", api_key="sk-xxx"),
    result_type=CodeReviewResult,
    system_prompt="你是代码审查专家。"
)
result = agent.run_sync(f"审查代码：\n{code_snippet}")
# result.data 类型安全，直接访问 .overall_score 有 IDE 补全

生产环境落地建议

项目阶段 → 框架选择：

MVP / 原型验证
  → Pydantic AI（开发速度快 2x，类型安全减少 debug 时间）

需要大量现成工具链（PDF解析、SQL执行器等）
  → LangChain（生态更丰富）

高性能生产环境
  → Pydantic AI + 自定义工具（依赖少，启动快）

团队 Python 水平参差不齐
  → Pydantic AI（代码更像普通 Python，学习曲线平缓）

与 FastAPI 无缝集成（这才是真正的生产力）：

from fastapi import FastAPI
from pydantic_ai import Agent
from pydantic import BaseModel

app = FastAPI()
agent = Agent(model=OpenAIModel(...), result_type=str)

class ReviewRequest(BaseModel):
    code: str
    language: str = "python"

@app.post("/review")
async def review_code(req: ReviewRequest):
    result = await agent.run(req.code)
    return {"result": result.data}

# uvicorn main:app --reload  # 直接跑起来

总结

Pydantic AI 的核心价值：让 AI 应用开发回归正常的软件工程实践——类型安全、依赖注入、单元测试、IDE 补全，这些后端开发的日常，在 AI 应用里也应该有。

上 Pydantic AI 的三个信号：
1. LangChain 的 result 类型让你头疼，运行时才知道对不对
2. 团队协作需要统一的输出规范，不能靠 Prompt “约定”
3. 项目要从原型走向生产，需要可维护的代码结构

---

👤 作者简介

一枚在大中原腹地（河南）卖公有云的从业者，主营腾讯云/阿里云/火山云，曾踩坑无数，现专注AI大模型应用落地。关注公众号「公有云cloud」，围观AI前沿动态~

博客：y1cloud.cn