human-in-the-loop

32 notes, read out of this brain and free to use. Each one was extracted from a source and is re-checked against its exam.

Human-in-the-loop for SQL query review using interrupts

LangGraph's human-in-the-loop features can pause execution before executing a SQL query to allow human review. The interrupt function is used to wrap the sql_db_query tool in a node that receives human input. The implementation allows for input to approve the tool call, edit its arguments, or provide user feedback. A checkpointer is required in the graph to pause and resume runs.

Tool interrupt example for human-in-the-loop SQL review

The human-in-the-loop SQL agent implementation follows the tool interrupt pattern documented in the broader human-in-the-loop guide, where interrupts are placed within tool nodes rather than at edges. This allows pausing specifically when a tool call needs human approval before execution.

User-fixable errors: pause with interrupt()

When missing information is needed from a user (like customer ID, order number, clarifications), use interrupt() to pause execution. Return Command with updated state and goto same node to retry. Code must come after interrupt() call, as anything before will re-run on resume.

Graph pauses at interrupt() and saves state to checkpointer

When execution reaches interrupt() in a node, the graph pauses indefinitely, saves all state to checkpointer, and waits. It can resume days later, picking up exactly where it left off. The thread_id in config ensures all state for conversation is preserved together.

interrupt() must come first in a node before any other code

Any code written before the interrupt() call will re-run when execution resumes. Place interrupt() first, then process the human's decision after the interrupt returns.

Route to human_review based on urgency and intent

In draft_response node, determine if needs_review: if urgency in ['high', 'critical'] or intent == 'complex', set goto = "human_review", else goto = "send_reply". This routes high-urgency and complex emails for human approval.

Human review interrupt with context for decision

In human_review node, call interrupt() with context needed for decision: email_id, original_email, draft_response, urgency, intent, action request. Human sees all relevant info and can approve or edit response. Returns humanDecision object with approved boolean and optional edited_response.

Resume graph with human decision after interrupt

When interrupt returns with approved=true, route to send_reply with either edited_response or original draft_response. If approved=false, route to END and human handles directly. Use config with thread_id for persistence: config = { configurable: { thread_id: "customer_123" } }.

Example: running email agent with billing issue

Create initial state: emailContent = "I was charged twice for my subscription! This is urgent!", senderEmail = "customer@example.com", emailId = "email_123". Run with config: { configurable: { thread_id: "customer_123" } }. Graph executes and pauses at human_review. Returns draft response for human review. Can resume later with human decision.

Human-in-the-loop with interrupt function

The interrupt() function pauses workflow execution to wait for human input. When called inside a task, it sends data to the user and waits for a Command response to resume. Prior task results are persisted so they are not re-run after interruption.

Review tool calls before execution

Create a review_tool_call function that calls interrupt() with tool call details. The human can then respond with an action: 'continue' (accept), 'update' (revise with new args), or 'feedback' (return ToolMessage). This enables human validation of tool usage before execution.

Command primitive for resuming workflows

The Command class is used to resume an interrupted workflow. Call Command(resume=value) to provide the data expected by the interrupted task, which allows execution to continue.

Example: human-in-the-loop workflow with interrupt

Example with three tasks: step_1 appends 'bar', human_feedback() calls interrupt() waiting for user input then appends it, step_3 appends 'qux'. Entrypoint chains them: result_1 = step_1().result(), result_2 = human_feedback(result_1).result(), result_3 = step_3(result_2).result().

Example: resuming human-in-the-loop with Command

After interrupt, resume with Command(resume='baz'). Execution continues through human_feedback which returns input appended to string, then step_3 completes workflow. Shows interrupt/resume cycle with custom user data.

Example: review_tool_call for tool validation

Function review_tool_call(tool_call) calls interrupt() with {'question': ..., 'tool_call': ...}. Returns action from human_review: 'continue' returns original tool_call, 'update' returns updated tool_call with new args, 'feedback' returns ToolMessage with feedback content.

Example: agent with tool review and iteration

Agent entrypoint calls model, checks model_response.tool_calls in while loop. For each tool call, review_tool_call() returns ToolCall or ToolMessage. Executes validated tool_calls in parallel, appends results to messages, calls model again until no tool_calls remain.

Replay with interrupts example

Python example showing replay with interrupts: After completing an interrupted workflow with graph.invoke({}, config) then graph.invoke(Command(resume='Alice'), config), use get_state_history to find the checkpoint before the interrupt node, then call graph.invoke(None, before_ask.config). The interrupt re-executes and pauses, waiting for new Command(resume=...).

Fork with interrupts example

Python example: After completing an interrupt, use get_state_history to find the checkpoint before the interrupt, call graph.update_state(before_ask.config, {'value': ['forked']}) to create fork_config, then graph.invoke(None, fork_config). The interrupt re-executes and pauses at a new interrupt, waiting for Command(resume=...). Then resume with graph.invoke(Command(resume='Bob'), fork_config).

Multiple interrupts: forking between them

If your graph collects input at several points (for example, a multi-step form), you can fork from between the interrupts to change a later answer without re-asking earlier questions. Use get_state_history to find the checkpoint after the first interrupt but before the second (where s.next == ('ask_age',)), then update_state to fork from that point. The first interrupt's result is preserved, and only the second interrupt re-executes.

Multiple interrupts fork example

Python example: After completing ask_name and hitting ask_age interrupt, find checkpoint with s.next == ('ask_age',) using get_state_history. Call graph.update_state(between.config, {'value': ['modified']}) to fork, then graph.invoke(None, fork_config). ask_name result is preserved ('name:Alice'), and ask_age pauses at interrupt waiting for new answer.

Human-in-the-loop with interrupt function

Example showing how to implement human-in-the-loop using the interrupt function: ```python from langgraph.types import interrupt def human_approval(state: MessagesState): answer = interrupt({"question": "Approve this action?"}) return {"messages": [{"role": "user", "content": answer}]} ```

interrupt() function purpose

The interrupt() function pauses execution and waits for human input, enabling human-in-the-loop interactions.

Agent authentication with scoped tokens

Call client.authenticate() with provider_id, scopes array, and user_id to get OAuth tokens. Tokens are scoped to the calling agent using the Assistant ID parameter. Optionally pass agent_id parameter to explicitly set agent scope. Returns auth_result with token accessible via auth_result.token.

LangGraph OAuth interrupt behavior

When authentication is required during LangGraph execution, the SDK throws an interrupt that pauses agent execution and presents the OAuth URL to the user. After the user completes OAuth authentication and the provider sends the callback, the agent resumes execution from the point it left off. Tokens are stored and refreshed for future use so subsequent authentication is not required.

Out-of-band OAuth authentication flow

For OAuth flows outside LangGraph context, call client.authenticate() and check if auth_result.status equals "pending". If pending, provide auth_result.url to the user. Call client.wait_for_completion(auth_result.auth_id) to wait for user to complete OAuth. If status is not pending, token is already available in auth_result.token.

Fixed HTTP input.respond validation for Event Streaming v2

Fixed HTTP input.respond validation for Event Streaming v2 to read pending interrupts from the durable thread row instead of rebuilding thread state. This prevents valid HITL resumes from incorrectly returning no_such_interrupt after reconnects, redeploys, or thread-state lookup failures.

input.respond forwards optional update and goto parameters in same Command

Fixed input.respond so optional update and goto parameters are forwarded into the same Command as the resume value.

Event Streaming v2 input.respond no_such_interrupt fixed

Fixed Event Streaming v2 input.respond returning no_such_interrupt for legitimate interrupts on the postgres backend over HTTP POST /commands.

A2A interrupt support with input-required state

Added native A2A interrupt support: input-required state is now returned when graphs are interrupted. Use the new command parameter in message/stream and message/send requests to resume with a Command payload.

Multiple interrupts support in task handling

Agent server v0.5.17 enhanced task handling to support multiple interrupts, aligning with open-source functionality.

Command resume for continuing after interrupt

Use Command(resume=...) (Python) or new Command({ resume: ... }) (JavaScript) to provide a value and resume graph execution after an interrupt. The value passed to resume becomes the return value of the interrupt() call inside the paused node.

Do not use Command for multi-turn conversations

Command(resume=...) is the only Command pattern intended as input to invoke() or stream() (optionally combined with update= to also apply a state change while resuming). Do not use Command(update=...) alone as input to continue multi-turn conversations, because passing any Command as input resumes from the latest checkpoint (the last step that ran, not __start__), so the graph will appear stuck if it already finished. To continue a conversation on an existing thread, pass a plain input dict instead.

Give your agent this brain