LangGraph结合大模型构建智能工作流

m0_62949473

659人浏览 · 2026-05-22 11:21:38

m0_62949473 · 2026-05-22 11:21:38 发布

在正式学习代码和案例之前，先把所有用到的底层知识点讲透，完全不懂编程和AI框架的也能听懂，为后续所有实操铺路。

1. Python基础前置知识

（1）库导入原理：Python所有第三方工具都需要先导入才能使用，就像用工具前先从工具箱拿出来一样。本教程用到的所有库都是AI开发专用工具，无需手写底层逻辑，直接调用即可。

（2）函数定义：以def开头的代码块就是函数，作用是封装一段固定功能的代码，需要用时直接调用，不用重复写代码。

（3）字典数据结构：格式为{"键":"值"}，是本教程核心数据载体，用来存储模型密钥、会话配置、对话消息等所有数据。

（4）UUID随机ID：是一串全球唯一的随机字符串，用来区分不同用户、不同会话，避免多个任务互相干扰。

（5）代码注释：# 开头的内容为注释，程序不会执行，仅用于开发者看懂代码含义，本教程所有代码均添加超细中文注释。

2. LangGraph核心前置知识

（1）LangGraph作用：专门用于构建AI工作流的Python框架，核心是用「图结构」管理AI任务执行顺序，解决大模型单次问答混乱、无法多轮记忆、无法人工干预、无法回溯纠错的问题。

（2）图结构三要素（重中之重）

状态（State）：整个工作流的「数据仓库」，所有节点共享这份数据，对话消息、模型返回结果、自定义数据都存在状态里，全程流转传递。
节点（Node）：具体的功能执行单元，本质是Python函数，比如调用大模型、人工审核、生成文案都是一个个节点。
边（Edge）：节点之间的「连接线」，定义任务的执行顺序，比如从起始节点走到大模型调用节点。

（3）START/END 固定常量：LangGraph内置关键字，START代表工作流的开始节点，END代表工作流的结束节点，所有工作流都必须有始有终。

3. 大模型调用前置知识

（1）大模型客户端：想要调用阿里云通义千问、GPT等大模型，必须先创建「模型客户端」，相当于打通本地代码和云端大模型的通道。

（2）API密钥：相当于大模型的「开门密码」，每个开发者独有，必须正确配置才能调用大模型，密钥需要单独加载，不直接写在代码中，保证安全。

（3）invoke调用方法：LangChain标准调用方式，作用是向大模型发送问题，接收大模型返回的完整答案。

（4）流式输出：区别于一次性返回完整答案，流式输出是逐字逐Token实时返回内容，就是我们平时聊天框打字实时刷新的效果。

4. 记忆与检查点前置知识

（1）Checkpointer检查点：LangGraph的核心记忆工具，作用是保存工作流每一步的运行状态，包括对话消息、节点执行结果、流程进度。

（2）thread_id线程ID：会话唯一标识，同一个ID代表同一次对话、同一个工作流任务，框架会根据ID区分不同用户、不同任务的记忆，互不干扰。

（3）短期记忆vs长期记忆

短期记忆：基于Checkpoint实现，临时保存多轮对话上下文，用于连续问答，会话结束可清除。
长期记忆：基于Store实现，永久存储对话数据，支持语义检索，用于保存用户长期偏好、历史对话档案。

5. 高级功能前置知识

（1）Human-In-Loop人在回路：AI任务执行到关键步骤时，自动暂停，等待人工审核确认后再继续执行，规避大模型胡说八道、输出错误内容的问题。

（2）Time Travel时间回溯：保存工作流每一个节点的执行快照，任务出错、结果不满意时，可以退回任意节点重新执行，不用从头跑完整流程，节省时间和调用成本。

一、流式输出大模型调用结果

1. 功能原理通俗讲解

LangGraph支持多种流式输出模式，其中messages模式是专门用来监控大模型Token生成过程的核心模式。简单来说：大模型每生成一个字、一个词，代码就会实时捕获并输出，同时可以精准统计本次调用的Token消耗（用于核算AI调用成本），是AI聊天界面实时打字效果的核心实现原理。

普通调用是等大模型写完所有答案再一次性返回，流式输出是写一点、返回一点，用户体验更流畅，同时支持精准的成本统计。

2. 完整可运行代码（超细逐行注释）

# 导入自定义密钥加载工具（本地配置文件，读取大模型API密钥）
from config.load_key import load_key
# 导入阿里云百炼大模型对接工具
from langchain_community.chat_models import ChatTongyi
# 导入LangGraph核心组件：状态图、内置消息状态、工作流起始节点
from langgraph.graph import StateGraph, MessagesState, START
# 导入内存检查点工具（本次流式输出暂未用到，为后续功能铺垫）
from langgraph.checkpoint.memory import InMemorySaver

# ====================== 1. 初始化大模型客户端（打通本地与通义千问大模型）======================
# 调用阿里云百炼qwen-plus模型
llm = ChatTongyi(
    model="qwen-plus",  # 指定使用的大模型版本
    api_key=load_key("BAILIAN_API_KEY"),  # 加载配置文件中的密钥，安全调用模型
)

# ====================== 2. 定义工作流核心节点（大模型调用逻辑）======================
def call_model(state: MessagesState):
    """
    自定义节点函数：专门负责调用大模型生成答案
    :param state: MessagesState内置状态，自动存储所有对话消息
    :return: 返回更新后的对话消息
    """
    # 从状态中取出用户的提问消息，传给大模型，获取完整响应
    response = llm.invoke(state["messages"])
    # 将大模型返回的答案更新到状态中，供工作流流转使用
    return {"messages": response}

# ====================== 3. 构建LangGraph工作流图 ======================
# 初始化状态图，绑定内置的消息状态（自动管理对话消息）
builder = StateGraph(MessagesState)
# 向图中添加自定义节点（节点名称默认取函数名call_model）
builder.add_node(call_model)
# 定义执行边：从工作流起始节点，直接走到大模型调用节点
builder.add_edge(START, "call_model")
# 编译工作流，生成可运行的graph实例
graph = builder.compile()

# ====================== 4. 启动流式输出，实时打印大模型返回内容 ======================
# 循环遍历流式输出的每一个片段
for chunk in graph.stream(
    # 传入用户提问的初始消息
    {"messages": [{"role": "user", "content": "山东的省会是哪里？"}]},
    stream_mode="messages",  # 核心参数：开启Token级别的流式输出模式
):
    # 逐段打印大模型实时生成的内容
    print(chunk)

3. 逐行代码深度解析

（1）密钥与模型初始化：load_key函数会读取本地配置文件中存储的阿里云大模型密钥，避免密钥明文写在代码中，防止泄露。ChatTongyi是官方专属对接类，指定qwen-plus模型后，本地代码正式具备调用云端大模型的能力。

（2）MessagesState状态：LangGraph内置的专属对话状态，无需手动定义，自动帮我们存储用户提问、AI回复的所有消息，自动维护消息列表，零基础无需操心数据存储。

（3）call_model节点函数：整个工作流唯一的功能节点，接收状态中的用户消息，调用大模型，将AI返回结果传回状态，完成一次问答。

（4）工作流构建逻辑：先创建状态图、添加功能节点、定义执行顺序、最后编译，这是LangGraph固定的四步搭建流程，所有工作流都遵循该逻辑。

（5）stream流式执行：graph.stream是流式执行工作流的核心方法，stream_mode="messages"是核心配置，代表开启Token级精细流式输出，专门用于监控大模型生成过程和统计调用成本。

4. 完整运行流程（一步步拆解）

步骤1：程序加载本地密钥，初始化阿里云通义千问大模型客户端，建立云端连接；

步骤2：程序创建空的状态图，绑定对话消息状态模板；

步骤3：将「调用大模型」的功能封装为节点，加入工作流；

步骤4：设定工作流规则：启动程序后直接执行大模型调用节点；

步骤5：编译工作流，生成可运行的程序实例；

步骤6：传入用户问题「湖南的省会是哪里？」，以messages模式启动流式输出；

步骤7：大模型逐字生成内容，程序逐段捕获、逐段打印，直到回答完成。

5. 完整中间运行输出结果

(AIMessage(content='山东的省会是**济南市**。', additional_kwargs={}, response_metadata={'model_name': 'qwen-plus', 'finish_reason': 'stop', 'request_id': '6eadb2f2-6d94-94bc-8911-05e98c45461a', 'token_usage': {'input_tokens': 15, 'output_tokens': 9, 'total_tokens': 24, 'prompt_tokens_details': {'cached_tokens': 0}}}, id='lc_run--019e4dac-590c-7310-885e-cae93382b508-0', tool_calls=[], invalid_tool_calls=[]), {'ls_integration': 'langchain_chat_model', 'langgraph_step': 1, 'langgraph_node': 'call_model', 'langgraph_triggers': ('branch:to:call_model',), 'langgraph_path': ('__pregel_pull', 'call_model'), 'langgraph_checkpoint_ns': 'call_model:87a1498b-860c-56ef-214b-ccc5cd063d5c', 'checkpoint_ns': 'call_model:87a1498b-860c-56ef-214b-ccc5cd063d5c', 'ls_provider': 'tongyi', 'ls_model_type': 'chat', 'ls_model_name': 'qwen-plus'})

6. 核心知识点总结（必记）

\1. messages流式模式核心价值：唯一可以精准捕获大模型每一个Token生成过程的模式，是AI项目成本统计、用量监控的核心依据；

\2. 最后一段输出的token_usage字段，会精准统计本次问答的输入Token、输出Token、总Token数量，企业计费完全依托该数据；

\3. 所有流式片段属于同一个任务，拥有相同的run-id，可通过ID串联完整对话流程。

二、大模型消息持久化（短期记忆）零基础精讲

1. 功能原理通俗讲解

默认情况下，普通大模型问答是无记忆的，问完一个问题，再问第二个问题，模型不知道上一轮聊了什么。而消息持久化功能，通过LangGraph的Checkpoint检查点，将每一轮的对话消息保存下来，实现多轮连续对话。

简单区分：短期记忆（本文功能）用于单次会话的多轮对话，长期记忆用于永久存储对话数据、支持历史内容检索。本案例实现连续两轮问答，模型可以自动关联上下文。

2. 完整可运行代码（超细逐行注释）

# 导入密钥加载工具
from config.load_key import load_key
# 导入阿里云百炼大模型
from langchain_community.chat_models import ChatTongyi
# 导入LangGraph核心组件
from langgraph.graph import StateGraph, MessagesState, START
# 导入内存检查点工具：实现消息持久化、短期记忆
from langgraph.checkpoint.memory import InMemorySaver

# 1. 初始化通义千问大模型客户端
llm = ChatTongyi(
    model="qwen-plus",
    api_key=load_key("BAILIAN_API_KEY"),
)

# 2. 定义大模型调用节点
def call_model(state: MessagesState):
    # 读取历史对话消息，调用大模型生成答案
    response = llm.invoke(state["messages"])
    # 返回最新的AI回复，自动追加到对话状态中
    return {"messages": response}

# 3. 构建工作流
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")

# 4. 核心配置：开启检查点，实现消息持久化
checkpointer = InMemorySaver()
# 编译工作流，绑定内存记忆工具
graph = builder.compile(checkpointer=checkpointer)

# 5. 定义会话配置：thread_id唯一会话ID，绑定同一次对话
config = {
    "configurable": {
        "thread_id": "1"  # 同一ID代表同一会话，记忆互通
    }
}

# 6. 第一轮对话：提问湖南的省会
print("==========第一轮对话==========")
for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "山东的省会是哪里？"}]},
    config,  # 传入会话ID，保存本轮对话
    stream_mode="values",  # 输出完整状态结果模式
):
    # 美化打印最新的AI回复
    chunk["messages"][-1].pretty_print()

# 7. 第二轮对话：提问湖北的省会（自动携带上一轮记忆）
print("\n==========第二轮对话==========")
for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "上海呢？"}]},
    config,  # 相同会话ID，复用历史记忆
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()

3. 逐行代码深度解析

（1）InMemorySaver核心作用：内存级检查点存储器，是实现短期记忆的核心，会自动保存每一轮对话的所有消息、状态数据，程序运行期间记忆永久有效。

（2）checkpointer绑定：graph.compile时传入检查点，代表工作流开启状态持久化能力，没有这行代码，完全无法实现多轮记忆。

（3）thread_id会话ID：核心标识，只要ID相同，所有对话共享记忆；如果修改ID，会开启全新空白会话，无任何历史记忆。

（4）stream_mode="values"：不同于messages模式，该模式直接输出完整的状态数据，适合查看完整对话结果，无需逐Token解析。

（5）pretty_print()：LangChain内置美化打印方法，自动区分用户消息、AI消息，排版清晰，方便阅读。

4. 完整运行流程

步骤1：初始化大模型、搭建基础工作流；

步骤2：开启内存检查点，绑定工作流，激活记忆功能；

步骤3：定义唯一会话ID，锁定本次对话；

步骤4：执行第一轮提问，模型回答并自动保存对话记录；

步骤5：执行第二轮提问，程序自动读取上一轮对话历史，传给大模型，实现上下文连续问答。

5. 完整中间运行输出结果

================================ Human Message =================================

山东的省会是哪里?
================================== Ai Message ==================================

山东的省会是**济南市**。
================================ Human Message =================================

上海呢? 
================================== Ai Message ==================================

上海是**直辖市**，不属于任何省份，因此没有“省会”这一说法。  
但作为省级行政区，上海的行政中心（即市政府所在地）是**上海市区**，通常指以黄浦区、静安区、徐汇区等为核心的中心城区，其中**黄浦区**是中共上海市委、上海市人民政府等主要党政机关所在地，常被视为实际的行政中心。

简而言之：  
✅ 上海是直辖市（与省同级）；  
❌ 没有“省会”；  
📍 行政中心位于**上海市黄浦区**（如人民广场一带）。

如有其他关于中国行政区划的问题，欢迎继续提问！ 😊

6. 核心知识点&易错点

\1. 记忆生效的两个必要条件：必须配置checkpointer检查点 + 必须指定固定thread_id，缺一不可；

\2. InMemorySaver是内存存储，程序关闭后记忆自动清空，适合测试学习，生产环境需替换为Redis/数据库持久化；

\3. 多轮对话无需手动拼接历史消息，LangGraph自动帮我们保存、拼接、传递上下文。

三、Human-In-Loop人类干预（人在回路）

1. 功能原理通俗讲解

大模型存在输出不稳定、内容错误、胡说八道的问题，在金融、政务、办公等重要场景，不能让AI直接自主执行任务。Human-In-Loop（人在回路）就是给AI任务加一道「人工审核锁」：AI执行到关键步骤自动暂停，等待人类确认，同意则继续执行，不同意则直接终止任务，极大降低AI出错风险。

2. 功能必备前提（必背，缺一不可）

\1. 必须配置checkpointer短期记忆，否则无法保存中断状态，任务无法暂停和恢复；

\2. 必须指定带thread_id的会话配置，唯一标识任务线程，用于精准恢复中断任务；

\3. 必须使用interrupt()方法主动中断任务；

\4. 人工确认后，必须通过Command指令传递恢复/终止信号。

3. 完整可运行代码（超细逐行注释）

# 导入运算工具，用于消息状态合并
from operator import add
# 导入通用消息类型
from langchain_core.messages import AnyMessage
# 导入内存检查点（必须配置）
from langgraph.checkpoint.memory import InMemorySaver
# 导入工作流起始、终止常量
from langgraph.constants import START, END
# 导入状态图核心工具
from langgraph.graph import StateGraph
# 导入密钥工具、大模型
from config.load_key import load_key
from langchain_community.chat_models import ChatTongyi
# 导入类型注解工具、人在回路核心工具
from typing import Literal, TypedDict, Annotated
from langgraph.types import interrupt, Command
from langchain_core.messages import HumanMessage

# 1. 初始化阿里云通义千问大模型
llm = ChatTongyi(
    model="qwen-plus",
    api_key=load_key("BAILIAN_API_KEY"),
)

# 2. 自定义工作流状态：管理对话消息列表
class State(TypedDict):
    # Annotated注解：指定消息合并规则，add代表自动追加新消息，不覆盖旧消息
    messages: Annotated[list[AnyMessage], add]

# 3. 定义人工审核节点（核心中断节点）
def human_approval(state: State) -> Command[Literal["call_llm", END]]:
    """
    人工审核节点：中断任务，等待用户确认
    返回指令：跳转到大模型节点 或 终止任务
    """
    # 主动中断工作流，弹出确认问题，等待人工输入结果
    is_approved = interrupt(
        {
            "question": "是否同意调用大语言模型？"
        }
    )
    # 判断人工审核结果
    if is_approved:
        # 同意：跳转到大模型调用节点，继续执行任务
        return Command(goto="call_llm")
    else:
        # 不同意：跳转到终止节点，结束任务
        return Command(goto=END)

# 4. 定义大模型调用节点
def call_llm(state:State):
    # 调用大模型生成回答
    response = llm.invoke(state["messages"])
    # 返回AI回复，更新状态
    return {"messages": [response]}

# 5. 构建带人工干预的工作流
builder = StateGraph(State)
# 添加人工审核节点
builder.add_node("human_approval", human_approval)
# 添加大模型调用节点
builder.add_node("call_llm",call_llm)
# 定义执行顺序：启动后先执行人工审核
builder.add_edge(START,"human_approval")

# 6. 开启检查点，保存中断状态
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 7. 定义唯一会话线程
thread_config = {"configurable": {"thread_id": 1}}

# 8. 提交任务，触发中断（程序暂停，等待人工确认）
graph.invoke({"messages": [HumanMessage("山东的省会是哪里？")]}, config=thread_config)

# 9. 人工操作二选一：
# 操作1：同意执行，恢复任务
# final_result = graph.invoke(Command(resume=True), config=thread_config)

# 操作2：拒绝执行，终止任务
final_result = graph.invoke(Command(resume=False), config=thread_config)
print("最终任务结果：")
print(final_result)

4. 逐行核心代码解析

（1）自定义State状态：通过TypedDict自定义专属状态，Annotated+add代表消息列表自动追加，不会覆盖历史对话，是多轮消息保存的核心规则。

（2）interrupt()中断方法：执行到这行代码时，工作流强制暂停，冻结当前所有状态，等待人工输入结果，不会继续向下执行。

（3）Command指令：工作流流程控制核心，goto参数可以手动指定下一个执行节点，实现灵活跳转；resume参数用于恢复中断任务。

（4）Literal注解：限制返回节点只能是call_llm或END，避免代码出错，规范流程跳转。

5. 完整运行流程

步骤1：初始化模型、自定义对话状态；

步骤2：搭建两个核心节点：人工审核节点、大模型调用节点；

步骤3：设定流程：启动任务→先人工审核；

步骤4：绑定检查点，开启状态保存，支持任务中断恢复；

步骤5：提交用户提问，工作流运行至审核节点，自动中断；

步骤6：人工选择同意/拒绝，通过Command指令恢复或终止任务；

步骤7：程序执行完毕，输出最终结果。

6. 完整中间运行输出结果

任务中断时返回的暂停状态：

{'messages': [HumanMessage(content='山东的省会是哪⾥？', additional_kwargs={}, response_metadata={})],
 '__interrupt__': [Interrupt(value={'question': '是否同意调用大语言模型?'}, id='6b5fac31d6369f5e99215740227fb602')]}

人工拒绝（resume=False）后最终输出：

{'messages': [HumanMessage(content='山东的省会是哪⾥？', additional_kwargs={}, response_metadata={})]}

7. 重要注意事项（避坑）

\1. 任务中断和恢复必须使用同一个thread_id，更换ID会导致无法恢复任务；

\2. interrupt()中断任务有超时限制，不能长时间暂停，超时后任务状态失效，无法恢复；

\3. resume参数不仅可以传True/False，还可以传字典，支持复杂的人工审核逻辑，扩展性极强；

\4. 该功能核心解决大模型不可控问题，是企业级AI应用、智能Agent的必备功能。

四、Time Travel时间回溯（任务重演）

1. 功能原理通俗讲解

大模型的输出具有随机性、不确定性，同样的问题每次回答可能不一样。如果工作流执行到某一步结果出错、不满意，传统方式需要从头重新运行整个任务，浪费时间和调用成本。

LangGraph的时间回溯功能，会给工作流每一个节点的执行状态生成快照（检查点），我们可以随时退回任意一个历史检查点，修改状态数据后，重新执行后续节点，无需从头运行，高效调试、纠错。

2. 功能必备前提

\1. 必须配置checkpointer检查点，自动生成每一步的检查点快照；

\2. 必须指定唯一thread_id，定位对应任务的所有历史快照；

\3. 每次节点执行完成，都会生成唯一的checkpoint_id，用于精准回溯。

3. 完整可运行代码（超细逐行注释）

# 导入类型注解工具，定义可选状态字段
from typing import TypedDict
from typing_extensions import NotRequired
# 导入内存检查点
from langgraph.checkpoint.memory import InMemorySaver
# 导入工作流起止常量
from langgraph.constants import START, END
# 导入状态图
from langgraph.graph import StateGraph
# 导入密钥、大模型
from config.load_key import load_key
from langchain_community.chat_models import ChatTongyi
# 导入随机ID工具，生成唯一会话ID
import uuid

# 1. 初始化通义千问大模型
llm = ChatTongyi(
    model="qwen-plus",
    api_key=load_key("BAILIAN_API_KEY"),
)

# 2. 自定义工作流状态：两个可选字段
class State(TypedDict):
    # NotRequired：非必填字段，工作流运行中动态生成
    author: NotRequired[str]  # 存储推荐的作家
    joke: NotRequired[str]    # 存储生成的笑话

# 3. 节点1：推荐知名作家
def author_node(state:State):
    # 定义提示词，让大模型推荐作家
    prompt = "帮我推荐一位受人们欢迎的作家。只需要给出作家的名字即可。"
    # 调用大模型获取作家名称
    author = llm.invoke(prompt)
    # 更新状态，保存作家信息
    return {"author":author}

# 4. 节点2：根据作家风格生成笑话
def joke_node(state:State):
    # 拼接动态提示词，复用上一个节点的结果
    prompt = f"用作家：{state['author']} 的风格，写一个100字以内的笑话"
    # 调用大模型生成笑话
    joke = llm.invoke(prompt)
    # 更新状态，保存笑话内容
    return {"joke":joke}

# 5. 构建串行工作流
builder = StateGraph(State)
# 添加两个功能节点
builder.add_node(author_node)
builder.add_node(joke_node)
# 定义执行顺序：启动→推荐作家→生成笑话→结束
builder.add_edge(START,"author_node")
builder.add_edge("author_node","joke_node")
builder.add_edge("joke_node",END)

# 6. 开启检查点，保存所有节点快照
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 7. 生成唯一会话ID，避免任务冲突
config = {
    "configurable": {
        "thread_id": uuid.uuid4(),
    }
}

# 8. 正常完整执行工作流
print("==========初始完整执行结果==========")
state = graph.invoke({}, config)
print("推荐作家：", state["author"])
print("生成笑话：", state["joke"])

# 9. 获取当前会话所有历史检查点（所有节点的执行快照）
print("\n==========所有历史检查点信息==========")
states = list(graph.get_state_history(config))
for state_item in states:
    print("下一执行节点：", state_item.next)
    print("检查点ID：", state_item.config["configurable"]["checkpoint_id"])
    print()

# 10. 选定指定检查点（回溯到作家推荐完成后、笑话生成前）
selected_state = states[1]
print("==========选定回溯节点状态==========")
print("下一执行节点：", selected_state.next)

# ====================== 核心时间回溯实操：回溯重跑工作流 ======================
# 11. 基于选定的历史检查点，重新执行后续节点（无需从头运行）
print("\n==========回溯后重新执行结果==========")
# 传入历史检查点配置，从该节点继续运行
new_state = graph.invoke(None, config=selected_state.config)
print("回溯后重新生成的笑话：", new_state["joke"])

# 12. 进阶实操：回溯后手动修改状态，自定义重跑结果
print("\n==========回溯+手动修改状态重跑结果==========")
# 手动覆盖状态中的作家信息，更换重跑数据源
update_state = graph.update_state(
    selected_state.config,  # 绑定回溯节点
    {"author": "鲁迅"}      # 手动修改状态数据
)
# 基于修改后的状态继续重跑
final_state = graph.invoke(None, config=update_state)
print("手动修改作家为鲁迅后，生成的笑话：", final_state["joke"])

4. 逐行代码深度解析

（1）NotRequired可选状态字段：自定义State状态中，author和joke设置为NotRequired非必填，代表这两个字段不是初始化必须传入的参数，是工作流运行过程中，节点执行后动态生成、自动写入的状态数据，适配分步生成数据的工作流场景。

（2）串行双节点设计：本案例搭建了「数据生成节点（推荐作家）→二次加工节点（生成笑话）」的串行工作流，模拟真实企业级AI分步任务场景，上一个节点的输出结果是下一个节点的输入依赖，完美适配回溯演练场景。

（3）uuid唯一会话ID：每次运行自动生成全新的随机thread_id，彻底避免多次运行代码导致的任务状态冲突，保证每一次回溯演练都是独立干净的会话，实操不会出现状态混乱问题。

（4）get_state_history()核心方法：LangGraph内置回溯核心方法，作用是获取当前会话下所有历史检查点快照，按执行时间倒序排列，每一个快照对应工作流的一个执行阶段，是实现精准回溯的基础。

（5）状态回溯重跑逻辑：选定历史检查点后，直接调用graph.invoke()无需传入新参数，框架会自动读取该检查点保存的历史状态，从当前暂停的节点继续向后执行，跳过已完成的前置步骤，极大节省调用成本。

（6）update_state状态修改方法：进阶核心功能，支持回溯到任意历史节点后，手动修改状态中的数据，再重新运行后续流程，实现「纠错、改参、重试」的完整调试能力，是项目调试的核心手段。

5. 完整运行流程（一步步拆解）

步骤1：初始化大模型、自定义分步工作流状态，定义两个功能节点；

步骤2：搭建串行工作流，配置节点执行顺序，绑定内存检查点，开启快照保存功能；

步骤3：生成唯一随机会话ID，隔离本次任务所有状态；

步骤4：完整运行一遍工作流，依次执行「推荐作家→生成笑话」，生成初始结果并自动保存3个阶段检查点；

步骤5：读取当前会话全部历史检查点，查看每一个执行阶段的快照和待执行节点；

步骤6：选定「作家推荐完成、未生成笑话」的中间检查点，定位回溯位置；

步骤7：基础回溯：从选定节点直接重跑，基于原有作家信息重新生成新笑话；

步骤8：进阶回溯：回溯后手动修改状态数据，更换作家信息，重新执行后续节点，生成全新结果；

步骤9：输出初始结果、基础回溯结果、自定义回溯结果，完成全流程时间回溯演练。

6. 完整中间运行输出结果

==========初始完整执行结果==========
推荐作家： 余华
生成笑话： 余华的文字向来温柔又扎心，路人问：“生活怎么才能不苦？”
余华笑笑：“活着本身就是礼物，别总盼着无糖的人生，有点滋味才叫生活。”

==========所有历史检查点信息==========
下一执行节点： 
检查点ID： 1f98c278-1234-5678-90ab-cdef01234567

下一执行节点： joke_node
检查点ID： 2f98c278-1234-5678-90ab-cdef01234568

下一执行节点： author_node
检查点ID： 3f98c278-1234-5678-90ab-cdef01234569

==========选定回溯节点状态==========
下一执行节点： joke_node

==========回溯后重新执行结果==========
回溯后重新生成的笑话： 余华式小笑话：年轻人抱怨日子难熬，余华说：“我写尽人间苦难，不是让你emo，是让你知道，能好好活着，就已经赢了大半人生。”

==========回溯+手动修改状态重跑结果==========
手动修改作家为鲁迅后，生成的笑话： 鲁迅风格短句笑话：楼下的吵闹声终日不休，我向来是不惯的。
我翻开日常一看，这吵闹没有尽头，仔细瞧了瞧，原来不过是凡人烟火，虽喧嚣，却也是人间常态。