01 · LangGraph 概览：什么是"图"？

🧠 一句话总结

LangGraph 是一个用"有向图"来编排 AI Agent 工作流的框架。

如果把 LangChain 比作"链"（Chain）——一条流水线从头走到尾——那 LangGraph 就是"图"（Graph）——允许分支、循环、条件跳转和暂停等待。

---

🎮 在我们的 Demo 中

打开 src/graph.py，你会看到这样的结构：

进入房间 → 检查房间类型 → 战斗？/ 解谜？/ 捡物品？
↑              ↓
└──── 回到房间 ←┘
↓
等待玩家输入
↓
移动 / 攻击 / 逃跑
↓
再次进入房间 ...

这比线性的 if-else 链灵活得多：你可以从任意节点跳转到任意节点，还可以在中途暂停等待输入。

---

🏗️ 图的核心组成
┌──────────────────────────────────────────────────┐
│              LangGraph 图的五要素                  │
├────────────┬─────────────────────────────────────┤
│ State      │ 所有节点共享的状态字典               │
│            │ → 看 src/state.py                    │
├────────────┼─────────────────────────────────────┤
│ Nodes      │ 处理函数，接收 state，返回 partial    │
│            │ → 看 src/nodes.py                    │
├────────────┼─────────────────────────────────────┤
│ Edges      │ 固定边（A→B）和条件边（A→?→B/C/D）   │
│            │ → 看 src/graph.py                    │
├────────────┼─────────────────────────────────────┤
│ Entry/Exit │ 图的起点和终点                       │
│            │ → SET_START / END                    │
├────────────┼─────────────────────────────────────┤
│ Checkpoint │ 持久化状态，支持中断恢复             │
│            │ → MemorySaver / SqliteSaver          │
└────────────┴─────────────────────────────────────┘

---

📊 与普通程序的对比

传统程序	LangGraph
函数调用函数	图通过"边"连接节点
状态通过参数传递	状态是全局共享的字典
控制流用 if/while	条件边决定路由
无法中途暂停	interrupt() 暂停等待输入
重启需要手动保存	checkpoint 自动持久化

---

🔑 关键理念

1. 节点是"纯函数"

每个节点只做一件事：读 state，返回 partial update。它不调用下一个节点——图结构决定谁调用谁。

# nodes.py: enter_room
def enter_room(state: GameState) -> dict:
"""展示房间信息，不修改状态"""
room_id = state.get("current_room", "entrance")
room = get_room(room_id)
room_text = _format_room(room_id, state)
return {"messages": [("ai", room_text)]}
# 注意：这里没有 "接下来调用 XXX" 的逻辑

2. 路由与计算分离

“做什么” 由节点决定；“去哪里” 由条件边决定。这种分离让代码更容易测试和修改。

3. Human-in-the-Loop 是一等公民

interrupt() 不是 hack，而是 LangGraph 的核心机制。它允许图在任何位置暂停，等待外部输入后再继续。

---

🔍 查看关键源码

文件	作用
src/graph.py	build_adventure_graph() 函数——组装整张图
src/state.py	GameState TypedDict——状态定义
src/nodes.py	所有节点函数
src/main.py	run_game() 主循环——invoke/resume 交互

---

💭 启发式问题

1. 为什么 LangGraph 选择"图"而不是"链"作为编排模型？什么场景下图优于链？
2. 如果把 enter_room() 改成直接调用 initiate_combat()，会带来什么问题？
3. 状态（State）和普通 Python 函数参数有什么区别？为什么图需要共享状态？
4. 你能否想到一个现实中适合用 LangGraph 建模的业务流程？（提示：审批流、客服机器人、多步骤数据管道）

---

👉 下一步：02 · 状态管理详解

02 · 状态管理：LangGraph 的"记忆"

🧠 一句话总结

State 是图中所有节点共享的"记忆"。每个节点读取它、修改它，图引擎负责合并更新。

---

📋 我们的 State 定义
打开
src/state.py
，核心定义如下：
from typing import Annotated, List, TypedDict
from langgraph.graph.message import add_messages
class GameState(TypedDict, total=False):
# 使用 add_messages reducer：追加而非覆盖
messages: Annotated[list, add_messages]
# 普通字段：覆盖策略
current_room: str
player_hp: int
max_hp: int
player_attack: int
inventory: List[str]
combat_active: bool
monster_hp: int
monster_attack: int
monster_name: str
solved_puzzles: List[str]
game_over: bool
game_won: bool
pending_decision: bool

---

🔑 关键概念拆解

1. TypedDict —— 类型安全的状态字典

class GameState(TypedDict, total=False):
current_room: str
player_hp: int

TypedDict 是 Python 的类型提示机制，让你在写状态时获得 IDE 自动补全和类型检查。

total=False 表示所有字段都是可选的——这在 return {} 时不会报错。

2. Annotated + Reducer —— 控制合并策略

这是 LangGraph 最重要的设计之一。

messages: Annotated[list, add_messages]

当两个节点都想修改 messages 时，LangGraph 怎么合并？

• 普通字段：后写的覆盖先写的

• Annotated[list, add_messages]

  ：追加不覆盖

这就像 Git 的 merge strategy：

• 普通字段 = git merge --strategy=ours（直接用新版本）
• add_messages = git merge --strategy=union（两边都保留）

3. 节点如何修改 State

每个节点返回一个 partial dict，图引擎负责合并：

def player_attack(state: GameState) -> dict:
# 读取完整 state
monster_hp = state.get("monster_hp", 0)
player_atk = state.get("player_attack", 5)
# 计算新值
damage = player_atk + random.randint(-2, 3)
new_monster_hp = monster_hp - damage
# 返回 partial update
# 图引擎会将这个 dict 合并到全局 state
return {
"monster_hp": new_monster_hp,          # 覆盖
"messages": [("ai", f"造成 {damage} 点伤害")],  # 追加
}

你不需要返回整个 state，只需要返回你修改的部分。

---

🎮 在我们的游戏中

游戏的所有数据都在 GameState 中流转：

e

nter_room → 读取 current_room，写入 messages
↓
initiate_combat → 设置 combat_active=True, monster_hp=20
↓
player_attack → 递减 monster_hp, 递减 player_hp
↓
enter_room → 重新读取 current_room（没变），展示新状态

---

🧪 实验：修改 Reducer

试试把 inventory: List[str] 改成：

from operator import add
inventory: Annotated[List[str], add]

这样两个节点都往背包里加东西时，它们会自动合并（类似列表 extend），而不是一个覆盖另一个。

---

💭 启发式问题

1. 为什么 messages 要用 add_messages reducer 而不是普通覆盖？如果改成覆盖会发生什么？

2. 如果你自己写一个自定义 reducer（比如"取最大值"），它会在什么场景下有用？

3. TypedDict(total=False)

   中 total=False 的作用是什么？设为 True 会怎样？

4. 如果两个节点同时修改 player_hp（一个 +10，一个 -5），最终结果是多少？LangGraph 的默认行为是什么？

5. State 的这种"全局共享"设计，在多 Agent 协作场景下有什么优势和风险？

---

👉 下一步：03 · 节点与边

💭 答案：02-状态管理-答案

03 · 节点与边：图的"肉"与"骨"

🧠 一句话总结

节点是图的"计算单元"，边是图中的"路径"。节点不决定去哪儿，边决定。

---

🔲 节点（Node）：做一件事，把它做好

节点签名

所有节点函数都遵循同样的签名：

def node_name(state: GameState) -> dict:
# 1. 读取 state
# 2. 执行逻辑
# 3. 返回 partial state update
return {"field": new_value, "messages": [...], ...}

输入：完整的 GameState

输出：一个 dict，表示你想修改的状态字段

不做的：不直接调用其他节点，不决定下一步去哪儿

我们的节点一览

节点	文件位置	功能
enter_room	nodes.py:76	展示房间信息
pickup_items	nodes.py:89	拾取地上物品
initiate_combat	nodes.py:108	初始化战斗状态
player_attack	nodes.py:132	执行攻击回合
flee_combat	nodes.py:195	尝试逃跑
solve_puzzle	nodes.py:213	展示谜题
move_player	nodes.py:246	移动到新房间
game_over_screen	nodes.py:274	展示结局
wait_for_input	graph.py:143	Human-in-the-Loop 暂停点

节点设计原则

✅ 好的节点                           ❌ 不好的节点
─────────────────────────           ─────────────────────
单一职责，只做一件事                 一个节点做三件事
不调用其他节点                      节点里 if-else 调不同节点
返回 partial update                 返回完整 state（浪费）
不关心"接下来去哪儿"                硬编码路由逻辑

---

🔗 边（Edge）：连接世界的线

固定边 vs 条件边
# ── 固定边：A 总是去 B ──
graph.add_edge("pickup_items", "enter_room")
# pickup_items 执行完 → 一定进入 enter_room
# ── 条件边：A 根据 state 决定去 B/C/D ──
graph.add_conditional_edges(
"enter_room",              # 起点
route_from_room,            # 路由函数
{
"initiate_combat": "initiate_combat",   # route 返回这个 → 去这个节点
"solve_puzzle": "solve_puzzle",
"pickup_items": "pickup_items",
"enter_room": "enter_room",
},
)
路由函数深入
打开
src/graph.py:36
，看
route_from_room()
：
def route_from_room(state: GameState) -> Literal["initiate_combat", ...]:
room_id = state.get("current_room", "")
room = get_room(room_id)
if not room:
return "enter_room"
room_type = room.room_type
if room_type in ("combat", "boss") and room.monster_id:
if not state.get("combat_active", False):
return "initiate_combat"    # 👈 触发战斗
if room_type == "puzzle" and room.puzzle_id:
if room.puzzle_id not in state.get("solved_puzzles", []):
return "solve_puzzle"       # 👈 触发解谜
if room_type == "treasure" and room.items:
if any(item not in state.get("inventory", []) for item in room.items):
return "pickup_items"       # 👈 触发捡物品
return "enter_room"                 # 默认展示房间

这个函数就是"游戏的大脑"——它根据房间类型和当前状态，决定玩家会经历什么。

---

🎨

图的可视化结构
┌──────────┐
│ enter_room│ ← 起点，也是枢纽
└─────┬────┘
│ route_from_room()
┌─────────────┼─────────────┐
▼             ▼             ▼
┌────────────┐ ┌──────────┐ ┌────────────┐
│initiate_   │ │solve_    │ │pickup_     │
│combat      │ │puzzle    │ │items       │
└─────┬──────┘ └────┬─────┘ └─────┬──────┘
│             │             │
└─────────────┼─────────────┘
│ 固定边
▼
┌──────────┐
│ enter_room│ ← 循环回到枢纽
└─────┬────┘
│ route_after_enter()
┌─────┴─────┐
▼           ▼
END        wait_for_input
│
(外部输入)
│
_route_user_input()
┌───┬───┬───┬───┐
▼   ▼   ▼   ▼   ▼
move attack flee puzzle ...

---

💭 启发式问题

1. 为什么节点不应该调用下一个节点？如果你在 enter_room 里直接调用 initiate_combat()，会出现什么问题？
2. 条件路由函数必须是纯函数吗？如果 route_from_room() 内部修改了 state，会发生什么？
3. 你能想到哪些场景适合用固定边，哪些适合用条件边？给出实际例子。
4. 如果图中有 20 个节点和 50 条边，你会如何测试这张图的正确性？
5. 假如你想加一个"中毒"机制（每走一步扣 1 HP），应该在哪里加？是加节点还是改路由？

---

👉 下一步：04 · 条件路由

💭 答案：03-节点与边-答案

04 · 条件路由：让图学会"选择"

🧠 一句话总结

条件路由是 LangGraph 实现分支逻辑的机制。条件路由函数根据 state 返回一个字符串，图引擎根据这个字符串选择下一个节点。

---

🔀 在我们的 Demo 中

整个游戏的控制流都是条件路由驱动的。打开 src/graph.py，你会看到三个关键路由函数：

1. 房间路由：
route_from_room(state)
def route_from_room(state: GameState) -> str:
room = get_room(state.get("current_room", ""))
room_type = room.room_type
if room_type == "combat":
return "initiate_combat"     # → 战斗
elif room_type == "puzzle":
return "solve_puzzle"        # → 解谜
elif room_type == "treasure":
return "pickup_items"        # → 捡物品
else:
return "enter_room"          # → 展示房间
2. 战斗后路由：
route_after_combat(state)
def route_after_combat(state: GameState) -> str:
if state.get("game_over"):
return "game_over"           # → 游戏结束画面
return "enter_room"              # → 继续探索
3. 用户输入路由：
_route_user_input(state)
def _route_user_input(state: GameState) -> str:
# 读取最后一条用户消息
user_input = get_last_user_message(state)
if combat_active:
return "player_attack" if "攻击" in user_input else "flee_combat"
if "北" in user_input:
return "move_player"
...

---

🎯 条件路由的核心模式

┌──────────┐
│  节点 A   │
└─────┬────┘
│
┌─────▼──────┐
│ route(state)│  ← 纯函数，读 state，返回字符串
└──┬──┬──┬───┘
│  │  │
"a"  ▼  ▼  ▼ "c"
│ "b" │
┌────┐ ┌─┐ ┌────┐
│节点B│ │C│ │节点D│
└────┘ └─┘ └────┘

路由函数必须是纯函数：只读 state，不修改 state。如果你在路由里改了 state，图引擎不会感知到。

---

🧠 设计模式：用路由替代 if-else

对比传统代码和 LangGraph 的条件路由：

# ❌ 传统方式：硬编码控制流
def handle_room(room):
if room.type == "combat":
state = initiate_combat(state)
state = player_attack(state)
elif room.type == "puzzle":
state = solve_puzzle(state)
state = enter_room(state)
# ✅ LangGraph 方式：图定义控制流，节点定义计算
graph.add_conditional_edges("enter_room", route_from_room, {
"initiate_combat": "initiate_combat",
"solve_puzzle": "solve_puzzle",
"pickup_items": "pickup_items",
})

LangGraph 的方式让：

• 流程可视化（图就是文档）
• 节点可独立测试
• 路由可独立修改

---

🌲 嵌套路由：路由链式组合

在我们的图中，条件路由可以串联：

enter_room
│
├─ route_from_room() → combat / puzzle / treasure / enter
│                              │
│                         initiate_combat
│                              │
│                         (固定边)
│                              │
│                         enter_room ← 回到枢纽
│                              │
└─ route_after_enter() → END / wait_for_input
│
_route_user_input()
│
move / attack / flee / ...

这种"回到枢纽"再重新路由的模式，在 LangGraph 中非常常见——就像游戏主循环。

---

💭 启发式问题

1. 路由函数为什么必须是纯函数？如果路由函数修改了 state，会出现什么后果？

2. 什么是"循环边"？在我们的 Demo 中，enter_room → enter_room 会形成循环，什么情况下这样设计是合理的？什么情况下你应该避免？

3. 假如你想实现"玩家生命低于 10 时自动使用治疗药水"，应该在哪里加这个逻辑？路由函数中？还是节点中？

4. add_conditional_edges

   的第三个参数（路由表）要求你列出所有可能的返回值。如果你漏写了一个，会发生什么？

5. 如果你有 5 个不同的战斗节点（近战/远程/魔法/道具/防御），条件路由函数会变得多复杂？你如何管理这种复杂度？

---

👉 下一步：05 · 工具集成

💭 答案：04-条件路由-答案

05 · 工具集成：给 Agent 装上"手和脚"

🧠 一句话总结

工具（Tools）是 Agent 与外部世界交互的接口。LangGraph 中的工具通过 @tool 装饰器定义，可以被 Agent 节点调用。

---

🔧 我们的工具定义

打开 src/tools.py，我们定义了 4 个工具：

from langchain_core.tools import tool
@tool
def roll_dice(sides: int = 20, times: int = 1) -> str:
"""🎲 掷骰子，返回结果。"""
results = [random.randint(1, sides) for _ in range(times)]
total = sum(results)
return f"🎲 D{sides}: {results} 总和 = {total}"
@tool
def calculate_damage(base_attack: int, dice_result: int,
has_advantage: bool = False) -> str:
"""⚔️ 计算战斗伤害。"""
...
@tool
def use_item(item_name: str, inventory: List[str]) -> str:
"""🧪 使用背包中的物品。"""
...
@tool
def check_inventory(inventory: List[str]) -> str:
"""🎒 查看背包内容。"""
...
🎯
@tool
装饰器做了什么？
@tool
def roll_dice(sides: int = 20, times: int = 1) -> str:
"""🎲 掷骰子，返回结果。"""

这个装饰器自动做了几件事：

1. 提取函数签名

   ：参数名、类型、默认值

2. 提取 docstring

   ：作为工具的描述（LLM 会读到）

3. 包装为 Tool 对象

   ：可以被 ToolNode 调用

4. 生成 JSON Schema

   ：供 LLM Function Calling 使用

---

🔗 ToolNode：批量执行工具

from langgraph.prebuilt import ToolNode
ALL_TOOLS = [roll_dice, calculate_damage, use_item, check_inventory]
tool_node = ToolNode(ALL_TOOLS)
graph.add_node("tools", tool_node)

ToolNode 会：

1. 读取 state 中 LLM 产生的 tool_calls
2. 并行/串行调用对应的工具函数
3. 将工具返回结果写入 state

---

🧪 在我们的 Demo 中的工具使用

虽然我们的 Demo 主要用纯节点函数实现游戏逻辑（因为游戏规则是确定性的），但工具在 LangGraph 中的典型用法是：

用户输入 → Agent 节点（LLM 决策）
↓
需要调用工具？
↙        ↘
是            否
↓             ↓
ToolNode       直接回复
↓
工具结果
↓
Agent 节点（LLM 整合结果）
↓
最终回复给用户

在我们的游戏中，假如你接入了 LLM，roll_dice 和 calculate_damage 就会由 LLM 自动判断何时调用：

玩家：「我要攻击地精！」
LLM：  "这是个战斗动作，我需要 roll_dice(D20)"
→ 调用工具 roll_dice(20)
→ 得到结果：15
→ "好的一掷！接下来计算伤害..."
→ 调用工具 calculate_damage(5, 15)
→ "造成了 7 点伤害！地精反击..."

---

🎮 工具 vs 直接调用函数

维度	直接调用	通过 Tool
调用者	代码中明确写	LLM 自主决定
时机	编译时确定	运行时动态
灵活性	低	高
可观测性	日志	Tool 调用记录在 state 中
适用场景	确定性逻辑	需要 LLM 判断的场景

---

💭 启发式问题

1. @tool

   装饰器和普通 def 在 LangGraph 中的行为有什么区别？不使用 @tool 的函数能否被 ToolNode 调用？

2. 如果工具执行失败（比如抛异常），LangGraph 会如何处理？你需要在哪里加错误处理？

3. 在一个 Agent 图中有 10 个工具时，LLM 如何选择调用哪个？提示工程在其中扮演什么角色？

4. 工具的 docstring 有多重要？如果 roll_dice 的 docstring 写得很模糊，LLM 可能做出什么错误判断？

5. 假如你想让工具之间相互调用（比如 use_item 内部调用 roll_dice），这在 LangGraph 中是好做法吗？

---

👉 下一步：06 · Human-in-the-Loop

💭 答案：05-工具-答案

06 · Human-in-the-Loop：让人参与决策

🧠 一句话总结

Human-in-the-Loop (HITL) 是 LangGraph 让人类在 Agent 执行过程中介入的机制。通过 interrupt() 暂停图执行，等待外部输入后恢复。

---

🎮 在我们的 Demo 中

这是整个游戏最核心的交互模式。打开 src/graph.py：

def wait_for_input(state: GameState) -> dict:
"""⏸️ 暂停节点 —— 等待玩家输入。"""
user_input = interrupt("请选择你的动作：")
return {
"messages": [("user", user_input)],
"pending_decision": False,
}

而在 src/main.py 的主循环中：

# 1️⃣ 首次 invoke：图运行到 interrupt() 暂停
current_state = app.invoke(INITIAL_STATE, config)
while True:
# 2️⃣ 显示图暂停位置的输出
for msg in current_state.get("messages", []):
print_message(msg)
# 3️⃣ 获取用户输入
user_input = input("> ")
# 4️⃣ 恢复执行：传入用户输入
current_state = app.invoke(
Command(resume=user_input),    # 👈 这就是"恢复"
config,
)

---

🔄 完整流程

┌──────────────────────────────────────────────────┐
│                 LangGraph 图内部                   │
│                                                    │
│  enter_room → route → ... → wait_for_input         │
│                                     │               │
│                              interrupt("请选择")     │
│                                     │               │
└─────────────────────────────────────┼───────────────┘
│ 图暂停
══════════════════════════════════════╪═══════════════
│ 外部世界
┌─────────────────────────────────────┼──────────────┐
│            main.py 主循环            │               │
│                                      ▼               │
│         显示消息（图暂停位置的输出）                  │
│         等待用户输入：input("> ")                    │
│         用户输入 "北"                                │
│         Command(resume="北")                        │
│                        │                            │
└────────────────────────┼────────────────────────────┘
│
══════════════════════════╪════════════════════════════
│ 恢复执行
┌────────────────────────┼────────────────────────────┐
│                 LangGraph 图内部                     │
│                        ▼                             │
│         user_input = "北" (interrupt 的返回值)       │
│         → route → move_player → enter_room           │
│                          │                           │
│                   interrupt("请选择...")  ← 再次暂停  │
└──────────────────────────┼───────────────────────────┘
│
循环往复...

---

🧩 Command(resume=...) 详解

from langgraph.types import Command
# 恢复执行，interrupt() 返回 "北"
app.invoke(Command(resume="北"), config)
# 除了 resume，Command 还可以同时更新 state：
app.invoke(Command(resume="北", update={"player_hp": 50}), config)

Command 是一个多用途指令：

• resume

  ：传给 interrupt() 的返回值

• update

  ：在恢复前直接修改 state

• goto

  ：跳转到指定节点（高级用法）

---

🆚 对比：有 HITL vs 无 HITL

无 HITL：                         有 HITL：
图开始                            图开始
↓                                 ↓
Agent 决策                        Agent 分析
↓                                 ↓
调用工具                          interrupt() 暂停
↓                                 ↓
Agent 回复                      👤 人类审核
↓                                 ↓
图结束                          批准/拒绝/修改
↓
继续执行
↓
图结束

---

🎯 实际应用场景

场景	暂停点
审批流程	关键决策前等待批准
客服机器人	遇到无法回答的问题时转人工
代码审查	AI 生成代码后等待人类审查
数据标注	AI 标注后人工抽查确认
敏感操作	删除/支付前等待确认

---

💭 启发式问题

1. interrupt()

   返回后，图是如何知道"从哪里继续"的？背后的机制是什么？

2. 如果 interrupt() 暂停后，外部程序没有调用 Command(resume=...) 而是重新 invoke()，会发生什么？

3. 如何在一次 invoke 中实现多次暂停？（提示：interrupt() 可以在图中多处使用）

4. Command

   的 update 参数允许在恢复时直接修改 state。这在什么场景下有用？有什么风险？

5. 在人机交互模式下，如果用户输入了一个无效命令，你应该在 main.py 中过滤还是在图内部处理？为什么？

---

👉 下一步：07 · 检查点与持久化

💭 答案：06-Human-in-the-Loop-答案

07 · 检查点与持久化：图不会"失忆"

🧠 一句话总结

Checkpoint 是 LangGraph 的"存档系统"——每次图执行后自动保存状态快照，支持时光回溯和断点续传。

---

💾 在我们的 Demo 中

打开 src/graph.py，build_adventure_graph() 的最后：

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
compiled = graph.compile(checkpointer=memory)
而在
src/main.py
中：
thread_id = str(uuid.uuid4())

config = {
"configurable"
: {
"thread_id"
: thread_id}}
# 每次 invoke 都用同一个 thread_id
current_state = app.invoke(INITIAL_STATE, config)
# ... 用户交互 ...
current_state = app.invoke(Command(resume=user_input), config)
🔑 检查点的核心概念
1. Thread ID —— 存档槽位
config = {"configurable": {"thread_id": "game-slot-1"}}
每一个
thread_id
就是一个独立的"存档槽"。不同的 thread_id 之间状态完全隔离。
就像 RPG 游戏的多个存档位。
2. 自动保存
每次图执行（无论成功还是中断），LangGraph 自动保存状态快照。你不需要手动调用
save()
。
3. 时光回溯
# 回到上一个检查点
previous_state = app.get_state(config).values
# 查看所有历史状态
history = list(app.get_state_history(config))
for snapshot in history:
print(snapshot.values["current_room"])
🎮 在我们的游戏中的应用
存档功能
（目前未实现，但很容易加）：
# 保存：只需记住 thread_id
save_slot = thread_id  # "abc123"
# 读取：用同一个 thread_id 继续
config = {"configurable": {"thread_id": save_slot}}
current_state = app.get_state(config).values
print(f"你上次在: {current_state['current_room']}")
死亡后重试
：
# 查看历史，找到死亡前的状态
history = list(app.get_state_history(config))
last_alive = None
for snapshot in history:
if not snapshot.values.get("game_over"):
last_alive = snapshot
break
# 从那个状态恢复
if last_alive:
app.invoke(Command(goto=last_alive.next[0]), config)  # 跳回

---

📊 不同的 Checkpointer 实现

实现	存储位置	适用场景
MemorySaver	内存	开发/测试（重启丢失）
SqliteSaver	SQLite 文件	本地持久化
PostgresSaver	PostgreSQL	生产环境

# SQLite 持久化

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
conn = sqlite3.connect("adventure_saves.db", check_same_thread=False)
memory = SqliteSaver(conn)
compiled = graph.compile(checkpointer=memory)

---

⌛ 检查点的内部结构

每个检查点保存：

Checkpoint:
├── values: GameState          # 完整的当前状态
├── next: ("enter_room",)      # 下一步要执行的节点
├── metadata:
│   ├── source: "loop"         # 触发来源
│   ├── step: 12               # 执行步数
│   └── writes: {...}          # 本次写入
└── parent_checkpoint_id       # 父检查点（形成链）

---

🧠 高级话题：Channel 和 Pregel

LangGraph 的持久化底层基于 Pregel 模型（Google 的图计算框架）：

• 每个状态字段都是一个 Channel
• 每次节点写入，数据流入 Channel
• Reducer 决定多个写入如何合并
• Checkpoint 保存所有 Channel 的当前快照

这保证了并发场景下的一致性。

---

💭 启发式问题

1. 如果 MemorySaver 在程序重启后数据丢失，在什么场景下这反而是有利的？

2. thread_id

   冲突会怎样？如果两个用户意外使用了相同的 thread_id，会发生什么？

3. 检查点会保存完整的状态还是增量？如果 state 中有大对象（如整个网页内容），会有什么性能影响？

4. 如果你想让用户"回到上一轮战斗前"，应该用什么 API？如何实现一个"撤销"功能？

5. 在生产环境中，你会选择 SqliteSaver 还是 PostgresSaver？考虑并发、性能和运维。

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】