Advanced

Step 4: Human-in-the-Loop

Not every agent decision should run autonomously. In this lesson, you will add approval gates that pause the workflow for human review, interrupt mechanisms for critical actions, and feedback collection so humans can steer agent behavior.

Why Human-in-the-Loop?

Even the best agents make mistakes. Human-in-the-loop controls are essential for:

• Safety: Code execution, API calls, and file writes should be reviewed before running in production.
• Accuracy: Research findings should be verified before being included in reports.
• Trust: Users trust systems more when they can review and correct agent decisions.
• Compliance: Many industries require human approval for automated actions.

LangGraph Checkpointing

LangGraph supports interrupts through its checkpointing system. When a workflow hits an interrupt, it saves its state and waits for human input before continuing.

# graph/human_review.py
"""Human-in-the-loop nodes for the multi-agent workflow."""
import json
from typing import Optional
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.types import interrupt, Command

from agents.state import AgentState


def human_approval_node(state: AgentState) -> dict:
    """Pause the workflow and ask for human approval.

    This node interrupts execution and presents the current state
    to the human for review. The human can approve, reject, or modify.

    The interrupt() call saves the graph state and returns control
    to the caller. When the caller resumes with a response, execution
    continues from this point.
    """
    # Build a summary of what needs approval
    results = state.get("results", {})
    next_agent = state.get("next_agent", "unknown")
    task = state.get("task", "No task specified")

    review_summary = {
        "action": f"Route task to {next_agent} agent",
        "task": task,
        "results_so_far": {k: v[:500] + "..." if len(v) > 500 else v for k, v in results.items()},
        "iteration": state.get("iteration", 0),
    }

    # This call pauses the workflow and waits for human input
    human_response = interrupt(
        {
            "type": "approval_required",
            "summary": review_summary,
            "prompt": "Do you approve this action? (approve/reject/modify)",
        }
    )

    # Process the human's response
    if isinstance(human_response, dict):
        action = human_response.get("action", "approve")
        feedback = human_response.get("feedback", "")
    elif isinstance(human_response, str):
        action = human_response.strip().lower()
        feedback = ""
    else:
        action = "approve"
        feedback = ""

    if action == "reject":
        return {
            "next_agent": "FINISH",
            "status": "completed",
            "messages": [AIMessage(content=f"Workflow stopped by human reviewer. Feedback: {feedback}")]
        }
    elif action == "modify":
        # Human wants to change the task
        return {
            "task": feedback if feedback else state["task"],
            "messages": [HumanMessage(content=f"Human modification: {feedback}")]
        }
    else:
        # Approved - continue as planned
        return {
            "messages": [HumanMessage(content="Human approved. Continuing workflow.")]
        }

Compile with Interrupts

Update the workflow to include interrupt points and a memory checkpointer:

# graph/workflow.py - Updated with human-in-the-loop
"""Multi-agent workflow with human-in-the-loop controls."""
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AIMessage

from agents.state import AgentState
from agents.supervisor import supervisor_node
from agents.researcher import researcher_node
from agents.coder import coder_node
from agents.analyst import analyst_node
from graph.routing import route_from_supervisor
from graph.human_review import human_approval_node


def aggregate_results(state: AgentState) -> dict:
    """Aggregate results from all agents into a final response."""
    results = state.get("results", {})

    if not results:
        return {
            "messages": [AIMessage(content="No results were generated.")],
            "status": "completed"
        }

    summary_parts = ["## Workflow Results\n"]
    for agent_name, result in results.items():
        summary_parts.append(f"### {agent_name.title()} Agent\n{result}\n")
    summary_parts.append(f"\n---\n*Completed in {state.get('iteration', 0)} iterations*")

    return {
        "messages": [AIMessage(content="\n".join(summary_parts))],
        "status": "completed"
    }


def should_require_approval(state: AgentState) -> str:
    """Decide whether human approval is needed before routing to an agent.

    Approval is required when:
    - The coder agent is about to execute code
    - More than 5 iterations have occurred (possible infinite loop)
    - The task involves sensitive operations
    """
    next_agent = state.get("next_agent", "")
    iteration = state.get("iteration", 0)

    # Always require approval for code execution
    if next_agent == "coder":
        return "human_approval"

    # Require approval after many iterations
    if iteration > 5:
        return "human_approval"

    # Otherwise, proceed without approval
    return "auto_route"


def route_after_approval(state: AgentState) -> str:
    """Route to the correct agent after human approval."""
    return route_from_supervisor(state)


def build_workflow_with_hitl(require_approval: bool = True):
    """Build the workflow with optional human-in-the-loop controls.

    Args:
        require_approval: If True, add human approval gates for critical actions.

    Returns:
        A compiled LangGraph application with checkpointing.
    """
    workflow = StateGraph(AgentState)

    # Add all nodes
    workflow.add_node("supervisor", supervisor_node)
    workflow.add_node("researcher", researcher_node)
    workflow.add_node("coder", coder_node)
    workflow.add_node("analyst", analyst_node)
    workflow.add_node("aggregate_results", aggregate_results)

    # Entry point
    workflow.add_edge(START, "supervisor")

    if require_approval:
        # Add the human approval node
        workflow.add_node("human_approval", human_approval_node)

        # Supervisor -> check if approval is needed
        workflow.add_conditional_edges(
            "supervisor",
            should_require_approval,
            {
                "human_approval": "human_approval",
                "auto_route": "researcher",  # Default auto-route
            }
        )

        # After approval -> route to the right agent
        workflow.add_conditional_edges(
            "human_approval",
            route_after_approval,
            {
                "researcher": "researcher",
                "coder": "coder",
                "analyst": "analyst",
                "aggregate_results": "aggregate_results",
            }
        )

        # For auto-routing (non-approval path), use conditional edges
        # Note: In practice, you would refine this further
    else:
        # No approval - direct routing from supervisor
        workflow.add_conditional_edges(
            "supervisor",
            route_from_supervisor,
            {
                "researcher": "researcher",
                "coder": "coder",
                "analyst": "analyst",
                "aggregate_results": "aggregate_results",
            }
        )

    # All agents loop back to supervisor
    workflow.add_edge("researcher", "supervisor")
    workflow.add_edge("coder", "supervisor")
    workflow.add_edge("analyst", "supervisor")

    # Aggregation leads to END
    workflow.add_edge("aggregate_results", END)

    # Compile with memory checkpointer for state persistence
    memory = MemorySaver()
    app = workflow.compile(checkpointer=memory)

    return app

Running with Human-in-the-Loop

Here is how to run the workflow with approval gates:

# main_hitl.py
"""Run the multi-agent workflow with human-in-the-loop controls."""
import os
import uuid
from dotenv import load_dotenv
from rich.console import Console
from rich.panel import Panel
from rich.markdown import Markdown
from rich.prompt import Prompt

load_dotenv()
console = Console()


def run_with_approval(task: str):
    """Run the workflow, pausing for human approval at interrupt points."""
    from graph.workflow import build_workflow_with_hitl

    app = build_workflow_with_hitl(require_approval=True)

    # Each conversation needs a unique thread ID for checkpointing
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    initial_state = {
        "messages": [],
        "next_agent": "",
        "task": task,
        "results": {},
        "status": "in_progress",
        "iteration": 0,
    }

    console.print(f"\n[dim]Starting workflow with approval gates...[/dim]\n")

    # Run until completion or interrupt
    while True:
        try:
            # Invoke or resume the workflow
            result = app.invoke(initial_state, config)

            # If we get here, the workflow completed without interruption
            final_messages = result.get("messages", [])
            if final_messages:
                console.print(Panel(
                    Markdown(final_messages[-1].content),
                    title="Final Result",
                    border_style="green"
                ))
            break

        except Exception as e:
            # Check if this is an interrupt (approval request)
            error_str = str(e)
            if "interrupt" in error_str.lower():
                # Get the interrupt payload
                state = app.get_state(config)
                interrupt_data = state.tasks

                if interrupt_data:
                    # Display what needs approval
                    for task_info in interrupt_data:
                        if hasattr(task_info, "interrupts"):
                            for intr in task_info.interrupts:
                                console.print(Panel(
                                    str(intr.value),
                                    title="Approval Required",
                                    border_style="yellow"
                                ))

                # Get human input
                action = Prompt.ask(
                    "Action",
                    choices=["approve", "reject", "modify"],
                    default="approve"
                )

                feedback = ""
                if action == "modify":
                    feedback = Prompt.ask("Enter modified task or feedback")

                # Resume the workflow with the human's response
                response = {"action": action, "feedback": feedback}
                app.invoke(Command(resume=response), config)

            else:
                console.print(f"[red]Error: {e}[/red]")
                break


def main():
    console.print(Panel(
        "[bold blue]Multi-Agent Workflow (Human-in-the-Loop)[/bold blue]\n"
        "The workflow will pause for your approval before critical actions.\n"
        "Type 'quit' to exit.",
        title="Welcome"
    ))

    while True:
        task = console.input("\n[bold green]Task:[/bold green] ").strip()
        if task.lower() in ("quit", "exit", "q"):
            break
        if not task:
            continue
        run_with_approval(task)


if __name__ == "__main__":
    main()

Feedback Collection Pattern

Beyond simple approve/reject, you can collect structured feedback that improves agent behavior:

# graph/human_review.py - Feedback collection (add to existing file)

def feedback_collection_node(state: AgentState) -> dict:
    """Collect structured feedback from the human reviewer.

    This is used at the end of a workflow to gather quality feedback
    that can be used to improve agent prompts and routing logic.
    """
    results = state.get("results", {})

    # Present results for review
    feedback = interrupt(
        {
            "type": "feedback_request",
            "results": {k: v[:1000] for k, v in results.items()},
            "questions": [
                "Rate the overall quality (1-5):",
                "Was the right agent used for each subtask?",
                "Any corrections or additions needed?",
            ]
        }
    )

    # Store feedback in state for later analysis
    if isinstance(feedback, dict):
        current_results = state.get("results", {})
        current_results["human_feedback"] = feedback
        return {
            "results": current_results,
            "messages": [HumanMessage(content=f"Feedback received: {json.dumps(feedback)}")]
        }

    return {
        "messages": [HumanMessage(content=f"Feedback: {feedback}")]
    }

Key Takeaways

• LangGraph's interrupt() function pauses workflow execution and waits for human input.
• The MemorySaver checkpointer preserves state across interrupts so the workflow can resume exactly where it stopped.
• Approval gates should be selective — gate irreversible actions, not every step.
• Structured feedback collection helps improve agent behavior over time.
• Each conversation needs a unique thread_id for independent state tracking.

What Is Next

In the next lesson, you will add monitoring and debugging — LangSmith tracing, cost tracking per agent run, and structured error handling to make your multi-agent system production-ready.