Agent/ˈeɪdʒənt/🔊 框架系统教程 + 全景手册

单文件教程 · 含 Part A 详细教程 + Part B 生态全景手册 · 共 §0–§19 + 附录 A–K

版本 2026.04 · 适配 LangChain 0.3.x / LangGraph 0.2.x · 适配 Claude Sonnet 4.6 / Opus 4.7

本书贯穿"智能体执行闭环 8 节点"作为主线 —— 所有原理、框架、特性、生态分类都挂到这条主线上。读者读完后形成"一条流程，多种实现"的统一心智模型。

Part A — 详细教程

本部分系统讲透三件套：推理范式 ReAct/riːˈækt/🔊（§1）+ 第一代框架 LangChain（§2-§3）+ 第二代框架 LangGraph（§4-§6），并以 §7 横向对比、§8 落地用户工程收尾。

§0 智能体的原理架构

本章是全书地基 —— 从最底层的 LLM/ɛl ɛl ɛm/🔊 工作原理，一路讲到生产级 Agent/ˈeɪdʒənt/🔊 架构的完整 6 层蓝图，重点覆盖记忆与存储体系。

§0.0 本章定位

§0.1 模型层：LLM 怎么工作

基础流程对应：节点 ② 推理的内部机制。

LLM 推理 4 步流水线

原文本 ──▶ Tokenizer ──▶ Token IDs ──▶ ① Embedding ──▶ ② Self-Attention ──▶ ③ 输出层 ──▶ 概率分布 ──▶ ④ Sample ──▶ 下一 token
"你好"    （切分+查词表） [123,456]    （查表得向量）    （N 层 Transformer）  （映射回词表）         （temp/top_p）  → "世"
            前置工序        ↑ 模型真正读的输入是这个       O(n²) 复杂度

每生成一个 token 重复一次这 4 步，直到达到 stop 条件或 max_tokens。

Token/ˈtoʊkən/🔊 IDs 到底是什么？

关键澄清：Token/ˈtoʊkən/🔊 IDs 既不是原文本字符，也不是向量 —— 它是一串整数。

具体来说，模型看不到字符（如 "你"），也看不到向量 —— 它只能读一串整数 ID（如 [15339, 1879, 99, ...]）。

完整数据流：

"你好世界"               ← 用户给的原文本字符串
    ↓ tokenizer.encode()
[123, 456, 789, 1011]   ← Token IDs（整数序列，一般每个 token 1 个整数）
    ↓ ① Embedding 查表
[[0.1, -0.3, 0.7, ...],  ← 每个整数 ID 查 embedding 表得到一个 d 维向量
 [0.5, 0.2, -0.1, ...],     （d = 4096-12288，取决于模型规模）
 [...], [...]]
    ↓ ② Self-Attention 加工
[[..., ..., ...], ...]   ← 经过 N 层 Transformer 处理后的"语境化向量"
    ↓ ③ 输出层映射回词表
[一组词表概率，如 0.3 是 "！"，0.2 是 "了"，0.1 是 "啊"...]
    ↓ ④ Sample
2048                     ← 选中的整数（"！"对应的 token ID）
    ↓ tokenizer.decode()
"！"                     ← 拼回字符串

为什么必须先变成整数？

Token ID 与 Embedding/ɪmˈbɛdɪŋ/🔊 向量的关系

模型权重里有一张大表（embedding matrix，形状 [词表大小, d]），输入 token ID 后从这张表"查行"得到对应向量。

Tokenizer/ˈtoʊkənaɪzər/🔊 如何切分：BPE 算法

模型不直接看字符。BPE（Byte Pair Encoding）算法把文字切成 token：

中文比英文贵 1.5-3 倍 —— 因为词表训练以英文优先，中文字符往往独占 1-2 个 token。

from anthropic import Anthropic
client = Anthropic()
resp = client.messages.count_tokens(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "你好世界"}]
)
print(resp.input_tokens)

Embedding：把 token 变成"模型可计算的向量"

模型看到 token ID（一个整数）没法直接思考，要先把它变成向量——一串浮点数（如 4096 个数字）。这串数字就是这个 token 的"含义指纹"。

Self-Attention（自注意力）：每个词"问问其他词"

直觉理解

人读句子时，理解某个词会回头看上下文里的相关词。Self-Attention（自注意力）就是模型版的"回头看"机制。

"小明 把 那本 红色 的 书 给了 小红"
                         ↑
当模型读到"书"时，它会：
  - 重点关注"红色"（描述书）
  - 重点关注"把"（涉及的动作）
  - 重点关注"小红"（接收者）
  - 弱关注其他词

每个 token "回头看"的过程是用 Q / K / V 三个向量做的：

工作流程（一次 attention）

1. 当前 token 用自己的 Q 跟所有其他 token 的 K 比对，得到一组"相关性分数"
2. 把这组分数过 softmax 归一化成概率分布（所有数加起来 = 1）
3. 用这个概率分布去加权融合所有 token 的 V，得到"在这个语境里，我（当前 token）的新含义"

结果：每个 token 不再是孤立的"词典义"，而是带上了上下文信息的"语境化含义"。

"Self" 的意思

"Self" 指 Q、K、V 都来自同一个序列（自己看自己）—— 区别于"cross attention"（Q 来自一个序列、K/V 来自另一个）。LLM 用的是 self-attention。

O(n²) 复杂度怎么来的

序列里 n 个 token，每个 token 都要看其他 n 个 token 一遍 —— 共 n × n = n² 次"看"。

这是上下文窗口有上限的根本原因 —— 不是技术做不到，是 O(n²) 太贵。

公式骨架（懂线性代数可看）

attention(Q, K, V) = softmax(Q · K^T / √d) · V

不懂线性代数没关系，记住三件事：

为什么 KV Cache/kæʃ/🔊 能优化

关键洞察：K 和 V 是 token 自己的属性，只取决于这个 token 自己，不取决于后面有谁。

所以一旦算过某个 token 的 K 和 V，永远不变 —— 后续新 token 来时，只需查询历史 token 的 K/V（不必重算），只算新 token 自己的 K/V。

详细机制见 §0.3.2。

多头注意力（Multi-Head Attention）：一个 token 多个角度同时看

单头不够

只用一组 Q/K/V 算 attention 太单调 —— 一个 token 只能从"一个角度"理解上下文。比如读"苹果"时，可能既需要关注它是水果（语义关系），也需要关注它在句子里是宾语（语法关系）。

多头机制

把 Q/K/V 切成 N 个"头"（典型 N = 16-128），每组独立做一次 attention，然后把 N 个结果拼起来：

单头：    Q × K^T  → 一个 attention 矩阵 → 一个加权融合结果
多头(8): Q1×K1^T  → attention_1 → 结果_1
         Q2×K2^T  → attention_2 → 结果_2
         ...
         Q8×K8^T  → attention_8 → 结果_8
         拼接 [结果_1, ..., 结果_8] → 投影回原维度

每个头的"分工"是训练时自动涌现的，不是手工设计。

工程参数

KV Cache 的现代优化（MQA / GQA）

每头都有独立 K/V → KV Cache 也按头数倍增。多头注意力让 KV Cache 占显存暴增。

GQA 是 2024 后几乎所有大模型的"省显存"标配。

位置编码（Positional Encoding）：让模型知道 token 顺序

不加位置编码模型分不清顺序

Self-Attention 本身对顺序不敏感 —— 不管你把 "我打他" 还是 "他打我"，每个 token 看其他 token 的方式一样。这显然不行。

解决：把"位置信息"也注入进来

每个 token 的 embedding 不只是"含义指纹"，还混入"位置指纹"：

主流 3 种位置编码

RoPE 为什么主流

长上下文（1M token Opus 4.7）的实现关键之一就是 RoPE 的位置外推优化（如 PI / NTK-aware / YaRN 等技巧）。

Transformer 一层结构：N 层堆叠

一层 = "Multi-Head Attention + FFN" + 残差 + 归一化

每一层 Transformer 的计算流：

输入向量 x
   ↓
   ┌──────────────────────┐
   │ Layer Norm           │  ← 归一化（防止数值爆炸）
   └────────┬─────────────┘
            ↓
   ┌──────────────────────┐
   │ Multi-Head Attention │  ← 让 token 互相"看"
   └────────┬─────────────┘
            ↓
   x + (上一步结果)            ← 残差连接（避免信息丢失）
            ↓
   ┌──────────────────────┐
   │ Layer Norm           │
   └────────┬─────────────┘
            ↓
   ┌──────────────────────┐
   │ FFN（前馈网络）       │  ← 两个全连接层 + 激活函数（让模型"思考"）
   │  W1 → GeLU → W2     │
   └────────┬─────────────┘
            ↓
   x' + (上一步结果)            ← 又一个残差连接
            ↓
   输出向量（给下一层）

总参数量怎么来的

层数 × 每层参数（attention + FFN）≈ 总参数量。

推理时各层之间是流水线

每个 token 走完 N 层后才输出概率分布。长 prompt + 多层 + 大维度 = 推理慢。

输出层 + 采样：概率分布展开

输出层（LM Head）

最后一层把每个 token 的隐藏向量乘以词表矩阵，得到词表大小的"分数向量"（logits）：

hidden_vector (d 维) × W_lm_head (d × vocab_size) → logits (vocab_size 维)

softmax：分数 → 概率分布

logits = [3.2, 1.1, -0.5, 4.7, ...]  ← 词表里每个 token 的分数
softmax(logits) = [0.18, 0.02, 0.005, 0.79, ...]  ← 归一化成概率（加起来 = 1）

采样策略详解

控制下一个 token 怎么从概率分布里挑出来：

temperature 的真实数学含义

原 logits: [3.2, 1.1, -0.5, 4.7]

T = 0.5（低温）：
  logits / 0.5 = [6.4, 2.2, -1.0, 9.4]  ← 拉大差距
  softmax = [0.05, 0.001, 0.000, 0.949]  ← 几乎只选 token 4 (确定性高)

T = 1.0（默认）：
  softmax([3.2, 1.1, -0.5, 4.7]) = [0.18, 0.02, 0.005, 0.79]

T = 2.0（高温）：
  logits / 2.0 = [1.6, 0.55, -0.25, 2.35]  ← 缩小差距
  softmax = [0.27, 0.10, 0.045, 0.59]  ← 概率更平均（创造性高）

T → 0：等价于 argmax（贪心）
T → ∞：等价于均匀分布（完全随机）

top_p（核采样）的算法

def top_p_sample(logits, p=0.9):
    probs = softmax(logits)
    sorted_probs, sorted_idx = sort_descending(probs)

    # 累积概率
    cum_probs = cumsum(sorted_probs)
    # 找到累积刚超 p 的位置
    cutoff = first_idx_where(cum_probs >= p)
    # 只在前 cutoff 个 token 里采样
    return sample_from(sorted_idx[:cutoff], sorted_probs[:cutoff])

实战推荐配置

关键洞察：模型本质是"统计下一个 token 的概率分布展开" —— 不是"逻辑推理"。

训练 vs 推理：两个完全不同的过程

训练（Training，发生在厂商内部）

结果：得到"模型权重文件"（Llama 3 70B 约 140 GB）。

推理（Inference/ˈɪnfərəns/🔊，发生在你调 API 时）

关键差异

为什么这区分重要

调 API 时你只用推理。所有"知识"都在权重里，权重在训练阶段就冻结了。

LLM 的能力与边界

记住这张表 —— 所有智能体框架做的事，都是在补 LLM 这些洞。

小结一行：LLM 是 4 步流水线（Token / Embedding / Attention / Sample），本质是概率分布展开 —— 它的 6 大局限定义了智能体框架的所有职责。

§0.2 接口层：API 协议与字节流

基础流程对应：节点 ① 上下文 → 节点 ② → 节点 ⑧ 输出的传输层。

一次 API 调用的 7 步流水线

客户端 ──HTTPS POST──▶ 厂商 API 网关 ──▶ 模型集群（GPU）──SSE 流──▶ 客户端
       1. 鉴权          2. 限流          3. Tokenize       4. 流式输出
                                          + Forward
                                          + Sample
                                          + Decode

关键洞察：首 token 慢（100-2000ms 走完整 prompt），后续 token 快（5-50ms 命中 KV Cache）。TTFT（Time To First Token）才是用户体感，不是 total latency。

实测吞吐量（2026.04）

工具调用协议：结构化解码

模型能输出结构化 JSON 调工具，不是魔法，是 grammar-constrained decoding：

三家协议字节级对比

Anthropic 字节级响应：

{
  "stop_reason": "tool_use",
  "content": [
    {"type": "text", "text": "我来查一下"},
    {"type": "tool_use", "id": "toolu_01abc",
     "name": "get_weather", "input": {"city": "北京"}}
  ]
}

回填给模型：

{"role": "user", "content": [
  {"type": "tool_result",
   "tool_use_id": "toolu_01abc",
   "content": "晴 25 度"}
]}

详细字节对照见 附录 K。

结构化解码内部机制（grammar-constrained decoding 怎么工作）

朴素做法 vs 约束解码

朴素做法（GPT-3 时代）：在 prompt 里写"请输出 JSON 格式"，靠模型听话。失败率 5-30%（模型经常输出错的 JSON、加多余文字、漏字段）。

结构化解码（GPT-4 / Claude 现代做法）：在采样阶段强制约束模型只能输出符合 schema 的 token。失败率 < 0.1%。

约束解码的字节级工作原理

把 schema 编译成"状态机 + 允许 token 集"，每生成一个 token 时：

当前状态：刚生成完 {"city":
   ↓
查 schema 状态机：下一位置必须是字符串值
   ↓
词表 50,000 个 token，但当前合法的只有约 2000 个（带引号开头的字符串 token）
   ↓
模型输出 logits（50,000 维）
   ↓
把不合法 token 的 logits 设为 -∞
   ↓
softmax → 只在合法 token 里采样
   ↓
生成下一 token （比如 `"`、`"北`）

实现思路

主流约束解码引擎

性能开销

工具调用 = 特殊 schema 的约束解码

# 厂商内部把工具描述转成 schema
schema = {
  "type": "object",
  "properties": {
    "name": {"type": "string", "enum": ["get_weather", "search"]},
    "arguments": {"type": "object", "properties": {...}}
  },
  "required": ["name", "arguments"]
}

# 在解码时强制约束
constrained_decode(model, prompt, schema)

这就是 OpenAI Function Calling / Anthropic Tool Use 不会输出格式错乱的根本原因 —— 不是"prompt 哄好了模型"，是采样层强制约束。

Anthropic Tool Use 的额外妙处

Claude 的 Tool Use 在训练阶段还把"工具调用 vs 文本回复"作为训练目标，让模型主动学会何时调用工具，约束解码只是兜底。这就是为什么 Claude 的工具调用准确率（选对工具 + 填对参数）业界最高。

流式输出：SSE 字节流

Server-Sent Events 是 HTTP 长连接 + 行级事件：

HTTP/1.1 200 OK
Content-Type: text/event-stream

event: content_block_delta
data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"北"}}

event: content_block_delta
data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"京"}}

event: message_stop
data: {"type":"message_stop"}

工具调用 + 流式的边界冲突：文本可 token 级吐字符，但工具调用块必须收齐整个 JSON 才能用：

event: content_block_delta
data: {"delta":{"type":"input_json_delta","partial_json":"{\"city\":"}}

event: content_block_delta
data: {"delta":{"type":"input_json_delta","partial_json":"\"北京\"}"}}

event: content_block_stop  ← 客户端在此才能解析 JSON 调用工具

小结一行：API 调用是 7 步流水线，工具调用靠结构化解码，三家协议字节级有差异 —— 跨厂商代码必须用 LangChain bind_tools 这类适配层。

§0.3 记忆层：完整存储架构

基础流程对应：节点 ⑥ 状态管理的全套实现。本节是全章重点。

§0.3.1 三层记忆模型总览

智能体的记忆按"时间尺度 + 持久化层级"分三层：

┌──────────────────────────────────────────────────────────────────────┐
│                        三层记忆模型（Memory Hierarchy）                │
├──────────────────────────────────────────────────────────────────────┤
│  ★ 短期（Short-term）★                                                │
│    范围：单次 API 调用内                                                │
│    实现：Context Window + KV Cache + Prompt Caching                  │
│    持久化：无（请求结束即销毁）                                         │
│    代表：模型内部 attention 缓存                                        │
├──────────────────────────────────────────────────────────────────────┤
│  ★ 中期（Mid-term）★                                                  │
│    范围：单次会话（同 thread_id）内                                    │
│    实现：消息历史 + LangChain Memory / LangGraph Checkpoint           │
│    持久化：进程内 / SQLite / Postgres                                  │
│    代表：MemorySaver / SqliteSaver / PostgresSaver                    │
├──────────────────────────────────────────────────────────────────────┤
│  ★ 长期（Long-term）★                                                 │
│    范围：跨会话、跨用户                                                │
│    实现：Vector DB + LangGraph Store + Anthropic Skills              │
│    持久化：永久数据库（Postgres / Pinecone / Qdrant）                 │
│    代表：BaseStore + embedding 索引 / Skills                          │
└──────────────────────────────────────────────────────────────────────┘

三层之间的协作关系

§0.3.2 短期记忆：Context Window + KV Cache + Prompt/prɑːmpt/🔊 Caching

Context Window 的工程现实

每次 API 调用塞进去的所有内容总和。容量看起来大但跑几轮就满：

KV Cache：续写为什么便宜

核心洞察

回到 §0.1：每个 token 都有 K（"我能提供什么"）和 V（"实际信息"）—— 这两者只取决于 token 自己，不取决于后面有谁。一旦算过，永远不变。

所以可以缓存：把每个 token 的 K、V 算出来后存到 GPU 显存里。下次生成下一个 token 时，只算新 token 的 K/V + 让它的 Q 去查询所有历史 K/V，不必重算历史。

不缓存 vs 缓存的对比

时序示意

首次调用：prompt = "今天天气真"
  Token 0 "今"  → 算 K0, V0  → 缓存
  Token 1 "天"  → 算 K1, V1  → 缓存（attend K0, V0）
  Token 2 "天"  → 算 K2, V2  → 缓存（attend K0..K1）
  Token 3 "气"  → 算 K3, V3  → 缓存（attend K0..K2）
  Token 4 "真"  → 算 K4, V4  → 缓存（attend K0..K3）

模型生成"好"：
  Token 5 "好"  → 算 K5, V5  → 缓存
                  attend K0..K4 时直接读 cache（不重算）

模型生成"！"：
  Token 6 "！"  → 算 K6, V6  → 缓存
                  attend K0..K5 时直接读 cache

每个 token 只算一次 K/V，永远缓存。这是为什么 LLM 能"流式打字"般快速生成 —— 后续 token 几乎是免费的。

工程后果

直觉总结：KV Cache = "每个 token 的属性只算一次、永远缓存"。这就是后续 token 能秒出的根本原因。

Prompt/prɑːmpt/🔊 Caching：把 KV Cache 持久化

Anthropic 2024.08 推出 —— 把"同 prefix 的 KV Cache"跨调用持久化：

client.messages.create(
    model="claude-sonnet-4-6",
    system=[{
        "type": "text",
        "text": "<10k token 的系统提示>",
        "cache_control": {"type": "ephemeral"}    # ★ 标记缓存
    }],
    messages=[{"role": "user", "content": "问题"}]
)

字节级证据（usage 字段）：

{
  "input_tokens": 50,
  "cache_creation_input_tokens": 0,
  "cache_read_input_tokens": 10000,
  "output_tokens": 200
}

价格折扣对比

工程后果

小结一行：短期记忆 = Context（数据）+ KV Cache（计算优化）+ Prompt Caching（持久化优化），稳定 prefix 前置 + cache_control 是降本的关键。

§0.3.3 中期记忆：会话历史的工程实现

LangChain Memory/ˈmɛməri/🔊 4 类（会话级，详见 §2.7）

四类的共同缺陷：会话级 + 进程内 —— 跨进程或跨会话失忆。

LangGraph Checkpoint：跨进程持久化

LangGraph 把 State 写到外部存储，让智能体跨进程、跨会话续跑。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver

# 三种 Saver 按规模选
checkpointer = MemorySaver()                    # 开发
checkpointer = SqliteSaver.from_conn_string("...")  # 单机
checkpointer = PostgresSaver.from_conn_string("...")  # 生产

agent = create_react_agent(llm, tools, checkpointer=checkpointer)

# thread_id 是会话标识
config = {"configurable": {"thread_id": "user-123"}}
agent.invoke({"messages": [...]}, config)   # 自动续上历史

Postgres 表结构

PostgresSaver.setup() 创建 4 张表：

写入时序

T+0    用户调 invoke(input, thread_id=X)
T+1ms  Checkpointer.load(thread_id=X) → 历史 State
       超步 1：Node A 跑
T+50ms Node A 返回更新
T+51ms Checkpointer.save(checkpoint_1) ← 屏障写入
       超步 2：Node B 跑
...

每个超步屏障写一次 checkpoint。

中期记忆的范围

小结一行：中期记忆的演化路径是 LangChain Memory（会话级）→ LangGraph Checkpoint（跨进程持久化）—— 后者是生产级标配。

§0.3.4 长期记忆：Vector DB + Store API + Skills

长期记忆的三种工程实现

Vector DB 工作原理

1. 写入：文本 → embedding 模型 → d 维向量 → 存入向量库
2. 检索：query → embedding → 与库中向量做 cosine 相似度 → 取 top-k
3. 注入：top-k 文档拼回 prompt（节点 ① 上下文增强）

Embedding 是什么（回顾 §0.1）

把一段文本（一句话 / 一段话 / 一篇文档）通过专门的"embedding 模型"（如 OpenAI text-embedding-3-small / Anthropic voyage-3 / 开源 bge-large）变成一串浮点数向量（如 1536 个数字）。

余弦相似度（Cosine/ˈkoʊsaɪn/🔊 Similarity）：怎么衡量"两个向量相不相关"

直觉：把每个向量看成 d 维空间里的一支箭头。两支箭头方向越接近、相关性越高。余弦相似度算的就是这两支箭头之间的"夹角余弦值"：

公式（看不懂没关系）：

cosine(A, B) = (A · B) / (|A| × |B|)

其中：
  A · B  = a1×b1 + a2×b2 + ... + ad×bd  （点积，每对分量相乘后加和）
  |A|    = sqrt(a1² + a2² + ... + ad²)   （A 的长度）
  |B|    = sqrt(b1² + b2² + ... + bd²)   （B 的长度）

只看方向不看长度 —— 这就是为什么用余弦而非欧氏距离：embedding 的"长度"可能反映文档篇幅而非语义。

为什么不用线性扫描

朴素做法：query 向量跟库里每一个向量都算 cosine。10 万条向量就要 10 万次乘加。100 万条就要 100 万次。线性 O(n) 太慢。

索引算法把 O(n) 降到 O(log n)

实战代码

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

# 1. 准备 embedding 模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")  # 1536 维

# 2. 写入：文本切片 + 转向量 + 入库
docs = ["公司退款政策...", "下单流程...", "物流时效..."]
db = FAISS.from_texts(docs, embeddings)

# 3. 检索：query → 向量 → 余弦相似度 → top-k
results = db.similarity_search("怎么退款？", k=3)
# 返回相关性最高的 3 条文档（按 cosine 降序）

主流向量库

Embedding 模型选型（决定召回质量的根因）

向量检索的召回率上限完全由 embedding 模型决定。模型越强、向量越能"理解"语义。

主流模型对比（2026.04）

MTEB（Massive Text Embedding Benchmark）= 业界主流评估，56 个任务平均分数。

选型决策树

Matryoshka/ˌmætrɪˈɒʃkə/🔊 Embeddings（俄罗斯套娃嵌入）

OpenAI text-embedding-3 系列支持降维而几乎不损失精度 —— 训练时把不同维度的子向量都优化得有意义：

# 全维度（3072）：最高精度
emb_full = openai.embed("文本", model="text-embedding-3-large")

# 截断到 1024 维：精度仅降 ~2%，存储省 3 倍
emb_compact = emb_full[:1024]

工程价值：高精度场景用 3072 维存底库，低延迟场景用 256 维快查 —— 一份模型支持多档需求。

混合检索（Hybrid/ˈhaɪbrɪd/🔊 Retrieval/rɪˈtriːvəl/🔊）：dense + sparse 结合

单纯向量检索的局限

关键字检索（Sparse/spɑːrs/🔊 / BM25）的强项

BM25（Best Match 25）是经典的关键字检索算法，强在精确匹配（如倒排索引）。

混合检索的标准做法

query = "OAuth 2.0 PKCE 安全流程"

并行：
  ├── Dense 检索（向量库）  → top 30 候选
  └── Sparse 检索（BM25）  → top 30 候选
       ↓
  合并 + 重排（Reciprocal Rank Fusion / RRF）：
    score(doc) = Σ 1 / (60 + rank_in_each_list)
       ↓
  Reranker 模型（可选，最强精度）→ 重新排序 top 30
       ↓
  取 top 5 注入 prompt

Reciprocal Rank Fusion/ˈfjuːʒən/🔊（RRF）算法

def rrf(rankings, k=60):
    """合并多个排序列表（dense + sparse）"""
    scores = defaultdict(float)
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking):
            scores[doc_id] += 1 / (k + rank + 1)
    return sorted(scores.items(), key=lambda x: -x[1])

Reranker/riːˈræŋkər/🔊（重排序）模型

进一步提升精度的"压舱石"。

Reranker 跟 embedding 不同 —— 它直接读 query + doc 一对判分，不是预先编码再算相似度。精度高但速度慢，适合 top 30 → top 5 的精排。

召回率提升幅度（实测）

LangGraph Store API（详见 §6.5）

from langgraph.store.memory import InMemoryStore
from langgraph.store.postgres import PostgresStore

store = InMemoryStore()  # 开发
# store = PostgresStore(conn_string)  # 生产

# namespace 是层级元组
namespace = ("user_memories", "user-123")

# 写入
store.put(namespace, key="diet", value={"preference": "vegetarian"})
store.put(namespace, key="lang", value={"primary": "中文"})

# 读取
mem = store.get(namespace, "diet")
print(mem.value)  # {"preference": "vegetarian"}

# 列出 namespace 内全部
all_mems = store.search(namespace)

加 embedding 索引：语义召回

from langchain_openai import OpenAIEmbeddings

store = InMemoryStore(
    index={"embed": OpenAIEmbeddings(), "dims": 1536}
)
store.put(namespace, "m1", {"text": "用户喜欢蓝色"})
store.put(namespace, "m2", {"text": "用户素食"})

# 语义搜索
results = store.search(namespace, query="食物偏好")
# 返回 m2（语义相关）排在前

在节点函数里使用 Store

def call_model(state, *, store):
    user_id = state["user_id"]
    namespace = ("user_memories", user_id)

    # 读相关记忆
    mems = store.search(namespace, query=state["messages"][-1].content)
    context = "\n".join(m.value["text"] for m in mems)

    # 注入 prompt
    enriched = [
        SystemMessage(f"已知信息：{context}"),
        *state["messages"]
    ]
    response = llm.invoke(enriched)

    # 自动抽取并存新记忆
    if "我喜欢" in state["messages"][-1].content:
        store.put(namespace, str(uuid.uuid4()), {"text": "..."})

    return {"messages": [response]}

Anthropic Skills（2025 新模式）

把"能力"打包成 Skill —— 含 prompt + tools + 示例 + memory：

详见 §16.3。

三种长期记忆的对比

小结一行：长期记忆三种实现互补 —— Vector DB 做文档检索、Store API 做结构化记忆、Skills 做能力封装。

§0.3.5 状态持久化与时间旅行

get_state_history：看历史快照

config = {"configurable": {"thread_id": "user-123"}}

for state in agent.get_state_history(config):
    print(f"超步 {state.metadata['step']}: 节点 {state.next}")
    print(f"  state: {state.values}")
    print(f"  checkpoint_id: {state.config['configurable']['checkpoint_id']}")

分支重放（Branching）

从历史某 checkpoint 改 state 后继续跑 —— 创出"平行宇宙"分支：

# 1. 找回到的 checkpoint
target = list(agent.get_state_history(config))[3]

# 2. 用它的 config 创建新分支
new_config = target.config

# 3. 修改 state 后继续
agent.update_state(new_config, {"messages": [HumanMessage("换问法")]})
result = agent.invoke(None, config=new_config)

工程价值

小结一行：时间旅行让 §0.3 的"中期记忆"从"现在"扩展到"历史 + 分支"，是 LangGraph 区别于 LangChain 的杀手级能力。

§0.3.6 多租户隔离的命名空间

生产环境必须按 workspace / 租户隔离记忆，避免数据泄漏。

三层隔离

config = {"configurable": {
    "thread_id": session_id,
    "checkpoint_ns": workspace_id,    # ★ 多租户隔离
    "user_id": user_id,
}}

# Store 也要带 workspace
namespace = ("user_memories", workspace_id, user_id)

Postgres 分区

-- 按 workspace_id 哈希分区
CREATE TABLE checkpoints (
    thread_id TEXT,
    checkpoint_ns TEXT,
    checkpoint_id TEXT,
    ...
) PARTITION BY HASH (checkpoint_ns);

CREATE TABLE checkpoints_p0 PARTITION OF checkpoints
FOR VALUES WITH (modulus 16, remainder 0);
-- ... 共 16 个分区

TTL 与清理

-- 删 30 天前 checkpoint
DELETE FROM checkpoints WHERE created_at < NOW() - INTERVAL '30 days';

大字段外挂

小结一行：多租户用 checkpoint_ns + Postgres 分区 + Store namespace 三层隔离，大字段必须外挂 S3。

§0.3.7 上下文工程：组装与截断

完整上下文构成（Anthropic API 实例）

client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4096,

    # 系统层（独立字段，可缓存）
    system=[{"type": "text", "text": "...",
             "cache_control": {"type": "ephemeral"}}],

    # 工具层（独立字段，可缓存）
    tools=[{"name": "...", "description": "...", "input_schema": {...}}],

    # 对话层
    messages=[
        # 历史
        {"role": "user", "content": "..."},
        {"role": "assistant", "content": [...]},
        # 当前
        {"role": "user", "content": [
            {"type": "text", "text": "<RAG 检索片段 1>"},
            {"type": "text", "text": "<RAG 检索片段 2>"},
            {"type": "text", "text": "用户当前问题"}
        ]}
    ]
)

Token 估算

4 种截断策略

Prompt Caching 命中条件

命中条件：从 prompt 起始位置开始的"连续相同 prefix"

工程建议：稳定不变的内容前置（系统 + 工具），变动的放最后（当前问题 + RAG）。

小结一行：上下文工程的核心是"稳定前置 + cache_control + 智能截断"，token 预算和 cache 命中是降本的两大杠杆。

§0.3.8 存储架构与成本（生产级深化）

三层记忆的存储成本估算

按 10 万 DAU × 平均 10 会话/天 × 平均 10 轮/会话 估算（中型 SaaS 智能体）：

关键洞察：Checkpoint 比 Vector DB 体积大 50-100×（因为存的是结构化 state，不是稀疏 embedding）。没有 TTL 清理就是定时炸弹。

Postgres Checkpoint 索引策略

LangGraph PostgresSaver.setup() 默认建：

CREATE INDEX checkpoints_thread_id_idx ON checkpoints(thread_id);
CREATE INDEX checkpoints_thread_id_step_idx ON checkpoints(thread_id, step);

生产环境推荐补加

-- 按用户查最近 N 条
CREATE INDEX checkpoints_thread_id_created_at_idx
ON checkpoints(thread_id, created_at DESC);

-- 按 workspace 分区（多租户）
ALTER TABLE checkpoints
PARTITION BY HASH (checkpoint_ns);

CREATE TABLE checkpoints_p0 PARTITION OF checkpoints
FOR VALUES WITH (modulus 16, remainder 0);
-- ... 共 16 分区

-- TTL 自动清理（cron job）
DELETE FROM checkpoints
WHERE created_at < NOW() - INTERVAL '30 days';

Vector DB 价格对比（2026.04 / 1M 条 1536-d embedding）

自部署 vs 云服务的盈亏平衡点：

向量索引算法对比

Hot / Warm / Cold 三层存储

Hot（< 7 天）     → Postgres 主库 + 内存索引   ~$0.10/GB/月
Warm（7-30 天）   → Postgres 只读副本           ~$0.05/GB/月
Cold（> 30 天）   → S3 / GCS 归档              ~$0.004/GB/月

实现：自动归档 cron

# 30 天前 checkpoint 导出 S3 后删主库
def archive_old_checkpoints():
    cutoff = datetime.now() - timedelta(days=30)
    rows = db.query("SELECT * FROM checkpoints WHERE created_at < %s", cutoff)
    s3.upload_jsonl(rows, key=f"archive/{cutoff.year}-{cutoff.month}.jsonl.zst")
    db.execute("DELETE FROM checkpoints WHERE created_at < %s", cutoff)

大字段外挂 S3 的容量优化

# 写入时
def my_node(state):
    if len(pdf_text) > 100_000:
        s3_key = f"pdf/{uuid.uuid4()}"
        s3.put(s3_key, pdf_text)
        return {"docs": [{"s3_key": s3_key, "type": "pdf"}]}
    return {"docs": [{"text": pdf_text, "type": "pdf"}]}

# 读取时
def use_doc(state):
    for doc in state["docs"]:
        text = s3.get(doc["s3_key"]) if "s3_key" in doc else doc["text"]
        ...

小结一行：存储成本三大杠杆 —— Postgres 索引 + 分区 + TTL 控制中期、向量库选型 + 三层冷热分层控制长期、大字段外挂 S3 控制 Postgres 体积。

§0.3.9 备份与灾备

Checkpoint 灾备方案

数据库迁移：把 Checkpoint 从 SQLite 升级到 Postgres

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver

# 旧
old = SqliteSaver.from_conn_string("checkpoints.db")
# 新
new = PostgresSaver.from_conn_string("postgres://...")
new.setup()

# 迁移
for cp in old.list(config=None):
    new.put(cp.config, cp.checkpoint, cp.metadata, new_versions={})

跨地域同步（多 region 部署）

us-east-1 (主)           ──CDC/Streaming Replication──▶  eu-west-1 (副)
  ↓                                                        ↓
agent v1 写 checkpoint                                     agent v2 读副本（只读）

灾备演练（Game Day）

小结一行：灾备 = PITR（小灾）+ 跨地域副本（中灾）+ S3 归档（大灾），按 RPO/RTO 要求选组合。

§0.4 智能体层：思考与执行原理

基础流程对应：节点 ② 推理 + ⑦ 循环的"为什么这样做"。

思维链（CoT）的实现机理

关键洞察：不是"模型在推理"，而是"token 序列展开 + KV cache 跨步复用"。

实证（GSM8K 数学题）：

CoT 让模型把中间计算 token 化、KV cache 化，最终答案 attend 这些中间步骤。

ReAct = CoT + 工具调用

ReAct 关键创新：Observation 文本进入 KV cache，相当于"把工具结果注入模型的思考"。

详细 ReAct 范式见 §1。

推理模型的内化（4 代）

对 ReAct 循环的影响

推理模型是怎么训练出来的（RL + Verifier）

普通 LLM 的训练

回顾 §0.1 训练 vs 推理：普通 LLM 走 预训练 → SFT → RLHF。RLHF 的奖励信号是人类对回复的偏好排序（"答案 A 比答案 B 好"）。

推理模型的训练增加了什么

OpenAI o1 / DeepSeek-R1 / Anthropic Extended Thinking 在 RLHF 之外，加了一个新阶段：

普通 LLM:        预训练 → SFT → RLHF（人类偏好奖励）
推理模型:        预训练 → SFT → RLHF → ★ RL with Verifier（可验证奖励）

核心创新：奖励不再来自人类偏好，而是来自自动可验证的信号：

训练流程（DeepSeek-R1 公开论文为例）

步骤 1: 给模型一道数学题
       "解 2x + 5 = 11"
       ↓
步骤 2: 让模型生成 N 个不同思考过程 + 答案
       Sample 1: <think>x = (11-5)/2 = 3</think> 答：3   ✓
       Sample 2: <think>x = 11/2 - 5 = 0.5</think> 答：0.5  ✗
       Sample 3: <think>x = 11 - 5 = 6, x = 6/2 = 3</think> 答：3  ✓
       ...
       ↓
步骤 3: Verifier 判分（不是人评，是脚本对照标准答案）
       Sample 1: +1（答对）
       Sample 2: -1（答错）
       Sample 3: +1（答对，且过程合理）
       ↓
步骤 4: 用强化学习（PPO / GRPO 算法）
       让模型更可能生成"答对"那种思考过程
       ↓
       重复几十万次 → 模型自动学会"长链思考"

涌现的能力

DeepSeek-R1 训练时没有显式教模型"该想多久"，但训练后模型自发学会：

这是一个模型涌现现象 —— 不是工程师写规则，而是 RL 训练逼出来的。

"Hidden Thinking" 是什么

用户看到的回复:
  "答：3"

模型内部生成的完整序列：
  <think>
  这道题是 2x + 5 = 11，要解 x。
  让我先把 5 移到右边：2x = 11 - 5 = 6
  再除以 2：x = 6 / 2 = 3
  让我验证：2*3 + 5 = 6 + 5 = 11 ✓
  </think>
  答：3

<think>...</think> 部分是 hidden thinking，用户看不到但要付费（因为占 GPU 计算）。OpenAI o1 把这部分完全隐藏，DeepSeek-R1 / Claude Extended Thinking 给开发者选项可以查看。

推理模型 vs 多步 ReAct 的本质区别

什么场景用推理模型才划算

推理模型的"思考预算"控制（reasoning_effort）

OpenAI o3 等模型支持 reasoning_effort 参数：

response = client.chat.completions.create(
    model="o3",
    messages=[...],
    reasoning_effort="medium"  # low / medium / high
)

Claude Extended Thinking 类似：thinking={"type": "enabled", "budget_tokens": 8000}。

智能体调用 = N 次 LLM API

T+0    API call 1  Input: 5k → Output: tool_use("search_order")  (50 token)
T+1.5s 执行 search_order → 返回

T+1.7s API call 2  Input: 6k → Output: tool_use("search_logistics")  (50 token)
T+3.2s 执行 search_logistics → 返回

T+3.5s API call 3  Input: 8k → Output: 最终回答  (200 token)
T+5s   返回用户

总耗时 ≈ 5s（LLM 3s + 工具 0.5s + 网络 1.5s）

Token 累积爆炸

每次调用都重发完整历史 —— 这是"上下文窗口爆炸"的实际机制。

成本估算公式（Claude Sonnet 4.6 / 2026.04）

Input 价格   = $3 / 百万 token
Output 价格  = $15 / 百万 token
Cache 命中   = $0.30 / 百万 token（90% 折扣）

5 步任务（无缓存）：
  ≈ 5 × (10k×$3/M + 500×$15/M) ≈ $0.19

5 步任务（开 Prompt Caching，5k 命中）：
  ≈ 5 × (5k×$0.30/M + 5k×$3/M + 500×$15/M) ≈ $0.12

进一步路由到 Haiku：可降到 $0.05

失败模式与节点级保护

每个都需要节点级保护（详见 §3.5 / §5.6）。

成本架构与 FinOps 深化

三家厂商完整定价表（2026.04，每百万 token）

关键价格规律

真实月度账单案例

案例 A：中型 SaaS 客服智能体

规模：10 万 DAU × 5 会话/天 × 5 步 × Sonnet 4.6
  100k × 5 × 5 = 250 万次 LLM 调用 / 天
  × 30 天 = 7,500 万次 / 月

每次调用平均：
  Input: 8k token
  Output: 300 token
  Cache 命中：70%（系统提示+工具描述固定）

月成本：
  Input（无缓存 30%）: 7500万 × 8k × 0.3 × $3/M = $54,000
  Input（缓存 70%）  : 7500万 × 8k × 0.7 × $0.30/M = $12,600
  Output           : 7500万 × 300 × $15/M = $33,750
  ─────────────────────
  总计              ≈ $100,350 / 月

不开 Caching： $234,000 / 月（多花 2.3×）
全用 Opus 4.7： $501,750 / 月（多花 5×）

案例 B：编程智能体（高 token 长循环）

规模：1 万开发者 × 50 次/天 × 30 步 × Opus 4.7
  10k × 50 × 30 = 1500 万次 / 天

每次调用平均：
  Input: 30k token（含大量代码上下文）
  Output: 800 token
  Cache 命中：80%（项目代码缓存）

月成本：
  Input（缓存 80%） : 1500万 × 30天 × 30k × 0.8 × $1.50/M = $16,200
  Input（无缓存 20%）: 1500万 × 30天 × 30k × 0.2 × $15/M  = $40,500
  Output           : 1500万 × 30天 × 800 × $75/M = $27,000
  ─────────────────────
  总计              ≈ $83,700 / 月

  人均月成本: $8.37
  按 $40/用户/月订阅 → 毛利 79%

6 大成本优化策略

模型路由实例（Sonnet + Haiku 混合）

def route_model(query: str) -> str:
    """简单分类用 Haiku，复杂推理用 Sonnet"""
    if len(query) < 100 and not any(k in query for k in ["分析", "规划", "代码"]):
        return "claude-haiku-4-5"        # 80% 流量走这里 ($0.80/$4)
    return "claude-sonnet-4-6"            # 20% 流量走这里 ($3/$15)

# 平均成本下降 60-70%（路由准确率 90%+ 时）

自部署 vs API 的盈亏平衡点

按 Llama 3.3 70B 自部署 vs Claude Sonnet API：

结论：日均 < 2.5 亿 input token 的项目，API 比自部署便宜。

关键 FinOps 指标（必上仪表盘）

异常监控（Cost Burst Detection）

# 简单异常检测：3-sigma 法则
def detect_burst(recent_costs: list[float]) -> bool:
    mean, std = statistics.mean(recent_costs[-100:]), statistics.stdev(recent_costs[-100:])
    current = recent_costs[-1]
    return current > mean + 3 * std   # 3σ 外异常

# 触发后立即：限流、报警、人工审查

延迟成本权衡（Pareto 前沿）

Pareto 前沿：Haiku → Gemini 2.5 Pro → Sonnet → Opus 是"价格/能力"最优组合。GPT-4o 在 Pareto 前沿上略劣（同价位 Sonnet 更强）。

小结一行：成本三大杠杆 —— Caching（省 60-90%）、模型路由（省 60-80%）、Batch API（省 50%）；FinOps 必看 7 项指标，不开仪表盘就是黑盒烧钱。

小结一行：智能体层的本质是 N 次 LLM API 调用 + 工具调用 + 累积状态 —— CoT/ReAct 是范式、推理模型是内化、成本和延迟是字节级账本。

§0.5 整体架构：现代主流 Agent 蓝图

基础流程对应：把所有节点组装成生产级架构。

6 层 Graph-based Architecture（Mermaid #31）

graph TB
    User([User 用户])
    User --> Planner["Planner 规划层
任务拆解 / 路径规划"]
    Planner --> GC[["Graph Controller 调度核心
状态机 / 流程控制"]]

    GC --> Executor["Executor 执行层
ReAct 节点
Thought-Action-Observation"]
    GC --> Critic["Critic 评估层
LLM-as-judge / Reflexion
判断对错 → 触发 retry"]

    Executor --> Tools["Tools 能力层
RAG / Code / API / DB / MCP"]
    Tools --> Memory[("Memory & State
短期 / 中期 / 长期 三层")]
    Critic -.判断.-> Executor

    Memory -.读写.-> GC
    Memory -.反馈.-> Planner

    style User fill:#fef3c7
    style Planner fill:#dbeafe,stroke:#3b82f6,stroke-width:3px
    style GC fill:#fce7f3,stroke:#ec4899,stroke-width:3px
    style Executor fill:#d1fae5,stroke:#10b981,stroke-width:3px
    style Critic fill:#e9d5ff,stroke:#a855f7,stroke-width:3px
    style Tools fill:#fed7aa
    style Memory fill:#ddd6fe

6 层职责详解

6 层 ↔ 8 节点 ↔ 5 代叠加 三向映射

三个视角互补，不冲突。

动态运行流程

1. User 输入任务
       ↓
2. Planner 拆任务 → 子任务列表
       ↓
3. Graph Controller 调度（状态机决定顺序）
       ↓
4. Executor 跑 ReAct（Thought → Action → Observation 循环）
       ↓ 调
5. Tools（搜索 / 代码 / API / DB）
       ↓ 写
6. Memory（短期 KV cache / 中期 Checkpoint / 长期 Store）
       ↓
7. Critic 评估（对吗？需要重试？）
       ↓ 触发
8. 不对 → 回 Executor / Planner replan
       对 → 继续下一子任务
       ↓ 完成所有子任务
9. 输出给 User

解决 ReAct 单体的 3 大短板

单 Agent → Multi-Agent 升级路径

Planner Agent
   ↓
Executor Agent A / Executor Agent B / ...   (并行 / 分工)
   ↓
Critic Agent
   ↓
Tool Agent（共享工具池）

详见 §6.3 多智能体模式（Supervisor / Plan-Execute / Swarm）。

一句话浓缩

现代 Agent 架构 = 🧠 Planner/ˈplænər/🔊（想做什么）+ 🕹 Graph Controller（什么时候做、做几次）+ ⚙️ Executor/ReAct（具体怎么做）+ 📚 Memory（记住）+ ✓ Critic（纠错）

把这张图记在心里，后续 §2-§6 都是在不同层填具体实现。

小结一行：业界 2026 落地蓝图就是 6 层 —— ReAct 是 Executor 内核、LangGraph 是 Graph Controller、MCP 是 Tools 协议、Checkpoint+Store+Vector DB 是 Memory 三层。

§0.6 全书坐标

本节是全书的"导航地图" —— 把 8 节点流程图、三件套对照、演进时间线、学习地图四张全局图集中放在这里，作为后续章节的统一参考。

8 节点基础流程图（Mermaid #1）

graph TB
    Start([用户输入]) --> N1["① 上下文组装
系统提示 + 历史 + 工具清单 + 状态"]
    N1 --> N2["② 大语言模型推理
读上下文 + 思考"]
    N2 --> N3{"③ 决策路由
调工具？直接答？"}
    N3 -->|直接回答| N8["⑧ 输出格式化
转成用户可读的回复"]
    N3 -->|调用工具| N4["④ 工具选择 + 安全调用"]
    N4 --> N5["⑤ 工具返回结果"]
    N5 --> N6["⑥ 观察 + 更新状态"]
    N6 --> N7{"⑦ 循环控制
该停了吗？"}
    N7 -->|继续| N2
    N7 -->|终止| N8
    N8 --> End([返回用户])

    style N1 fill:#fef3c7,stroke:#f59e0b
    style N2 fill:#dbeafe,stroke:#3b82f6
    style N3 fill:#fce7f3,stroke:#ec4899
    style N4 fill:#d1fae5,stroke:#10b981
    style N5 fill:#d1fae5,stroke:#10b981
    style N6 fill:#fef3c7,stroke:#f59e0b
    style N7 fill:#fce7f3,stroke:#ec4899
    style N8 fill:#dbeafe,stroke:#3b82f6

8 节点工程问题

三件套在 8 节点中各管什么

结论：ReAct = 范式（精神层），LangChain = 组件库（货架层），LangGraph = 编排引擎（蓝图层），三者分层堆叠，业界主流玩法是组合用。

演进时间线

学习地图（Mermaid #2）

graph LR
    Z["§0 原理架构"]
    A["§1 ReAct"]
    B["§2 LangChain 心智"]
    C["§3 LangChain 实战"]
    D["§4 LangGraph 心智"]
    E["§5 LangGraph 实战"]
    F["§6 LangGraph 进阶"]
    G["§7 三者对比"]
    H["§8 落地 custom_agent"]
    PartB["Part B
§9-§19
生态全景"]

    Z --> A --> B --> C --> D --> E --> F --> G --> H
    H -.通往 Part B.-> PartB

    style Z fill:#fef3c7
    style A fill:#dbeafe
    style B fill:#d1fae5
    style C fill:#d1fae5
    style D fill:#fce7f3
    style E fill:#fce7f3
    style F fill:#fce7f3
    style G fill:#e9d5ff
    style H fill:#e9d5ff
    style PartB fill:#fed7aa

§0.7 本章小结

§0.8 反模式速记

§0.9 术语速查

§0.10 推荐下一章

下一章：§1 ReAct：把基础流程节点 ②③ 显式化 —— §0 给了完整原理架构，§1 深入讲 ReAct 这个最核心的范式如何形成、它的生态位置、以及 5 代演进图。

§1 ReAct：把基础流程节点 ②③ 显式化

§1.0 本章定位

§1.1 没有 ReAct 之前：纯提示词时代的窘境

在基础流程中的位置：节点 ②（推理）单独运转的状态。

场景引入

2022 年中，给大模型加"工具"的玩法非常原始。开发者想让模型查天气，会这样写 prompt：

你是一个助手。如果用户问天气，回答：TOOL_CALL: weather(city=北京)
否则正常回答。

用户：今天北京天气怎么样？
助手：

模型可能输出 TOOL_CALL: weather(city=北京) —— 但也可能输出：

痛点逐节点拆解

简化流程图（Mermaid #3）

graph LR
    Q[用户问题] --> P["纯 Prompt
无 Thought 字段"]
    P --> M["大语言模型
节点 ②"]
    M --> A["输出（混杂）
答案 / 调用 / 编造"]
    A --> Out[?]

    style M fill:#fde68a,stroke:#f59e0b,stroke-width:3px
    style A fill:#fecaca,stroke:#ef4444

小结一行：纯 Prompt 时代节点 ② 是个黑盒，输出格式不稳定 → 节点 ③ 解析全靠运气。

§1.2 ReAct 论文核心贡献：把"思考"和"行动"解耦

在基础流程中的位置：把节点 ② 推理拆成"显式思考"+"显式行动"两步。

论文背景

ReAct 论文全称 ReAct: Synergizing Reasoning and Acting in Language Models，2022 年 10 月由 Princeton 的 Shunyu Yao 等人提出（ICLR 2023）。

ReAct 这个名字是个双关： - Reasoning + Acting = ReAct（推理 + 行动） - 同时也呼应英文单词 react（反应、应对）

核心贡献：把模型输出强制分三段

ReAct 的精髓 —— 在 prompt 里规定模型必须按以下三段交替输出：

Thought: 我需要先查一下北京今天的天气
Action: weather
Action Input: 北京
Observation: 北京今天晴朗 25 度
Thought: 我已经拿到了天气，可以回答了
Action: Final Answer
Action Input: 北京今天晴朗 25 度

为什么"显式 Thought"如此重要

ReAct 之前已有思维链（CoT）—— 让模型一步一步思考。但 CoT 只解决了节点 ②（让推理质量上升），没解决节点 ③④⑤（行动）。

ReAct 论文的核心论点 —— 让模型把"为什么要调这个工具"先说出来再调，会大幅降低错误率。

论文实验结果

ALFWorld 上 +29 个百分点，证明带"行动"的任务里 ReAct 极强。

小结一行：ReAct = 思维链 + 工具调用，让模型的推理过程从"看不见摸不着"变成"可观察可干预"。

§1.3 思考-行动-观察循环：三步缺一不可

在基础流程中的位置：节点 ②③⑤ 形成的最小闭环。

三步的工程含义

内循环图示（Mermaid #4）

graph TB
    Start([用户输入]) --> T1["Thought 1
我应该先查 X"]
    T1 --> A1["Action 1
调 search(X)"]
    A1 --> O1["Observation 1
X 是 ..."]
    O1 --> T2{"Thought 2
够了吗？"}
    T2 -->|不够| A2["Action 2
调 calculator(...)"]
    A2 --> O2["Observation 2
结果是 ..."]
    O2 --> T2
    T2 -->|够了| FA["Action: Final Answer
返回用户"]
    FA --> End([输出])

    style T1 fill:#dbeafe
    style T2 fill:#dbeafe
    style A1 fill:#d1fae5
    style A2 fill:#d1fae5
    style FA fill:#d1fae5
    style O1 fill:#fef3c7
    style O2 fill:#fef3c7

为什么三步缺一不可

ReAct 的本质 —— 强制把"想什么"和"做什么"分开写出来，让框架能介入每一步。

小结一行：三步循环让节点 ② → ③ → ④ → ⑤ → ② 形成可观察的反馈环。

§1.4 ReAct 提示词模板逐字段拆解

在基础流程中的位置：节点 ① 上下文组装的具体模板。

标准 ReAct 模板

回答以下问题。可用以下工具：

{tools_description}

使用以下格式：

Question: 用户的输入问题
Thought: 你应该思考下一步做什么
Action: 选择上述工具之一
Action Input: 工具的输入参数
Observation: 工具返回的结果
... (Thought / Action / Action Input / Observation 可以重复多次)
Thought: 我现在知道最终答案了
Final Answer: 给用户的最终答案

开始！

Question: {input}
{agent_scratchpad}

关键字段解释

少样本示例（few-shot examples）怎么选

ReAct 论文里发现 —— prompt 里给 1-3 个范例比纯 zero-shot 效果好得多。范例选择原则：

少样本示例不是越多越好 —— 放 5 个之后边际效益急剧下降，且 token 成本线性上升。

节点 ① 上下文组装顺序

[System Prompt: ReAct 模板说明]
[Tools Description: 工具列表]
[Few-shot Example 1]
[Few-shot Example 2]
[Few-shot Example 3]
[Question: 当前问题]
[Agent Scratchpad: 历史 Thought/Action/Observation]

小结一行：ReAct 模板的精髓在"格式约束 + 少样本示范"，节点 ① 上下文组装的胶水代码全在这里。

§1.5 ReAct 兄弟范式：补什么不足

在基础流程中的位置：扩展 ReAct 在节点 ②③⑥⑦ 的能力。

ReAct 只是众多智能体推理范式中最早爆火的一个，2023-2024 年陆续出现了一系列"补丁"。

计划-执行（Plan-and-Execute）

反思（Reflexion）

思维树（Tree-of-Thoughts，ToT）

兄弟范式与 ReAct 对照

小结一行：ReAct 是"基线"，其他范式都在 ReAct 上某个节点做加法 —— 工程实践中先 ReAct，发现具体痛点再选对应兄弟范式。

§1.6 ReAct 的"原生竞争方案"：Function Calling 与 Tool Use

在基础流程中的位置：节点 ③④ 跳过字符串解析。

问题：ReAct 的字符串解析很脆

ReAct 让模型输出 Action: weather 这种字符串，框架要正则提取工具名。一旦模型输出 Action: 调用 weather 或 Action: weather() 就解析失败。

2023.06 OpenAI Function Calling 的解法

OpenAI 2023.06 发布的 Function Calling 是这么做的：

1. 在 API 请求里给一个 functions 数组（每个函数有 name + description + JSON Schema 参数）
2. 模型不再输出文本，而是返回一个特殊的 function_call 字段（包含 name 和 arguments）
3. 这个字段是结构化 JSON，不是字符串解析

协议对照

节点级别的影响

现状

2026 年的实际玩法： - 新项目几乎都用 Function Calling / Tool Use 原生协议，不再用 ReAct 字符串解析 - ReAct 的精神（Thought + Action + Observation）依然适用 —— 只是用 JSON 字段实现而非字符串 - LangChain 的 Tool Calling Agent（§3.2）就是这套现代做法

小结一行：ReAct 是范式（精神层），Function Calling 是协议（接口层），两者相辅相成 —— 新项目用 Function Calling 实现 ReAct 思想。

§1.7 历史名作：AutoGPT 与 BabyAGI

在基础流程中的位置：节点 ⑦ 循环控制的两种典型设计。

2023.03 AutoGPT 走红

AutoGPT 由 Toran Bruce Richards 于 2023.03 在 GitHub 发布，几周内拿到 100k+ star —— 第一个引爆 ReAct 范式的开源项目。

遗产：定义了"自主智能体"这个词的大众认知 —— 给个目标就自己跑。 坑：没有 token 限制，跑一晚上花 100 美元；经常卡死循环；任务完成质量低。

2023.04 BabyAGI 出现

Yohei Nakajima 几乎同时发布的 BabyAGI 走的是 Plan-Execute 路线：

Objective → 生成 Task List → 取第一个 Task → 执行
                ▲                              │
                └──────────────────────────────┘
                     根据执行结果重排 Task List

遗产：把"多角色协作"思路普及 —— 后来的 Supervisor / Plan-Execute / Swarm 模式都受它启发。

这两个项目的影响

小结一行：AutoGPT 让"自主智能体"破圈，BabyAGI 启发"多角色协作" —— 它们的局限直接催生了 LangGraph 这一代图式编排引擎。

§1.8 ReAct 局限性：每个局限对应基础流程的哪个节点

在基础流程中的位置：揭示 ReAct 在节点 ①⑤④⑦⑥ 的全面短板。

小结一行：ReAct 的 6 大局限完美映射基础流程的 6 个节点 —— LangChain（§2-§3）补节点级，LangGraph（§4-§6）补流程级。

§1.9 业界用 ReAct 的真实场景

在基础流程中的位置：节点 ②③⑤ 在产品中的具体形态。

真实场景的共同特征

小结一行：ReAct 的真实战场是"步骤不定、工具固定、需要中间反馈"的任务 —— 这正是 LangGraph 优化的核心。

§1.10 代际演进图：从 ReAct 到 2026（5 代叠加模型）

在基础流程中的位置：把整个 Agent 范式发展史压成一条时间轴。理解 ReAct 在 5 代演进中是哪一代、被怎样叠加、为什么没死。

场景引入

读者读完前 9 节会有个困惑："ReAct 这么多坑，业界真在用吗？"答案是 —— 不直接用 ReAct，但它是所有现代范式的底层原语。本节用"5 代演进 + 6 层叠加"模型把这点说透。

5 代演进总览（Mermaid #28）

graph LR
    G0["第 0 代
2022.01
Chain-of-Thought
纯思考"]
    G1["第 1 代
2022.10
ReAct
思考+工具"]
    G2["第 2 代
2023
Plan-Execute
Reflexion / ToT
修补 ReAct"]
    G3["第 3 代
2024
LangGraph
多智能体 / MCP
系统级架构"]
    G4["第 4 代
2024.12+
Reasoning Models
Computer Use
模型重塑 Agent"]
    G5["第 5 代
2026 形成中
A2A / Skills
Agent as Code
网络化 + 工程化"]

    G0 --> G1 --> G2 --> G3 --> G4 --> G5

    style G0 fill:#fef3c7
    style G1 fill:#dbeafe
    style G2 fill:#d1fae5
    style G3 fill:#fce7f3,stroke-width:3px
    style G4 fill:#e9d5ff
    style G5 fill:#fed7aa

每代的核心创新

每代在 8 节点上的着力（速读版）

每代的优缺点对照

关键洞察 1：分层叠加而非替代（6 层模型）

每一代不替代上一代，而是叠加在不同层。这就是为什么 ReAct 没死。

读法：一个真实的 2026 Agent 是这 6 层的组合，不是某一代的"纯产品"。

关键洞察 2：演进的 5 个核心规律

2026 业界主流组合（按场景）

一句话总结

ReAct 没死，它降级成了"原语" —— 第 4 代推理模型 + 第 3 代 LangGraph 编排 + 第 5 代 MCP 协议 才是 2026 严肃 Agent 的真实形态。

详见 附录 L Agent 范式代际演进总表 —— 包含 5 代百科速查、6 层映射详细矩阵、custom_agent 代际定位。

§1.11 本章小结

§1.12 反模式速记

§1.13 术语速查

§1.14 推荐下一章

下一章：§2 LangChain：第一代框架的心智模型 —— ReAct 是范式，LangChain 是把它工程化的第一代尝试。

§2 LangChain：第一代框架的心智模型

§2.0 本章定位

§2.1 LangChain 解决了什么 — 2022 末的工程痛点

在基础流程中的位置：节点 ① 到 ⑧ 全部 — 第一次给每个节点提供"组件级"工具箱。

场景引入

2022 年 10 月，开发者要做一个能查天气的智能体，得自己手写：

总计 350+ 行胶水代码，且每家厂商一份。

LangChain 的工程目标

Harrison Chase 2022 年 10 月开源 LangChain，目标是把这些"胶水代码"做成可复用的组件库：

小结一行：LangChain 解决的核心问题是"用统一抽象消除厂商和工具的胶水代码"。

§2.2 历史脉络：v0.0 → v0.3 的四次重构

在基础流程中的位置：理解版本演进，知道哪些 API 已废弃。

版本时间线

版本变化的工程含义

升级到 v0.3 的破坏性改动清单

小结一行：LangChain 三年四次大变身，新项目从 v0.3 起步、老项目按官方迁移指南升级（详见 §3.9）。

§2.3 核心抽象六件套对应到流程节点

在基础流程中的位置：每个组件管哪个节点。

六件套总览

每个组件的最小代码示例

# ② LLM
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ① Prompt
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手"),
    ("user", "{question}"),
])

# ⑧ OutputParser
from langchain_core.output_parsers import StrOutputParser
parser = StrOutputParser()

# 用 LCEL 串起来
chain = prompt | llm | parser
result = chain.invoke({"question": "今天天气怎么样？"})

LLM vs ChatModel 的区分

新项目只用 ChatModel。

小结一行：六件套是 LangChain 的"乐高积木"，LCEL 是把它们粘起来的"胶水"。

§2.4 LCEL 表达式语言：管道符的设计哲学

在基础流程中的位置：把节点 ①②⑧ 串起来的胶水。

LCEL（LangChain Expression Language）是什么

LangChain 表达式语言（LangChain Expression Language，简称 LCEL）—— 用 Python 的 | 管道符把组件串起来：

chain = prompt | llm | parser

读作"把 prompt 输出喂给 llm，再喂给 parser"。

为什么用管道符

LCEL 与传统写法对照（Mermaid #5）

graph LR
    subgraph Old["旧写法（LLMChain，已废弃）"]
        A1[Question] --> B1["LLMChain
llm + prompt 黑盒包装"]
        B1 --> C1[Answer]
    end

    subgraph New["新写法（LCEL）"]
        A2[Question] --> P[prompt]
        P --> L[llm]
        L --> Pa[parser]
        Pa --> C2[Answer]
    end

    style B1 fill:#fecaca,stroke:#ef4444
    style P fill:#d1fae5
    style L fill:#dbeafe
    style Pa fill:#fef3c7

LCEL 提供的辅助原语

一个稍复杂的 LCEL 示例

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

retrieve_chain = (
    RunnableParallel({
        "context": retriever | format_docs,    # RAG 检索结果
        "question": RunnablePassthrough(),     # 透传问题
    })
    | prompt
    | llm
    | parser
)

读作：先并行检索文档 + 透传问题，组装到 prompt，过 llm，过 parser。

小结一行：LCEL 把"组件 + 管道符"当一等公民，让流式 / 异步 / 批处理 / 观测自动获得。

§2.5 包结构三层：core / langchain / community / partners

在基础流程中的位置：理解依赖管理。

四个包的职责

推荐安装策略

# 最小配置（只用 OpenAI）
pip install langchain langchain-openai

# 完整配置（用多家 + 社区集成）
pip install langchain langchain-openai langchain-anthropic langchain-community

为什么拆包

小结一行：v0.1 起的拆包架构让 LangChain 从"巨石"变"瑞士军刀" —— 按需选包。

§2.6 Runnable 接口：LCEL 背后的统一抽象

在基础流程中的位置：所有节点的组件背后的统一契约。

Runnable 是什么

Runnable 是 langchain-core 里的抽象基类，所有 LangChain 组件都实现这个接口。它定义了一组标准方法：

class Runnable:
    def invoke(self, input): ...           # 同步单次
    def ainvoke(self, input): ...          # 异步单次
    def batch(self, inputs): ...           # 同步批量
    def abatch(self, inputs): ...          # 异步批量
    def stream(self, input): ...           # 同步流式
    def astream(self, input): ...          # 异步流式
    def __or__(self, other): ...           # 管道符

为什么 Runnable 重要

不理解 Runnable 不算理解 LCEL：

自定义 Runnable 的两种方式

# 方式 1：用 RunnableLambda 包装函数
from langchain_core.runnables import RunnableLambda

def my_func(x: dict) -> str:
    return x["question"].upper()

my_runnable = RunnableLambda(my_func)
chain = my_runnable | llm

# 方式 2：继承 Runnable
from langchain_core.runnables import Runnable

class MyRunnable(Runnable):
    def invoke(self, input, config=None):
        return input.upper()

小结一行：Runnable 是 LangChain 的"USB 接口" —— 任何符合这个接口的东西都能插进 LCEL 管道。

§2.7 记忆模块四种深入对比

在基础流程中的位置：节点 ⑥ 状态管理在 LangChain 时代的实现。

四种 Memory 的工程定位

详细对比

选哪种 Memory 的决策树

对话轮数？
├── < 10 轮 → ConversationBufferMemory（最简）
├── 10-50 轮，不需要细节 → ConversationSummaryMemory
├── 实体关系密集 → ConversationKGMemory
└── 长对话需细节召回 → VectorStoreRetrieverMemory

现代写法：RunnableWithMessageHistory

v0.2 起官方推荐用 RunnableWithMessageHistory 取代直接挂 Memory：

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

store = {}

def get_session_history(session_id: str) -> ChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="question",
    history_messages_key="history",
)

result = chain_with_history.invoke(
    {"question": "..."},
    config={"configurable": {"session_id": "user-123"}},
)

Memory 的根本缺陷

不管哪种 Memory，都是会话级（同一 session 内）。跨会话失忆。这就是为什么需要 LangGraph 的 Store API（§6.5）。

小结一行：LangChain Memory 解决会话内的"上下文工程"，但跨会话失忆 —— 这洞要等 LangGraph Store API 来补。

§2.8 文档加载器与文本分割器

在基础流程中的位置：节点 ① 上下文增强（RAG）的入口。

Document Loaders 加载文档

LangChain 内置 100+ 种文档加载器（在 langchain-community）：

Text Splitters 切片

一个完整 RAG 数据准备示例

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

docs = PyPDFLoader("manual.pdf").load()                    # 加载
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(docs)                    # 切片
db = FAISS.from_documents(chunks, OpenAIEmbeddings())      # 向量化
retriever = db.as_retriever(search_kwargs={"k": 4})        # 检索器

小结一行：Document Loader + Text Splitter + Embedding + VectorStore 是 LangChain 的"RAG 四件套"，各组件之间通过 Document 对象统一。

§2.9 输出解析器三类与现代 with_structured_output

在基础流程中的位置：节点 ⑧ 输出格式化。

三类 OutputParser

老写法：手动用 PydanticOutputParser

from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel

class WeatherInfo(BaseModel):
    city: str
    temperature: float
    condition: str

parser = PydanticOutputParser(pydantic_object=WeatherInfo)
prompt = ChatPromptTemplate.from_messages([
    ("system", "提取天气信息。{format_instructions}"),
    ("user", "{input}"),
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | llm | parser
result = chain.invoke({"input": "北京今天 25 度晴"})
# WeatherInfo(city='北京', temperature=25.0, condition='晴')

问题：要手动管 format_instructions，复杂且易错。

现代写法：with_structured_output

v0.2 起官方推 with_structured_output()：

structured_llm = llm.with_structured_output(WeatherInfo)
result = structured_llm.invoke("北京今天 25 度晴")
# WeatherInfo(city='北京', temperature=25.0, condition='晴')

好处：

小结一行：节点 ⑧ 输出格式化从"prompt 哄模型"演化到"API 强约束"，新项目用 with_structured_output。

§2.10 缓存：LLM cache + 嵌入 cache

在基础流程中的位置：节点 ② / 节点 ① 增强的成本优化。

为什么需要缓存

LLM 调用缓存

from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache

# 内存缓存（开发用）
set_llm_cache(InMemoryCache())

# 持久化缓存（开发或低频生产）
set_llm_cache(SQLiteCache(database_path=".llm_cache.db"))

设完后所有 LLM 调用自动走缓存。命中条件：完全相同的输入（包括 temperature 等参数）。

嵌入缓存（CacheBackedEmbeddings）

from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore

store = LocalFileStore("./embedding_cache")
underlying = OpenAIEmbeddings()
cached_embed = CacheBackedEmbeddings.from_bytes_store(
    underlying, store, namespace=underlying.model
)

# 第一次 embedding 算并缓存
vec1 = cached_embed.embed_query("hello")
# 第二次直接命中
vec2 = cached_embed.embed_query("hello")

生产级缓存

小结一行：缓存是节点 ② / 节点 ① 的成本杠杆，开发用内存、生产用 Redis/ˈrɛdɪs/🔊 或语义缓存。

§2.11 流式：astream / stream

在基础流程中的位置：节点 ⑧ 输出格式化的体验关键。

为什么要流式

LLM 生成 200 字的回答可能要 3-5 秒。流式让用户每生成一个 token 就能看到，体感从"等 5 秒"变"看打字"。

LCEL 全链路流式

chain = prompt | llm | StrOutputParser()

# 同步流式
for chunk in chain.stream({"question": "讲个笑话"}):
    print(chunk, end="", flush=True)

# 异步流式
async for chunk in chain.astream({"question": "讲个笑话"}):
    print(chunk, end="", flush=True)

流式背后的机制

流式的限制

小结一行：流式让 chat-style 智能体的体感大幅提升 —— 但结构化输出和工具调用流式有坑（§3.x 会讲）。

§2.12 重试与回退：with_retry / with_fallbacks

在基础流程中的位置：横切节点 ② / ④ 的稳定性手段。

with_retry：失败重试

chain_with_retry = chain.with_retry(
    retry_if_exception_type=(ConnectionError, TimeoutError),
    wait_exponential_jitter=True,
    stop_after_attempt=3,
)

with_fallbacks：失败回退到备选

primary = ChatOpenAI(model="gpt-4o")
fallback = ChatAnthropic(model="claude-sonnet-4-6")

chain = (prompt | primary | parser).with_fallbacks(
    [prompt | fallback | parser]
)

主链失败（如 OpenAI 限流）就跑备选链（Anthropic）。

何时用什么

小结一行：重试管"瞬时错"，回退管"持续错"，组合用更稳。

§2.13 bind_tools 现代工具绑定

在基础流程中的位置：节点 ② 推理 + ④ 工具调用的现代衔接。

老写法：Tool Calling Agent + AgentExecutor

# 老（v0.1-v0.2）
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
result = executor.invoke({"input": "..."})

现代写法：bind_tools + 自己跑循环

# 现代（v0.3）
llm_with_tools = llm.bind_tools(tools)
response = llm_with_tools.invoke([HumanMessage(content="...")])
# response.tool_calls 是结构化字段

bind_tools() 把工具列表绑到 LLM，模型输出会带 tool_calls 字段。不再需要 AgentExecutor 黑盒 —— 用 LangGraph 的图式编排（详见 §5.1）显式控制循环。

bind_tools 的工作机制

LangChain 自动适配三家协议（详见附录 K）。

小结一行：bind_tools 是新版工具绑定的标准 API —— 它替代了老的 Tool 类 + AgentExecutor 组合。

§2.14 LangSmith 观测平台

在基础流程中的位置：横切所有节点的运行追踪。

LangSmith 是什么

LangChain 团队的商业观测平台，与 LangChain 框架解耦（不用 LangChain 也能用 LangSmith）。

启用 LangSmith

export LANGSMITH_API_KEY=...
export LANGSMITH_TRACING=true
export LANGSMITH_PROJECT=my-agent

启用后所有 LCEL 调用自动产生 trace，无需改代码。

LangSmith 在基础流程的作用

小结一行：LangSmith 是 LangChain 生态的"观测中枢"，但不绑死 —— 严肃团队都开。

§2.15 业界目前怎么用 LangChain

在基础流程中的位置：业界对 LangChain 的真实评估。

三类主流玩法

各部分的去留

小结一行：LangChain 已从"Agent 框架的代名词"退化为"组件胶水库"，新项目的默认是组件库 + LangGraph 编排。

§2.16 本章小结

§2.17 反模式速记

§2.18 术语速查

§2.19 推荐下一章

下一章：§3 用 LangChain 实现 ReAct（流程跑通实战） —— 把 §2 的组件库真的用起来，用 LangChain 写个完整的 ReAct 智能体并暴露其硬伤。

§3 用 LangChain 实现 ReAct（流程跑通实战）

§3.0 本章定位

§3.1 老接口（已废弃）：AgentExecutor + initialize_agent

在基础流程中的位置：节点 ③ 决策路由 + ⑦ 循环控制的老黑盒。

历史背景

2022.10 - 2024.05 期间，LangChain 智能体的"标准写法"是：

# 已废弃，只为理解历史
from langchain.agents import initialize_agent, AgentType
from langchain_openai import OpenAI

agent = initialize_agent(
    tools=[search_tool, calculator_tool],
    llm=OpenAI(temperature=0),
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
)
result = agent.invoke("北京今天比上海温度高几度？")

AgentExecutor 黑盒结构（Mermaid #6）

graph TB
    subgraph AE["AgentExecutor (黑盒)"]
        Prompt["Prompt 模板
含 ReAct 格式"]
        LLM["LLM 节点 ②"]
        Parser["AgentOutputParser
正则提取 Action"]
        Tools["Tools 字典 节点 ④"]
        Loop["while not done:
判断 Final Answer"]
    end

    Q[用户输入] --> AE
    AE --> R[最终答案]

    Prompt --> LLM
    LLM --> Parser
    Parser -->|Action: 工具名| Tools
    Tools --> Loop
    Loop -->|未结束| Prompt
    Loop -->|Final Answer| R

    style AE fill:#fecaca,stroke:#ef4444,stroke-width:3px
    style Parser fill:#fef3c7,stroke:#f59e0b

黑盒在哪些节点出问题

AgentType 的几种"个性"

老 initialize_agent 支持多种 Agent 类型：

v0.3 起 initialize_agent 已发出 deprecation 警告。

小结一行：老 AgentExecutor 把节点 ③⑦ 黑盒化、字符串解析脆弱、不支持 checkpoint —— 这就是为什么需要 LangGraph。

§3.2 现代做法：Tool Calling Agent + LCEL

在基础流程中的位置：节点 ③ 用结构化字段替代字符串解析。

现代写法的演化

v0.2 起，LangChain 推荐：

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个能调用工具的助手"),
    ("placeholder", "{chat_history}"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "北京今天比上海温度高几度？"})

现代做法的全流程图（Mermaid #7）

graph LR
    Q[用户输入] --> P["ChatPromptTemplate
节点 ①"]
    P --> L["llm.bind_tools(tools)
节点 ②"]
    L --> J{"node ③
有 tool_calls？"}
    J -->|否| F["Final Output
节点 ⑧"]
    J -->|是| T["ToolNode 并发调用
节点 ④⑤"]
    T --> M["Add ToolMessage 到
agent_scratchpad 节点 ⑥"]
    M --> P

    style L fill:#dbeafe
    style J fill:#fce7f3
    style T fill:#d1fae5

与老接口的关键差异

create_tool_calling_agent 的内部机制

它本质上是一个 LCEL 链：

# 等价于
agent = (
    {
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_to_tool_messages(x["intermediate_steps"]),
    }
    | prompt
    | llm.bind_tools(tools)
    | ToolsAgentOutputParser()  # 解析 tool_calls 字段，不再正则
)

小结一行：现代做法用结构化 tool_calls 字段替代 ReAct 字符串，节点 ③ 决策路由从"正则猜"变成"字段读"。

§3.3 完整代码：50 行从 prompt 到 agent 到调用

在基础流程中的位置：把节点 ① 到 ⑧ 全跑通。

"""LangChain 现代 ReAct 智能体完整实现 (50 行)"""

# 1. 导入
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.globals import set_llm_cache
from langchain.cache import SQLiteCache

# 2. 开缓存（节省调试成本，节点 ② 优化）
set_llm_cache(SQLiteCache(database_path=".llm_cache.db"))

# 3. 定义工具（节点 ④）
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。city 是中文城市名。"""
    fake_db = {"北京": "晴 25 度", "上海": "多云 22 度", "广州": "雨 28 度"}
    return fake_db.get(city, f"未找到 {city} 的天气")

@tool
def calculator(expression: str) -> str:
    """计算数学表达式。expression 是 Python 可执行的字符串如 '25 - 22'。"""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"计算错误：{e}"

tools = [get_weather, calculator]

# 4. 大语言模型（节点 ②）
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 5. 提示词模板（节点 ①）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，可调用提供的工具。"
               "如需多步操作，可多次调工具；最后再给出回答。"),
    ("placeholder", "{chat_history}"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),  # 节点 ⑥ 中间状态
])

# 6. 智能体 + 执行器（节点 ③⑦）
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,             # 打印每步
    max_iterations=8,          # 节点 ⑦ 循环上限
    return_intermediate_steps=True,  # 调试用
)

# 7. 调用（跑完整流程）
result = executor.invoke({
    "input": "北京今天比上海温度高几度？",
    "chat_history": [],
})

# 8. 输出（节点 ⑧）
print("最终答案:", result["output"])
print("中间步骤数:", len(result["intermediate_steps"]))

期望的执行轨迹

> Entering new AgentExecutor chain...
[Tool call] get_weather(city='北京')
Tool result: 晴 25 度
[Tool call] get_weather(city='上海')
Tool result: 多云 22 度
[Tool call] calculator(expression='25 - 22')
Tool result: 3
[Final response] 北京今天 25 度，上海 22 度，北京比上海高 3 度。
> Finished chain.

50 行代码各部分对应基础流程

小结一行：50 行就能跑出可工作的 LangChain 智能体 —— 但每个节点都有问题，下面 §3.5 拆解。

§3.4 节点 ② 输出可见性：怎么看模型的"思考"过程

在基础流程中的位置：节点 ② 推理的可观测性。

verbose=True 的输出格式

> Entering new AgentExecutor chain...

Invoking: `get_weather` with `{'city': '北京'}`
晴 25 度

Invoking: `get_weather` with `{'city': '上海'}`
多云 22 度

Invoking: `calculator` with `{'expression': '25 - 22'}`
3

> Finished chain.

拿到中间步骤的两种方式

# 方式 1: return_intermediate_steps=True
executor = AgentExecutor(agent=agent, tools=tools, return_intermediate_steps=True)
result = executor.invoke({"input": "..."})
for action, observation in result["intermediate_steps"]:
    print(f"[Action] {action.tool}({action.tool_input})")
    print(f"[Obs] {observation}")

# 方式 2: 用 LangSmith
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "..."
# 之后所有 invoke 都自动产生 trace，去网页看

流式查看中间过程

# astream_events 可以看到每个内部事件
async for event in executor.astream_events({"input": "..."}, version="v2"):
    if event["event"] == "on_tool_start":
        print(f"[Tool start] {event['name']}({event['data']['input']})")
    elif event["event"] == "on_tool_end":
        print(f"[Tool end] {event['data']['output']}")
    elif event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

可见性局限

要看节点 ② 的"思考过程"，要么模型本身支持 reasoning 输出（如 Claude Extended Thinking、o1），要么用"Thought + Action + Observation"格式（即回到老 ReAct prompt）。

小结一行：现代 Tool Calling Agent 输出结构化但缺"思考过程"——要原汁原味的 ReAct Thought 字段，得改 prompt 强制要求。

§3.5 LangChain 实现 ReAct 的硬伤逐节点拆解

在基础流程中的位置：本节是引出 LangGraph 的关键。

硬伤 1：节点 ⑥ 状态污染

# 假设这是真实场景
result1 = executor.invoke({"input": "查北京天气", "chat_history": []})
result2 = executor.invoke({"input": "查上海天气", "chat_history": []})  # 没接上 history

问题：两次 invoke 之间状态丢失。要持续对话得自己管 chat_history，且只是会话级 —— 进程重启或换实例就忘。

硬伤 2：节点 ③ AgentExecutor 黑盒难调试

硬伤 3：节点 ⑦ 循环只能粗放管

max_iterations=8 是唯一控制。没有： - 中途暂停问人类（HITL） - 卡死时优雅退出（细粒度信号） - 跳到指定步骤（time travel） - 失败时从某 checkpoint 重试

硬伤 4：节点 ⑤ 错误工具调用污染会话

@tool
def buggy_search(q: str) -> str:
    raise ValueError("API quota exceeded")

# 模型调 buggy_search，AgentExecutor 把异常 stringify 进 scratchpad
# 后续推理被"API quota exceeded"误导

异常进入 prompt 历史，模型可能反复尝试或彻底跑偏。没有节点级的"错误隔离"机制。

硬伤 5：节点 ⑥ 没法时间旅行

# 跑了 10 步，发现第 5 步走错了
# AgentExecutor 提供方式：完全重跑（贵 + 不确定）
# 想要的：从第 4 步的 state 改个参数继续
# AgentExecutor: ✗ 不支持
# LangGraph: ✓ 支持（详见 §5.7）

五大硬伤汇总

小结一行：LangChain 实现 ReAct 在每个节点都有可观察的硬伤，这五大硬伤就是 LangGraph 的设计动机。

§3.6 回调（Callbacks）系统

在基础流程中的位置：横切节点 ② / ④ / ⑤ 的观测手段。

Callbacks 是什么

不开 LangSmith 时，LangChain 提供本地的回调机制 —— 在每个组件运行时触发钩子。

内置事件

自定义 Callback Handler

from langchain_core.callbacks import BaseCallbackHandler

class TokenCountHandler(BaseCallbackHandler):
    def __init__(self):
        self.total_tokens = 0

    def on_chat_model_end(self, response, **kwargs):
        usage = response.llm_output.get("token_usage", {})
        self.total_tokens += usage.get("total_tokens", 0)

handler = TokenCountHandler()
result = executor.invoke({"input": "..."}, config={"callbacks": [handler]})
print(f"共消耗 {handler.total_tokens} tokens")

Callbacks vs LangSmith 的选型

小结一行：Callbacks 是 LangSmith 的"低配版"，能不能上云决定选哪个。

§3.7 错误处理：OutputParserException 捕获

在基础流程中的位置：节点 ③ 解析 + 节点 ⑤ 工具返回的稳定性。

三类常见错误

OutputParserException 处理

老 ReAct 字符串解析失败时（如模型输出 Action: 调用搜索）：

from langchain.agents import AgentExecutor

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=True,  # 自动捕获并把错误回填给模型让它重试
    # 或自定义：
    # handle_parsing_errors="请按格式重写：Action: <name>\nAction Input: <json>"
)

handle_parsing_errors 的策略： - True：用默认提示让模型修正 - str：把字符串作为修正提示 - Callable：自定义函数返回修正提示

工具错误处理

@tool
def safe_search(q: str) -> str:
    """搜索，失败时返回错误信息让模型决策"""
    try:
        return real_search(q)
    except Exception as e:
        return f"[搜索失败] {str(e)[:200]}"  # 截断防污染

# 工具内部捕获，不让框架捕获

自我修复解析器

from langchain.output_parsers import OutputFixingParser

parser = OutputFixingParser.from_llm(
    parser=PydanticOutputParser(pydantic_object=MySchema),
    llm=llm,
)
chain = prompt | llm | parser

解析失败时，OutputFixingParser 自动让 LLM 重写。

重试与回退（参考 §2.12）

chain_robust = chain.with_retry(
    retry_if_exception_type=(RateLimitError, APITimeoutError),
    stop_after_attempt=3,
).with_fallbacks([backup_chain])

小结一行：节点 ③⑤ 的错误处理三件套是 handle_parsing_errors + OutputFixingParser + with_retry/with_fallbacks，组合用最稳。

§3.8 业界目前怎么用 LangChain：迁出老接口

在基础流程中的位置：业界对 LangChain ReAct 的真实评估。

真实情况

业界案例对比

何时仍该用 LangChain（无 LangGraph）

小结一行：LangChain 仍在用，但场景缩小到"简单管道 / 组件库" —— Agent 工作迁到 LangGraph。

§3.9 从 LangChain 老接口迁移到 LangGraph 的官方迁移指南

在基础流程中的位置：节点 ③⑦ 的迁移路径。

官方迁移文档位置

LangGraph 文档站有专门的 langgraph/how-tos/migrate-from-langchain 章节。核心是把 AgentExecutor 替换成 LangGraph 的 create_react_agent。

迁移对照表

最小迁移示例

# 老
from langchain.agents import create_tool_calling_agent, AgentExecutor
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "查天气"})

# 新（一行）
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(llm, tools, prompt=prompt)
result = agent.invoke({"messages": [{"role": "user", "content": "查天气"}]})

迁移后立刻得到的好处

详见 §5.2 - §5.4。

小结一行：迁移官方提供一行替换，迁移后立刻拿到 checkpoint / HITL / 可视化四大好处。

§3.10 本章小结

§3.11 反模式速记

§3.12 术语速查

§3.13 推荐下一章

下一章：§4 LangGraph：第二代框架的心智模型 —— §3 暴露的五大硬伤逐节点对应 LangGraph 的核心改进。

§4 LangGraph：第二代框架的心智模型

§4.0 本章定位

§4.1 LangGraph 起源：2024 年 LangChain 团队的反思

在基础流程中的位置：直面 §3 暴露的五大硬伤。

起源故事

2024 年 1 月，LangChain 团队在博客里反思过去一年：

"We realized that for production agents, the chain-of-X mental model is not enough. Agents need to be graphs — with cycles, conditional branches, and explicit state."

直译：链式心智模型不够 —— 智能体需要图（有循环、有条件分支、有显式状态）。

五大硬伤 → 解法对照（Mermaid #8）

graph LR
    subgraph LC["LangChain 老 AgentExecutor 硬伤"]
        H1["⑥ 状态会话级 + 进程内"]
        H2["③ AgentExecutor 黑盒"]
        H3["⑦ 循环只有 max_iterations"]
        H4["⑤ 错误污染会话"]
        H5["⑥ 不能时间旅行"]
    end

    subgraph LG["LangGraph 解法"]
        S1["Checkpoint 持久化
+ Store API 跨会话"]
        S2["条件边显式可见"]
        S3["recursion_limit + interrupt
+ 条件边精细控制"]
        S4["节点内 try/except
+ retry 边"]
        S5["get_state_history
+ 分支重放"]
    end

    H1 -.->|节点 ⑥| S1
    H2 -.->|节点 ③| S2
    H3 -.->|节点 ⑦| S3
    H4 -.->|节点 ⑤| S4
    H5 -.->|节点 ⑥| S5

    style H1 fill:#fecaca,stroke:#ef4444
    style H2 fill:#fecaca,stroke:#ef4444
    style H3 fill:#fecaca,stroke:#ef4444
    style H4 fill:#fecaca,stroke:#ef4444
    style H5 fill:#fecaca,stroke:#ef4444
    style S1 fill:#d1fae5,stroke:#10b981
    style S2 fill:#d1fae5,stroke:#10b981
    style S3 fill:#d1fae5,stroke:#10b981
    style S4 fill:#d1fae5,stroke:#10b981
    style S5 fill:#d1fae5,stroke:#10b981

时间线

设计动机的根本

LangChain 的 AgentExecutor 是"管道 + while 循环"，本质是线性序列。但智能体的真实形态是图：

小结一行：LangGraph 是 LangChain 团队对"链式不够"的官方回应 —— 升级心智模型到图。

§4.2 核心模型转变：从"链"到"图"

在基础流程中的位置：基础流程的"形状"从线变成图。

链 vs 图的工程含义

状态图（State Graph）的工程含义

LangGraph 的核心抽象是状态图（State Graph，简称 SG）：

图思维的核心好处

小结一行：从链到图不是工程美学问题，是必要性问题 —— 真实智能体的拓扑需要循环和分支。

§4.3 三要素深入：状态 / 节点 / 边

在基础流程中的位置：State 对应节点 ⑥，Node 对应节点 ②④⑤，Edge 对应节点 ③⑦。

State：图的"共享内存"

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

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # 消息列表（自动追加）
    counter: int                                # 计数器（覆盖式更新）
    metadata: dict                              # 元数据（覆盖）

Node：状态变换函数

def my_node(state: AgentState) -> dict:
    # 读 state
    messages = state["messages"]
    # 做事
    new_msg = llm.invoke(messages)
    # 返回 state 的部分更新（不需要返回完整 state）
    return {"messages": [new_msg], "counter": state["counter"] + 1}

Edge：连接节点的"路径"

三要素与基础流程的对应

小结一行：State / Node / Edge 三要素一一对应基础流程的关键节点，把"管道 + while"换成"图 + 状态机"。

§4.4 超步并行执行模型（Pregel）

在基础流程中的位置：节点级别的并发执行机制。

Pregel 的来源

Pregel 是 Google 2010 年发表的图计算论文（用于 PageRank 这类大规模图算法）。LangGraph 借用了它的执行模型：

超步执行的工作流程

Superstep 1: 所有"被激活"节点并行执行 → 写 State 更新 → 屏障同步
Superstep 2: 上轮 State 更新激活下一批节点 → 并行执行 → 屏障同步
...
直到没有节点被激活，结束

工程含义：节点之间天然并行

与传统执行模型对比

Pregel 模型对开发者意味着什么

小结一行：Pregel 给 LangGraph 提供了"同步并行"的执行模型 —— 简单图也能拿到自动并发。

§4.5 五行最小图（Hello World）

在基础流程中的位置：用最小代码示例演示节点 ②⑥⑦⑧。

最小可运行图

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

# 1. 定义 State
class State(TypedDict):
    value: int

# 2. 定义节点函数
def add_one(state: State) -> dict:
    return {"value": state["value"] + 1}

# 3. 构建图（这就是 5 行核心）
graph = StateGraph(State)
graph.add_node("incrementer", add_one)
graph.add_edge(START, "incrementer")
graph.add_edge("incrementer", END)
app = graph.compile()

# 4. 调用
result = app.invoke({"value": 0})
print(result)  # {"value": 1}

三要素关系（Mermaid #9）

graph TB
    subgraph SG["StateGraph"]
        S["State
共享字典"]
        N1["Node A
读 state, 返回更新"]
        N2["Node B"]
        E1["Edge
静态 or 条件"]
    end

    Start([START]) --> N1
    N1 -->|更新写回| S
    S -->|读取| N2
    N2 -->|更新写回| S
    N1 --- E1 --- N2
    N2 --> End([END])

    style S fill:#fef3c7,stroke:#f59e0b,stroke-width:3px
    style N1 fill:#dbeafe,stroke:#3b82f6
    style N2 fill:#dbeafe,stroke:#3b82f6
    style E1 fill:#fce7f3,stroke:#ec4899

升级到带条件分支的图

def router(state: State) -> str:
    return "doubler" if state["value"] > 5 else "incrementer"

def double(state: State) -> dict:
    return {"value": state["value"] * 2}

graph = StateGraph(State)
graph.add_node("incrementer", add_one)
graph.add_node("doubler", double)
graph.add_conditional_edges(
    START,
    router,
    {"incrementer": "incrementer", "doubler": "doubler"},
)
graph.add_edge("incrementer", END)
graph.add_edge("doubler", END)
app = graph.compile()

add_conditional_edges 三个参数：源节点、路由函数（返回字符串）、字符串到节点名的映射。

节点函数的几种返回值类型

小结一行：5 行最小图揭示了 StateGraph 的三个核心 API（add_node / add_edge / compile） + 条件边 / Command 是稍进阶的工具。

§4.6 编译与可视化：StateGraph vs CompiledGraph

在基础流程中的位置：理解构建期和执行期的区别。

两个对象的区别

调用 compile() 后图变不可变，所有节点和边都被锁定，开始可执行。

CompiledGraph 的执行接口

config 参数的作用

result = app.invoke(
    {"value": 0},
    config={
        "configurable": {"thread_id": "user-123"},  # 检查点用
        "recursion_limit": 25,                       # 循环上限
        "tags": ["dev"],                             # LangSmith 标记
        "metadata": {"user_id": "abc"},
        "callbacks": [my_callback],                  # 回调
    },
)

可视化：自带 Mermaid 渲染

print(app.get_graph().draw_mermaid())
# 或保存图片
app.get_graph().draw_mermaid_png("graph.png")

输出可直接贴到 Mermaid 编辑器看图。调试和文档双用。

异步全栈支持

# 同步节点
def sync_node(state):
    ...

# 异步节点
async def async_node(state):
    result = await some_async_call()
    ...

# 一张图可以混用同步和异步节点，框架自动处理

调用方根据需要选 invoke 或 ainvoke：

推荐：节点尽量写成 async def，调 ainvoke —— 在 Web 服务里能拿满并发。

小结一行：compile() 是构建期到执行期的分界线，CompiledGraph 提供 6 种执行接口 + 可视化。

§4.7 业界目前怎么用 LangGraph：典型案例

在基础流程中的位置：业界采用真实数据。

Klarna：客服智能体

Replit：编程智能体

LinkedIn：招聘助手

Elastic / Uber：内部知识助手

共同模式

小结一行：业界采用 LangGraph 的真实场景集中在"多工具 + 长流程 + 高可控"的生产智能体。

§4.8 本章小结

§4.9 反模式速记

§4.10 术语速查

§4.11 推荐下一章

下一章：§5 用 LangGraph 实现 ReAct（同任务对比 §3） —— 把 §4 的概念落到代码，用 LangGraph 重写 §3 的同一个任务。

§5 用 LangGraph 实现 ReAct（同任务对比 §3）

§5.0 本章定位

§5.1 手写 ReAct 循环：State + 条件边

在基础流程中的位置：用 LangGraph 重做节点 ②③④⑤⑥⑦。

场景引入

§3 用 LangChain 实现了 ReAct 智能体（查天气 + 计算）。本节用 LangGraph 重写完全相同的任务，看代码差异。

完整代码：手写 ReAct 循环

"""LangGraph 手写 ReAct 智能体（与 §3.3 相同任务）"""

from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.prebuilt import ToolNode

# 1. 工具定义（节点 ④，跟 §3 完全相同）
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。"""
    fake_db = {"北京": "晴 25 度", "上海": "多云 22 度", "广州": "雨 28 度"}
    return fake_db.get(city, f"未找到 {city} 的天气")

@tool
def calculator(expression: str) -> str:
    """计算数学表达式。"""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"计算错误：{e}"

tools = [get_weather, calculator]

# 2. 大语言模型（节点 ②）+ 绑工具
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# 3. 定义 State（节点 ⑥）
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# 4. 定义节点函数
def call_model(state: AgentState) -> dict:
    """节点 ②：调用大语言模型"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# ToolNode 是 langgraph.prebuilt 提供的"调工具"节点（节点 ④⑤）
tool_node = ToolNode(tools)

# 5. 定义条件路由函数（节点 ③）
def should_continue(state: AgentState) -> str:
    """看最新消息是否有 tool_calls"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# 6. 构图
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    END: END,
})
graph.add_edge("tools", "agent")  # 节点 ⑦ 循环回去

app = graph.compile()

# 7. 调用
result = app.invoke({
    "messages": [HumanMessage(content="北京今天比上海温度高几度？")],
})

# 8. 输出（节点 ⑧）
for msg in result["messages"]:
    print(f"[{type(msg).__name__}]", msg.content[:100])

图的可视化（Mermaid #10）

graph TB
    Start([START]) --> Agent["agent 节点 ②
call_model"]
    Agent -->|tool_calls 存在| Tools["tools 节点 ④⑤
ToolNode"]
    Agent -->|无 tool_calls| End([END])
    Tools -->|节点 ⑦ 循环| Agent

    style Agent fill:#dbeafe,stroke:#3b82f6
    style Tools fill:#d1fae5,stroke:#10b981

这段代码与 §3.3 LangChain 实现的关键差异

小结一行：手写 LangGraph ReAct 把每个节点和路由都暴露出来，可控可调试 —— 但前提是写图时清晰。

§5.2 prebuilt create_react_agent

在基础流程中的位置：把 §5.1 的 30 行图压缩到 1 行。

一行版本

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    llm,
    tools,
    prompt="你是一个智能助手，可调用工具完成任务。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "北京今天比上海温度高几度？"}],
})

create_react_agent 的内部就是 §5.1

打开源码可以看到，create_react_agent 内部就是 §5.1 那张图（StateGraph + agent 节点 + tools 节点 + 条件边 + 循环）+ 一些便利封装。

何时用 prebuilt vs 自己手写

prebuilt 的可配置参数

agent = create_react_agent(
    model=llm,
    tools=tools,
    prompt="...",                      # 系统提示
    state_schema=CustomState,           # 自定义 State
    checkpointer=MemorySaver(),         # 检查点（详见 §5.3）
    interrupt_before=["tools"],         # 静态 breakpoint（详见 §5.4）
    interrupt_after=[],
    debug=True,
)

小结一行：create_react_agent 是"标准 ReAct 一行版"，不满足时改用手写图（见 §6 进阶）。

§5.3 检查点：三种保存器 + thread_id 续跑

在基础流程中的位置：节点 ⑥ 状态管理升级为持久化。

检查点（Checkpoint）解决什么

回顾 §3.5 硬伤 1：LangChain 的 Memory 是会话级 + 进程内。LangGraph 的检查点机制把 State 写到外部存储，让智能体跨进程、跨会话续跑。

三种内置 Saver

MemorySaver 示例

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
agent = create_react_agent(llm, tools, checkpointer=memory)

# 第一轮，给个 thread_id
config = {"configurable": {"thread_id": "user-123"}}
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "查北京天气"}]},
    config=config,
)

# 第二轮，同 thread_id，自动续上历史
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "再查上海"}]},
    config=config,
)
# result2 的 messages 包含第一轮所有上下文

Postgres 生产级示例

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

DB_URI = "postgresql://user:pass@localhost:5432/checkpoints"
pool = ConnectionPool(DB_URI)

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 第一次跑建表

    agent = create_react_agent(llm, tools, checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "user-123"}}
    result = agent.invoke({"messages": [...]}, config=config)

setup() 会在 Postgres 中创建 4 张表： - checkpoints：State 快照 - checkpoint_writes：节点写入记录 - checkpoint_blobs：大字段二进制 - checkpoint_migrations：版本管理

Checkpoint 写入时序（Mermaid #11）

sequenceDiagram
    participant U as 用户
    participant A as Agent (CompiledGraph)
    participant N1 as Node A
    participant N2 as Node B
    participant CP as Checkpointer

    U->>A: invoke(input, thread_id=X)
    A->>CP: load(thread_id=X) 是否有历史？
    CP-->>A: 有，返回历史 State

    Note over A: 超步 1 开始

    A->>N1: 调用 (state)
    N1-->>A: 返回 update
    A->>CP: save(checkpoint_1)

    Note over A: 超步 2 开始

    A->>N2: 调用 (state)
    N2-->>A: 返回 update
    A->>CP: save(checkpoint_2)

    A-->>U: 返回最终 state

thread_id 的语义

config 与 configurable 运行时

config 是每次调用时传的运行时配置：

result = agent.invoke(
    input,
    config={
        "configurable": {
            "thread_id": "user-123",       # checkpoint 用
            "checkpoint_ns": "weather",     # 命名空间（多任务时分开）
            "user_id": "abc",               # 自定义字段，节点里可读
        },
        "recursion_limit": 50,             # 节点 ⑦ 循环上限
        "tags": ["prod"],
    },
)

# 节点里读 config
def my_node(state, config: RunnableConfig):
    user_id = config["configurable"]["user_id"]
    ...

小结一行：Checkpoint + thread_id 让节点 ⑥ 从"会话内"升级到"跨会话持久化"，三种 Saver 按规模选。

§5.4 人在回路：interrupt() 与 NodeInterrupt

在基础流程中的位置：节点 ⑦ 循环控制 + 节点 ④ 工具调用前的人工审批。

为什么需要 HITL

静态 breakpoint：interrupt_before / interrupt_after

最简单的 HITL —— 在编译图时声明"工具节点前必须暂停"：

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
agent = create_react_agent(
    llm, tools,
    checkpointer=memory,
    interrupt_before=["tools"],   # 工具节点前暂停
)

config = {"configurable": {"thread_id": "user-123"}}

# 第一次 invoke（会停在 tools 节点前）
result = agent.invoke(
    {"messages": [{"role": "user", "content": "退款 1000 元"}]},
    config=config,
)
# result 包含到目前为止的 state，但 tools 还没跑

# 人工审批
state = agent.get_state(config)
if state.next == ("tools",):
    print(f"待执行的工具：{state.values['messages'][-1].tool_calls}")
    user_approve = input("批准？(y/n): ")

    if user_approve == "y":
        # 继续：传 None 表示"从中断处继续"
        agent.invoke(None, config=config)
    else:
        # 改 state：拒绝并回复
        agent.update_state(
            config,
            {"messages": [ToolMessage(content="用户拒绝", tool_call_id=...)]},
        )
        agent.invoke(None, config=config)

动态中断：interrupt() 函数

LangGraph 0.2 起支持节点内动态中断（不必在编译时静态声明）：

from langgraph.types import interrupt, Command

def review_node(state):
    last_msg = state["messages"][-1]
    if "退款" in last_msg.content:
        # 动态中断，把要审批的内容传给客户端
        approval = interrupt({"action": "退款审批", "amount": 1000})
        if approval == "approved":
            return {"messages": [ToolMessage(content="已退款", ...)]}
        else:
            return {"messages": [ToolMessage(content="拒绝退款", ...)]}
    return {}

# 客户端继续：用 Command(resume=...)
agent.invoke(Command(resume="approved"), config=config)

暂停-恢复时序（Mermaid #12）

sequenceDiagram
    participant U as 用户
    participant App as Web App
    participant A as Agent
    participant CP as Checkpointer

    U->>App: 提交退款请求
    App->>A: invoke(input, thread_id=X)
    A->>A: 跑到工具节点前
    A->>CP: save(state with next=tools)
    A-->>App: 返回 (status=interrupted)
    App-->>U: 显示"等待审批"

    Note over U: 用户点击"批准"

    U->>App: 批准
    App->>A: invoke(None, config) 恢复
    A->>CP: load(state)
    A->>A: 跑 tools 节点
    A->>A: 跑 agent 节点（合成回答）
    A-->>App: 返回最终结果
    App-->>U: "已退款 1000 元"

三种 HITL 模式

小结一行：HITL 三件套（静态 breakpoint / 动态 interrupt / update_state）覆盖"暂停 - 审批 - 修正"全场景。

§5.5 与 §3 LangChain 实现的逐行对比表

在基础流程中的位置：把两种实现拉通看每个节点的差异。

同任务两种实现

总行数：LangChain ~50 行 / LangGraph ~60 行（因为节点 ②③ 显式化）

各节点的"可控性"对比

调试体感对比