LangGraph 人机协同：用 Interrupt 完成审批、编辑和恢复

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 Command, interrupt


class State(TypedDict):
    device_id: str
    reason: str
    idempotency_key: str
    result: dict


ORDERS: dict[str, dict] = {}


def approve_and_create(state: State) -> dict:
    decision = interrupt({
        "action": "create_work_order",
        "device_id": state["device_id"],
        "reason": state["reason"],
        "allowed_decisions": ["approve", "reject", "edit"],
    })
    if decision["type"] == "reject":
        return {"result": {"status": "rejected"}}

    reason = decision.get("reason", state["reason"])
    key = state["idempotency_key"]
    if key not in ORDERS:
        ORDERS[key] = {"status": "created", "device_id": state["device_id"], "reason": reason}
    return {"result": ORDERS[key]}


builder = StateGraph(State)
builder.add_node("approve_and_create", approve_and_create)
builder.add_edge(START, "approve_and_create")
builder.add_edge("approve_and_create", END)
graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "u-17:task-88"}}
paused = graph.invoke({
    "device_id": "AX-3",
    "reason": "设备离线",
    "idempotency_key": "task-88:create-order",
    "result": {},
}, config)
print(paused["__interrupt__"])

finished = graph.invoke(Command(resume={"type": "approve"}), config)
print(finished["result"])