AI Agent 工程师 JD · 第一性原理拆解:pydantic-ai 与 langgraph 怎么把 JD 的每一条变成工程

这份文档要回答的问题:那张 JD 上的要求——状态/上下文管理、Tool Calling 稳定性、多 Agent 流程、系统工程、token 成本、ReAct/Plan-Execute、生产 vs demo——不是一堆零散的关键词,而是一个 Agent 系统从第一性原理出发,绕不开的同一组问题。本文先从最小内核推出"绕不开的问题清单"(Part 0),再逐个 JD 问题展开,每个问题都讲透 pydantic-ai 和 langgraph 各自怎么实现、为什么这么实现、什么时候用哪个。

怎么读:从 Part 0 顺着读。Part 0 建立一个贯穿全文的主心智模型——「两个设计中心」。后面六个 Part 都是这个模型在具体问题上的展开。每个 Part 的结构固定:① 第一性原理(为什么这是个不可回避的问题)→ ② pydantic-ai 怎么做 → ③ langgraph 怎么做 → ④ 对照与判断 → ⑤ 上生产要补什么。

取材与时效(诚实披露):本文所有框架机制都对着本地 clone 的源码核验过(pydantic-ai/ 与 langgraph/,截至 2026-05)。这两个 clone 是比较前沿的开发版,有几个 API 和稳定发行版不同,文中会逐一标注(例如 pydantic-ai 的 capabilities 体系、langgraph 的 error_handler 参数、recursion_limit 默认值 10007)。市场采用数据(谁在用)是时间点快照,会变。读到具体 API 时,以你实际安装的版本为准——文中给了源码路径,自己 grep 一遍最稳。

目录

• Part 0 · 全局视角:Agent 系统的第一性原理 —— 最小内核 → 绕不开的问题清单 → 两个设计中心 → 主对照表
• Part 1 · 状态与上下文管理(JD: memory / context window)
• Part 2 · 工具调用稳定性与错误处理(JD: timeout / retry / partial failure)
• Part 3 · 多 Agent 流程设计(JD: planner-executor / 并行 / 异步)
• Part 4 · 系统工程:稳定性 / 失败率 / 延迟 / 日志 / 调试(JD: 系统工程意识)
• Part 5 · Token 与调用成本(JD: token / 调用成本)
• Part 6 · 哪些适合生产、哪些只适合 demo(JD: 清晰判断)
• 附录 A · 一页速查表 ｜ 附录 B · 源码索引 ｜ 附录 C · 真实案例 ｜ 附录 D · 自测题

Part 0 · 全局视角:Agent 系统的第一性原理

0.1 Agent 的最小内核:它不是 while 循环,是状态机

先把 Agent 这个词剥到只剩骨头。一个 LLM,本质上是一个无状态的纯函数:

LLM: (一段文本) → (一段文本)

它不会记住上一次说了什么,不会真的"执行"任何动作,它只会生成文本。所谓"Agent 调用了工具""Agent 查了数据库""Agent 改了订单",全都是一层错觉——真实发生的是:

1. 你(框架)把目标 + 历史 + 可用工具的描述,拼成一段文本,喂给 LLM;
2. LLM 生成一段文本,里面夹着一个结构化的"我想调用 search_sku(query='排骨')"的意图(tool call);
3. 框架解析出这个意图,框架真的去执行那个 Python 函数,拿到结果;
4. 框架把结果(observation)再拼回文本,喂给 LLM,回到第 1 步。

这就是 ReAct 循环(Reason → Act → Observe)。很多人把它想成一个 while 循环:

while not done:
    thought, action = llm(context)   # Reason
    observation = execute(action)     # Act
    context += observation            # Observe

这个 while 循环的心智模型是错的,或者说是不够的。 一旦你要管"这一步失败了怎么办""调到一半要等用户确认怎么办""进程崩了怎么从中间恢复""并行调三个工具其中一个挂了怎么办"——while 循环就不够用了。正确的心智模型是状态机(state machine):每一步是一个有名字的节点,节点之间的转移是显式的,当前"走到哪了 + 手里有什么状态"是可以被快照、被检查、被恢复的。

这不是比喻,是两个框架的真实实现。 - pydantic-ai 的一次 agent.run(),在源码里字面意义上就是一个图(graph)实例在跑:build_agent_graph() 构造出一个由 UserPromptNode → ModelRequestNode → CallToolsNode(循环)→ End 组成的图(pydantic_ai_slim/pydantic_ai/_agent_graph.py:2066)。 - langgraph 干脆把"图"放进了名字里,它的执行器是一个 Pregel/BSP 超步(super-step) 引擎(libs/langgraph/langgraph/pregel/_loop.py)。

记住这一点:"Agent = 状态机" 是后面一切的地基。 JD 里每一条要求,本质上都是"这个状态机的某个维度要怎么工程化"。

0.2 从最小内核,推出绕不开的问题清单

现在做一件事:盯着上面那个最小内核,不看任何框架,纯粹问"要让这玩意儿在真实世界里跑起来、跑得住,我被迫要解决哪些问题?" 你会发现 JD 上那些词,一条条自己冒出来:

这张表就是把 JD"串起来"的钥匙。 JD 不是 HR 拍脑袋列的关键词,它是一个写过生产 Agent 的人,把"Agent 系统的物理学"翻译成了招聘语言。你能从最小内核推出这张表,就说明你真懂这个领域——而不是背了一堆框架 API。

JD 里还有两个加分项没进上表,它们是"横切"的: - ReAct / Plan-Execute / Multi-Agent 在真实系统应用过:这不是独立问题,而是 Part 3 的"方法论"维度——同一个多 agent 问题,用哪种编排范式落地。本文在 Part 3 集中讲,Part 1/2 里也会提到 ReAct 循环本身。 - 金融 / 量化 / 风控背景:领域加分,本文在 Part 5(成本模型)和附录 C(真实交易 agent 模式)轻量带过,不展开。

0.3 主心智模型:两个「设计中心」

到这里,关键问题来了:pydantic-ai 和 langgraph 都要解决上面那六个问题,但它们给出的答案系统性地不同。如果你一个 API 一个 API 地背,会觉得"怎么这么多差异、记不住"。但如果你抓住每个框架的一个"设计中心"(design center),后面所有差异都能从这个中心推出来。

pydantic-ai 的设计中心 = 类型安全的函数 / 循环(type-safe function & loop)

它的世界观:一次 agent 运行,就是一个图实例;这个图的状态,永远是一条消息列表 list[ModelMessage] + 你的业务依赖 deps。 业务数据不进框架状态,而是通过依赖注入(RunContext[DepsT])从旁边喂进来。pydantic 在这里被用到极致:每一次工具调用的入参、每一次结构化输出,都走 pydantic 的校验引擎(validate_python),在热路径上跑。

源码证据: - 一次运行的状态容器是 GraphAgentState(一个 dataclass,不是 pydantic model),里面是 message_history: list[ModelMessage]、usage、run_step、conversation_id……(_agent_graph.py:118) - 依赖单独放在 RunContext[T].deps 里,和图状态分离(_run_context.py:35)

# pydantic-ai:agent 即"类型安全的函数",输入输出都被 pydantic 锁死
from pydantic_ai import Agent
from pydantic import BaseModel

class CityInfo(BaseModel):
    city: str
    country: str

agent = Agent('openai:gpt-5.2', output_type=CityInfo)   # 输出 schema 由 pydantic 保证
result = agent.run_sync('The windy city in the US of A.')
print(result.output)   # CityInfo(city='Chicago', country='USA') —— 已校验、强类型

langgraph 的设计中心 = 并发的 reducer 状态机(concurrent reducer state machine)

它的世界观:一切都是图;状态是你自己定义的一组"通道(channel)",每个通道带一个 reducer(归并函数),决定并发写入怎么合并;执行靠 Pregel 超步推进,每个超步结束就可以做一次 checkpoint。 pydantic 在这里只被当成类型注解的载体——langgraph 读你的 TypedDict / BaseModel 的注解来知道有哪些通道、用哪个 reducer,但几乎不调用 pydantic 的校验(runtime 代码里 model_validate 出现 0 次,校验交给 langchain-core)。

源码证据: - Pregel 超步循环在 pregel/_loop.py(内部把一步叫 superstep) - 状态通道与 reducer:Annotated[list, add_messages] 这种写法,add_messages 是个按消息 ID 归并的 reducer,不是简单 append(libs/langgraph/langgraph/graph/message.py) - 持久化靠 checkpointer(BaseCheckpointSaver + InMemorySaver/SqliteSaver/PostgresSaver),按 thread_id 存取

# langgraph:一切都是图;状态是带 reducer 的通道;执行靠超步
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # reducer 决定并发写入怎么合并

builder = StateGraph(State)
builder.add_node("chatbot", lambda s: {"messages": [("assistant", "Hi")]})
builder.set_entry_point("chatbot")
builder.set_finish_point("chatbot")
graph = builder.compile()      # 必须 compile 才能跑
graph.invoke({"messages": []})

一句话记住两个中心

LangGraph 用的是 pydantic 的"类型注解",不是它的"校验引擎";pydantic-ai 用的是 pydantic 的全部。

更精确地说:它们解决的是不同的根问题,所以 pydantic 在两边扮演完全不同的角色,所以下游一切都不同。这不是"谁更强",是"设计中心不同"。

设计中心 → 下游一切差异(因果图)

                    ┌─────────────────────────────────────────────────────────────┐
                    │  pydantic-ai 设计中心:类型安全的函数/循环                       │
                    │  (state = list[ModelMessage] + deps;pydantic 在热路径)         │
                    └─────────────────────────────────────────────────────────────┘
                                              │ 推出
   ┌──────────────────────┬──────────────────┼───────────────────┬──────────────────────┐
   ▼                      ▼                  ▼                   ▼                      ▼
状态:schema 收敛       工具:入参/输出      重试:ModelRetry      并行:asyncio.wait      成本:UsageLimits
(业务进 deps,         强校验(热路径)     喂回 LLM 自纠正      保留 sibling          一等硬约束 + cost()
 框架状态恒定)                                                  (不 fail-fast)

                    ┌─────────────────────────────────────────────────────────────┐
                    │  langgraph 设计中心:并发的 reducer 状态机                       │
                    │  (state = channels + reducers;pydantic 只是类型注解)           │
                    └─────────────────────────────────────────────────────────────┘
                                              │ 推出
   ┌──────────────────────┬──────────────────┼───────────────────┬──────────────────────┐
   ▼                      ▼                  ▼                   ▼                      ▼
状态:schema 灵活       工具:校验交给        重试:RetryPolicy     并行:Send + 超步       成本:不原生跟踪
(任意 channel,         langchain-core      整 node 重跑;       (asyncio.gather       (callback/LangSmith
 reducer 定合并)        ToolMessage 喂回    error_handler        fan-out)              外挂观测)

把这张图记在脑子里。后面每个 Part,我都会回到"这是哪个设计中心推出来的"。

0.4 主对照表(全文索引)

下面这张表是全文的"地图":每一行是一个 JD 问题,每一列是一个框架的"招法"。后面每个 Part 就是展开其中一行。

Part 1 · 状态与上下文管理

JD 原文:状态与上下文管理(memory / context window)

1.1 第一性原理:为什么"状态"是 Agent 的头号问题

回到最小内核:LLM 是无状态纯函数。它不记得上一句话。所以"对话能连续""agent 记得你三轮前说的预算是 5000"——这些全都是框架在每一轮把历史重新拼进 prompt 制造出来的幻觉。

这立刻逼出两个硬约束:

1. 你必须自己存历史(LLM 不存)。存哪、存什么格式、谁来负责持久化,是工程问题。
2. context window 有限且贵(见 Part 5)。历史不可能无限往里塞。于是"在有限的窗口里塞下够用的信息"成了核心技艺——这就是 context engineering。

那么一个 Agent 在任意时刻,到底要"hold"住哪些状态?可以拆成四类(这个分类来自对 pydantic-ai 源码的剖析,但它是框架无关的):

两个框架最根本的区别,就是"会话事实"和"业务依赖"怎么放。 记住这个分类,下面看两边怎么实现。

1.2 pydantic-ai 怎么做:schema 收敛 + 依赖注入

pydantic-ai 的设计中心(类型安全的函数)直接决定了它的状态模型:会话事实永远是一条 list[ModelMessage];业务依赖走 deps 注入,和会话事实严格分开。

(1) 多轮对话 = 把上一轮的消息列表传进下一轮

from pydantic_ai import Agent, ModelMessagesTypeAdapter

agent = Agent('openai:gpt-5.2')
r1 = agent.run_sync('Tell me a joke.')
r2 = agent.run_sync('Explain it.', message_history=r1.new_messages())  # 续上文

# 持久化:消息列表序列化成 JSON 存库,下次读回来
blob = ModelMessagesTypeAdapter.dump_json(r2.all_messages())   # bytes,可入库
restored = ModelMessagesTypeAdapter.validate_json(blob)        # 读回来续跑

• result.all_messages() 给你整段历史;result.new_messages() 只给这一轮新增的。
• ModelMessagesTypeAdapter(就是 TypeAdapter(list[ModelMessage]))负责 JSON 双向序列化——这是"自己存历史"的标准动作。
• 关键 gotcha:框架不会自动持久化。 agent.run() 跑完,消息就在内存里;是你负责把 all_messages() 写进 Postgres/Redis。框架只给你序列化工具,不替你存。

(2) 业务依赖走 RunContext[DepsT],不进会话历史

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class Deps:
    merchant_id: str
    db: "DBConn"

agent = Agent('openai:gpt-5.2', deps_type=Deps)

@agent.tool
async def search_sku(ctx: RunContext[Deps], query: str) -> list[str]:
    # merchant_id 从 deps 拿,不是从 LLM 的参数拿 —— 这点对防注入很关键(见 Part 2)
    return await ctx.deps.db.search(ctx.deps.merchant_id, query)

result = agent.run_sync('找排骨', deps=Deps(merchant_id='M123', db=conn))

这就是"schema 收敛":无论业务多复杂,框架状态永远是 messages,业务的东西都在 deps 里。好处是框架状态稳定、可序列化、可预测;坏处是你得自己设计 deps。

(3) context window 管理 = ProcessHistory 能力(capability)

历史会越来越长,迟早撑爆窗口。pydantic-ai 用能力(capability)机制在"每次请求模型前"插一个钩子来裁剪/压缩历史:

from pydantic_ai import Agent, ModelMessage, RunContext
from pydantic_ai.capabilities import ProcessHistory

def keep_recent(ctx: RunContext[None], messages: list[ModelMessage]) -> list[ModelMessage]:
    if ctx.usage.total_tokens > 1000:
        return messages[-6:]      # token 超阈值就只留最近几条
    return messages

agent = Agent('openai:gpt-5.2', capabilities=[ProcessHistory(keep_recent)])

版本提示(本文 clone 的前沿版):在这个 clone 里,历史处理器通过 capabilities=[ProcessHistory(fn)] 装配(源码 pydantic_ai_slim/pydantic_ai/capabilities/process_history.py),ProcessHistory 实现了 before_model_request 钩子。更早的稳定版用的是 Agent(..., history_processors=[fn]) 这个参数形式。功能等价,装配方式不同;以你装的版本为准。压缩策略本身(只留最近 N 条 / 摘要旧消息 / 检索式召回)是框架无关的。

(4) 动态 system prompt + conversation_id

@agent.system_prompt(dynamic=True)
async def sys(ctx: RunContext[Deps]) -> str:
    return f'当前商家:{ctx.deps.merchant_id}'

conversation_id 每次首跑自动生成、随 message_history 继承,可显式覆盖。它是"把多轮 run 串成一个会话"的关联键(Part 4 观测会用到)。

1.3 langgraph 怎么做:带 reducer 的通道 + checkpointer

langgraph 的设计中心(并发 reducer 状态机)决定了它的状态模型完全不同:状态是你自定义的一组通道,每个通道带一个 reducer 决定"并发写入怎么合并";持久化靠 checkpointer 按 thread_id 自动存。

(1) 状态 = StateGraph(schema),通道用 Annotated[T, reducer] 声明合并规则

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 这个通道:用 add_messages 合并
    draft_total: float                        # 没标 reducer:默认 LastValue(后写覆盖)

builder = StateGraph(State)

• add_messages 不是简单 append,而是按消息 ID 归并/去重(graph/message.py)。
• 三种核心通道语义:LastValue(取最新,默认)、BinaryOperatorAggregate(用你给的二元操作符合并,如 operator.add 做列表拼接)、Topic(PubSub 广播)。
• 为什么需要 reducer? 因为 langgraph 允许多个节点在同一个超步并发写同一个通道。两个并发节点都往 messages 写,框架必须知道"怎么合并这两份写入"——这就是 reducer 存在的根本原因。这是它"并发"设计中心的直接产物。

(2) 持久化 = checkpointer + thread_id

from langgraph.checkpoint.memory import InMemorySaver        # 仅 dev/test
# from langgraph.checkpoint.postgres import PostgresSaver    # 生产

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "conv-123"}}
graph.invoke(input1, config)     # 自动存 checkpoint
graph.invoke(input2, config)     # 同一 thread_id → 自动接着上次的状态跑

• 这是和 pydantic-ai 最大的体感差别:持久化是内建的。你给一个 checkpointer + thread_id,框架就在每个超步后自动把整个状态快照存下来,下次同 thread_id 自动恢复。你不用手动 dump_json 再写库。
• 关键 gotcha:InMemorySaver 进程一重启就全没了,源码 docstring 明说"只用于 debug/test,生产请用 PostgresSaver"(libs/checkpoint/.../memory/__init__.py:33)。"用了 checkpointer" ≠ "持久化了",得看是哪个 saver。

1.4 对照与判断

怎么判断用哪个?

• 你的状态本质上就是"一段对话 + 一些业务句柄",不需要复杂的并发状态合并 → pydantic-ai 的 schema 收敛更省心,框架状态永远可预测。
• 你的状态是个复杂的、多节点并发读写的工作流状态(多个 agent 往同一块状态写) → langgraph 的 channel + reducer 是为此而生,而且 checkpointer 白送你"跨进程恢复"。
• 一个常见反模式(来自源码剖析的洞察):langgraph 的"啥都塞 TypedDict"会随业务复杂度膨胀,state schema 越长越乱;pydantic-ai 的"state = messages,业务进 deps"把框架状态钉死,业务复杂度的增长不污染框架状态。两种取舍,没有绝对优劣。

context window 工程(框架无关的通用招):无论哪个框架,长会话都得在"塞进窗口的信息"上动手——① 滑动窗口(只留最近 N 条);② 摘要(把旧消息让 LLM 压成一段 summary);③ 检索式召回(把旧消息存向量库,按当前问题召回 top-k);④ 结构化外置(把"已确认的事实"抽成结构化状态,不靠原始对话承载)。pydantic-ai 用 ProcessHistory 挂这些策略,langgraph 在节点里手写。

1.5 上生产要补什么

• 持久化层:pydantic-ai 必须自己接(Postgres/Redis 存 message list);langgraph 把 InMemorySaver 换成 PostgresSaver。
• 会话并发控制:同一会话同时来两条消息怎么办?(锁 / 串行化,框架都不管)。
• 裁剪的完整性:切历史时,tool call 和它的 tool return 必须成对,切碎了模型会报错。这是两边都有的隐坑。
• 长会话的压缩策略:不能等撑爆才处理,要有 token 阈值触发的摘要/召回。

Part 2 · 工具调用稳定性与错误处理

JD 原文:Tool Calling 稳定性与错误处理(timeout / retry / partial failure)

2.1 第一性原理:工具是 Agent 伸向不可靠世界的手

回到最小内核第 3 步:框架真的去执行那个函数。这一步是整个 Agent 系统里"现实"侵入的地方——前面的 reason 都在 LLM 的文本世界里,只有 tool call 真的会碰网络、碰数据库、碰第三方 API、碰钱。所以一切不确定性、一切失败,都集中在这里。

而且失败不是一种,是四层,每层的处理方式不同(这个分层来自对 pydantic-ai 的剖析,框架无关):

核心洞察(也是面试高频考点):协议层和业务层的失败,不该当异常崩掉,而该变成喂回 LLM 的"observation",让这个状态机继续转、自我纠正。传输层和供应商层才是传统的"重试/降级"。两个框架对"协议/业务层失败"的处理哲学,正好分道扬镳——这是 Part 2 的戏眼。

还有一个独立维度:partial failure(部分失败)。一个 LLM 响应可能一次要调 3 个工具。如果第 2 个挂了,第 1、3 个的结果还要不要?能不能让 LLM 看到"1、3 成功,2 失败"然后继续?这就是并行工具的容错语义,两个框架的实现也不同。

2.2 pydantic-ai 怎么做:ModelRetry 自纠正回路

pydantic-ai 的设计中心(类型安全 + 把错误喂回 LLM)在这里体现得最淋漓尽致。

(1) 协议/业务层:ModelRetry —— 框架主动把错误变成 LLM 的 observation

from pydantic_ai import Agent, RunContext, ModelRetry

agent = Agent('openai:gpt-5.2')

@agent.tool(retries=3)                       # 这个工具最多重试 3 次
def place_order(ctx: RunContext, sku: str, qty: int) -> str:
    if qty <= 0:
        raise ModelRetry(f'数量必须是正数,你给的是 {qty},请重新调用')  # ← 喂回 LLM
    if past_cutoff():
        raise ModelRetry('已过下单截止时间,请改约明天或询问用户')        # 业务层也走它
    return do_place(sku, qty)

• 工具里 raise ModelRetry(reason) → 框架把 reason 包成 RetryPromptPart 塞回消息流 → 模型看到"上次错在哪",重新生成 tool call。这是一个框架强制的自纠正回路。
• 协议层校验是免费的:工具入参由 pydantic 自动校验(设计中心),qty="二十" 这种会自动触发 retry,你不用写校验代码。
• 默认 retries=1(源码 agent/__init__.py:137,tools 和 output 都默认 1)。复杂 schema 下 1 次往往不够——这是 Part 6 的"must-know"之一。

(2) 防死循环:UsageLimits 在执行前就卡住

from pydantic_ai.usage import UsageLimits

result = agent.run_sync(
    '把这单处理完',
    usage_limits=UsageLimits(
        request_limit=50,          # 默认就是 50,不是无限
        tool_calls_limit=10,       # 最多 10 次工具调用
        total_tokens_limit=20_000,
    ),
)

• tool_calls_limit 是执行前检查(usage.py:check_before_tool_call):如果这一轮模型要调的工具数会超限,一个都不执行,直接抛 UsageLimitExceeded。这是 ReAct 循环跑飞的硬刹车。

(3) 传输层:HTTP 重试是 opt-in 的(必须自己加)

from pydantic_ai.retries import AsyncTenacityTransport, RetryConfig, wait_retry_after
from tenacity import retry_if_exception_type, stop_after_attempt
from httpx import HTTPStatusError

transport = AsyncTenacityTransport(RetryConfig(
    retry=retry_if_exception_type(HTTPStatusError),
    wait=wait_retry_after(max_wait=300),   # 尊重 429 的 Retry-After
    stop=stop_after_attempt(5),
))
# 把 transport 装到 httpx client,再传给 model —— 详见 docs/retries.md

• 关键 gotcha:pydantic-ai 默认没有自动 HTTP 重试(源码核验:model 层找不到内建重试逻辑)。传输层重试要你主动用 TenacityTransport/AsyncTenacityTransport 包 httpx。这也是 Part 6 的 must-know。
• HTTP 重试和工具重试是正交的两层:都开的话,实际尝试次数可能是两者相乘,要算清预算。

(4) partial failure:asyncio.wait 而非 gather,刻意保留 sibling

这是 pydantic-ai 一个精妙且高频被问的设计。一个模型响应里的多个 tool call,默认是并行执行的(get_parallel_execution_mode 默认 'parallel',源码 _agent_graph.py:1858),实现方式是 asyncio.create_task 起任务,再用 asyncio.wait 协调——而不是 asyncio.gather,也不是 asyncio.TaskGroup。

# _agent_graph.py:1884-1897 的真实逻辑(简化)
if parallel_execution_mode == 'parallel_ordered_events':
    await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
else:
    while pending:
        done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
        # 每个 task 的错误单独 try/except 处理,不影响其他 task

• 为什么不用 gather/TaskGroup? 因为 TaskGroup(以及 gather 默认行为)在一个任务失败时会取消其他兄弟任务。而 Agent 要的是相反的语义:第 2 个工具挂了,第 1、3 个的结果仍然有效,要一起喂回给 LLM,让它看到"1、3 成功、2 失败"再决策。用 asyncio.wait + 逐任务捕获,就能保留部分成功(partial-failure tolerance)。
• 想要顺序执行?给工具标 sequential=True,或用 sequential_tool_calls() 上下文。

(5) 工具抛了非 ModelRetry 的普通异常会怎样?

@agent.tool
def risky(ctx, x: int) -> int:
    return external_call(x)   # 如果这里抛 ConnectionError(不是 ModelRetry)?

源码核验(tool_manager.py:330-335):普通 Exception(非 ModelRetry、非 CallDeferred/ApprovalRequired 这些控制流)会被交给 on_tool_execute_error 能力钩子。默认钩子的行为是把原异常再抛出去(propagate)(capabilities/abstract.py:686 文档明确:"Raise the original error to propagate it")——也就是默认会让这次 run 失败。一个 capability 可以 override 这个钩子来"返回一个值压制错误"或"raise ModelRetry 转成自纠正"。

must-know(Part 6 会重申):框架不会把工具里的普通业务异常自动变成 retry——只有 ModelRetry 才触发自纠正。所以工具要么对可恢复的问题 raise ModelRetry(...),要么自己 try/except 把业务错误翻译好,否则默认就是抛出去、run 失败。

2.3 langgraph 怎么做:ToolNode 把异常转成 ToolMessage

langgraph 的设计中心(中立的状态机)决定了它的哲学相反:框架不替你决定怎么纠错,它把工具错误转成一条普通的状态更新(ToolMessage)喂回去,让 LLM 自己决定下一步。

(1) 协议/业务层:ToolNode 捕获异常 → ToolMessage(status='error')

from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool

@tool
def place_order(sku: str, qty: int) -> str:
    if qty <= 0:
        raise ValueError('数量必须是正数')
    return do_place(sku, qty)

# handle_tool_errors 可传 bool / str / Callable / 异常类型
def fmt_err(e: Exception) -> str:
    return f'工具出错:{e},请调整后重试'

tool_node = ToolNode([place_order], handle_tool_errors=fmt_err)

• ToolNode 捕获工具异常,格式化成一条 ToolMessage(角色 tool、带 tool_call_id)追加进 messages 通道(源码 libs/prebuilt/.../tool_node.py:749)。LLM 下一轮看到这条 error message,自己决定是重试、换工具还是告诉用户。框架不强制纠正回路。
• 一批工具里部分失败:ToolNode 用 asyncio.gather 并行跑,逐工具把成功/失败都收集成各自的 ToolMessage,一起回到状态——这点和 pydantic-ai 殊途同归(都保留部分结果),但实现是 gather(fail-fast bubbling)而非 asyncio.wait。

(2) 传输层:RetryPolicy,绑在节点上(整个 node 重跑)

from langgraph.types import RetryPolicy

policy = RetryPolicy(
    initial_interval=0.5, backoff_factor=2.0, max_interval=128.0,
    max_attempts=3, jitter=True, retry_on=Exception,
)
builder.add_node("tools", tool_node, retry_policy=policy)

• RetryPolicy(指数退避 + 抖动)由 Pregel 执行器在节点级应用(libs/langgraph/langgraph/types.py:406)。
• 关键 gotcha:重试粒度是整个节点,不是单个工具。一个节点里 3 个 tool call,只要触发重试,整个节点(3 个 call)全部重跑。如果工具有副作用又不幂等,这会重复执行——必须配幂等键(见 2.5)。
• add_node(..., retry=) 这个老参数已废弃,用 retry_policy=(源码 graph/state.py:753 有 deprecation 警告)。

(3) 节点/图级 error_handler(本文 clone 的较新能力)

def fallback(state, exc):
    return {"error": str(exc)}   # 节点抛错时兜底

builder.add_node("risky", risky_node, error_handler=fallback)        # 节点级
# 或在构造时设全图默认:StateGraph(State, error_handler=fallback)

版本提示:在这个 clone 里,error_handler 是真实参数——既能作为 StateGraph(..., error_handler=...) 的全图默认,也能在 add_node(..., error_handler=...) 设节点级(源码 graph/state.py:276, 672)。注意:没有 add_error_handler() 这个方法(那是常见的误记),也不是 compile() 的参数。稳定发行版里 langgraph 的错误处理更多靠 RetryPolicy + ToolNode(handle_tool_errors=) + 用条件边路由到"错误处理节点"。以你的版本为准。

2.4 对照与判断

怎么判断?

• 你想要框架替你把"参数错/业务错"自动变成模型的自我纠正信号、并且对工具入参强类型校验有刚需 → pydantic-ai 的 ModelRetry + pydantic 校验是最顺手的。
• 你想要完全掌控"出错后走哪条边"(出错路由到人工、路由到降级节点、记录后继续)→ langgraph 的 ToolMessage + 条件边/error_handler给你的是状态机级别的控制。
• 注意 retry 粒度的坑:langgraph 整 node 重跑,副作用工具必须幂等;pydantic-ai per-tool 重试也一样要幂等(任何 retry 机制的前提)。

2.5 上生产要补什么

无论哪个框架,以下这些框架都不替你做,必须自己补:

• 幂等(idempotency):任何会重试的、有副作用的工具,都要带幂等键(Idempotency-Key),让"重复执行一次"无害。这是 retry 能用的前提。
• 熔断(circuit breaker):两个框架都没有内建熔断。下游服务挂了还猛重试会雪崩,要自己在工具层或网关层加。
• 补偿(compensation),而不是回滚:这是最深的一个坑。框架的 checkpoint / 状态快照,回滚不了工具已经造成的外部副作用(钱已经退了、订单已经下了)。框架能恢复的是"它自己的状态",不是"外部世界"。所以"撤销"必须在业务层做补偿——记一条 Action Journal(动作日志:存 pre_state + 补偿函数 + 过期时间),用户点撤销时跑对应的补偿动作。这正是真实生产 agent(如附录 C 的 max-ai)花大力气做的事:upsert 的补偿是 remove,下单 的补偿是 取消(若还没过 cutoff),并给 30 秒撤销窗口。这条是区分"跑过生产"和"只做过 demo"的硬指标。
• 超时:工具级超时要自己设(LLM 不会主动 timeout 一个卡住的 HTTP 调用)。

Part 3 · 多 Agent 流程设计

JD 原文:多 Agent 流程设计(planner-executor / 并行 / 异步);加分项:ReAct / Plan-Execute / Multi-Agent 在真实系统应用过

3.1 第一性原理:为什么要多 Agent,以及它的两个根问题

回到最小内核:单个 LLM 循环的能力和注意力是有界的。当任务复杂到——工具几十个、要分阶段规划、不同子任务需要不同的系统提示/人格/权限——塞进一个 prompt + 一个工具池,LLM 会注意力涣散、工具选错、上下文互相干扰。这时才需要"多 Agent"。

但"多 Agent"是个被严重滥用的词。从第一性原理看,任何多 Agent / 编排设计,本质上只在回答两个根问题:

1. 执行编排(orchestration):下一步走哪,是 LLM 自己决定(自主、ReAct 式),还是代码确定性地路由(deterministic、状态机式)?越靠 LLM 自主,越灵活也越不可控;越靠代码路由,越可控也越死板。
2. 状态传播(state propagation):agent 之间怎么传信息——是共享一块状态(shared state / blackboard),还是像管道一样把输出传给下一个(pipe-and-filter / handoff)?

把这两个问题想清楚,再看任何框架的"多 Agent 功能",你都能一眼归类。

还要破一个迷思——多 Agent 有强弱之分,别动不动就说"我们做了多 agent":

JD 说"planner-executor / 并行 / 异步",指的是中到强。但面试里更值钱的判断是:"什么时候不需要多 agent"——这往往比堆 agent 更体现工程成熟度(见 3.5)。

3.2 pydantic-ai 怎么做:委派 / handoff / pydantic-graph

pydantic-ai 把多 agent 拆成递进的复杂度级别,核心是"agent 是无状态、可复用的 callable,可以被组合"。

(1) Agent 委派(agent-as-tool):一个 agent 在工具里调另一个 agent

from pydantic_ai import Agent, RunContext

planner = Agent('openai:gpt-5.2')
worker  = Agent('openai:gpt-5.2')

@planner.tool
async def do_subtask(ctx: RunContext[None], task: str) -> str:
    r = await worker.run(task, usage=ctx.usage)   # ← usage 传下去,成本累计到同一账本
    return r.output

• usage=ctx.usage 是关键:把父 agent 的 usage 账本传给子 agent,这样嵌套调用的 token/成本累计到同一个预算里(呼应 Part 5)。
• 这天然就是 planner-executor / Plan-Execute 模式:planner 把 worker 当工具用。

(2) 程序化 handoff:应用代码按序调多个 agent,传递 message_history

async def pipeline(usage):
    r1 = await search_agent.run(prompt, usage=usage)
    # 把上一段对话作为下一个 agent 的历史传过去
    r2 = await booking_agent.run('确认预订', message_history=r1.all_messages(), usage=usage)
    return r2.output

• 这是"中"层多 agent:编排逻辑在你的应用代码里,不是框架里。简单、可控、好测。

(3) pydantic-graph:需要显式状态机时

当流程复杂到需要显式的节点/边(条件分支、循环、回到某节点),pydantic-ai 提供底层的 pydantic-graph(就是驱动 agent loop 本身的那个图库):

from dataclasses import dataclass
from pydantic_graph import BaseNode, GraphRunContext, End

@dataclass
class Charge(BaseNode[OrderState, None, str]):   # BaseNode[StateT, DepsT, RunEndT]
    async def run(self, ctx: GraphRunContext[OrderState]) -> "Ship | End[str]":
        if ctx.state.paid:
            return Ship()
        return End('支付失败')

• 关键认知:pydantic-graph 是手写状态机的底层工具(BaseNode/GraphRunContext/End),不是 langgraph 那种"高层编排框架"。你要自己定义节点和转移。它适合"我需要精确控制流程"的场景,但没有 langgraph 的 Send fan-out、checkpointer 那一套开箱即用的编排糖。

(4) 并行 / 异步:agent.run() 本身是 async 的,多个独立 agent 的并行 = 应用层 asyncio.gather:

results = await asyncio.gather(agent_a.run(x), agent_b.run(y), agent_c.run(z))

而单次 run 内部、一个模型响应里的多个工具,默认是并行的(见 Part 2.2)。pydantic-ai 没有"框架原生的跨 agent fan-out 原语",并行靠 asyncio。

3.3 langgraph 怎么做:Send fan-out / subgraph / interrupt

langgraph 的设计中心(并发状态机)让它在"强多 agent 编排"上是主场:一切都是图拓扑,节点间用边连,fan-out 用 Send,子 agent 用 subgraph,人在环用 interrupt。

(1) 条件路由 + Send 做 map-reduce / 并行 fan-out

from langgraph.types import Send

def fan_out(state):
    # 返回一组 Send:每个是 (目标节点, 给它的状态切片) —— 框架并发执行它们
    return [Send("gen_joke", {"subject": s}) for s in state["subjects"]]

builder.add_conditional_edges("START", fan_out)   # 动态分派
builder.add_node("gen_joke", lambda s: {"jokes": [f'about {s["subject"]}']})
builder.add_edge("gen_joke", "END")

• Send(node, arg) 是"被物化的任务(reified task)":它把"去哪个节点 + 带什么状态"打包成一个对象。Pregel 执行器把一组 Send 当成一等控制流原语,在一个超步里并发调度,结果再通过通道的 reducer 归并(所以你需要一个 Annotated[list, operator.add] 之类的可合并通道来收集结果)。这就是 langgraph 的 fan-out / map-reduce。Send 从 langgraph.types 导入。

(2) Subgraph:每个 agent 是一张编译好的图,作为节点嵌入

• 多 agent 系统里,每个 agent 可以是一个独立编译的 StateGraph,作为父图的一个节点。subgraph 维持自己隔离的状态,通过输入/输出映射和父图交互。supervisor 模式 = 一个路由节点读共享状态、用 Send 分派给各 worker subgraph、再归并。

(3) Human-in-the-loop:interrupt() + Command(resume=...)

from langgraph.types import interrupt, Command

def ask_human(state):
    answer = interrupt("请确认这笔退款?")   # ← 图在这里暂停,状态被 checkpoint
    return {"approved": answer}

# 外部拿到中断后,带着人的输入恢复:
for chunk in graph.stream(Command(resume="yes"), config):
    ...

• interrupt() 让图暂停并持久化,等外部用 Command(resume=...) 恢复——这是"等用户确认/审批"这类异步流程的一等支持。前提:必须配 checkpointer + thread_id,否则 interrupt() 会抛 GraphInterrupt 无法恢复。这是 langgraph 相对 pydantic-ai 的一个结构性优势:长时间等待人类、跨进程恢复,它原生支持。

3.4 ReAct / Plan-Execute / Supervisor 各自怎么落地

面试点:create_react_agent 被废弃这件事本身就是个信号——langgraph 在往"用 langchain.agents.create_agent 做高层 agent、用 langgraph 做底层编排"收敛。知道这个,说明你跟着生态在走,而不是停在两年前的教程。

3.5 对照与判断:什么时候不需要多 Agent

最值钱的判断——先问"能不能用单 agent 解决":

• 真实经验里(参考本目录的 langgraph 调研),一个 Agent 该不该拆成多 agent,可以用几个触发信号自检:① 子任务是否需要不同的系统提示/人格/权限边界;② 是否需要真正的并发(而非顺序);③ 子任务是否天然跨多轮/跨会话;④ 单 agent 的工具池是否已经大到 LLM 选不对工具。四条都不满足,就别拆——单 agent + 好的工具设计(让 LLM 自己组合工具)往往更稳、更便宜、更好调试。强行多 agent 会引入状态同步、成本级联、失败传播一堆新问题。
• 一个反例校准(附录 C 的 max-ai):43 个工具的生产 agent,仍然用单 agent,因为它的子任务不满足上面四条——靠 LLM 在单循环里组合工具就够了。这恰恰是"对 demo vs 生产有清晰判断"的体现:多 agent 不是越多越牛。

3.6 上生产要补什么

• fan-out 背压:Send 可以一次炸出几百个并发任务,但没有内建并发上限/背压。要自己限并发(否则打爆下游或限流)。
• 失败传播:并发子任务里挂了几个怎么办?要明确"部分失败"语义(类似 Part 2 的 partial failure,但在 agent 粒度)。
• 成本级联:嵌套委派/fan-out 会让 token 成本指数放大。pydantic-ai 用 usage=ctx.usage 把预算串起来;langgraph 要自己在 callback 里汇总(见 Part 5)。
• 死锁/无限委派:agent A 调 B、B 调 A 的环要防;pydantic-ai 的 UsageLimits、langgraph 的 recursion_limit 是最后的刹车。

Part 4 · 系统工程:稳定性 / 失败率 / 延迟 / 日志 / 调试

JD 原文:具备系统工程意识,关注稳定性、失败率、延迟、日志与调试能力

4.1 第一性原理:可观测性 = 重建"到底发生了什么"的能力

把 Part 0 的最小内核和 Part 2 的失败分层叠加,你会得到一个冷酷的事实:Agent 是一个分布式、异步、每一步都不确定(LLM 随机)、每一层都会失败的系统。 它比传统 Web 服务难调试得多——同样的输入,两次跑出来的 tool call 序列可能不一样;一个 bug 可能要特定的 LLM 输出才触发。

所以"系统工程意识"在 Agent 语境下,核心是一件事:任何时刻,你都要能重建"这次运行到底发生了什么"——模型看到了什么 prompt、生成了什么、调了哪些工具、每步花了多少时间和 token、哪一步失败了、为什么。这就是可观测性(observability)。没有它,生产事故你只能靠猜。

具体拆成四个工程能力(对应 JD 那串词): - 稳定性 / 失败率:能分类失败(Part 2 的四层)、能统计失败率、有降级路径。 - 延迟:知道延迟来自哪(模型 API?工具?串行轮次?),能测、能砍。 - 日志:每一步都有结构化记录,能按 run/会话关联起来。 - 调试:出问题时能拿到完整的消息交换、能从中间状态复现。

两个框架给的"重建能力"形态不同:pydantic-ai 靠消息流 + OTel span,langgraph 靠checkpoint 快照 + 多模式 stream。

4.2 pydantic-ai 怎么做:OTel/Logfire + 消息捕获 + 类型化异常

(1) 可观测性:OpenTelemetry(经 Logfire),用 gen_ai.* 语义约定

import logfire
from pydantic_ai import Agent

logfire.configure()
logfire.instrument_pydantic_ai()     # 一行,自动给 agent run/model 请求/工具调用打 OTel span

agent = Agent('openai:gpt-5.2')
result = agent.run_sync('...')        # 自动产生 span:含 token 用量、延迟、tool call

• 用的是 OpenTelemetry 的 GenAI 语义约定(gen_ai.* 属性),所以能接任何 OTel 后端(Logfire、Jaeger、Grafana……),不锁定。
• 每条消息都带 run_id / conversation_id / timestamp,这是把分散的 span 关联成"一次运行 / 一个会话"的关联键。
• 关键 gotcha:可观测性是 opt-in。不调 logfire.instrument_pydantic_ai()(或 Agent.instrument_all(InstrumentationSettings(...)))就没有任何 span——零开销,但也零可见性。

(2) 调试:capture_run_messages() 抓完整消息交换

from pydantic_ai import capture_run_messages, Agent

agent = Agent('openai:gpt-5.2')
with capture_run_messages() as messages:
    try:
        result = agent.run_sync('foobar')
    except Exception:
        print(messages)   # ← 出错时把"模型到底看到/生成了什么"全打出来
        raise

• 这是 Agent 调试的核心动作:异常发生时,把完整的消息流 dump 出来,你才能看清是 prompt 的问题、模型输出的问题、还是工具的问题。

(3) 稳定性:类型化异常 + FallbackModel 降级

from pydantic_ai.exceptions import UnexpectedModelBehavior, UsageLimitExceeded, ModelHTTPError
from pydantic_ai.models.fallback import FallbackModel

# 供应商挂了自动切下一个
model = FallbackModel('anthropic:claude-opus-4-8', 'openai:gpt-5.2')
agent = Agent(model)

try:
    agent.run_sync('...')
except UsageLimitExceeded:  ...   # 跑飞/超预算
except ModelHTTPError:      ...   # 供应商 HTTP 错
except UnexpectedModelBehavior: ...  # 模型行为异常(如超过 max retries)

• 错误是类型化的异常(不是字符串错误码),能结构化处理。FallbackModel 提供供应商层降级(对应 Part 2 的第四层失败)。

(4) 长跑 / 持久执行:durable_exec 子模块原生支持 Temporal / DBOS / Prefect(pydantic_ai_slim/pydantic_ai/durable_exec/ 下有 temporal/ dbos/ prefect/)。需要"工作流崩了能从中间精确恢复"时用(见 4.4)。

4.3 langgraph 怎么做:checkpoint 即审计轨迹 + 多模式 stream

langgraph 的设计中心(超步状态机)给了它一个独特的观测形态:每个超步结束都有一个不可变的 checkpoint 快照,这些快照天然就是一条完整的审计轨迹。

(1) checkpoint = 不可变审计轨迹

• 每个超步后,框架把整个状态 + 元数据(source、step、run_id、parent ids)存成一个 Checkpoint。这串快照可以回放,可以从任意一个恢复。这是 langgraph 把"持久化"和"可观测性"统一在一个机制里的妙处。

(2) get_state / get_state_history:检查任意时刻的状态

config = {"configurable": {"thread_id": "1"}}
snap = graph.get_state(config)          # StateSnapshot: values, next, interrupts, tasks, metadata
for s in graph.get_state_history(config):   # 整条历史,可时间旅行
    print(s.values, s.next, s.metadata)

(3) stream_mode:按需要的粒度观测

for event in graph.stream(input, config, stream_mode="debug", durability="sync"):
    ...

• stream_mode 七种:values(每步完整状态)、updates(每步增量)、checkpoints、tasks(任务开始/结束含错误)、debug(checkpoint+task 都给)、messages(LLM token 级流式)、custom。调试用 debug/tasks,前端实时用 messages。
• durability 三档:"sync"(下一步前阻塞落盘)、"async"(后台落盘)、"exit"(到结束才落)。这是稳定性 vs 延迟的旋钮。

(4) 在哪暂停:compile(interrupt_before=[...], interrupt_after=[...]) 可在指定节点前后断点(配合 Part 3 的 HITL)。

4.4 对照与判断:两种"重建"心智 + durable execution 何时真需要

两种 debug 心智:pydantic-ai 是"重放消息"——出问题时看那条 list[ModelMessage],因为状态就是消息;langgraph 是"重放快照"——出问题时看某个超步的 StateSnapshot,因为状态是通道快照。前者轻、直观;后者强、能精确定位到"第几个超步、哪个通道变成了啥"。

durable execution(Temporal / DBOS)什么时候才真需要? 别一上来就上。一个务实的升级阶梯(参考本目录 langgraph 调研的结论):

1. 先用框架自带的持久化:pydantic-ai 自己存 message list(或 pydantic-graph 的 FullStatePersistence);langgraph 用 checkpointer。这能覆盖"进程内崩了重启接着跑"。
2. 需要"等几小时/几天的人工审批"再恢复 → langgraph 的 interrupt/Command 已能覆盖一部分;更硬的跨服务、超长时 → 上 DBOS(零额外基础设施,把状态存你的 Postgres)。
3. 需要跨多个服务的、强一致的长事务编排(不只是 agent 内部) → 上 Temporal(独立的 workflow 引擎,但要求 workflow 函数确定性:不能直接调时间/随机/外部 API,得通过 activity)。

背后是 5 个正交的"持久 agent"原语,任何持久化方案(LangGraph/Temporal/DBOS)都是这五个的组合:① 状态机(走到哪)② checkpointer(把进度落盘)③ interrupt(挂起+唤醒)④ reducer(并发写怎么合)⑤ 条件边(下一步去哪)。理解这五个,你就不会被"该上 LangGraph 还是 Temporal"绕晕——它们只是在不同层、用不同强度实现同样的五件事。

4.5 延迟工程

Agent 的延迟来自三处,按大小排:① 模型 API 调用(最大头,尤其大模型 + 长 context)→ ② 工具执行(网络/DB)→ ③ 串行轮次(ReAct 每多一轮就多一次模型往返)。砍延迟的手段:

• 轮次:能一轮搞定别两轮(好的工具设计、清晰的指令);并行工具(Part 2.2,pydantic-ai 默认并行)。
• 模型:简单步骤路由到小模型(Haiku 级,见 Part 5);prompt caching 降低首 token 延迟。
• 流式:run_stream(pydantic-ai)/ stream_mode="messages"(langgraph)让用户先看到 token,感知延迟大降(实际延迟没变,体验变好)。
• 测量:OTel span / checkpoint 时间戳告诉你每段花多久——先测再砍,别猜。
• 注意 recursion_limit ≠ 延迟/成本预算:它是 langgraph 防无限循环的超步上限(本文 clone 默认 10007,可经 LANGGRAPH_DEFAULT_RECURSION_LIMIT 改;历史稳定版常见默认是 25——以你的版本为准)。它防的是"跑飞",不是控成本。

Part 5 · Token 与调用成本

JD 原文(截图顶部被截断的部分):token / 调用成本等

5.1 第一性原理:成本随轮次超线性增长

回到最小内核:LLM 无状态,每一轮都要把完整历史重新喂进去。这意味着第 N 轮的输入 token ≈ 前 N-1 轮所有内容之和。所以一个 10 轮的 ReAct 对话,token 消耗不是线性的,是接近 N² 的(每轮的 input 都在变长)。再叠加:多 agent 委派/fan-out 会让调用次数乘性放大(Part 3.6)。

于是"成本"在 Agent 里是一等工程问题,不是事后报销: - token = 钱,而且供应商之间价差能到 10-100 倍。 - context 有限(Part 1),塞太多既贵又可能降质。 - 必须能:计量 → 设硬上限 → 压低 → 路由。

两个框架对成本的态度截然不同——这是设计中心的又一个直接产物:pydantic-ai 把成本当一等约束(内建 usage + 硬上限 + 算钱);langgraph 把成本当 observable(自己外挂观测)。

5.2 pydantic-ai 怎么做:RunUsage + UsageLimits 硬闸 + cost()

(1) 计量:RunUsage

result = agent.run_sync('...')
u = result.usage
print(u.input_tokens, u.output_tokens, u.cache_read_tokens, u.cache_write_tokens, u.requests)

RunUsage 字段(源码 usage.py:182):input_tokens / output_tokens / cache_read_tokens / cache_write_tokens / requests / tool_calls / details。缓存读写单独计——这对算 prompt caching 省了多少很关键。

(2) 硬上限:UsageLimits(执行前就卡)

from pydantic_ai.usage import UsageLimits

agent.run_sync('...', usage_limits=UsageLimits(
    request_limit=50,            # 默认 50(不是无限!)
    input_tokens_limit=20_000,
    output_tokens_limit=4_000,
    total_tokens_limit=30_000,
    tool_calls_limit=10,
    # count_tokens_before_request=True,  # 请求前先数 token(会多一次 API 调用)
))

• 超限抛 UsageLimitExceeded。这是把"成本预算"变成运行时硬约束——agent 永远花不超这个数(最多超一次模型调用的量)。
• gotcha:response_tokens_limit 是每次请求的上限,不是累计;tool_calls_limit 执行前检查(超了一个工具都不跑)。

(3) 算钱:result.cost()(基于 genai-prices)

price = result.cost()        # 用 genai-prices 库的价格快照,返回 PriceCalculation
print(price.total_price)     # USD

• gotcha:用的是版本化的价格快照(genai-prices 库),不是实时价。新模型/调价后可能不准,生产要自己更新价格表或对账。

(4) 跨 agent 累计:usage=ctx.usage 把子 agent 的消耗累计进同一账本(Part 3.2),这样 UsageLimits 能管住整棵委派树的总成本。

(5) prompt caching:按供应商支持(Anthropic/Google/Azure OpenAI 等),命中的 input token 走 cache_read_tokens,大幅降价。

5.3 langgraph 怎么做:原生不跟踪,靠外挂

源码核验:langgraph 全仓库 grep cost/usage/tokens 没有原生的成本/ token 跟踪 API。 这是它"成本与执行流正交"的设计选择——把计费/配额完全交给外部。所以在 langgraph 里管成本要靠:

• langchain callbacks / LangSmith:用 BaseCallbackHandler 的 on_llm_end 读 usage_metadata 自己累加;或接 LangSmith 后端聚合。
• checkpoint metadata:把累计 token 塞进 checkpoint 的 metadata,随状态持久化。
• 执行边界(不是成本预算):
• recursion_limit(本文 clone 默认 10007):超步上限,防无限循环,不是 token 预算(一个超步可能就烧几千 token)。
• RemainingSteps(managed value,stop - step):节点里读它来主动停。
• CachePolicy(key_func, ttl):节点输出的记忆化缓存——注意这是缓存"节点返回值",不是 LLM 的 prompt cache,两回事。

from langgraph.types import CachePolicy
builder.add_node("retrieve", retrieve_node,
                 cache_policy=CachePolicy(key_func=lambda s: s["query"], ttl=3600))

5.4 对照与判断

判断:对"成本必须可控、要按用户/租户记账、要设硬上限"的场景(几乎所有 B2B 生产系统),pydantic-ai 的内建 usage 体系省很多事;langgraph 你得自己搭一套观测+预算。但 langgraph 的 CachePolicy(节点级记忆化)在"同样的中间步骤反复跑"的工作流里能省真金白银,是 pydantic-ai 没有的。

5.5 成本工程的通用招(框架无关)

不管哪个框架,压成本的核心手段就这几招:

1. 模型路由:按任务复杂度分级——分类/路由/抽取这类简单活给小模型(Haiku 级),只有主推理给大模型(Sonnet/Opus 级)。这通常是最大的省钱杠杆。
2. prompt caching:把稳定不变的部分(system prompt、工具定义、长 context)缓存,命中后 input 大幅降价。
3. context 压缩:Part 1 的 ProcessHistory/摘要/召回,直接砍每轮的 input token(对抗 N² 增长)。
4. batch API:能离线批处理的(非实时)走 batch,通常半价。
5. 把 LLM 换成确定性代码:能用规则/查表/算法搞定的步骤,别用 LLM——这是成本和稳定性的双赢。

   金融线轻提(JD 加分项):在真实的量化/交易 agent 系统里,这一招被用到极致——比如组合管理用 Python 先算出"允许的动作边界",LLM 只在边界内做选择(constrained LLM);风控 agent 干脆是纯 Python 确定性逻辑,不调 LLM。一个跑 19 个并行分析 agent 的系统,成本模型大致是 单价 × 调用数 × 标的数,所以把其中能确定化的 agent 抽成非 LLM,既省钱又可回测、可确定性重放。判断"哪些 agent 必须是 LLM、哪些该是代码",本身就是成本工程的核心。 (详见附录 C)

Part 6 · 哪些适合生产、哪些只适合 demo

JD 原文(加分项):对哪些方法适合生产、哪些只适合 demo 有清晰判断

6.1 第一性原理:production 四要素 vs demo 三特征

把前五个 Part 串起来,"生产级"和"demo"的分界线其实是可以精确定义的:

一句话定义:production = observable + cost-bounded + fault-tolerant + state-persistent;demo = functional + in-memory + happy-path。 "能 demo"只证明了 functional 这一项,剩下三项才是工程量所在,也是 3-6 个月教不会、要靠真踩过坑的能力——这正是 JD 把它列为加分项、且是"高级 vs 中级"分水岭的原因。

判断一个方法/框架特性是不是生产级,就过这四条。下面把它落到两个框架的具体默认值上——因为默认值是不是被改对了,是"有没有真跑过生产"的最快指纹。

6.2 pydantic-ai 的生产 litmus test:5 个 must-know

这五个点,每一个都是"默认值是给 hello-world 的,不是给生产的"。能脱口说出这五个,基本等于证明了你真把 pydantic-ai 跑上过线(全部源码核验过):

1. retries 默认是 1(agent/__init__.py:137,tools 和 output 都是)。复杂 schema / 容易出错的工具,1 次自纠正往往不够,生产里要按工具调到 3 左右。
2. HTTP 重试是 opt-in:pydantic-ai 默认不自动重试 HTTP。传输层的 429/503/timeout 要你自己用 TenacityTransport/AsyncTenacityTransport 包 httpx(docs/retries.md)。不加 = 一遇限流就崩。
3. 图持久化默认只存最新快照:pydantic-graph 默认用 SimpleStatePersistence(源码 docstring:"just hold the latest snapshot... used by default")。要"从任意中间步恢复 / 看完整历史",必须显式换成 FullStatePersistence(存快照列表)或 FileStatePersistence(落盘)。
4. 工具的业务异常框架不会自动转 retry:只有 raise ModelRetry(...) 才触发自纠正回路;其他普通异常默认走 on_tool_execute_error 钩子并propagate(run 失败)(tool_manager.py:330)。所以每个工具要么对可恢复错误 raise ModelRetry,要么自己 try/except 翻译业务错误。
5. 长对话要主动压缩:不挂 ProcessHistory(或老版 history_processors),历史会一直涨到撑爆窗口 + 成本 N² 爆炸。生产里必须有 token 阈值触发的窗口/摘要策略。

6.3 langgraph 的生产 litmus test

1. InMemorySaver 只配 dev:源码 docstring 明说生产用 PostgresSaver/SqliteSaver。用了 checkpointer ≠ 持久化了,得看是哪个 saver(进程重启 InMemory 全丢)。
2. 错误处理要显式设计:节点要配 RetryPolicy(注意 add_node(retry=) 已废弃,用 retry_policy=);工具节点配 ToolNode(handle_tool_errors=...);关键路径用 error_handler 或条件边路由到降级/人工节点。默认啥都没有 = 一个节点抛错整个图崩。
3. recursion_limit 要按场景调:本文 clone 默认 10007(很高),但它只防无限循环、不控成本/token;多 agent 深委派时反而可能需要调低做保护。别把它当预算。
4. HITL 的 interrupt/resume 契约要清晰:interrupt() 必须配 checkpointer + thread_id,且要想清楚"恢复时状态从哪来、谁来 resume、超时怎么办"。
5. 成本要自己接:langgraph 不原生跟踪 token,生产必须接 callback/LangSmith 做用量+预算(Part 5)。

6.4 选型决策树 + 真实生产现状

怎么选(基于本目录已有的跨框架调研 + 源码事实):

你的核心痛点是什么?
├─ 工具入参/结构化输出的"类型可靠性"是刚需,且想轻量嵌进现有 API/服务
│     → pydantic-ai(设计中心:类型安全的函数;pydantic 用全部;ModelRetry 自纠正)
│
├─ 复杂多 agent 编排 + 长跑工作流 + 人在环审批 + 跨进程/时间旅行恢复
│     → langgraph(设计中心:并发 reducer 状态机;checkpointer/interrupt/Send 原生)
│
└─ 两者都要:用 langgraph 做编排骨架,把"需要强类型的那个 agent 节点"用 pydantic-ai 实现
      → 混用(langgraph 编排 + pydantic-ai 做 type-safe node)

真实生产现状(时间点快照,会变):据本目录调研,LangGraph 的生产用户更多、更偏企业级复杂工作流(被引用的有 Klarna、Replit、Uber、JPMorgan 等量级);pydantic-ai 更年轻、更偏"嵌进 API 的轻量 agent"(MindsDB、Vercel 等)。关键判断不是"pydantic-ai 不成熟",而是两者处在不同的成熟度曲线、面向不同的客户重心。pydantic-ai 背靠 Pydantic/Logfire 团队,类型安全和可观测是它的基因;langgraph 背靠 LangChain 生态,编排和长跑是它的基因。

6.5 框架无关的 demo 陷阱

最后,有些东西无论用什么框架都偏 demo,识别它们也是"清晰判断"的一部分(参考本目录 handbook 的 Tier 3 分析,附带时效 caveat:框架生态变化快,下面是调研时点的判断,自己复核):

• CrewAI / AutoGen 这类"高层多 agent 编排":做 demo 惊艳(几行就能跑一个"多 agent 协作"),但工业级落地案例相对少,生产里对失败/成本/可观测的控制力往往不够。
• Streamlit 当生产前端:做原型极快,但不是为生产并发/状态/鉴权设计的。
• 直接手撸 LLM API 拼循环:学习理解很好(知道框架在帮你做什么),但生产里你会把上面五个 Part 的轮子全重新造一遍——除非你有非常特殊的理由。
• "能 demo"的多 agent:见 Part 3.1 的弱/中/强——很多"我们做了 multi-agent"其实是弱语义(一个 agent 多次调用),别被唬住,也别这么唬别人。

附录 A · 一页速查表(面试/复习前快速过)

附录 B · 源码索引(本文每个论断的出处)

路径相对 pydantic-ai/(包在 pydantic_ai_slim/pydantic_ai/、图库在 pydantic_graph/)和 langgraph/(主包在 libs/langgraph/langgraph/)。

pydantic-ai - agent run = graph 实例 / 节点:_agent_graph.py:2066(build_agent_graph)、:234-1381(UserPromptNode/ModelRequestNode/CallToolsNode/SetFinalResult) - 状态容器:_agent_graph.py:118(GraphAgentState,dataclass,含 message_history: list[ModelMessage]) - 依赖注入:_run_context.py:35(RunContext[T].deps) - 历史处理:capabilities/process_history.py:28(ProcessHistory,before_model_request 钩子) - 序列化:messages.py(ModelMessagesTypeAdapter) - 重试:exceptions.py:40(ModelRetry)、agent/__init__.py:137(默认 retries=1)、tool_manager.py:330(普通异常→on_tool_execute_error,默认 propagate)、capabilities/abstract.py:686(钩子默认 raise) - 并行工具:_agent_graph.py:1858(get_parallel_execution_mode 默认 parallel)、:1884-1897(asyncio.wait ALL/FIRST_COMPLETED) - 用量/成本:usage.py:182(RunUsage)、:262(UsageLimits,默认 request_limit=50)、:412(check_before_tool_call)、messages.py(ModelResponse.cost()) - HTTP 重试:retries.py:117/215(TenacityTransport/AsyncTenacityTransport) - 可观测:docs/logfire.md(instrument_pydantic_ai/Agent.instrument_all)、capture_run_messages、models/fallback.py(FallbackModel) - 持久执行:durable_exec/{temporal,dbos,prefect}/ - 图持久化:pydantic_graph/persistence/in_mem.py:32(SimpleStatePersistence,默认,仅最新)/:86(FullStatePersistence,全历史)、persistence/file.py:30(FileStatePersistence) - 多 agent:docs/multi-agent-applications.md(委派 usage=ctx.usage、程序化 handoff)、pydantic_graph/basenode.py(BaseNode/GraphRunContext/End)

langgraph - 超步执行:libs/langgraph/langgraph/pregel/_loop.py(PregelLoop,内部 superstep) - 状态/通道:graph/state.py:130(StateGraph)、graph/message.py:60(add_messages,按 ID 归并)、channels/{last_value,binop,topic}.py - 校验:runtime 中 model_validate 出现 0 次(校验交 langchain-core) - 检查点:libs/checkpoint/.../base/__init__.py:176(BaseCheckpointSaver)、memory/__init__.py:33(InMemorySaver,仅 dev)、checkpoint-sqlite/checkpoint-postgres - 工具/重试:libs/prebuilt/.../tool_node.py:749(ToolNode/handle_tool_errors,并行用 asyncio.gather)、types.py:406(RetryPolicy)、graph/state.py:753(retry= 已废弃) - 错误处理:graph/state.py:276(StateGraph(..., error_handler=))、:672(add_node(..., error_handler=));无 add_error_handler() 方法 - 多 agent / HITL:types.py:654(Send)、:748(Command)、:801(interrupt);prebuilt/chat_agent_executor.py:274(create_react_agent 已 @deprecated) - 观测/持久:types.py:87(Durability=Literal['sync','async','exit'])、:120(StreamMode 7 种)、get_state/get_state_history→StateSnapshot、compile(interrupt_before/after) - 成本相关:_internal/_config.py:31(DEFAULT_RECURSION_LIMIT=10007)、managed/is_last_step.py:18(RemainingSteps)、types.py(CachePolicy,节点输出缓存)

附录 C · 真实案例速览(把抽象落到工程)

C.1 max-ai(pydantic-ai 跑生产的样子) —— 本目录 max-ai-design-v1.md - 选 pydantic-ai 而非 fork 别的框架,核心理由是可逆性(pip 装可控、fork 锁定);路径 A(fork)≈22 天 vs 路径 B(pydantic-ai)≈15 天。 - 43 个原子工具,仍用单 agent(呼应 Part 3.5:不满足多 agent 触发条件就别拆)。 - 风险分层 L0-L4 当 UX 一等原语:只读自动执行 / 草稿改动 30 秒撤销 / 订单改动需确认 / 不可逆动作多重确认 / 纠纷强制转人工。 - Action Journal + 补偿(呼应 Part 2.5):每个有副作用的动作存 pre_state + 补偿函数 + 过期时间,30 秒内可撤销——因为框架 checkpoint 回不了外部副作用。 - 工具调度中间件从 JWT 注入 merchant_id,覆盖 LLM 给的参数——防 prompt 注入(呼应 Part 1.2 deps 注入)。

C.2 ai-hedge-fund(多 agent + 确定性 agent,JD 金融线轻提) —— 本目录 ai-hedge-fund-learnings.md - constrained LLM:Python 先算出"允许的动作边界",LLM 只在边界内选——把安全前置,而非事后补救。 - 确定性 agent:风控等 agent 是纯 Python,不调 LLM——省钱 + 可回测 + 可确定性重放(呼应 Part 5.5)。 - 确定性重放测试:MockConfigurableAgent + decision_sequence 把"非确定的 LLM"变成"可重放的 mock",从而能对 agent 组合逻辑做回归测试(解决 Part 4 "LLM 随机导致难测"的痛点)。 - 19 个并行分析 agent 的成本 ≈ 单价 × 调用数 × 标的数 —— 所以"哪些 agent 该确定化"是核心成本/稳定性决策。

附录 D · 自测题(Feynman 式:能从第一性原理推出来,才算真懂)

不要背 API。每题先合上文档,试着从"Agent = 状态机 / LLM 无状态 / 工具碰真实世界"这些第一性原理推导,再对照正文。

1. (Part 0) 为什么说"Agent 不是 while 循环,是状态机"?这个区别在工程上具体让你能做哪三件 while 循环做不到的事?
2. (Part 0) 用一句话说出 pydantic-ai 和 langgraph 的"设计中心"分别是什么,并由此推出它们在"状态持久化"上的差异。
3. (Part 1) LLM 是无状态的,那"多轮对话"是怎么实现的?为什么长对话成本是 N² 而不是 N?
4. (Part 1) pydantic-ai 把业务数据放 deps 而不放进会话历史,这个选择带来什么好处和什么负担?
5. (Part 2) "协议层/业务层失败"为什么不该当异常崩掉,而该喂回 LLM?两个框架对这件事的哲学差异是什么?
6. (Part 2) pydantic-ai 并行执行工具时为什么刻意用 asyncio.wait 而不是 asyncio.gather/TaskGroup?这保护了什么语义?
7. (Part 2) 为什么"框架的 checkpoint 回滚不了工具的外部副作用"?那"撤销"该在哪一层、怎么做?
8. (Part 3) 给出四个"应该用单 agent 而不是多 agent"的信号。43 个工具的 max-ai 为什么还是单 agent?
9. (Part 3) langgraph 的 Send 为什么被称为"被物化的任务"?它和"直接函数调用"的本质区别是什么?
10. (Part 4) "可观测性 = 重建发生了什么的能力"——pydantic-ai 和 langgraph 各自靠什么机制提供这个能力?
11. (Part 4) 什么时候才真的需要 Temporal/DBOS 这种 durable execution?5 个"持久 agent 原语"是哪五个?
12. (Part 5) langgraph 原生不跟踪成本,这是疏忽还是设计选择?对应它的什么设计中心?
13. (Part 6) 给你一份别人写的 Agent 代码,你怎么用"默认值改对没有"在 5 分钟内判断作者有没有真跑过生产?(分别说 pydantic-ai 和 langgraph 各看哪几个默认值)
14. (综合) 面试官说"我们要做一个能自动处理客户退款、要能等人工审批、要严格控成本、退款动作要能撤销"的 agent,你会怎么选型、每个 JD 维度怎么落地?(把 6 个 Part 串起来答)

版本与时效再次提醒:本文 API 基于 2026-05 的 pydantic-ai / langgraph 开发版 clone 核验。标了"版本提示"的地方(capabilities/ProcessHistory、error_handler、recursion_limit=10007、create_react_agent 废弃)在你装的版本里可能不同;市场采用数据是时间点快照。落地前以你实际版本的源码/文档为准——附录 B 给了所有源码路径,自己 grep 最稳。