LangGraph 入门手册

LangGraph 入门指南

一、LangGraph 是什么

1.1 命名由来

Lang：

• 生态纽带：延续 LangChain 命名体系，是 LangChain 生态的官方核心组件
• 领域聚焦：明确指向 LLM 驱动的智能体（Agent）开发
• 用户认知：降低学习门槛，让 LangChain 开发者快速理解其定位

Graph：

• 计算图：将 AI 工作流建模为有向图，节点表示执行单元，边表示执行路径
• 状态图：内置强状态管理，图中流动的是包含上下文、历史、执行状态的完整"状态对象"
• 突破线性限制：天然支持循环迭代、并行执行、动态分支、断点续跑

1.2 与 LangChain 的关系

链是图的特例，图是链的超集，二者共同构成 LangChain 生态的"基础层 + 高级层"架构。

1.3 回顾 LangChain 六大核心组件

二、LangGraph核心概念：四大基石

三、快速上手

3.1 需求

先问个问题，根据问题分类成闲聊和技术类，然后闲聊类走闲聊分支，技术类走技术分支。

+-----------+| __start__ |+-----------+ * * *+-------------------+| classify_question |+-------------------+ ........+------------+ +------------+| chat_reply | | tech_reply |+------------+ +------------+ ******** ****+---------+| __end__ |+---------+

3.2 完整代码

from langchain_core.output_parsers import StrOutputParserfrom langgraph.graph import START, END, StateGraphfrom llm import llmfrom typing import TypedDict# ==================== 定义全局状态 State ====================class ChatState(TypedDict):question: str # 用户输入的问题answer: str # 最终生成的回答q_type: str # 问题分类结果：chat（闲聊）或 tech（技术）# ==================== 定义节点函数 ====================def classify_question(state: ChatState) -> ChatState:chain = llm | StrOutputParser()res = chain.invoke(f"判断问题类型，返回 chat 或 tech：{state['question']}")return {"q_type": res}def chat_reply(state: ChatState) -> ChatState:chain = llm | StrOutputParser()res = chain.invoke(f"友好闲聊回复：{state['question']}")return {"answer": res}def tech_reply(state: ChatState) -> ChatState:chain = llm | StrOutputParser()res = chain.invoke(f"专业简洁解答技术问题：{state['question']}")return {"answer": res}# ==================== 定义条件路由 ====================NODE_CLASSIFY_QUESTION = "classify_question"NODE_CHAT_REPLY = "chat_reply"NODE_TECH_REPLY = "tech_reply"nodes = [NODE_CHAT_REPLY, NODE_TECH_REPLY]path_map = {node: node for node in nodes}def route_by_type(state: ChatState) -> str:if state["q_type"] == "chat":return NODE_CHAT_REPLYelse:return NODE_TECH_REPLY# ==================== 组装并编译图 ====================graph = (StateGraph(ChatState).add_node(NODE_CLASSIFY_QUESTION, classify_question).add_node(NODE_CHAT_REPLY, chat_reply).add_node(NODE_TECH_REPLY, tech_reply).add_edge(START, NODE_CLASSIFY_QUESTION).add_conditional_edges(NODE_CLASSIFY_QUESTION,route_by_type,path_map).add_edge(NODE_CHAT_REPLY, END).add_edge(NODE_TECH_REPLY, END).compile())# ==================== 图结构可视化 ====================mindmap = graph.get_graph()mindmap.print_ascii()print(mindmap.draw_mermaid())png_bytes = mindmap.draw_mermaid_png()with open("mindmap.png", "wb") as f:f.write(png_bytes)# ==================== 运行测试 ====================res1 = graph.invoke({"question": "今天天气怎么样？"})print("【闲聊问题回复】n", res1["answer"], "n")res2 = graph.invoke({"question": "Python 列表和元组的区别？"})print("【技术问题回复】n", res2["answer"])

3.3 构建图的固定流程

定义 State → 定义节点函数 → 定义路由函数(可选) → 组装图(add_node/add_edge) → compile() → invoke()

四、State 详解

4.1 两种定义方式

方式一：TypedDict（简单覆盖）

默认行为：节点返回的 dict 会整体覆盖 state 中对应字段的值。

from typing import TypedDictclass MyState(TypedDict):question: stranswer: str

方式二：Annotated + Reducer（追加/合并）

当某个字段需要累加而非覆盖时（如消息列表），使用 Annotated[类型, reducer函数]：

from typing import Annotated, TypedDictfrom langgraph.graph.message import add_messagesclass ChatState(TypedDict):messages: Annotated[list, add_messages]# 追加消息而非覆盖question: str # 普通覆盖

Reducer 的工作原理：

• add_messages(left, right)：将新消息追加到已有消息列表，相同 ID 的消息会被更新
• 也可以自定义 reducer：Annotated[int, lambda old, new: old + new] 实现累加

4.2 内置 State：MessagesState

LangGraph 提供了内置的 MessagesState，等价于：

class MessagesState(TypedDict):messages: Annotated[list[AnyMessage], add_messages]

直接使用：

from langgraph.graph import MessagesState, StateGraphgraph = StateGraph(MessagesState)

4.3 StateGraph 构造函数参数

StateGraph(state_schema, # 必填：State 类型（TypedDict 类）context_schema=None,# 可选：上下文 schema，用于子图间共享只读上下文*,input_schema=None,# 可选：图的输入 schema，不填则用 state_schemaoutput_schema=None, # 可选：图的输出 schema，不填则用 state_schema)

input_schema / output_schema 示例：

class InternalState(TypedDict):question: stranswer: strq_type: str # 内部中间状态，不想暴露给外部class InputState(TypedDict):question: str # 只接收 questionclass OutputState(TypedDict):answer: str # 只返回 answergraph = StateGraph(InternalState,input_schema=InputState,output_schema=OutputState,)# 调用者只需传 {"question": "..."}，只能拿到 {"answer": "..."}# q_type 是内部实现细节，对外不可见

五、Node 详解

5.1 节点函数的签名

节点函数接收两个参数（第二个可选）：

def my_node(state: MyState, config: RunnableConfig) -> dict:#↑ 必填↑ 可选return {"field": new_value}

• state：当前全局状态
• config：LangChain 的 RunnableConfig，可获取 config["configurable"] 中的运行时参数

5.2 节点返回值规则

5.3 add_node 完整参数

graph.add_node(node,# 节点名称(str) 或 直接传函数(自动取函数名)action=None, # 当 node 是 str 时，此参数为处理函数*,defer=False, # 是否延迟执行（等到运行即将结束时才执行）metadata=None, # 节点元数据 dict，用于调试/追踪input_schema=None, # 节点专属输入 schema（默认用图的 state_schema）retry_policy=None, # 重试策略cache_policy=None, # 缓存策略error_handler=None,# 节点级错误处理函数destinations=None, # 声明节点可能跳转的目标（仅用于图渲染，不影响执行）timeout=None,# 超时设置（秒 / timedelta / TimeoutPolicy）)

两种注册方式：

# 方式1：名称 + 函数分开传graph.add_node("my_node", my_function)# 方式2：直接传函数，自动用函数名作为节点名graph.add_node(my_function)# 节点名 = "my_function"

常用参数详解：

5.4 重试策略 RetryPolicy

当节点函数抛出异常时，LangGraph 可以按策略自动重试，避免因临时网络抖动、API 限流等瞬态错误导致整个图执行失败。

参数详解

from langgraph.types import RetryPolicyretry = RetryPolicy(max_attempts=3,# 最大尝试次数（含首次执行），默认 3initial_interval=0.5,# 首次重试前的等待秒数，默认 0.5backoff_factor=2.0,# 退避倍数，默认 2.0max_interval=128.0,# 最大等待秒数上限，默认 128.0jitter=True, # 是否添加随机抖动，默认 Trueretry_on=Exception,# 触发重试的异常条件，默认所有 Exception)

退避时间计算

以 initial_interval=1.0, backoff_factor=2.0 为例：

第 1 次执行失败 → 等待 1s → 第 2 次重试第 2 次重试失败 → 等待 2s → 第 3 次重试第 3 次重试失败 → 等待 4s → 第 4 次重试（如果 max_attempts 允许）...等待时间不会超过 max_interval（默认 128s）

retry_on 的三种用法

# 方式一：指定异常类型（只有该类型异常才重试）retry = RetryPolicy(retry_on=ConnectionError)# 方式二：指定多个异常类型retry = RetryPolicy(retry_on=(ConnectionError, TimeoutError))# 方式三：自定义判断函数（最灵活）retry = RetryPolicy(retry_on=lambda e: "rate limit" in str(e).lower())

使用示例

retry = RetryPolicy(max_attempts=3, initial_interval=1.0, backoff_factor=2.0)graph = (StateGraph(State).add_node("call_llm", call_llm, retry_policy=retry).add_node("process", process)# 不需要重试的节点不传 retry_policy...)

5.5 缓存策略 CachePolicy

当相同输入状态再次进入节点时，CachePolicy 可以让节点直接返回缓存结果，跳过函数体执行。这对 LLM 调用等耗时操作特别有价值。

两个条件缺一不可

节点缓存生效需要同时满足两个条件：

⚠️ 只设 cache_policy 不传 cache，缓存不会生效！ 这是初学者最常犯的错误。

from langgraph.cache.memory import InMemoryCache as GraphCachefrom langgraph.types import CachePolicy# ✅ 正确：两个条件都满足graph = (StateGraph(State).add_node("answer", answer_qs, cache_policy=CachePolicy(ttl=60))....compile(cache=GraphCache())# 必须传入！)# ❌ 错误：只设了 cache_policy，没有存储后端，缓存不生效graph = (StateGraph(State).add_node("answer", answer_qs, cache_policy=CachePolicy(ttl=60))....compile()# 缺少 cache=...)

CachePolicy 参数详解

from langgraph.types import CachePolicycache_policy = CachePolicy(ttl=60, # 缓存过期时间（秒），None 表示永不过期key_func=...# 缓存 key 生成函数，默认用 pickle 序列化输入后哈希)

存储后端选择

# 内存缓存（开发调试）from langgraph.cache.memory import InMemoryCache as GraphCachegraph = ... .compile(cache=GraphCache())# Redis 缓存（生产环境）from langgraph.cache.redis import RedisCachegraph = ... .compile(cache=RedisCache(redis_url="redis://localhost:6379"))

缓存 key 的工作原理

1. 节点执行前，用 key_func 对节点的输入状态计算缓存 key
2. 在 cache 存储后端中查找该 key
3. 如果命中且未过期 → 直接返回缓存结果，函数体不执行
4. 如果未命中 → 正常执行函数，将输出结果写入缓存

5.6 三层机制对比

LLM 缓存、节点缓存、检查点持久化容易混淆，下表帮你彻底理清：

需要"相同问题不重复调 LLM"→ LLM 缓存（第一层）需要"相同输入不重复执行节点函数" → 节点缓存（第二层）需要"记住执行进度，支持暂停/继续"→ 检查点持久化（第三层）

5.7 缓存与重试完整代码示例

import operatorimport timefrom typing import Annotated, TypedDict# 第一层：LLM 缓存（全局设置，对所有 LLM 调用生效）from langchain_core.caches import InMemoryCachefrom langchain_core.globals import set_llm_cache# 第二层：节点缓存所需的存储后端from langgraph.cache.memory import InMemoryCache as GraphCachefrom langgraph.types import CachePolicy# 第三层：检查点持久化from langgraph.checkpoint.memory import MemorySaver# 重试策略from langgraph.types import RetryPolicy# 其他导入from langchain_core.output_parsers import StrOutputParserfrom langgraph.graph import START, END, StateGraph, add_messagesfrom llm import llm# 启用第一层：LLM 缓存set_llm_cache(InMemoryCache())class StateMessage(TypedDict):messages: Annotated[list, add_messages]listNumber: Annotated[float, operator.add]question: strdef answer_qs(state):chain = llm | StrOutputParser()return {"messages": [chain.invoke(state["question"])]}def smile1(state):return {"listNumber": 1}def smile2(state):return {"listNumber": 2}NODE_QUESTION_QS = "answer_qs"NODE_SMILE1 = "smile1"NODE_SMILE2 = "smile2"# 重试策略：最多 3 次，首次等 1s，退避倍数 2（1s → 2s → 4s）retry = RetryPolicy(max_attempts=3, initial_interval=1.0, backoff_factor=2.0)# 缓存策略：缓存 60 秒cache_policy = CachePolicy(ttl=60)graph = (StateGraph(StateMessage).add_node(NODE_QUESTION_QS, answer_qs, retry_policy=retry, cache_policy=cache_policy).add_node(NODE_SMILE1, smile1).add_node(NODE_SMILE2, smile2).add_edge(START, NODE_QUESTION_QS).add_edge(NODE_QUESTION_QS, NODE_SMILE1).add_edge(NODE_SMILE1, NODE_SMILE2).add_edge(NODE_SMILE2, END)# checkpointer=MemorySaver()→ 第三层：检查点持久化# cache=GraphCache()→ 第二层：节点缓存的存储后端.compile(checkpointer=MemorySaver(), cache=GraphCache()))config = {"configurable": {"thread_id": "1"}}t1 = time.time()print("第一次执行：")print(graph.invoke({'question': '1+2=? 直接返回答案就行'}, config))print(f"耗时: {time.time()-t1:.2f}s")print("n1111111111111111111111111111")t2 = time.time()print("第二次运行（利用缓存，瞬间返回）：")print(graph.invoke({'question': '1+2=? 直接返回答案就行'}, config))print(f"耗时: {time.time()-t2:.2f}s")

六、Edge 详解

6.1 普通边 add_edge

graph.add_edge(start_key,# 起始节点名（str）或 节点名列表（list[str]）end_key,# 目标节点名（str）)

多节点汇聚：当 start_key 是列表时，表示等待所有起始节点都完成后，才执行目标节点：

# research 和 analysis 都完成后，才进入 summarygraph.add_edge(["research", "analysis"], "summary")

6.2 条件边 add_conditional_edges

当需要条件分支时，定义路由函数来决定下一步走哪个节点。

路由函数的返回值

路由函数返回下一个节点的名称（字符串），或返回 'END' 直接结束：

def route_fn(state: MyState) -> str:if state["q_type"] == "chat":return "chat_reply"elif state["q_type"] == "tech":return "tech_reply"else:return END# 直接结束，不进入任何节点

add_conditional_edges 完整参数

graph.add_conditional_edges(source, # 源节点名称（str）path, # 路由函数（Callable）或 Runnablepath_map=None # 路径映射（dict 或 list），可选)

path_map 的四种写法：

# 写法1：dict 映射（路由返回值 → 目标节点名）path_map = {"chat": "chat_reply", "tech": "tech_reply"}# 写法2：dict 映射（返回值和节点名相同时的简写）path_map = {"chat_reply": "chat_reply", "tech_reply": "tech_reply"}# 写法3：list（每个元素既是返回值又是节点名）path_map = ["chat_reply", "tech_reply"]# 写法4：不传 path_map# 此时路由函数的返回值直接作为节点名# ⚠️ 缺点：图可视化时无法确定可能的分支，会假设可以跳到任何节点

path 也可以是 Runnable

from langchain_core.runnables import RunnableLambdaroute_runnable = llm | StrOutputParser()# 用 LLM 来决定路由graph.add_conditional_edges("classify",route_runnable,path_map={"chat": "chat_reply", "tech": "tech_reply"})

七、Compile 详解

graph.compile(checkpointer=None, # 持久化检查点，支持断点续跑*,cache=None,# 缓存后端store=None,# 跨线程的长期记忆存储interrupt_before=None, # 在指定节点执行前中断（Human-in-the-loop）interrupt_after=None,# 在指定节点执行后中断debug=False, # 调试模式name=None, # 编译后的图名称transformers=None, # 流式输出转换器)

checkpointer 示例：

from langgraph.checkpoint.memory import MemorySavergraph = StateGraph(MyState).add_node(...).add_edge(...).compile(checkpointer=MemorySaver())# 必须传 thread_idconfig = {"configurable": {"thread_id": "thread-1"}}result = graph.invoke({"question": "你好"}, config)# 同一 thread_id 下，下次调用会保留上次的 stateresult2 = graph.invoke({"question": "继续"}, config)

interrupt_before 示例：

graph = builder.compile(checkpointer=MemorySaver(),interrupt_before=["send_email"]# 发邮件前暂停)# 第一次调用会在 send_email 前暂停result = graph.invoke(inputs, config)# 人工确认后，传 None 继续result = graph.invoke(None, config)

八、高级控制原语

前面我们掌握了 LangGraph 的基础构建块：State、Node、Edge、Checkpointer。这些组件覆盖了大多数场景，但当图变得更复杂时，你会遇到三个基础组件无法优雅解决的问题：

1. 边是静态的 —— 编译前必须确定"A → B"，但运行时才知道要分发到几个节点怎么办？
2. 节点只能返回状态更新或走静态边 —— 想同时更新状态 + 动态跳转怎么办？
3. 节点只能拿到 state 和 config —— 需要访问用户身份、持久化存储、流式输出等运行时能力怎么办？

LangGraph 提供了三个高级原语来解决这些问题：

它们和已有概念的关系：

┌─────────────────────────────────────────────────────────────┐│LangGraph 构建块体系 │├──────────────┬──────────────────┬───────────────────────────┤│ 基础组件│增强配置 │高级控制原语│├──────────────┼──────────────────┼───────────────────────────┤│ State│ RetryPolicy│ Send（动态扇出） ││ Node │ CachePolicy│ Command（状态+路由一体化） ││ Edge │ Checkpointer │ Runtime Context（随身工具箱）││ 条件边│ Store│ │└──────────────┴──────────────────┴───────────────────────────┘基础组件：构建图的最小单元，必须掌握增强配置：让图更健壮（重试、缓存、持久化），按需添加高级原语：解决基础组件无法优雅处理的复杂场景，进阶使用

8.1 Send —— 动态扇出（Map-Reduce 模式）

解决什么问题？

普通的边（Edge）是静态的：你在编译前就必须确定"A → B"。但有些场景，你不知道要执行几次下游节点。

经典例子：Map-Reduce

用户输入 3 个主题 → 每个主题各生成一个笑话 → 汇总所有笑话

问题在于：3 个主题是运行时才知道的，你没法提前写 3 条边。Send 就是解决这个问题的。

适用场景：

• 文档批量分片摘要：一篇长文拆成 10 段，并行调用摘要节点 10 次
• 多工具并行查询：同时查天气、数据库、搜索引擎
• 多子问题拆解：把用户大问题拆成 N 个子问题，并行求解再汇总

怎么理解？

把 Send 想象成一个动态分发器：

普通边：A ──→ B（1 对 1，写死的）Send：A ──→ B ──→ B ──→ B（1 对 N，运行时决定 N 是几，且并行执行）

代码示例

from langgraph.types import Send# 状态中有 subjects: ["猫", "狗", "程序员"]def continue_to_jokes(state: OverallState):# 为每个 subject 动态创建一条"边"，发送到 generate_joke 节点# 每个 Send 携带不同的输入状态return [Send("generate_joke", {"subject": "猫"}),Send("generate_joke", {"subject": "狗"}),Send("generate_joke", {"subject": "程序员"}),]# 用 add_conditional_edges 注册，和普通条件边一样graph.add_conditional_edges("node_a", continue_to_jokes)

关键点：

• Send("节点名", 状态) = 向指定节点发送一份独立的状态
• 返回列表中有几个 Send，下游节点就执行几次
• 每次执行拿到的状态互不相同（各自只收到自己的那份）
• 并行执行：多个 Send 会并发运行，不是串行

Send 与 add_conditional_edges 的配合机制

很多初学者的困惑是：Send 明明是"动态分发"，为什么要用 add_conditional_edges 来注册？

核心原因：Send 本质上是一种特殊的条件边返回值。add_conditional_edges 是 LangGraph 中"从节点出发，运行时决定下一步去哪"的唯一机制，Send 只是让这个"决定"从"去一个节点"变成了"去 N 个节点"。

三种条件边返回值对比

# add_conditional_edges 的 path 函数可以返回三种东西：# ① 返回字符串 → 走 1 条分支（最普通用法）def route(state):if state["type"] == "A":return "node_a"# 去节点 Areturn "node_b"# 去节点 B# ② 返回 Send → 走 1 条分支，但携带自定义状态def route(state):return Send("node_a", {"custom": "data"}) # 去节点 A，且注入自定义状态# ③ 返回 Send 列表 → 走 N 条分支（动态扇出）def route(state):return [Send("node_a", {"subject": "猫"}),Send("node_a", {"subject": "狗"}),Send("node_a", {"subject": "程序员"}),] # 节点 A 执行 3 次，每次拿到不同的状态

逐步拆解：从普通条件边到 Send

第一步：普通条件边（1 → 1）

def route(state):return "node_b" # 返回节点名，走一条路graph.add_conditional_edges("node_a", route, ["node_b", "node_c"])

第二步：条件边返回 Send（1 → 1，但携带自定义状态）

def route(state):return Send("node_b", {"subject": "猫"})graph.add_conditional_edges("node_a", route, ["node_b"])

第三步：条件边返回 Send 列表（1 → N，动态扇出）

def route(state):return [Send("node_b", {"subject": "猫"}),Send("node_b", {"subject": "狗"}),]graph.add_conditional_edges("node_a", route, ["node_b"])

add_conditional_edges 的第三个参数

graph.add_conditional_edges("source_node", path_func, path_map)#↑↑ ↑#从哪个节点出发 路由函数 可能的目标节点列表

• path_map 是声明式的：告诉 LangGraph "路由函数可能返回的目标节点有哪些"
• 当路由函数返回 Send 时，path_map 中必须包含 Send 指向的节点名
• 如果省略 path_map，LangGraph 会尝试自动推断，但显式声明更安全

Superstep 机制：并行结果如何"等齐"再汇合？

一个常见的疑问：add_edge("generate_joke", "collect_jokes") 只是一条普通边，collect_jokes 怎么知道要等 3 个并行结果全部完成？

答案是 LangGraph 的 Superstep（超步）机制——框架自动保证，不需要手动处理。

LangGraph 的执行按超步推进，一个超步内的所有并行任务必须全部完成，才会进入下一个超步：

Superstep 1:START → 条件边返回 3 个 Send↓Superstep 2:generate_joke × 3（并行执行）┌─ Send1: {"subject":"狗"}→ 返回 {"jokes": ["狗笑话"]}├─ Send2: {"subject":"猫"}→ 返回 {"jokes": ["猫笑话"]}└─ Send3: {"subject":"程序员"} → 返回 {"jokes": ["程序员笑话"]}↓⚡ 3 个全部完成后，add reducer 累加结果jokes = ["狗笑话", "猫笑话", "程序员笑话"]↓Superstep 3:collect_jokes（此时 jokes 已包含全部 3 个结果）

两个关键保证：

如果缺少 add reducer：

# ❌ 不用 add reducer，jokes 只是 list[str]class OverallState(TypedDict):jokes: list[str]# 默认 last writer wins（覆盖）# 3 个并行结果只保留最后一个，前两个被覆盖！# collect_jokes 只能看到 1 个笑话，而不是 3 个

简单记忆：Superstep 保证"等齐"，add reducer 保证"攒齐"，两者缺一不可。

Superstep 等的是"批次"而非"节点"

如果 Send 发到不同的节点，Superstep 也会等所有任务完成吗？答案是会。

def route(state):return [Send("node_a", {"data": 1}),Send("node_a", {"data": 2}),Send("node_b", {"data": 3}),]

Superstep 1:条件边返回 3 个 Send↓Superstep 2:node_a × 2（并行）+node_b × 1（并行）┌─ node_a(data=1) ──→ 下游 node_c├─ node_a(data=2) ──→ 下游 node_c└─ node_b(data=3) ──→ 下游 node_d↓⚡ 3 个任务全部完成，才进入下一个 Superstep（node_c 不会因为 node_a 先完成就提前执行，必须等 node_b 也完成）↓Superstep 3:node_c 和 node_d 同时开始

Superstep 是按"批次"隔离的，不是按"节点"隔离的。 同一批 Send 出去的所有任务（不管目标节点是否相同），必须全部完成才能推进到下一批。

和普通条件边的区别

常见错误

# ❌ 错误 1：在节点函数中返回 Send（节点必须返回 dict）def my_node(state):return [Send("target", {"data": 1})] # TypeError!# ✅ 正确：在条件边函数中返回 Senddef my_router(state):return [Send("target", {"data": 1})]graph.add_conditional_edges("my_node", my_router, ["target"])# ❌ 错误 2：path_map 中漏掉了 Send 指向的节点graph.add_conditional_edges("node_a", router, ["other_node"])# 如果 router 返回 Send("target", ...)，但 "target" 不在 path_map 中 → 报错# ✅ 正确：path_map 包含所有可能的目标graph.add_conditional_edges("node_a", router, ["target"])

完整代码示例

"""Send 最小示例：Map-Reduce 模式图结构：START ──(条件边+Send)──→ generate_joke × 3（并行）│▼collect_jokes│▼END"""from typing import Annotated, TypedDictfrom operator import addfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.types import Sendfrom llm import llmfrom langchain_core.output_parsers import StrOutputParser# OverallState：图的全局状态，所有节点共享# jokes 用 Annotated[list[str], add] 声明，多个并行节点返回的 jokes 列表会自动累加class OverallState(TypedDict):subjects: list[str]jokes: Annotated[list[str], add]# JokeState：Send 传给 generate_joke 的私有状态# 每次并行执行时，节点只收到 Send 携带的那份数据class JokeState(TypedDict):subject: strdef generate_joke(state: JokeState) -> dict:subject = state["subject"]chain = llm | StrOutputParser()joke = chain.invoke(f"请讲一个关于{subject}的笑话，只需返回笑话内容")return {"jokes": [joke]}def collect_jokes(state: OverallState) -> dict:print(f"n汇总 {len(state['jokes'])} 个笑话:")for i, joke in enumerate(state["jokes"], 1):print(f"{i}. {joke}")return {}graph = (StateGraph(OverallState).add_node("generate_joke", generate_joke).add_node("collect_jokes", collect_jokes).add_conditional_edges(START,lambda state: [Send("generate_joke", {"subject": subject})for subject in state["subjects"]],["generate_joke"],).add_edge("generate_joke", "collect_jokes").add_edge("collect_jokes", END).compile())if __name__ == "__main__":print("Send 示例：Map-Reduce 模式")result = graph.invoke({"subjects": ["猫", "狗", "程序员"]})

8.2 Command —— 状态更新 + 路由一体化

解决什么问题？

传统 LangGraph 分离逻辑：

• 节点只能返回状态字典，不能控制跳转；
• 跳转必须单独写 add_conditional_edges + 独立路由函数，逻辑拆分两处。

Command 让你一步到位：既更新状态，又控制跳转。

但更重要的是，Command 解决了两个条件边完全做不到的场景：

Command 的价值不在"省一行条件边"，而在打通了条件边够不到的两个地方：

1. 图的边界 —— interrupt 让图能暂停等外部世界，resume 让外部世界能注入数据继续
2. 工具的内部 —— 工具不再只是"干活的"，还能"指路的"

简单记忆：Command 的杀手级场景 = 暂停恢复 + 工具指路，其他场景用条件边就好。

四个参数

使用场景一：中断后恢复（resume）

"""Command 杀手级场景一：interrupt + resume（中断恢复）图结构：START ──→ submit ──→ review(中断等审批) ──Command──→ execute(执行) ──→ END└──→ reject(拒绝) ──→ END执行时序：第1次 invoke: {"request": "申请休假"} → 执行到 review → interrupt 暂停第2次 invoke: Command(resume="批准")→ review 恢复→ 走 execute 分支或第2次 invoke: Command(resume="拒绝")→ review 恢复→ 走 reject 分支"""from typing import Literalfrom typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.types import Command, interruptfrom langgraph.checkpoint.memory import MemorySaverclass State(TypedDict):request: str # 用户提交的请求approval: str# 审批结果："批准" 或 "拒绝"result: str# 最终处理结果def submit(state: State) -> dict:"""提交请求（只是打印，不做实际处理）"""print(f"? 提交请求: {state['request']}")return {}def review(state: State) -> Command[Literal["execute", "reject"]]:"""人工审批节点 —— 整个 demo 的核心执行流程：1. interrupt("请审批此请求") 暂停整个图 - 图的状态被 checkpointer 保存 - invoke() 返回，包含 __interrupt__ 信息 - 此时图"冻结"在 review 节点2. 外部决定审批结果后，再次 invoke(Command(resume="批准/拒绝")) - interrupt() 的返回值就是 resume 传入的值 - 图从冻结点恢复，继续执行 review 节点的剩余代码3. 根据 approval 值返回 Command，同时更新状态 + 控制跳转 - "批准" → Command(update={"approval": "批准"}, goto="execute") - 其他 → Command(update={"approval": "拒绝"}, goto="reject")"""print("⏸️等待人工审批...")# interrupt() 暂停执行，返回值由外部 Command(resume=...) 传入# 第一次执行到这里时，图暂停，approval 还没有值# 第二次用 Command(resume="批准") 恢复时，approval = "批准"approval = interrupt("请审批此请求")if approval == "批准":return Command(update={"approval": "批准"}, # 更新状态：记录审批结果goto="execute"# 跳转：去执行请求)else:return Command(update={"approval": "拒绝"}, # 更新状态：记录审批结果goto="reject" # 跳转：去拒绝请求)def execute(state: State) -> dict:"""执行请求（审批通过后走这里）注意：execute 不是"同意"的意思，而是"执行"- review 节点负责审批（同意/拒绝）- execute 节点负责把同意的请求落地执行- reject 节点负责处理被拒绝的请求"""print(f"✅ 执行请求: {state['request']}（已{state['approval']}）")return {"result": f"已执行: {state['request']}"}def reject(state: State) -> dict:"""拒绝请求（审批不通过走这里）"""print(f"❌ 拒绝请求: {state['request']}（已{state['approval']}）")return {"result": f"已拒绝: {state['request']}"}graph = (StateGraph(State).add_node("submit", submit).add_node("review", review).add_node("execute", execute).add_node("reject", reject).add_edge(START, "submit").add_edge("submit", "review")# review 返回 Command，自带路由信息，不需要 add_conditional_edges.add_edge("execute", END).add_edge("reject", END)# ⚠️ 必须加 checkpointer！# interrupt 依赖 checkpointer 保存中断时的状态# 没有 checkpointer，interrupt 无法工作.compile(checkpointer=MemorySaver()))graph.get_graph().print_ascii()if __name__ == "__main__":# ── 第一次调用：执行到 review 节点时被 interrupt 暂停 ──print("=" * 50)print("第一次调用：提交请求，遇到 interrupt 暂停")print("=" * 50)# config 中的 thread_id 标识一个执行线程# 同一个 thread_id 的多次 invoke 共享状态（由 checkpointer 管理）config = {"configurable": {"thread_id": "1"}}result1 = graph.invoke({"request": "申请休假3天"}, config)# 此时图在 review 节点暂停了# result1 包含 __interrupt__ 信息，表示图被中断了print(f"暂停后的状态: {result1}")# ── 第二次调用：用 Command(resume=...) 传入审批结果，恢复执行 ──print("n" + "=" * 50)print("第二次调用：传入审批结果，恢复执行")print("=" * 50)# Command(resume="批准") 做了两件事：# 1. 把 "批准" 作为 interrupt() 的返回值注入# 2. 从 checkpointer 恢复之前保存的状态，继续执行 review 节点# 试试改成 Command(resume="拒绝") 看不同结果result2 = graph.invoke(Command(resume="批准"), config)print(f"n最终结果: {result2['result']}")# ── 再演示一次拒绝的场景 ──print("n" + "=" * 50)print("再试一次：拒绝场景")print("=" * 50)# 用不同的 thread_id，开启一个新的执行线程config2 = {"configurable": {"thread_id": "2"}}graph.invoke({"request": "申请加薪50%"}, config2)result3 = graph.invoke(Command(resume="拒绝"), config2)print(f"n最终结果: {result3['result']}")

⚠️ 重要注意事项：返回 Command 时必须加类型注解 Command[Literal["node_b", "node_c"]]

interrupt() 的本质机制

interrupt() 不是普通函数，它是一个特殊的控制流指令，效果类似 Python 的 yield：

第一次 invoke 执行到 interrupt() 时：approval = interrupt("请审批此请求")# ↑# 到这里就"断电"了！# 下面的 if/else 根本不会执行# 图的状态被 checkpointer 保存# invoke() 立即返回第二次 invoke(Command(resume="批准")) 时：approval = interrupt("请审批此请求")# ↑# 这里"来电"了！resume 的值 "批准" 成为 interrupt() 的返回值# approval = "批准"# 继续执行下面的 if/else

interrupt() 做了三件事：

1. 保存现场 —— 把当前状态写入 checkpointer（所以必须加 MemorySaver）
2. 抛出中断信号 —— 图的执行引擎捕获这个信号，停止推进
3. 等待恢复 —— 下次 invoke(Command(resume=...)) 时，resume 的值作为 interrupt() 的返回值注入，从断点继续执行

简单记忆：interrupt() = 图的"暂停键"，Command(resume=...) = 图的"播放键"，checkpointer = 保存暂停位置的"存档卡"。

使用场景二：从工具返回

from langgraph.types import Command@tooldef my_tool(some_input: str) -> Command[Literal["next_node"]]:result = do_something(some_input)return Command(update={"tool_result": result},# 更新状态goto="next_node"# 工具执行完跳到指定节点)

什么时候用 Command，什么时候用条件边？

8.3 Runtime Context —— 节点的随身工具箱

解决什么问题？

之前节点函数只能拿到 state 和 config。但有些信息不属于业务状态，却又是节点执行时需要的：

Runtime Context 就是把这些"不属于 State 但节点又需要的东西"统一打包，以类型安全的方式注入节点。

怎么理解？

State 是节点间传递的"货物"，Runtime Context 是节点执行时的"随身工具箱"。

货物（State）会随着图的执行流动、变化；工具箱（Runtime Context）是每次执行时注入的辅助信息，节点只读不改。

Runtime 包含什么？

from langgraph.runtime import Runtime, get_runtime# 在节点函数中，runtime 包含以下内容：runtime.context # ? 自定义上下文（如 user_id, db_conn）runtime.store # ? 持久化存储（BaseStore）runtime.stream_writer # ? 自定义流式输出runtime.previous# ⏮️ 上一次该线程的返回值runtime.execution_info# ℹ️ 执行元信息（只读）runtime.heartbeat # ? 心跳（防止长任务超时）runtime.control # ?️ 运行控制（如请求优雅停止）

最常用：自定义上下文 context

这是你自己定义的只读上下文，类似依赖注入：

from dataclasses import dataclassfrom langgraph.graph import StateGraphfrom langgraph.runtime import Runtime# 第一步：定义上下文结构@dataclassclass Context:user_id: strdb_conn: strclass State(TypedDict):response: str# 第二步：告诉 StateGraph 上下文的结构graph = StateGraph(state_schema=State, context_schema=Context)# 第三步：节点函数第二个参数接收 runtimedef my_node(state: State, runtime: Runtime[Context]) -> dict:user_id = runtime.context.user_id# 类型安全！IDE 有提示db_conn = runtime.context.db_connreturn {"response": f"Hello, {user_id}!"}graph.add_node("my_node", my_node)# ...# 第四步：调用时传入上下文result = graph.invoke({}, context=Context(user_id="alice", db_conn="postgres://..."))

和 State 的区别：

• State：节点可读可写，在节点间流动
• Context：节点只读，整个运行期间不变

和 config 的区别：

• config：无类型提示，config["configurable"]["user_id"] 容易拼错
• context：有类型提示，runtime.context.user_id IDE 自动补全

持久化存储 store

跨线程/跨会话的长期记忆，不同于 State（单次运行）和 Checkpointer（单线程状态）：

from langgraph.store.memory import InMemoryStorestore = InMemoryStore()store.put(("users",), "alice", {"name": "Alice"})def my_node(state: State, runtime: Runtime) -> dict:if memory := runtime.store.get(("users",), "alice"):name = memory.value["name"]return {"response": f"Hello, {name}!"}graph = ... .compile(store=store)

自定义流式输出 stream_writer

在节点执行过程中，向客户端推送自定义数据（不等节点执行完）：

def my_node(state: State, runtime: Runtime) -> dict:runtime.stream_writer({"progress": "50%"}) # 中途推送进度# ... 做一些耗时操作 ...runtime.stream_writer({"progress": "100%"})return {"result": "done"}# 调用端用 stream_mode="custom" 接收for chunk in graph.stream({"input": "..."}, stream_mode="custom"):print(chunk)# {"progress": "50%"}→{"progress": "100%"}

执行元信息 execution_info

当前执行的元数据，调试和日志时很有用：

def my_node(state: State, runtime: Runtime) -> dict:info = runtime.execution_infoprint(f"线程: {info.thread_id}")print(f"检查点: {info.checkpoint_id}")print(f"任务: {info.task_id}")print(f"第几次重试: {info.node_attempt}")# 配合 RetryPolicy 使用return {}

两种访问方式

# 方式一：作为节点函数参数（推荐，类型安全）def my_node(state: State, runtime: Runtime[Context]) -> dict:user_id = runtime.context.user_id# 方式二：在函数内部调用 get_runtime()（适用于无法改签名的场景）from langgraph.runtime import get_runtimedef my_node(state: State) -> dict:runtime = get_runtime(Context)user_id = runtime.context.user_id

三种"数据载体"对比

简单记忆：State 是"货物"，Runtime Context 是"工具箱"。货物在站点间流转，工具箱是每个工人随身携带的。

8.4 三个高级原语的选用指南

需要"运行时动态分发到 N 个节点"→ Send需要"同时更新状态 + 控制跳转"→ Command需要"在节点中访问运行时能力" → Runtime Context

九、速查表

9.1 核心方法参数速查

9.2 三层机制速查

9.3 高级原语速查