LangGraph 持久执行：用 Checkpoint、Thread、重试和状态历史恢复任务

python -m pip install "langgraph>=1,<2"

from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy


class State(TypedDict):
    device_id: str
    status: str
    attempts: int


def read_device(state: State) -> dict:
    return {"status": "offline", "attempts": state["attempts"] + 1}


def summarize(state: State) -> dict:
    return {"status": f"{state['device_id']}:{state['status']}"}


builder = StateGraph(State)
builder.add_node(
    "read_device",
    read_device,
    retry_policy=RetryPolicy(max_attempts=3),
)
builder.add_node("summarize", summarize)
builder.add_edge(START, "read_device")
builder.add_edge("read_device", "summarize")
builder.add_edge("summarize", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "u-17:task-88"}}

print(graph.invoke({"device_id": "AX-3", "status": "", "attempts": 0}, config))
print(graph.get_state(config).values)

for snapshot in graph.get_state_history(config):
    print(snapshot.next, snapshot.values)