Recap — Adding Human Oversight to Agents

# Adding Human Oversight to Agents Some actions are too important to leave to the LLM: - Sending an email. - Issuing a refund. - Deleting data. - Posting to social media. Human-in-the-loop (HITL) pauses execution before these actions, gets approval, then resumes. LangGraph makes this clean. ## Pattern 1 — Interrupt before a node The simplest pattern. Compile with `interrupt_before`: ```python from langgraph.checkpoint.postgres import PostgresSaver checkpointer = PostgresSaver.from_conn_string("postgresql://...") graph = builder.compile( checkpointer=checkpointer, interrupt_before=["execute_refund"], ) ``` When execution reaches the `execute_refund` node, it stops. State is saved. The caller gets control: ```python config = {"configurable": {"thread_id": "ticket-123"}} # Run until interrupt result = graph.invoke({"messages": [user_msg]}, config) # Check if we're paused state = graph.get_state(config) if state.next: # there's a next node to run next_node = state.next[0] print(f"Agent wants to execute: {next_node}") print(f"State: {state.values}") # Get human approval user_approves = ask_human(state.values) if user_approves: # Resume execution graph.invoke(None, config=config) else: # Update state to reflect rejection, then resume graph.update_state( config, {"messages": [HumanMessage("Refund rejected by reviewer")]}, ) graph.invoke(None, config=config) ``` The agent loops back into its decision node with the rejection note. It can revise its approach. ## Pattern 2 — Dynamic interrupts (interrupt() function) For more flexibility, interrupt within a node: ```python from langgraph.types import interrupt def execute_refund_node(state): refund_data = { "order_id": state["order_id"], "amount": state["refund_amount"], "reason": state["refund_reason"], } # Interrupt and wait for human input approval = interrupt({ "type": "refund_approval", "details": refund_data, }) if approval == "approved": # Actually execute the refund result = refund_service.create(**refund_data) return {"messages": [AIMessage(f"Refund issued: {result['refund_id']}")]} else: return {"messages": [AIMessage("Refund rejected by reviewer")]} ``` When the node hits `interrupt()`, execution pauses. The state is saved with `interrupt_value` set to whatever was passed. Resumption: ```python graph.invoke( {"approval": "approved"}, # this becomes interrupt()'s return value config=config ) ``` This is more flexible than `interrupt_before` — you can interrupt mid-node, pass structured data both ways, and have multiple interrupts per node. ## Pattern 3 — Conditional human-in-the-loop Only require approval for certain conditions: ```python def maybe_request_approval(state): if state["refund_amount"] > 1000: return "human_review" return "execute_refund" builder.add_conditional_edges("agent", maybe_request_approval) builder.add_node("human_review", human_review_node) builder.add_node("execute_refund", execute_refund_node) ``` For low-stakes (refund < $1000), auto-execute. For high-stakes, pause for review. ## Pattern 4 — Feedback loop Let the human suggest changes, then re-run: ```python def draft_email_node(state): draft = state["email_draft"] feedback = interrupt({ "type": "draft_review", "draft": draft, "action_options": ["approve", "edit", "regenerate"], }) if feedback["action"] == "approve": return {"final_email": draft} elif feedback["action"] == "edit": return {"email_draft": feedback["edited_content"], "iteration": state["iteration"] + 1} else: # regenerate return {"messages": [HumanMessage(feedback["regenerate_instructions"])]} ``` The graph routes back to `draft_email_node` if the human asked for changes. Loops until approved. ## Building the human-side UI The HITL pattern requires a UI to: - Show pending approvals. - Display agent state and planned action. - Provide approve/reject/edit buttons. - Submit decision and resume. ```python # Pseudo-code for a Flask endpoint @app.route("/agent/approve/ ", methods=["POST"]) def approve_action(thread_id): config = {"configurable": {"thread_id": thread_id}} decision = request.json["decision"] if decision == "approved": graph.invoke({"approval": "approved"}, config=config) else: graph.invoke({"approval": "rejected", "reason": request.json["reason"]}, config=config) return {"status": "resumed"} ``` A real system has: - Queue of pending approvals (rows from `graph.get_state_history` where `state.next` is non-empty). - Email/Slack notifications to reviewers. - Audit log of all approvals. - Timeout handling (if no human acts in N hours, auto-escalate). ## When to use HITL Always HITL for: - Financial transactions (refunds, payme