重点学习,:

1. 图运行流程中断和状态保存 是如何实现的
2. 使用thread_id 对图的执行过程进行恢复

以下内容参照代码学习
1. 中断和状态保存 是如何实现的
在构建图的函数中(build_graph)使用
memory = InMemorySaver()
return builder.compile(
checkpointer=memory,
interrupt_before=[“generate_answer”],
)
其中 checkpointer=memory 是将断点状态保存在memory中;
interrupt_before=[“generate_answer”] 是说在流程进入generate_answer进行中断.
这里会生成一个特殊的事件 {“interrupt”: ()}

注意事项: checkpointer的实现方式
InMemorySaver: 内存型 适合学习使用
SQLite: 本地使用
PostgreSQL: 平台服务(官方建议)
还可以使用Redis/MongoDB, mysql需要自己实现BaseCheckpointSaver

2. 使用thread_id 对图的执行过程进行恢复
其实是学习graph.stream() / graph.invoke()的参数
当前重点
input: 输入, 即状态GraphState, 新启动一条执行链 传入input,
从 checkpoint 恢复：传 None, graph会根据config从checkpointer读取中断时的状态;

config: 运行配置，最常用的是放 thread_id, 作用是用来告诉LangGraph这次执行属于哪条线程

durability: 持久化策略,sync 同步写入(当前节点处理完先落盘, 再进行下个节点),
async异步写入(后台异步写入,线程级异常,或者系统级异常上个节点状态可能来不及写入),
exit执行结束保存

其他:
interrupt_before/interrupt_after 在某些节点前/后暂停, 此次代码在compile() 阶段配置
output_keys: 输出字段, 默认图通常返回完整状态
context: 额外运行上下文，不一定进状态

注意事项:

即使 thread_id 相同，如果再次执行时重新传入新的输入状态，这次调用也会被当成一次新的输入驱动执行，而不是从断点继续。

3. invoke() 和 stream() 的区别
graph.stream()

逐步产出事件
适合学习、调试、观察中间过程
能看到 __interrupt__

graph.invoke()

直接返回最终结果
不适合观察中间事件
更像“正常调用图拿结果”

代码如下,

"""Day 2：Checkpointer 与断点恢复最小实现。

本 demo 在 Day 1 的最小状态图基础上加入内存型 checkpointer，
重点观察三件事：

1. 第一次执行时，图在指定节点前被挂起。
2. 复用同一个 thread_id 时，图会从上次 checkpoint 继续。
3. 换一个新的 thread_id 时，会重新开启一条新执行链路。

注意：这里使用 InMemorySaver 只用于学习 checkpoint 机制，
并不代表生产环境的持久化恢复方案。
"""

from typing import Any, Literal, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, StateGraph
from langgraph.graph.state import CompiledStateGraph
from langgraph.types import RunnableConfig, StateSnapshot


class GraphState(TypedDict):
    """最小状态定义，沿用 Day 1 的 4 个核心字段。"""

    question: str
    need_tool: bool
    answer: str
    steps: list[str]


def analyze_question(state: GraphState) -> dict[str, Any]:
    """分析问题是否需要工具检索。"""
    need_tool_words = ["搜索", "查找", "计算", "翻译"]
    need_tool = any(word in state["question"] for word in need_tool_words)
    print(f'[analyze_question] 问题："{state["question"]}"')
    print(f"[analyze_question] 判断结果：need_tool={need_tool}")

    return {
        "need_tool": need_tool,
        "steps": state["steps"] + ["analyze_question"],
    }


def retrieve_info(state: GraphState) -> dict[str, Any]:
    """模拟工具检索。"""
    print(f"[retrieve_info] 模拟检索：{state['question']}")
    result = f'[检索结果] 关于“{state["question"]}”的参考资料'
    print(f"[retrieve_info] 检索到：{result}")

    return {
        "answer": result,
        "steps": state["steps"] + ["retrieve_info"],
    }


def generate_answer(state: GraphState) -> dict[str, Any]:
    """基于当前状态生成回答。"""
    if state["answer"]:
        answer = f"综合检索结果回答：{state['answer']}"
    else:
        answer = f"直接回答：{state['question']} 是一个基础问题，直接回答即可。"
    print(f"[generate_answer] 生成回答：{answer}")

    return {
        "answer": answer,
        "steps": state["steps"] + ["generate_answer"],
    }


def finalize(state: GraphState) -> dict[str, Any]:
    """终结节点，输出最终结果快照。"""
    print(f"[finalize] 最终 answer = {state['answer']}")
    print(f"[finalize] 执行路径 steps = {state['steps']}")

    return {
        "steps": state["steps"] + ["finalize"],
    }


def route_decision(state: GraphState) -> Literal["retrieve", "direct"]:
    """根据 need_tool 决定走检索分支还是直接回答分支。"""
    if state.get("need_tool", False):
        print("[route_decision] 需要工具 -> 走 retrieve 分支")
        return "retrieve"
    print("[route_decision] 不需要工具 -> 走 direct 分支")
    return "direct"


def build_graph() -> CompiledStateGraph:
    """构建带 checkpointer 的可执行图。

    这里额外加入 interrupt_before=["generate_answer"]，目的是让图在
    generate_answer 执行前静态挂起，方便观察 checkpoint 保存与恢复。
    """
    builder = StateGraph(GraphState)

    builder.add_node("analyze_question", analyze_question)
    builder.add_node("retrieve_info", retrieve_info)
    builder.add_node("generate_answer", generate_answer)
    builder.add_node("finalize", finalize)

    builder.set_entry_point("analyze_question")
    builder.add_conditional_edges(
        "analyze_question",
        route_decision,
        {"retrieve": "retrieve_info", "direct": "generate_answer"},
    )
    builder.add_edge("retrieve_info", "generate_answer")
    builder.add_edge("generate_answer", "finalize")
    builder.add_edge("finalize", END)

    memory = InMemorySaver()
    return builder.compile(
        checkpointer=memory,
        interrupt_before=["generate_answer"],
    )


def build_initial_state(question: str) -> GraphState:
    """构造初始状态。"""
    return {
        "question": question,
        "need_tool": False,
        "answer": "",
        "steps": [],
    }


def build_config(thread_id: str) -> RunnableConfig:
    """构造 thread_id 配置。"""
    return {"configurable": {"thread_id": thread_id}}


def print_stream_events(
    graph: CompiledStateGraph,
    graph_input: GraphState | None,
    config: RunnableConfig,
) -> None:
    """逐步打印 stream 事件。"""
    for event in graph.stream(graph_input, config):
        if "__interrupt__" in event:
            print("--- 命中静态断点：图已在 generate_answer 前挂起 ---")
            print(f"  {event}")
            print()
            continue

        for node_name, state_update in event.items():
            print(f"--- 节点 {node_name} 产出的状态更新 ---")
            for key, value in state_update.items():
                print(f"  {key} = {value}")
            print()


def print_snapshot(thread_id: str, snapshot: StateSnapshot) -> None:
    """打印当前 checkpoint 快照的关键信息。"""
    print(f"[checkpoint] thread_id = {thread_id}")
    print(f"[checkpoint] 当前完整状态 values = {snapshot.values}")
    print(f"[checkpoint] 下一步待执行节点 next = {snapshot.next}")
    print()


def run_resume_demo() -> None:
    """演示相同与不同 thread_id 下的 checkpoint 行为差异。"""
    # 构建图
    graph = build_graph()
    question = "搜索一下 Python 语法教程"

    first_thread_id = "lesson-thread-1"
    second_thread_id = "lesson-thread-2"

    print("=" * 60)
    print("Day 2：Checkpointer 与断点恢复")
    print("演示目标：同一 thread_id 继续执行，新 thread_id 重新开始")
    print("=" * 60)
    print()

    print(">>> 第 1 次执行：首次运行，在 generate_answer 前挂起")
    print(f">>> 输入：{question}")
    first_config = build_config(first_thread_id)  
    print_stream_events(graph, build_initial_state(question), first_config)  # 打印和运行图
    print_snapshot(first_thread_id, graph.get_state(first_config))

    print(">>> 第 2 次执行：复用同一个 thread_id，从 checkpoint 继续")
    print(">>> 这一次不再传入初始状态，而是直接继续上次链路")
    print_stream_events(graph, None, first_config)
    print_snapshot(first_thread_id, graph.get_state(first_config))

    print(">>> 第 3 次执行：换一个新的 thread_id，重新开启流程")
    second_config = build_config(second_thread_id)
    print_stream_events(graph, build_initial_state(question), second_config)
    print_snapshot(second_thread_id, graph.get_state(second_config))

    print(">>> 观察结论：")
    print("1. 同一个 thread_id 会命中同一条 checkpoint 链路。")
    print("2. 第二次执行传入 None，表示从已有状态继续，而不是重新喂一份初始状态。")
    print("3. 新 thread_id 会得到一条全新的执行链路，因此会再次在断点处挂起。")
    print("4. InMemorySaver 仅保存在当前进程内，重启进程后不会保留。")


if __name__ == "__main__":
    run_resume_demo()