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.pybuild_adventure_graph()函数——组装整张图src/state.pyGameStateTypedDict——状态定义src/nodes.py所有节点函数src/main.pyrun_game()主循环——invoke/resume 交互 启发式问题为什么 LangGraph 选择图而不是链作为编排模型什么场景下图优于链如果把enter_room()改成直接调用initiate_combat()会带来什么问题状态State和普通 Python 函数参数有什么区别为什么图需要共享状态你能否想到一个现实中适合用 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, totalFalse): # 使用 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, totalFalse): current_room: str player_hp: intTypedDict是 Python 的类型提示机制让你在写状态时获得 IDE 自动补全和类型检查。totalFalse表示所有字段都是可选的——这在return {}时不会报错。2. Annotated Reducer —— 控制合并策略这是 LangGraph 最重要的设计之一。messages: Annotated[list, add_messages]当两个节点都想修改messages时LangGraph 怎么合并普通字段后写的覆盖先写的Annotated[list, add_messages]追加不覆盖这就像 Git 的 merge strategy普通字段 git merge --strategyours直接用新版本add_messages git merge --strategyunion两边都保留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中流转enter_room → 读取 current_room写入 messages ↓ initiate_combat → 设置 combat_activeTrue, monster_hp20 ↓ player_attack → 递减 monster_hp, 递减 player_hp ↓ enter_room → 重新读取 current_room没变展示新状态 实验修改 Reducer试试把inventory: List[str]改成from operator import add inventory: Annotated[List[str], add]这样两个节点都往背包里加东西时它们会自动合并类似列表 extend而不是一个覆盖另一个。 启发式问题为什么messages要用add_messagesreducer 而不是普通覆盖如果改成覆盖会发生什么如果你自己写一个自定义 reducer比如取最大值它会在什么场景下有用TypedDict(totalFalse)中totalFalse的作用是什么设为True会怎样如果两个节点同时修改player_hp一个 10一个 -5最终结果是多少LangGraph 的默认行为是什么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_roomnodes.py:76展示房间信息pickup_itemsnodes.py:89拾取地上物品initiate_combatnodes.py:108初始化战斗状态player_attacknodes.py:132执行攻击回合flee_combatnodes.py:195尝试逃跑solve_puzzlenodes.py:213展示谜题move_playernodes.py:246移动到新房间game_over_screennodes.py:274展示结局wait_for_inputgraph.py:143Human-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 ... 启发式问题为什么节点不应该调用下一个节点如果你在enter_room里直接调用initiate_combat()会出现什么问题条件路由函数必须是纯函数吗如果route_from_room()内部修改了 state会发生什么你能想到哪些场景适合用固定边哪些适合用条件边给出实际例子。如果图中有 20 个节点和 50 条边你会如何测试这张图的正确性假如你想加一个中毒机制每走一步扣 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 中非常常见——就像游戏主循环。 启发式问题路由函数为什么必须是纯函数如果路由函数修改了 state会出现什么后果什么是循环边在我们的 Demo 中enter_room → enter_room会形成循环什么情况下这样设计是合理的什么情况下你应该避免假如你想实现玩家生命低于 10 时自动使用治疗药水应该在哪里加这个逻辑路由函数中还是节点中add_conditional_edges的第三个参数路由表要求你列出所有可能的返回值。如果你漏写了一个会发生什么如果你有 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: 掷骰子返回结果。这个装饰器自动做了几件事提取函数签名参数名、类型、默认值提取 docstring作为工具的描述LLM 会读到包装为 Tool 对象可以被ToolNode调用生成 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会读取 state 中 LLM 产生的 tool_calls并行/串行调用对应的工具函数将工具返回结果写入 state 在我们的 Demo 中的工具使用虽然我们的 Demo 主要用纯节点函数实现游戏逻辑因为游戏规则是确定性的但工具在 LangGraph 中的典型用法是用户输入 → Agent 节点LLM 决策 ↓ 需要调用工具 ↙ ↘ 是 否 ↓ ↓ ToolNode 直接回复 ↓ 工具结果 ↓ Agent 节点LLM 整合结果 ↓ 最终回复给用户在我们的游戏中假如你接入了 LLMroll_dice和calculate_damage就会由 LLM 自动判断何时调用玩家「我要攻击地精」 LLM 这是个战斗动作我需要 roll_dice(D20) → 调用工具 roll_dice(20) → 得到结果15 → 好的一掷接下来计算伤害... → 调用工具 calculate_damage(5, 15) → 造成了 7 点伤害地精反击... 工具 vs 直接调用函数维度直接调用通过 Tool调用者代码中明确写LLM 自主决定时机编译时确定运行时动态灵活性低高可观测性日志Tool 调用记录在 state 中适用场景确定性逻辑需要 LLM 判断的场景 启发式问题tool装饰器和普通def在 LangGraph 中的行为有什么区别不使用tool的函数能否被ToolNode调用如果工具执行失败比如抛异常LangGraph 会如何处理你需要在哪里加错误处理在一个 Agent 图中有 10 个工具时LLM 如何选择调用哪个提示工程在其中扮演什么角色工具的 docstring 有多重要如果roll_dice的 docstring 写得很模糊LLM 可能做出什么错误判断假如你想让工具之间相互调用比如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.pydef 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(resumeuser_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) # 除了 resumeCommand 还可以同时更新 state app.invoke(Command(resume北, update{player_hp: 50}), config)Command是一个多用途指令resume传给interrupt()的返回值update在恢复前直接修改 stategoto跳转到指定节点高级用法 对比有 HITL vs 无 HITL无 HITL 有 HITL 图开始 图开始 ↓ ↓ Agent 决策 Agent 分析 ↓ ↓ 调用工具 interrupt() 暂停 ↓ ↓ Agent 回复 人类审核 ↓ ↓ 图结束 批准/拒绝/修改 ↓ 继续执行 ↓ 图结束 实际应用场景场景暂停点审批流程关键决策前等待批准客服机器人遇到无法回答的问题时转人工代码审查AI 生成代码后等待人类审查数据标注AI 标注后人工抽查确认敏感操作删除/支付前等待确认 启发式问题interrupt()返回后图是如何知道从哪里继续的背后的机制是什么如果interrupt()暂停后外部程序没有调用Command(resume...)而是重新invoke()会发生什么如何在一次invoke中实现多次暂停提示interrupt()可以在图中多处使用Command的update参数允许在恢复时直接修改 state。这在什么场景下有用有什么风险在人机交互模式下如果用户输入了一个无效命令你应该在main.py中过滤还是在图内部处理为什么 下一步07 · 检查点与持久化 答案06-Human-in-the-Loop-答案07 · 检查点与持久化图不会失忆 一句话总结Checkpoint 是 LangGraph 的存档系统——每次图执行后自动保存状态快照支持时光回溯和断点续传。 在我们的 Demo 中打开src/graph.pybuild_adventure_graph()的最后from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() compiled graph.compile(checkpointermemory) 而在 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(resumeuser_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(gotolast_alive.next[0]), config) # 跳回 不同的 Checkpointer 实现实现存储位置适用场景MemorySaver内存开发/测试重启丢失SqliteSaverSQLite 文件本地持久化PostgresSaverPostgreSQL生产环境# SQLite 持久化from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 conn sqlite3.connect(adventure_saves.db, check_same_threadFalse) memory SqliteSaver(conn) compiled graph.compile(checkpointermemory)⌛ 检查点的内部结构每个检查点保存Checkpoint: ├── values: GameState # 完整的当前状态 ├── next: (enter_room,) # 下一步要执行的节点 ├── metadata: │ ├── source: loop # 触发来源 │ ├── step: 12 # 执行步数 │ └── writes: {...} # 本次写入 └── parent_checkpoint_id # 父检查点形成链 高级话题Channel 和 PregelLangGraph 的持久化底层基于Pregel模型Google 的图计算框架每个状态字段都是一个Channel每次节点写入数据流入 ChannelReducer 决定多个写入如何合并Checkpoint 保存所有 Channel 的当前快照这保证了并发场景下的一致性。 启发式问题如果MemorySaver在程序重启后数据丢失在什么场景下这反而是有利的thread_id冲突会怎样如果两个用户意外使用了相同的thread_id会发生什么检查点会保存完整的状态还是增量如果 state 中有大对象如整个网页内容会有什么性能影响如果你想让用户回到上一轮战斗前应该用什么 API如何实现一个撤销功能在生产环境中你会选择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%免费】
LangGraph企业级应用教程- 用“有向图“来编排 AI Agent 工作流
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.pybuild_adventure_graph()函数——组装整张图src/state.pyGameStateTypedDict——状态定义src/nodes.py所有节点函数src/main.pyrun_game()主循环——invoke/resume 交互 启发式问题为什么 LangGraph 选择图而不是链作为编排模型什么场景下图优于链如果把enter_room()改成直接调用initiate_combat()会带来什么问题状态State和普通 Python 函数参数有什么区别为什么图需要共享状态你能否想到一个现实中适合用 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, totalFalse): # 使用 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, totalFalse): current_room: str player_hp: intTypedDict是 Python 的类型提示机制让你在写状态时获得 IDE 自动补全和类型检查。totalFalse表示所有字段都是可选的——这在return {}时不会报错。2. Annotated Reducer —— 控制合并策略这是 LangGraph 最重要的设计之一。messages: Annotated[list, add_messages]当两个节点都想修改messages时LangGraph 怎么合并普通字段后写的覆盖先写的Annotated[list, add_messages]追加不覆盖这就像 Git 的 merge strategy普通字段 git merge --strategyours直接用新版本add_messages git merge --strategyunion两边都保留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中流转enter_room → 读取 current_room写入 messages ↓ initiate_combat → 设置 combat_activeTrue, monster_hp20 ↓ player_attack → 递减 monster_hp, 递减 player_hp ↓ enter_room → 重新读取 current_room没变展示新状态 实验修改 Reducer试试把inventory: List[str]改成from operator import add inventory: Annotated[List[str], add]这样两个节点都往背包里加东西时它们会自动合并类似列表 extend而不是一个覆盖另一个。 启发式问题为什么messages要用add_messagesreducer 而不是普通覆盖如果改成覆盖会发生什么如果你自己写一个自定义 reducer比如取最大值它会在什么场景下有用TypedDict(totalFalse)中totalFalse的作用是什么设为True会怎样如果两个节点同时修改player_hp一个 10一个 -5最终结果是多少LangGraph 的默认行为是什么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_roomnodes.py:76展示房间信息pickup_itemsnodes.py:89拾取地上物品initiate_combatnodes.py:108初始化战斗状态player_attacknodes.py:132执行攻击回合flee_combatnodes.py:195尝试逃跑solve_puzzlenodes.py:213展示谜题move_playernodes.py:246移动到新房间game_over_screennodes.py:274展示结局wait_for_inputgraph.py:143Human-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 ... 启发式问题为什么节点不应该调用下一个节点如果你在enter_room里直接调用initiate_combat()会出现什么问题条件路由函数必须是纯函数吗如果route_from_room()内部修改了 state会发生什么你能想到哪些场景适合用固定边哪些适合用条件边给出实际例子。如果图中有 20 个节点和 50 条边你会如何测试这张图的正确性假如你想加一个中毒机制每走一步扣 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 中非常常见——就像游戏主循环。 启发式问题路由函数为什么必须是纯函数如果路由函数修改了 state会出现什么后果什么是循环边在我们的 Demo 中enter_room → enter_room会形成循环什么情况下这样设计是合理的什么情况下你应该避免假如你想实现玩家生命低于 10 时自动使用治疗药水应该在哪里加这个逻辑路由函数中还是节点中add_conditional_edges的第三个参数路由表要求你列出所有可能的返回值。如果你漏写了一个会发生什么如果你有 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: 掷骰子返回结果。这个装饰器自动做了几件事提取函数签名参数名、类型、默认值提取 docstring作为工具的描述LLM 会读到包装为 Tool 对象可以被ToolNode调用生成 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会读取 state 中 LLM 产生的 tool_calls并行/串行调用对应的工具函数将工具返回结果写入 state 在我们的 Demo 中的工具使用虽然我们的 Demo 主要用纯节点函数实现游戏逻辑因为游戏规则是确定性的但工具在 LangGraph 中的典型用法是用户输入 → Agent 节点LLM 决策 ↓ 需要调用工具 ↙ ↘ 是 否 ↓ ↓ ToolNode 直接回复 ↓ 工具结果 ↓ Agent 节点LLM 整合结果 ↓ 最终回复给用户在我们的游戏中假如你接入了 LLMroll_dice和calculate_damage就会由 LLM 自动判断何时调用玩家「我要攻击地精」 LLM 这是个战斗动作我需要 roll_dice(D20) → 调用工具 roll_dice(20) → 得到结果15 → 好的一掷接下来计算伤害... → 调用工具 calculate_damage(5, 15) → 造成了 7 点伤害地精反击... 工具 vs 直接调用函数维度直接调用通过 Tool调用者代码中明确写LLM 自主决定时机编译时确定运行时动态灵活性低高可观测性日志Tool 调用记录在 state 中适用场景确定性逻辑需要 LLM 判断的场景 启发式问题tool装饰器和普通def在 LangGraph 中的行为有什么区别不使用tool的函数能否被ToolNode调用如果工具执行失败比如抛异常LangGraph 会如何处理你需要在哪里加错误处理在一个 Agent 图中有 10 个工具时LLM 如何选择调用哪个提示工程在其中扮演什么角色工具的 docstring 有多重要如果roll_dice的 docstring 写得很模糊LLM 可能做出什么错误判断假如你想让工具之间相互调用比如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.pydef 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(resumeuser_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) # 除了 resumeCommand 还可以同时更新 state app.invoke(Command(resume北, update{player_hp: 50}), config)Command是一个多用途指令resume传给interrupt()的返回值update在恢复前直接修改 stategoto跳转到指定节点高级用法 对比有 HITL vs 无 HITL无 HITL 有 HITL 图开始 图开始 ↓ ↓ Agent 决策 Agent 分析 ↓ ↓ 调用工具 interrupt() 暂停 ↓ ↓ Agent 回复 人类审核 ↓ ↓ 图结束 批准/拒绝/修改 ↓ 继续执行 ↓ 图结束 实际应用场景场景暂停点审批流程关键决策前等待批准客服机器人遇到无法回答的问题时转人工代码审查AI 生成代码后等待人类审查数据标注AI 标注后人工抽查确认敏感操作删除/支付前等待确认 启发式问题interrupt()返回后图是如何知道从哪里继续的背后的机制是什么如果interrupt()暂停后外部程序没有调用Command(resume...)而是重新invoke()会发生什么如何在一次invoke中实现多次暂停提示interrupt()可以在图中多处使用Command的update参数允许在恢复时直接修改 state。这在什么场景下有用有什么风险在人机交互模式下如果用户输入了一个无效命令你应该在main.py中过滤还是在图内部处理为什么 下一步07 · 检查点与持久化 答案06-Human-in-the-Loop-答案07 · 检查点与持久化图不会失忆 一句话总结Checkpoint 是 LangGraph 的存档系统——每次图执行后自动保存状态快照支持时光回溯和断点续传。 在我们的 Demo 中打开src/graph.pybuild_adventure_graph()的最后from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() compiled graph.compile(checkpointermemory) 而在 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(resumeuser_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(gotolast_alive.next[0]), config) # 跳回 不同的 Checkpointer 实现实现存储位置适用场景MemorySaver内存开发/测试重启丢失SqliteSaverSQLite 文件本地持久化PostgresSaverPostgreSQL生产环境# SQLite 持久化from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 conn sqlite3.connect(adventure_saves.db, check_same_threadFalse) memory SqliteSaver(conn) compiled graph.compile(checkpointermemory)⌛ 检查点的内部结构每个检查点保存Checkpoint: ├── values: GameState # 完整的当前状态 ├── next: (enter_room,) # 下一步要执行的节点 ├── metadata: │ ├── source: loop # 触发来源 │ ├── step: 12 # 执行步数 │ └── writes: {...} # 本次写入 └── parent_checkpoint_id # 父检查点形成链 高级话题Channel 和 PregelLangGraph 的持久化底层基于Pregel模型Google 的图计算框架每个状态字段都是一个Channel每次节点写入数据流入 ChannelReducer 决定多个写入如何合并Checkpoint 保存所有 Channel 的当前快照这保证了并发场景下的一致性。 启发式问题如果MemorySaver在程序重启后数据丢失在什么场景下这反而是有利的thread_id冲突会怎样如果两个用户意外使用了相同的thread_id会发生什么检查点会保存完整的状态还是增量如果 state 中有大对象如整个网页内容会有什么性能影响如果你想让用户回到上一轮战斗前应该用什么 API如何实现一个撤销功能在生产环境中你会选择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%免费】
