Case 4 — Human-in-the-Loop Approval Flow for High-Stakes Actions

# Case 4 — Human-in-the-Loop Approval Flow > *"Your agent issues refunds, sends emails to customers, and updates customer accounts. Add HITL: high-stakes actions pause for human approval before execution."* --- ## Step 1 — Identify what needs approval Decision matrix: | Action | Requires approval? | |---|---| | Look up order status | No (read-only) | | Look up customer profile | No | | Search knowledge base | No | | Create a ticket | Maybe (auto-approve low priority) | | Send email to customer | YES (every one) | | Issue refund > $100 | YES | | Issue refund <= $100 | Maybe (auto-approve if standard reason) | | Cancel order | YES | | Update account info | YES | Read-only: no approval. Write/destructive: approval. Configure thresholds. ## Step 2 — State design ```python class AgentState(TypedDict): messages: Annotated[list[BaseMessage], operator.add] user_id: str iteration: int pending_approval: Optional[dict] # action awaiting human approval_decision: Optional[str] # "approved" | "rejected" | None ``` `pending_approval` captures the action context; `approval_decision` is set by the human. ## Step 3 — Tools that auto-flag for approval ```python @tool def initiate_refund(order_id: str, amount: float, reason: str) -> dict: '''Initiate a refund. May require human approval for amounts > $100.''' # Auto-approve under threshold for standard reasons standard_reasons = ["defective", "wrong size", "didn't arrive"] if amount <= 100 and any(r in reason.lower() for r in standard_reasons): result = refund_service.create(order_id, amount, reason) return {"success": True, "refund_id": result.id, "auto_approved": True} # Otherwise flag for approval return { "requires_approval": True, "action": "refund", "details": {"order_id": order_id, "amount": amount, "reason": reason}, } @tool def send_customer_email(customer_id: str, subject: str, body: str) -> dict: '''Send an email to a customer. ALWAYS requires human approval.''' return { "requires_approval": True, "action": "send_email", "details": {"customer_id": customer_id, "subject": subject, "body": body}, } ``` Tools don't execute high-stakes actions directly. They return a "requires_approval" flag with action details. ## Step 4 — Agent node with approval detection ```python def agent_node(state): response = llm_with_tools.invoke([SystemMessage(SYSTEM_PROMPT)] + state["messages"]) return {"messages": [response], "iteration": state["iteration"] + 1} def tool_node(state): last_msg = state["messages"][-1] tool_messages = [] pending = None for tc in last_msg.tool_calls: tool = TOOL_MAP[tc["name"]] result = tool.invoke(tc["args"]) if isinstance(result, dict) and result.get("requires_approval"): pending = result # Stash for human review tool_messages.append(ToolMessage( content=json.dumps({"status": "pending_approval", "action": result["action"]}), tool_call_id=tc["id"], )) break # Stop processing other tools; wait for approval else: tool_messages.append(ToolMessage(content=json.dumps(result), tool_call_id=tc["id"])) return {"messages": tool_messages, "pending_approval": pending} ``` ## Step 5 — Conditional routing for approval ```python def should_continue(state): if state.get("pending_approval"): return "wait_for_approval" # Will trigger interrupt last = state["messages"][-1] if hasattr(last, "tool_calls") and last.tool_calls: return "tools" return END def apply_approval(state): pending = state["pending_approval"] decision = state["approval_decision"] if decision == "approved": # Actually execute the action if pending["action"] == "refund": result = refund_service.create(**pending["details"]) tool_msg = f"Refund approved and issued: {result.id}" elif pending["action"] == "send_email": email_service.send(**pending["details"]) tool_msg = f"Email sent to customer {pending['details']['customer_id']}" else: tool_msg = f"Action rejected by reviewer. Reason: {state.get('rejection_reason', 'unspecified')}" return { "messages": [SystemMessage(tool_msg)], "pending_approval": None, "approval_decision": None, } ``` ## Step 6 — Wire up the graph with interrupts ```python from langgraph.checkpoint.postgres import PostgresSaver builder = StateGraph(AgentState) builder.add_node("agent", agent_node) builder.add_node("tools", tool_node) builder.add_node("apply_approval", apply_approval) builder.set_entry_point("agent") builder.add_conditional_edges("agent", should_continue) builder.add_edge("tools", "agent") builder.add_edge("apply_approval", "agent") checkpointer = PostgresSaver.from_conn_string("postgresql://...") # Interrupt before