Human in the Loop
Module Summary

# ── 依赖导入 ──────────────────────────────────────────────
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]}

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 打印

# ════════════════════════════════════════════════
# 【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)

# ════════════════════════════════════════════════
# 【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)  # 现在可以正常继续

# ════════════════════════════════════════════════
# 【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_approved