Building a Ride-Share Zone-Balancing Agent with LangGraph - Part 4: Letting a Human Step In

This is Part 4 of a 5-part series. Part 3 gave the agent memory. It can now run for hours on its own, cycle after cycle, without forgetting what it already tried. That's exactly the problem. A genuinely severe deficit might call for an aggressive surge multiplier, or a costly driver bonus. Right now, the agent applies whatever it decides immediately. Nobody has looked at it first. Nothing pauses for a second opinion, no matter how expensive the call. Part 4 adds that pause. It uses LangGraph's dynamic interrupt() : - A node calls interrupt(payload) - but only conditionally, when it actually wants to pause. - Execution stops at that exact point. - app.invoke(Command(resume=answer), config=thread) continues from there.answer becomesinterrupt() 's return value, right at the call site. Two gated pauses, not one: - Early - request_data_edit . Gated oncycle_number == 1 . A human validates or corrects the raw starting snapshot once, beforedetect_imbalance even runs. Later cycles use the agent's own carried-forward numbers, not a fresh unverified reading - so re-confirming every cycle would just be noise. - Late - request_approval . Gated onseverity == "critical" . A wrong automatic call is most expensive in a critical deficit, so that's where a human checkpoint earns its cost. Mild, moderate, balanced, and surplus cycles still run fully autonomously, exactly like Part 3. One property worth naming: no new fields got added to the state. Every interrupt here only reads state Part 3 already has, and writes back a field Part 3 already has - zone , recommended_policy , explanation . Human review is a new control-flow capability. It's not new data. The Graph start_cycle ──▶ apply_scheduled_conditions ──▶ request_data_edit ──▶ detect_imbalance ──▶ classify_severity ──▶ set_candidates (gate: cycle_number == 1) │ ┌──────────┴──────────┐ ▼ (balanced) ▼ (deficit/surplus) trivial_do_nothing reconcile_inputs ← LLM #1 │ ▼ │ resolved_imbalance │ ▼ │ choose_best_policy │ ▼ │ generate_explanation ← LLM #2 └──────────┬──────────┘ ▼ request_approval (gate: severity == "critical") ▼ simulate_and_report Everything from detect_imbalance through generate_explanation is unchanged from Part 3 - same nodes, reused directly. Two new nodes bracket that core: request_data_edit right after the cycle starts, request_approval right before the outcome is simulated. Both are plain Python. And both return {} on any cycle where their gate doesn't apply - so an ungated cycle looks exactly like Part 3. The Nodes That Pause Each one is a plain function with an if guard at the top. No LangGraph machinery decides whether to pause. The node itself does, then calls interrupt(payload) when it wants to: def request_data_edit(state: AgentState) -> AgentState: if state["cycle_number"] != 1: return {} zone = state["zone"] answer = interrupt({ "kind": "data_edit", "question": ( f"Review the starting snapshot for {zone['zone_name']}. " "Resume with {'corrections': {...}} to fix fields, or {'corrections': {}} to accept it." ), "zone": zone, }) corrections = answer.get("corrections", {}) if not corrections: return {} return {"zone": {**zone, **corrections}} interrupt() 's return value becomes whatever gets passed to Command(resume=...) - here, a corrections dict. One real gotcha, worth knowing before you use this: Command(resume={}) is falsy in Python. LangGraph treats a falsy resume as if no answer was given at all. It just re-fires the same interrupt, instead of continuing. Always resume with a non-empty dict - {"corrections": {}} is how you accept the snapshot as-is. The late pause has the same shape, but a real decision to make: def request_approval(state: AgentState) -> AgentState: if state["severity"] != "critical": return {} answer = interrupt({ "kind": "approval", "question": ( f"Critical imbalance in {state['zone']['zone_name']} (cycle {state['cycle_number']}). " "Approve, reject, or override the recommended policy." ), "zone_name": state["zone"]["zone_name"], "cycle": state["cycle_number"], "imbalance_ratio": state["imbalance_ratio"], "recommended_policy": state["recommended_policy"], "explanation": state["explanation"], "policy_evaluations": state["policy_evaluations"], "candidate_policies": state["candidate_policies"], }) action = (answer or {}).get("action", "approve") if action == "reject": return { "recommended_policy": "do_nothing", "explanation": "Rejected by human reviewer - falling back to do_nothing.", } if action == "override": chosen = answer["policy"] return { "recommended_policy": chosen, "explanation": f"Overridden by human reviewer: {chosen} chosen instead of " f"{state['recommended_policy']}.", } return {} # approve - no change Three possible resume values: - {"action": "approve"} - keep the recommended policy. - {"action": "reject"} - fall back todo_nothing . - {"action": "override", "policy": " "} - force a specific policy. Building It Same graph as Part 3, with these two nodes inserted at the right points. It's wrapped in a build(checkpointer) function, rather than built inline once - the next section needs a second graph object sharing the exact same checkpointer. The structure never changes. Only whether it's the same Python object does: def build(checkpointer): llm = ChatOllama(model="qwen2.5:14b", temperature=0) llm_with_reconcile_tool = llm.bind_tools([report_context_and_schedule]) def _reconcile(state): return reconcile_inputs(state, llm_with_reconcile_tool) def _explain(state): return generate_explanation(state, llm) g = StateGraph(AgentState) g.add_node("start_cycle", start_cycle) g.add_node("apply_scheduled_conditions", apply_scheduled_conditions) g.add_node("request_data_edit", request_data_edit) g.add_node("detect_imbalance", detect_imbalance) g.add_node("classify_severity", classify_severity) g.add_node("set_candidates", set_candidates) g.add_node("trivial_do_nothing", trivial_do_nothing) g.add_node("reconcile_inputs", _reconcile) g.add_node("resolved_imbalance", resolved_imbalance) g.add_node("choose_best_policy", choose_best_policy) g.add_node("generate_explanation", _explain) g.add_node("request_approval", request_approval) g.add_node("simulate_and_report", simulate_and_report) g.add_edge(START, "start_cycle") g.add_edge("start_cycle", "apply_scheduled_conditions") g.add_edge("apply_scheduled_conditions", "request_data_edit") g.add_edge("request_data_edit", "detect_imbalance") g.add_edge("detect_imbalance", "classify_severity") g.add_edge("classify_severity", "set_candidates") g.add_conditional_edges("set_candidates", route_llm_or_skip, { "trivial_do_nothing": "trivial_do_nothing", "reconcile_inputs": "reconcile_inputs", }) g.add_edge("reconcile_inputs", "resolved_imbalance") g.add_edge("resolved_imbalance", "choose_best_policy") g.add_edge("choose_best_policy", "generate_explanation") g.add_edge("generate_explanation", "request_approval") g.add_edge("trivial_do_nothing", "request_approval") g.add_edge("request_approval", "simulate_and_report") g.add_edge("simulate_and_report", END) return g.compile(checkpointer=checkpointer) memory = MemorySaver() app = build(memory) Correcting the Starting Snapshot Cycle 1 always pauses at request_data_edit . Downtown Core's raw snapshot understates its driver count here - it's recorded as 9, but it was actually 15. Reaching the pause: downtown = get_zone(zones, DEFAULT_ZONE_NAME, driver_count=9) r1 = app.invoke(make_initial_state(downtown), config=thread_edit) PAUSED - kind=data_edit question: Review the starting snapshot for Downtown Core. Resume with {'corrections': {...}} to fix fields, or {'corrections': {}} to accept it. zone as recorded: driver_count=9, rider_request_count=16 uncorrected ratio: 1.78 At 1.78, that's a deficit. But it's a wrong one - a data problem, not a real supply problem. Correcting it and resuming: MY_CORRECTION = {"driver_count": 15} r2 = app.invoke(Command(resume={"corrections": MY_CORRECTION}), config=thread_edit) (no interrupt - cycle ran straight through) ratio now: 1.07 (balanced) [Cycle 1] [Downtown Core] ratio=1.07 | severity=none | policy=do_nothing | wait 3.3min → 4.6min | resolved=N/A One correction, and the zone flips from deficit to balanced. That's the entire point of this pause - catching a bad reading before it drives a real decision. request_data_edit is gated on cycle_number == 1 , so it doesn't fire again on cycle 2 of the same thread. The edit gate only ever applies once per thread. Reviewing a Critical Decision request_approval only fires when severity == "critical" - a ratio bad enough that the wrong automatic call is expensive. Reaching the pause, on a zone with 4 drivers against 80 rider requests: PAUSED - kind=approval question: Critical imbalance in Airport (cycle 1). Approve, reject, or override the recommended policy. severity=critical | recommended=surge_pricing | candidates=['surge_pricing', 'driver_bonus', 'demand_redirect', 'do_nothing'] explanation: The surge pricing policy was selected for the Airport zone in Cycle 1 because it generated the highest profit of $142.93, even though it did not resolve the imbalance. Approving and resuming - but against a different graph object this time: MY_DECISION = {"action": "approve"} app_resume = build(memory) # a different graph object, sharing only memory r3 = app_resume.invoke(Command(resume=MY_DECISION), config=thread_approve) final policy: surge_pricing [Cycle 1] [Airport] ratio=20.0 | severity=critical | policy=surge_pricing | wait 5.6min → 13.3min | resolved=NO That app_resume detail is worth pausing on. It's built fresh from the same build() function, and it only shares one thing with the original app : the MemorySaver . The paused state isn't living inside the app Python object - it's living in the checkpointer. Any graph object built the same way, sharing that checkpointer, can pick the pause back up. A Third Kind of Pause: Debugging request_data_edit and request_approval are both gated. They only pause under a specific condition, and the resume value changes what happens next. A