Human-in-the-Loop Patterns

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver

# Build graph with human interrupt
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "end": END,
})
graph.add_edge("tools", "agent")

# Compile with interrupt BEFORE tool execution
memory = SqliteSaver.from_conn_string(":memory:")
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["tools"],  # Pause here for human review
)

# Run until interrupt
config = {"configurable": {"thread_id": "user-1"}}
result = app.invoke({"messages": ["Send email to boss"]}, config)

# Human reviews the pending tool call...
print("Agent wants to:", result["messages"][-1].tool_calls)

# Human approves - resume execution
result = app.invoke(None, config)  # Continue from checkpoint

def confidence_router(state: AgentState) -> str:
    """Route based on agent confidence level."""
    last_msg = state["messages"][-1]

    # Extract confidence from agent's reasoning
    confidence = extract_confidence(last_msg.content)

    if confidence >= 0.9:
        return "auto_execute"    # High confidence: proceed
    elif confidence >= 0.6:
        return "human_approve"   # Medium: ask for approval
    else:
        return "human_takeover"  # Low: hand off entirely

def escalate_to_human(state: AgentState) -> AgentState:
    """Transfer to human with full context."""
    context = {
        "conversation": state["messages"],
        "agent_reasoning": state.get("reasoning", ""),
        "attempted_actions": state.get("actions", []),
        "failure_reason": state.get("error", "Low confidence"),
    }
    notify_human_agent(context)
    return {"status": "escalated"}