从流式输出到时间旅行，Module 3 构建了一套完整的
人机协作与可控 AI 执行体系。
默认的 invoke() 是阻塞式的——图跑完才返回。流式输出让你在图执行的每一步都能实时获取数据，是 Human-in-the-loop 的感知基础。
断点（Breakpoint）是在编译期固定下来的暂停位置，每次执行到该节点就无条件中断，等待外部决策后再继续——这是最直接的人机协作方式。
None 作为输入，从断点处恢复执行invoke(None) 中的 None 意思是"不新增输入，直接用保存的状态继续往下走"。断点只能"批准或拒绝"，而 update_state() 让你能在暂停期间直接修改图内部 State——纠正错误数据、注入人工输入、然后带着新状态继续执行。
add_messages 至关重要）update_state 时，传入相同 ID 的消息会覆盖原消息，而非追加as_node 解决了一个微妙问题：LangGraph 需要知道"是哪个节点执行了这次 State 更新"，才能正确计算下一步应该走哪条边。静态断点无条件暂停，而动态断点让节点内部代码根据运行时数据自行决定是否触发中断，并可携带自定义的中断原因信息传给外部。
value（字符串或字典），可用于说明中断原因interrupt()，否则节点正常执行invoke(None) 会重新触发同一中断；正确做法是先 update_state 写入处理结果再恢复LangGraph 的 Checkpointer 在每个节点执行后自动保存快照。时间旅行让你能回溯到任意历史时刻——重放原路径或分叉走向全新分支。
checkpoint_idconfig 传给 invoke()，按原来相同路径再走一遍update_state(past_config, new_values) 修改历史状态，再 invoke() 开启新执行分支不能实时观察图在做什么，就无法在合适的时机介入。stream_mode="updates" 是最轻量的方式，astream_events 是最细粒度的方式，按需选用。
对所有高风险操作（发邮件、写数据库、调用付费 API）设置 interrupt_before，是最简单的安全保障。配合 MemorySaver 才能正常工作。
精准修改单个字段，不影响其他状态。as_node 参数让 LangGraph 能正确计算更新后的下一步路由，别忘了它。
让数据决定是否暂停，而不是让代码决定。interrupt(value) 还能携带中断原因，让外部系统知道"为什么需要人工介入"。
只要挂载了 Checkpointer，每个节点都是自动保存点。时间旅行让"回退重试"从不可能变成一行代码的事，这是 LangGraph 区别于普通 Agent 框架的核心优势之一。
分叉执行创造了独立的执行历史分支，原来的 thread 不受影响。这是实验不同策略、进行 A/B 测试、以及从错误中恢复的强力工具。
一个集流式进度展示、人工审批、状态编辑、动态风险中断与历史回溯于一体的合同处理工作流。将 Module 3 全部 5 个核心知识点有机整合，模拟真实企业级审批场景。
# ── 依赖导入 ──────────────────────────────────────────────
from typing import Literal, Optional
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt # 【L4】动态断点函数
# ════════════════════════════════════════════════
# State：继承 MessagesState，扩展合同专用字段
# ════════════════════════════════════════════════
class ContractState(MessagesState):
contract_text: str # 合同全文
risk_score: float # 风险分数 0.0-1.0
key_clauses: list[str] # 提取的关键条款
decision: str # 审批决定：pending/approved/rejected
llm = ChatOpenAI(model="gpt-4o-mini")
memory = MemorySaver() # 【L2/L5】Checkpointer：断点+时间旅行的必要前提
# ── 节点函数 ──────────────────────────────────────────────
def analyze_document(state: ContractState):
"""解析合同，提取结构与关键条款"""
response = llm.invoke([
SystemMessage(content="你是一个专业的合同分析师。提取合同中的关键条款和风险点。"),
HumanMessage(content=f"请分析以下合同：\n{state['contract_text']}"),
])
clauses = response.content.split("\n")[:5]
return {"key_clauses": clauses, "messages": [response]}流式输出的每个 chunk 包含该节点对 State 的增量更新。ContractState 中每个字段的变化都会通过 stream_mode="updates" 实时推送，让外部能看到合同文本什么时候写入、风险分数何时计算完成。
断点触发后，图的执行状态必须被保存下来，等待恢复时才能从断点处继续。MemorySaver 就是这个状态存储——没有它，interrupt_before 会报错。同时它也是 L5 时间旅行的数据来源。
import asyncio
config = {"configurable": {"thread_id": "contract_review_001"}}
# ── 【L1】stream_mode="updates"：节点粒度实时输出 ────────
# 每个节点执行完后，立即推送该节点对 State 的增量变化
# 不需要等整个图跑完，用户能实时看到每一步的进展
for chunk in graph.stream(
{"contract_text": contract_text},
config=config,
stream_mode="updates",
):
node_name = list(chunk.keys())[0]
print(f"[{node_name}] 执行完成 → {list(chunk[node_name].keys())}")
# 输出示例：
# [analyze_document] 执行完成 → ['key_clauses', 'messages']
# ⏸ 图暂停（interrupt_before="submit_contract"）
# ── 【L1】astream_events：Token 级别流式输出 ─────────────
# 捕获 LLM 每生成一个 token 就立刻推送的事件
# 适合聊天界面需要"打字机"效果的场景
async def stream_tokens():
async for event in graph.astream_events(
{"contract_text": contract_text},
config=config,
version="v2",
):
if event["event"] == "on_chat_model_stream":
token = event["data"]["chunk"].content
print(token, end="", flush=True) # 逐 token 打印"updates"：增量字典，最小化数据量，生产首选。"values"：完整 State，调试时方便查看全貌。"messages"：只推消息列表的增量，专为聊天场景优化。三者性能开销依次递增。
事件名称遵循 on_<component>_<lifecycle> 格式。on_chat_model_stream 是 LLM 流式 token；on_tool_start/end 是工具调用；on_chain_start/end 是节点生命周期。用 event["name"] 过滤特定节点的事件。
# ════════════════════════════════════════════════
# 【L2】静态断点：编译期固定，每次提交前无条件暂停
# · interrupt_before=["submit_contract"]
# · 必须挂载 checkpointer，否则断点无法保存状态
# ════════════════════════════════════════════════
graph = builder.compile(
checkpointer=memory,
interrupt_before=["submit_contract"], # 在 submit 节点执行前暂停
)
# ── 第一次运行：触发断点后图自动暂停 ──
graph.invoke(
{"contract_text": contract_text},
config=config,
)
# ── 【L2】断点触发后：检查当前 State ──
snapshot = graph.get_state(config)
print("当前 State：", snapshot.values)
print("下一步节点：", snapshot.next) # ('submit_contract',)
# ── 【L3】用 update_state 修改合同条款（人工审核后纠错）──
# 相当于人类审核员直接在 State 上"划改"了一条条款
# as_node 告诉 LangGraph 这次更新来自哪个节点（影响路由）
graph.update_state(
config,
{"contract_text": "[已修正] 合同总金额：100万元（原：150万元）..."},
as_node="analyze_document", # 声明此更新来源于 analyze_document 节点
)
# ── 【L2】恢复执行：传入 None 作为新输入 ──
# None 意味着"不增加新输入，使用已保存的 State 继续"
# 图会从断点处（submit_contract 节点）继续往下执行
final_state = graph.invoke(None, config)graph.invoke(None, config) 中 None 表示"没有新的用户输入"。LangGraph 看到这个信号，会去 Checkpointer 里加载最新快照，从断点处的下一个节点继续执行。这是恢复执行的标准姿势。
LangGraph 用节点名来决定执行完后走哪条边。当你 update_state 时，必须告诉图"这次更新等价于哪个节点刚执行完"，否则图不知道下一步去哪。对消息字段使用 add_messages Reducer 时，as_node 还影响消息的追加 vs 覆盖行为。
# ════════════════════════════════════════════════
# 【L4】动态断点：运行时根据条件决定是否触发中断
# · 使用 interrupt() 函数，在节点内部条件性调用
# · 可携带任意信息作为中断原因传给外部审核人员
# ════════════════════════════════════════════════
def risk_assessment(state: ContractState):
"""计算风险分数，高风险时动态触发中断"""
# 简化的风险评分（生产中用专门的风险模型）
text = state["contract_text"]
score = 0.9 if "担保" in text or "不可抗力" in text else 0.3
# ── 【L4】只有高风险（score > 0.7）才触发中断 ──
# interrupt() 调用后，节点执行立即挂起
# 传入的字典会被存入检查点，外部可通过 get_state() 读取
if score > 0.7:
human_decision = interrupt({
"reason": "检测到高风险条款，需要法律顾问审核",
"risk_score": score,
"flagged_keywords": ["担保", "不可抗力"],
"contract_preview": text[:300],
})
# interrupt() 返回值是外部通过 update_state 注入的人工决定
return {"risk_score": score, "decision": human_decision}
# 低风险：直接通过，不打扰审批人员
return {"risk_score": score, "decision": "auto_approved"}
# ── 外部代码处理动态中断 ──────────────────────────────────
graph.invoke({"contract_text": risky_contract}, config=config)
# 读取中断原因
snapshot = graph.get_state(config)
interrupt_info = snapshot.tasks[0].interrupts[0].value
print(f"中断原因：{interrupt_info['reason']}")
print(f"风险分数：{interrupt_info['risk_score']}")
# ── 【L4】正确的恢复方式：先写入审批结果，再 invoke ──
# 直接 invoke(None) 会重新执行 risk_assessment 节点，再次触发 interrupt
# 正确做法：先用 update_state 向 State 写入人工决定，再恢复
graph.update_state(config, {"decision": "approved_with_amendments"})
graph.invoke(None, config) # 现在可以正常继续静态断点（L2）：每次经过该节点都暂停，无论数据内容。动态断点（L4）：只有满足运行时条件才暂停，低风险合同直接通过，完全不打扰审批人员。这让 Agent 系统更智能、审批效率更高。
动态断点被触发后，该节点处于"执行中途挂起"状态。如果直接 invoke(None)，LangGraph 会重新执行该节点，再次遇到 interrupt()，死循环。正确做法：先用 update_state 写入一个值，让节点"感知到"人工决定已到达，再恢复。
# ════════════════════════════════════════════════
# 【L5】时间旅行：浏览 + 重放 + 分叉
# · 每个节点执行后 Checkpointer 自动保存快照
# · 所有快照构成一条时间线，每条有唯一 checkpoint_id
# ════════════════════════════════════════════════
# ── 浏览完整执行历史 ──────────────────────────────────────
history = list(graph.get_state_history(config)) # 最新在前
for i, checkpoint in enumerate(history):
ckpt_id = checkpoint.config["configurable"]["checkpoint_id"]
decision = checkpoint.values.get("decision", "pending")
risk = checkpoint.values.get("risk_score", "N/A")
print(f"[{i}] checkpoint_id={ckpt_id[:12]}... decision={decision} risk={risk}")
# ── Replay（重放）：从某个历史检查点重新走相同路径 ─────────
# 取 analyze_document 节点刚执行完的那个快照
analyze_done_ckpt = history[-2] # 倒数第2个 = 分析完成时的快照
# 传入历史快照的 config（包含 checkpoint_id）来重放
replayed = graph.invoke(None, analyze_done_ckpt.config)
# 结果与历史一致，因为 State 数据相同，图的决策路径也相同
# ── Fork（分叉）：修改历史状态后开启新的执行分支 ───────────
# 场景：回到"分析完成"那个时刻，改变合同文本，看结果如何变化
# Step 1：找到目标历史检查点
target = analyze_done_ckpt
# Step 2：在历史快照上执行 update_state，修改合同内容
# 这会创建一个新检查点（分叉点），原来的历史时间线不受影响
forked_config = graph.update_state(
target.config,
{"contract_text": "[修改版] 已去除担保条款的低风险合同..."},
as_node="analyze_document",
)
# Step 3：从分叉点继续执行，走向全新的执行分支
forked_result = graph.invoke(None, forked_config)
print(f"分叉后风险分数：{forked_result['risk_score']}") # 预计 0.3（低风险）
print(f"分叉后决定：{forked_result['decision']}") # auto_approvedReplay：不修改历史 State，直接从某个检查点重新执行。结果与历史相同（确定性 Agent 下）。用于验证行为、调试偶发问题。
Fork：先修改历史 State，再执行。产生一条全新的执行分支，不影响原时间线。用于纠错和 A/B 测试。
Fork 的关键在于 update_state(historical_config, ...) 返回一个新的 config（包含新的 checkpoint_id）。这个新 config 代表分叉点，后续对它调用 invoke(None) 会从分叉点继续，而不是从原来的历史检查点。Git 类比：这就是 git checkout <old-commit> + git commit，创建了新分支。