💾 检查点与人在回路

class BaseCheckpointSaver(Generic[V]):
    """创建图 checkpointer 的基类。

    Checkpointer 让 LangGraph agent 在多次交互间持久化状态。

    配置 checkpointer 后，调用图时要在 config 里传 thread_id：
        config = {"configurable": {"thread_id": "my-thread"}}
        graph.invoke(inputs, config)

    thread_id 是存取 checkpoint 的主键。没有它，
    checkpointer 无法保存状态、从中断恢复、或启用时间旅行调试。

    thread_id 怎么选：
    - 单次工作流：每次用唯一 ID（如 uuid4）
    - 对话记忆：同一对话复用同一 thread_id 累积状态
    """

    # 核心方法：
    def get_tuple(self, config): ...   # 取最新的 checkpoint
    def put(self, config, checkpoint, metadata, new_versions): ...  # 存 checkpoint
    def put_writes(self, config, writes, task_id): ...  # 存待写入
    def list(self, config): ...        # 列出历史 checkpoint（时间旅行用）
    # 异步版：aget_tuple / aput / aput_writes / alist

# pip install langgraph langchain-openai
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

model = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(MessagesState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot"); graph.add_edge("chatbot", END)

# 关键：编译时传入 checkpointer
app = graph.compile(checkpointer=InMemorySaver())

# 同一对话复用同一 thread_id → 状态累积（对话记忆！）
config = {"configurable": {"thread_id": "conversation-1"}}

# 第一轮
app.invoke({"messages": [HumanMessage("我叫小明")]}, config)

# 第二轮（同一 thread_id，记得上一轮）
result = app.invoke({"messages": [HumanMessage("我叫什么？")]}, config)
print(result["messages"][-1].content)  # → "你叫小明"
# 因为 checkpointer 把第一轮的消息历史存了，第二轮自动加载

def interrupt(value: Any) -> Any:
    """在节点内用可恢复的异常中断图。

    实现【人在回路】：暂停执行，把 value 暴露给客户端。
    value 可以传达上下文，或请求恢复所需的信息。

    在一个节点内，首次调用此函数会抛出 GraphInterrupt 异常，
    暂停执行。value 随异常发送给执行图的客户端。

    客户端恢复时，必须用 Command 指定一个值来继续执行。
    图会从【节点开头】重新执行所有逻辑（重要！）。

    如果一个节点有多个 interrupt，LangGraph 按顺序匹配恢复值。

    ⚠️ 使用 interrupt 必须启用 checkpointer（依赖状态持久化）。
    """

@dataclass
class Command:
    """更新图状态并发送消息的指令。

    Args:
        update: 要应用到状态的更新
        resume: 恢复执行用的值（配合 interrupt 使用）
            - 可以是 {interrupt_id: resume_value} 映射
            - 或单个值（恢复下一个 interrupt）
        goto: 下一步导航到的节点名
        graph: 发送到哪个图（Command.PARENT 发往父图）
    """

# pip install langgraph
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command

class State(TypedDict):
    email_content: str
    approved: bool

def write_email(state) -> dict:
    """节点1：生成邮件草稿。"""
    return {"email_content": "尊敬的客户，您的订单已确认..."}

def human_review(state) -> Command:
    """节点2：暂停，等人工审核。"""
    # interrupt() 暂停执行，把邮件内容暴露给客户端
    # 第二次进入（恢复时），这里会【返回】人工的输入值
    user_decision = interrupt({
        "question": "这封邮件可以发送吗？",
        "email": state["email_content"],
    })
    # user_decision 是人工通过 Command(resume=...) 传进来的
    return Command(update={"approved": user_decision == "yes"})

def send_email(state) -> dict:
    if state["approved"]:
        return {"email_content": state["email_content"] + " [已发送]"}
    return {"email_content": "[发送被拒绝]"}

# 构建图
graph = StateGraph(State)
graph.add_node("write", write_email)
graph.add_node("review", human_review)
graph.add_node("send", send_email)
graph.add_edge(START, "write")
graph.add_edge("write", "review")
graph.add_edge("review", "send")
graph.add_edge("send", END)

app = graph.compile(checkpointer=InMemorySaver())  # interrupt 必须有 checkpointer！
config = {"configurable": {"thread_id": "email-1"}}

# 第一次调用：会在 review 节点 interrupt 暂停
result = app.invoke({"email_content": ""}, config)
print("暂停时的状态:", result)   # 注意：执行到 interrupt 就停了

# 查看中断信息
state = app.get_state(config)
print("中断值:", state.tasks[0].interrupts[0].value)  # → 邮件内容 + 问题

# 人工决策后，用 Command 恢复
result = app.invoke(Command(resume="yes"), config)   # 批准发送
print("最终:", result["email_content"])   # → ...[已发送]

# 在 "send_email" 节点【之前】暂停
app = graph.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["send_email"],   # ← 声明式中断
)

# 第一次调用：执行到 send_email 前暂停
result = app.invoke(inputs, config)

# 人工检查 state 后，再次调用【同样的 invoke】即可继续
result = app.invoke(None, config)   # 传 None 表示"从断点继续"

# 列出所有历史 checkpoint
for state in app.get_state_history(config):
    print(f"step {state.metadata.get('step')}: {state.values}")

# 回溯到某个历史状态，从那里重新跑
app.invoke(None, history_state.config)   # 用历史 config 从该点继续
# 用于：调试、What-if 分析、撤销操作