PydanticAI 与 LangGraph 深度对比
本章以 LangChain 首席科学家的视角,客观分析两个框架的设计理念、技术特点和适用场景。
本章概览
本章将从多个维度对比 PydanticAI 和 LangGraph:
- 设计哲学和核心理念
- 架构和实现方式
- 代码风格和开发体验
- 功能特性对比
- 性能和生态系统
- 选型建议
1. 设计哲学对比
1.1 核心定位
┌─────────────────────────────────────────────────────────────────┐
│ 框架定位对比 │
├────────────────────────────┬────────────────────────────────────┤
│ PydanticAI │ LangGraph │
├────────────────────────────┼────────────────────────────────────┤
│ │ │
│ "FastAPI for AI Agents" │ "Orchestration for AI Agents" │
│ │ │
│ 类型安全 + 简洁 │ 状态管理 + 精确控制 │
│ │ │
│ 声明式 Agent 定义 │ 显式图结构定义 │
│ │ │
│ 验证驱动开发 │ 状态驱动开发 │
│ │ │
└────────────────────────────┴────────────────────────────────────┘1.2 设计理念
PydanticAI 的理念:
- 简单优先:让简单的事情保持简单
- 类型安全:编译时捕获错误
- 验证驱动:输出必须符合预期
- 隐式控制流:框架管理执行逻辑
LangGraph 的理念:
- 控制优先:开发者完全掌控
- 状态中心:明确的状态流转
- 图驱动:可视化的执行路径
- 显式控制流:每个决策都可定制
1.3 哲学对比表
| 维度 | PydanticAI | LangGraph |
|---|---|---|
| 控制粒度 | 粗粒度(框架托管) | 细粒度(开发者控制) |
| 学习曲线 | 平缓 | 较陡 |
| 灵活性 | 适中 | 极高 |
| 代码量 | 少 | 较多 |
| 调试难度 | 简单 | 需要理解图结构 |
| 可视化 | 无内置 | 内置图可视化 |
2. 架构对比
2.1 核心架构差异
PydanticAI 架构 LangGraph 架构
────────────── ──────────────
┌─────────┐ ┌─────────────────────┐
│ Agent │ │ StateGraph │
│ │ │ ┌───────────────┐ │
│ ┌─────┐ │ │ │ State │ │
│ │Tools│ │ │ │ (TypedDict) │ │
│ └─────┘ │ │ └───────┬───────┘ │
│ │ │ │ │
│ ┌─────┐ │ │ ┌───────▼───────┐ │
│ │Deps │ │ │ │ Nodes │ │
│ └─────┘ │ │ │ (Functions) │ │
│ │ │ └───────┬───────┘ │
│ ┌─────┐ │ │ │ │
│ │Output│ │ │ ┌───────▼───────┐ │
│ └─────┘ │ │ │ Edges │ │
└────┬────┘ │ │ (Transitions) │ │
│ │ └───────────────┘ │
▼ └─────────────────────┘
隐式执行流 │
▼
显式图结构2.2 组件映射
| PydanticAI | LangGraph | 说明 |
|---|---|---|
Agent | StateGraph | 核心控制单元 |
@agent.tool | Node function | 可执行的操作 |
RunContext | State | 运行时上下文 |
output_type | State schema | 输出定义 |
instructions | System message in state | 系统指令 |
run() | graph.invoke() | 执行入口 |
run_stream() | graph.stream() | 流式执行 |
| 隐式 | add_edge() | 状态转移 |
| 隐式 | add_conditional_edges() | 条件路由 |
3. 代码对比
3.1 简单 Agent 实现
PydanticAI 版本:
python
from pydantic_ai import Agent
agent = Agent(
'openai:gpt-4o',
instructions='你是一个有帮助的助手。'
)
result = agent.run_sync('什么是人工智能?')
print(result.output)LangGraph 版本:
python
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
llm = ChatOpenAI(model="gpt-4o")
def chatbot(state: State):
return {"messages": [llm.invoke(state["messages"])]}
graph = StateGraph(State)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
result = app.invoke({
"messages": [("system", "你是一个有帮助的助手。"),
("user", "什么是人工智能?")]
})
print(result["messages"][-1].content)代码量对比:PydanticAI 约 7 行,LangGraph 约 20 行。
3.2 带工具的 Agent
PydanticAI 版本:
python
from pydantic_ai import Agent
import httpx
agent = Agent('openai:gpt-4o')
@agent.tool_plain
async def get_weather(city: str) -> str:
"""获取城市天气"""
async with httpx.AsyncClient() as client:
resp = await client.get(f"https://api.weather.com/{city}")
return resp.text
result = await agent.run('北京今天天气怎么样?')LangGraph 版本:
python
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
import httpx
class State(TypedDict):
messages: Annotated[list, add_messages]
@tool
async def get_weather(city: str) -> str:
"""获取城市天气"""
async with httpx.AsyncClient() as client:
resp = await client.get(f"https://api.weather.com/{city}")
return resp.text
tools = [get_weather]
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)
def chatbot(state: State):
return {"messages": [llm.invoke(state["messages"])]}
def should_continue(state: State):
last = state["messages"][-1]
if last.tool_calls:
return "tools"
return END
graph = StateGraph(State)
graph.add_node("chatbot", chatbot)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "chatbot")
graph.add_conditional_edges("chatbot", should_continue)
graph.add_edge("tools", "chatbot")
app = graph.compile()
result = await app.ainvoke({
"messages": [("user", "北京今天天气怎么样?")]
})3.3 结构化输出
PydanticAI 版本:
python
from pydantic import BaseModel
from pydantic_ai import Agent
class Analysis(BaseModel):
sentiment: str
score: float
keywords: list[str]
agent = Agent('openai:gpt-4o', output_type=Analysis)
result = agent.run_sync('分析这段文本的情感...')
print(result.output.sentiment) # 类型安全!
print(result.output.score)LangGraph 版本:
python
from langchain_openai import ChatOpenAI
from langchain_core.pydantic_v1 import BaseModel
class Analysis(BaseModel):
sentiment: str
score: float
keywords: list[str]
llm = ChatOpenAI(model="gpt-4o").with_structured_output(Analysis)
# 直接使用,无需图
result = llm.invoke("分析这段文本的情感...")
print(result.sentiment)注意:对于简单的结构化输出,LangGraph 用户通常直接使用 LangChain 的
with_structured_output,无需构建图。
4. 功能特性对比
4.1 特性矩阵
| 特性 | PydanticAI | LangGraph |
|---|---|---|
| 基础能力 | ||
| 单 Agent 对话 | ✅ 优秀 | ✅ 支持 |
| 工具调用 | ✅ 优秀 | ✅ 优秀 |
| 流式输出 | ✅ 优秀 | ✅ 优秀 |
| 结构化输出 | ✅ 原生支持 | ✅ 通过 LangChain |
| 高级能力 | ||
| 多 Agent 协作 | ⚠️ 基础支持 | ✅ 核心能力 |
| 复杂状态管理 | ⚠️ 有限 | ✅ 强大 |
| 条件路由 | ❌ 隐式 | ✅ 显式定义 |
| 循环控制 | ⚠️ 自动 | ✅ 精确控制 |
| 人类介入 | ✅ 支持 | ✅ 优秀 |
| 检查点/恢复 | ⚠️ 基础 | ✅ 强大 |
| 开发体验 | ||
| 类型安全 | ✅ 核心特性 | ⚠️ 可选 |
| IDE 支持 | ✅ 优秀 | ⚠️ 一般 |
| 测试支持 | ✅ 内置 TestModel | ⚠️ 需自行实现 |
| 可视化 | ❌ 无 | ✅ 内置 |
| 生态集成 | ||
| MCP 支持 | ✅ 原生 | ⚠️ 社区方案 |
| 可观测性 | ✅ Logfire 原生 | ✅ LangSmith 原生 |
| 模型支持 | ✅ 20+ | ✅ 100+ |
4.2 多 Agent 能力对比
PydanticAI 多 Agent:
python
from pydantic_ai import Agent
# 创建专家 Agent
researcher = Agent('openai:gpt-4o', instructions='你是研究专家')
writer = Agent('openai:gpt-4o', instructions='你是写作专家')
# 委托模式
main_agent = Agent('openai:gpt-4o')
@main_agent.tool
async def research(ctx, topic: str) -> str:
"""调用研究专家"""
result = await researcher.run(topic)
return result.output
@main_agent.tool
async def write(ctx, content: str) -> str:
"""调用写作专家"""
result = await writer.run(content)
return result.outputLangGraph 多 Agent:
python
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
class TeamState(TypedDict):
messages: Annotated[list, add_messages]
next: str
researcher_llm = ChatOpenAI(model="gpt-4o")
writer_llm = ChatOpenAI(model="gpt-4o")
def supervisor(state: TeamState):
# 决定下一个 Agent
response = supervisor_llm.invoke(...)
return {"next": response.content}
def researcher(state: TeamState):
response = researcher_llm.invoke(state["messages"])
return {"messages": [response]}
def writer(state: TeamState):
response = writer_llm.invoke(state["messages"])
return {"messages": [response]}
def route(state: TeamState):
return state["next"]
# 构建图
graph = StateGraph(TeamState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route, {
"researcher": "researcher",
"writer": "writer",
"FINISH": END,
})
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
team = graph.compile()关键差异:
- PydanticAI:通过工具委托,简单但不够灵活
- LangGraph:显式图结构,复杂但控制精确
5. Graph 能力对比
5.1 PydanticAI Graph
PydanticAI 也提供了 Graph 能力(通过 pydantic-graph):
图:PydanticAI Graph 官方文档
python
from pydantic_graph import BaseNode, End, Graph
from dataclasses import dataclass
@dataclass
class State:
value: int
class IncrementNode(BaseNode[State, None, int]):
async def run(self, ctx) -> "DecrementNode | End[int]":
ctx.state.value += 1
if ctx.state.value >= 5:
return End(ctx.state.value)
return DecrementNode()
class DecrementNode(BaseNode[State, None, int]):
async def run(self, ctx) -> IncrementNode:
ctx.state.value -= 1
return IncrementNode()
graph = Graph(nodes=[IncrementNode, DecrementNode])
result = await graph.run(IncrementNode(), state=State(value=0))5.2 LangGraph Graph
python
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class State(TypedDict):
value: int
def increment(state: State):
return {"value": state["value"] + 1}
def decrement(state: State):
return {"value": state["value"] - 1}
def should_continue(state: State):
if state["value"] >= 5:
return END
return "decrement"
graph = StateGraph(State)
graph.add_node("increment", increment)
graph.add_node("decrement", decrement)
graph.add_edge(START, "increment")
graph.add_conditional_edges("increment", should_continue)
graph.add_edge("decrement", "increment")
app = graph.compile()
result = app.invoke({"value": 0})5.3 Graph 能力对比
| 特性 | PydanticAI Graph | LangGraph |
|---|---|---|
| 定义方式 | 类继承 | 函数 + 配置 |
| 类型安全 | 编译时检查 | 运行时检查 |
| 状态传递 | 通过 context | 通过 state dict |
| 边定义 | 返回值推断 | 显式 add_edge |
| 条件路由 | 返回类型 | conditional_edges |
| 可视化 | Mermaid 导出 | 内置可视化 |
| 成熟度 | 较新 | 成熟 |
6. 性能对比
6.1 启动开销
| 方面 | PydanticAI | LangGraph |
|---|---|---|
| 导入时间 | ~50ms | ~200ms |
| Agent 创建 | ~5ms | ~10ms |
| 首次调用 | ~100ms | ~150ms |
注:实际性能受环境影响,以上为参考值。
6.2 运行时开销
两者的运行时开销主要来自 LLM API 调用,框架本身开销可忽略。
6.3 内存占用
| 场景 | PydanticAI | LangGraph |
|---|---|---|
| 基础 Agent | ~20MB | ~50MB |
| 带工具 Agent | ~25MB | ~60MB |
| 多 Agent | ~40MB | ~100MB |
7. 生态系统对比
7.1 可观测性
PydanticAI + Logfire:
python
import logfire
from pydantic_ai import Agent
logfire.configure()
logfire.instrument_pydantic_ai()
agent = Agent('openai:gpt-4o')LangGraph + LangSmith:
python
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "..."
# 自动追踪7.2 测试支持
PydanticAI:
python
from pydantic_ai.models.test import TestModel
agent = Agent(TestModel()) # 无需真实 API
result = agent.run_sync("test")LangGraph:
python
# 需要 mock 或使用 fake LLM
from langchain_community.llms.fake import FakeListLLM
fake_llm = FakeListLLM(responses=["test response"])7.3 社区和文档
| 方面 | PydanticAI | LangGraph |
|---|---|---|
| GitHub Stars | ~5K | ~10K |
| 发布时间 | 2024 | 2023 |
| 文档质量 | 优秀 | 优秀 |
| 示例数量 | 较少 | 丰富 |
| 社区活跃度 | 增长中 | 活跃 |
| Stack Overflow | 较少 | 较多 |
8. 选型建议
8.1 选择 PydanticAI 的场景
✅ 推荐使用 PydanticAI:
1. 快速原型开发
└── 最少代码实现功能
2. 类型安全要求高
└── IDE 支持、编译时检查
3. 单 Agent + 工具场景
└── 简单直接的交互模式
4. FastAPI 项目集成
└── 设计理念一致
5. 需要结构化输出
└── Pydantic 原生支持
6. 团队熟悉 Pydantic
└── 学习成本最低8.2 选择 LangGraph 的场景
✅ 推荐使用 LangGraph:
1. 复杂多 Agent 系统
└── Supervisor、层级架构
2. 需要精确控制流程
└── 条件路由、循环控制
3. 复杂状态管理
└── 检查点、恢复、回滚
4. 企业级应用
└── 成熟、稳定、生态丰富
5. 人类介入工作流
└── Human-in-the-loop 原生支持
6. 需要流程可视化
└── 图结构直观展示8.3 决策树
开始
│
├── 需要复杂的多 Agent 协作?
│ ├── 是 ──► LangGraph
│ └── 否 ──┐
│ │
├── 需要精确控制执行流程?
│ ├── 是 ──► LangGraph
│ └── 否 ──┐
│ │
├── 类型安全是首要考虑?
│ ├── 是 ──► PydanticAI
│ └── 否 ──┐
│ │
├── 快速原型/MVP 开发?
│ ├── 是 ──► PydanticAI
│ └── 否 ──┐
│ │
├── 已有 LangChain 基础设施?
│ ├── 是 ──► LangGraph
│ └── 否 ──► 根据团队偏好选择
│
结束9. 关键洞察
作为 LangChain 首席科学家,以下是我的客观分析:
9.1 PydanticAI 的创新
- 类型系统的深度整合:真正的编译时类型安全
- FastAPI 设计理念:让 AI 开发像 Web 开发一样简单
- 验证优先:输出验证是一等公民
- 依赖注入:优雅的运行时依赖管理
9.2 LangGraph 的优势
- 精确控制:图结构提供完全的执行控制
- 状态管理:强大的检查点和恢复能力
- 成熟生态:丰富的集成和社区支持
- 可视化:直观的流程理解
9.3 互补而非竞争
两个框架解决不同层次的问题:
┌─────────────────────────────────────────────────────────────────┐
│ 应用复杂度谱系 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 简单 ◄────────────────────────────────────────────────► 复杂 │
│ │
│ ┌─────────────────────────────┐ │
│ │ PydanticAI │ │
│ │ • 单 Agent │ │
│ │ • 简单工具链 │ │
│ │ • 结构化输出 │ │
│ └─────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────┐│
│ │ LangGraph ││
│ │ • 多 Agent 协作 ││
│ │ • 复杂状态机 ││
│ │ • 企业级编排 ││
│ └───────────────────────────────────────────┘│
│ │
└─────────────────────────────────────────────────────────────────┘10. 小结
| 维度 | PydanticAI | LangGraph |
|---|---|---|
| 核心优势 | 类型安全、简洁 | 控制精确、功能强大 |
| 适合场景 | 快速开发、简单应用 | 复杂系统、企业级 |
| 学习曲线 | 平缓 | 较陡 |
| 代码风格 | 声明式 | 图式 |
| 成熟度 | 新兴 | 成熟 |
两个框架各有所长,选择取决于:
- 项目复杂度
- 团队技术栈
- 控制粒度需求
- 类型安全重要性
在下一章,我们将探讨如何让这两个框架协同工作,发挥各自优势。