Human in the loop

from typing import TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
 
class State(TypedDict):
    order_id: str
    refund: float
    status: str
 
def check_order(state: State) -> dict:
    return {"refund": 129.0}                       # a real app would look the order up
 
def approve(state: State) -> dict:
    # Pause the graph and wait for a human. The state is saved by the checkpointer.
    decision = interrupt({"question": f"Refund {state['refund']} GBP for {state['order_id']}?"})
    return {"status": "approved" if decision == "yes" else "rejected"}
 
graph = StateGraph(State)
graph.add_node("check_order", check_order)
graph.add_node("approve", approve)
graph.add_edge(START, "check_order")
graph.add_edge("check_order", "approve")
graph.add_edge("approve", END)
app = graph.compile(checkpointer=InMemorySaver())  # checkpoints after every step
 
config = {"configurable": {"thread_id": "ticket-42"}}   # one thread = one saved conversation
first = app.invoke({"order_id": "A1234", "refund": 0, "status": "new"}, config)
print("paused, asking:", first["__interrupt__"][0].value["question"])
print("saved state   :", app.get_state(config).values)
 
final = app.invoke(Command(resume="yes"), config)   # later, maybe from another server
print("after resume  :", final)

paused, asking: Refund 129.0 GBP for A1234?
saved state   : {'order_id': 'A1234', 'refund': 129.0, 'status': 'new'}
after resume  : {'order_id': 'A1234', 'refund': 129.0, 'status': 'approved'}