Human-in-the-Loop Agents

Human-in-the-Loop (HITL)

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

# Interrupt graph for human approval
def needs_approval(state: AgentState) -> str:
    last = state["messages"][-1]
    # Check if action is sensitive (delete, payment, email)
    if any(w in str(last) for w in ["delete","send email","payment","purchase"]):
        return "human_review"
    return "execute"

def human_review_node(state: AgentState) -> AgentState:
    """Pause here for human input via interrupt."""
    # LangGraph will pause and save state
    # Human resumes via: graph.update_state(config, {"approved": True})
    return state

# Checkpointing for pause/resume
memory = SqliteSaver.from_conn_string("checkpoints.db")
graph  = StateGraph(AgentState)
graph.add_node("agent",        call_agent)
graph.add_node("human_review", human_review_node)
graph.add_node("execute",      execute_node)
graph.add_conditional_edges("agent", needs_approval)

app = graph.compile(checkpointer=memory, interrupt_before=["human_review"])

# Run until interrupt
config = {"configurable":{"thread_id":"task-001"}}
result = app.invoke(initial_state, config)
# Resume after human approves
app.invoke(None, config)  # continue from checkpoint