RSS 中文
⌕

lesson-13

19 MIN READ | UPDATED: 2026-05-07

Subtitle: Before the Writer starts drafting based on the outline, use interrupt_before to wait for a human to click "Approve" before proceeding.

Welcome back, geeks, to the LangGraph Multi-Agent Masterclass. I'm your old friend, the AI architect who's been rolling around in code for a decade.

Over the past 12 parts, our "AI Content Agency" has really started taking shape. We have a strategic Planner, a well-read Researcher, and a hardworking Writer. Watching them automatically pass tasks back and forth in the code, don't you feel a bit of that cyber-capitalist thrill?

However, reality often hits you with a harsh wake-up call.

Yesterday, my Planner had a wild idea and created an outline for an article on "Quantum Computing" that included a chapter titled "How to Use Quantum Computing for Stock Trading and Achieve Financial Freedom." Then, without hesitation, the Writer followed this absurd outline and churned out 3,000 words of pure nonsense. The result? Not only did it waste a massive amount of Tokens (which is hard-earned cash!), but it also produced a pile of unusable garbage.

What does this tell us? A completely runaway AI is both dangerous and expensive. At critical decision-making junctures, we need to introduce human intelligence for quality control. This brings us to today's topic: Human in the Loop (HITL).

Today, we are going to introduce a real "Chief Editor" to our Agency—that is, you behind the screen. After the Planner writes the outline and before the Writer starts drafting, the system must pause and wait for you to give the "Proceed!" command before the workflow can continue.

🎯 Learning Objectives

Through today's hands-on practice, you will master the following skills:

  1. Understand the underlying logic of HITL: Why is the persistence of the graph's State a prerequisite for implementing Human in the Loop?
  2. Master the magic of interrupt_before: How to set precise breakpoints during the LangGraph compilation phase, allowing the workflow to hit the brakes at the edge of the cliff.
  3. Proficiently use Thread IDs: Learn how to wake up a sleeping (paused) graph and let it continue its sprint from its previous state.
  4. Refactor the Agency workflow: Build an impenetrable "human approval wall" between the Planner and the Writer.

📖 Principle Analysis

In traditional code logic, a program is either running or it has finished. If you want a program to stop halfway and wait for you, you usually have to block the thread using input(). However, this is absolutely unworkable in modern asynchronous web services (like Web APIs).

How does LangGraph elegantly solve this problem? The answer is: State Snapshot + Checkpointer mechanism.

Imagine you are playing a single-player game (like Black Myth: Wukong). You just defeated the first Boss (the Planner generated the outline), and an even harder Boss lies ahead (the Writer starts writing a long article). To prevent a total wipeout, what do you do? Save Game!

LangGraph's HITL follows this exact logic:

  1. When the graph execution reaches your set breakpoint (e.g., interrupt_before=["writer"]).
  2. LangGraph uses a Checkpointer (like MemorySaver or a database) to take a snapshot of all current States and save it.
  3. Then, the graph's execution terminates directly (suspends), freeing up computing resources.
  4. Later, after a human reviews the outline on the frontend UI and clicks the "Approve" button.
  5. We call the graph again, bringing along the thread_id (the save slot) from that time. LangGraph reads the snapshot and seamlessly resumes execution from the breakpoint.

As always, a picture is worth a thousand words. Let's look at our refactored Agency workflow for today:

stateDiagram-v2
    %% Define styles
    classDef aiNode fill:#e1f5fe,stroke:#0288d1,stroke-width:2px,color:#000
    classDef humanNode fill:#fff3e0,stroke:#f57c00,stroke-width:2px,stroke-dasharray: 5 5,color:#000
    classDef stateNode fill:#e8f5e9,stroke:#388e3c,stroke-width:2px,color:#000

    %% Node definitions
    Start((START))
    Planner[Planner Agent\n(Generate Outline)]:::aiNode
    Checkpoint[(Checkpointer\nSave State Snapshot)]:::stateNode
    Human{Human Approval\n(HITL)}:::humanNode
    Writer[Writer Agent\n(Draft Article)]:::aiNode
    End((END))

    %% Flow connections
    Start --> Planner : Receive user topic
    Planner --> Checkpoint : Outline generated
    
    Checkpoint --> Human : interrupt_before='Writer'
    
    Human --> Writer : Human clicks "Approve"\n(Pass thread_id to continue)
    Human --> Planner : Human clicks "Reject"\n(Modify state, regenerate)
    
    Writer --> End : Article drafted

Take note! The core of this diagram isn't the Agents, but rather the dashed human node and the Checkpointer. Without a Checkpointer, the graph suffers from amnesia; even if you approve it, it won't remember what the previous outline was.

💻 Practical Code Walkthrough

Enough talk, show me the code. Today, we will use Python along with LangGraph's latest API to implement this feature. To help everyone see the essence clearly, I will use Mock LLM calls, focusing entirely on the control flow of the graph.

Prerequisites: Please ensure you have langgraph installed. pip install langgraph

1. Define State and Nodes

First, we define the Agency's state, along with the Planner and Writer nodes.

import time
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

# ==========================================
# 1. Define State - Our "game save" data structure
# ==========================================
class AgencyState(TypedDict):
    topic: str              # User's initial topic
    outline: Optional[str]  # Outline generated by Planner
    article: Optional[str]  # Article generated by Writer
    approval_status: str    # Human approval status: 'pending', 'approved', 'rejected'

# ==========================================
# 2. Define Nodes - Our AI workers
# ==========================================
def planner_node(state: AgencyState) -> dict:
    """
    Planner Node: Responsible for generating an outline based on the topic.
    """
    print("\n[Planner 👨‍💼] 正在疯狂构思大纲中...")
    time.sleep(1) # Simulate thinking time
    
    topic = state["topic"]
    # Simulate outline generated by LLM
    mock_outline = f"""
    主题：《{topic}》
    1. 什么是{topic}？
    2. {topic}的核心原理
    3. {topic}的未来展望
    """
    print("[Planner 👨‍💼] 大纲生成完毕！")
    
    # Return updated state
    return {
        "outline": mock_outline,
        "approval_status": "pending" # Set to pending approval
    }

def writer_node(state: AgencyState) -> dict:
    """
    Writer Node: Responsible for writing the article based on the outline.
    """
    print("\n[Writer ✍️] 收到主编的通过指令，开始根据大纲码字...")
    time.sleep(1)
    
    outline = state["outline"]
    # Simulate LLM writing article based on outline
    mock_article = f"这是关于大纲【{outline.strip()}】的详细正文...\n(此处省略一万字)"
    print("[Writer ✍️] 文章撰写完毕！可以下班了！")
    
    return {"article": mock_article}

2. Build the Computation Graph with Breakpoints

This is the highlight of today's lesson. Watch closely how we combine MemorySaver and interrupt_before.

# ==========================================
# 3. Graph Construction
# ==========================================
builder = StateGraph(AgencyState)

# Add nodes
builder.add_node("planner", planner_node)
builder.add_node("writer", writer_node)

# Define edges (workflow routing)
builder.add_edge(START, "planner")
builder.add_edge("planner", "writer") # Note: Although this connects directly to writer, we will interrupt it during compilation
builder.add_edge("writer", END)

# Core Step 1: Instantiate a Checkpointer (Memory version, use PostgresSaver in production)
memory = MemorySaver()

# Core Step 2: Compile the graph and set breakpoints!
# interrupt_before=["writer"] means: Pause graph execution [BEFORE] the writer node executes.
graph = builder.compile(
    checkpointer=memory, 
    interrupt_before=["writer"]
)

print("✅ Agency 工作流编译成功，已开启 HITL 模式。")

3. Simulate Actual Execution: From Pause to Resume

Now, let's simulate a real business scenario: User submits a request -> Graph pauses -> Terminal prompts user for confirmation -> User inputs approval -> Graph resumes execution.

# ==========================================
# 4. Simulation Loop
# ==========================================
def run_agency():
    # thread_id must be configured; it is the unique credential the graph uses to identify the "save slot"
    # thread_id is crucial. It's the unique identifier for the "save slot".
    thread_config = {"configurable": {"thread_id": "agency_task_001"}}
    
    initial_state = {
        "topic": "LangGraph 多智能体架构",
        "outline": None,
        "article": None,
        "approval_status": "none"
    }

    print("\n🚀 [系统] 启动 Agency，接收到新任务...")
    
    # Phase 1 execution: It will automatically stop before the writer node
    for event in graph.stream(initial_state, config=thread_config):
        for node_name, node_state in event.items():
            pass # Nodes already print info internally, no extra processing needed here
            
    # At this point, the graph is paused. We can check its current state
    current_state = graph.get_state(thread_config)
    
    # The 'next' attribute tells you which node will execute next if resumed
    # If 'next' has a value, it means the graph was interrupted
    if current_state.next:
        print(f"\n⏸️ [系统] 工作流已暂停。下一个将要执行的节点是: {current_state.next}")
        print(f"📄 [系统] 当前生成的大纲内容如下：\n{current_state.values.get('outline')}")
        
        # Simulate human approval process
        user_input = input("👨‍⚖️ [人类主编] 大纲如上，是否同意让 Writer 开始撰写？(输入 'y' 同意，其他退出): ")
        
        if user_input.strip().lower() == 'y':
            print("\n👨‍⚖️ [人类主编] 审批通过！放行！")
            
            # Core Step 3: Resume execution!
            # Pass None as input state, meaning "do not modify current state, just continue"
            # Must pass the same thread_config to find the previous save slot
            for event in graph.stream(None, config=thread_config):
                pass
        else:
            print("\n👨‍⚖️ [人类主编] 审批驳回！任务终止。")
    else:
        print("\n✅ [系统] 任务已全部执行完毕。")

# Run our test
if __name__ == "__main__":
    run_agency()

Execution Output Demonstration

When you run this code, your terminal will display the following interaction:

✅ Agency 工作流编译成功，已开启 HITL 模式。

🚀 [系统] 启动 Agency，接收到新任务...

[Planner 👨‍💼] 正在疯狂构思大纲中...
[Planner 👨‍💼] 大纲生成完毕！

⏸️ [系统] 工作流已暂停。下一个将要执行的节点是: ('writer',)
📄 [系统] 当前生成的大纲内容如下：

    主题：《LangGraph 多智能体架构》
    1. 什么是LangGraph 多智能体架构？
    2. LangGraph 多智能体架构的核心原理
    3. LangGraph 多智能体架构的未来展望
    
👨‍⚖️ [人类主编] 大纲如上，是否同意让 Writer 开始撰写？(输入 'y' 同意，其他退出): y

👨‍⚖️ [人类主编] 审批通过！放行！

[Writer ✍️] 收到主编的通过指令，开始根据大纲码字...
[Writer ✍️] 文章撰写完毕！可以下班了！

See that? The graph hit the brakes hard right before writer executed! The entire context (the outline content) was perfectly preserved in thread_id: agency_task_001. When you inputted y and called stream(None, config) again, it woke up straight from the breakpoint and finished the remaining work.

Pitfalls & Prevention Guide

When introducing HITL, many beginners (and even some veterans) fall into deep traps. As your mentor, I must sound the alarm here:

💣 Pitfall 1: Forgetting the Checkpointer, causing the graph to get "amnesia"

Symptoms: You set interrupt_before, but after the graph finishes the first phase and you try to resume, it throws an error saying the state cannot be found, or it starts running from the very beginning. Diagnosis: You didn't pass a checkpointer into compile(). Prevention: HITL must rely on state persistence. Without MemorySaver (or Postgres/Redis Saver), LangGraph has absolutely no idea where the graph paused. Remember the formula: HITL = Checkpointer + interrupt_before/after.

💣 Pitfall 2: Messed up Thread IDs causing "crossed wires"

Symptoms: Alice clicks approve, but Bob's article gets published. Diagnosis: When resuming the graph, the wrong thread_id was passed. Prevention: In actual web backend development (e.g., using FastAPI), you must strongly bind the thread_id to your database's Task ID or User ID. When the graph suspends, the frontend polls or waits; after the user clicks the button, the frontend sends the Task ID back to the backend, and the backend assembles it into thread_config to wake up the graph. Never hardcode thread_id in a production environment!

💣 Pitfall 3: Passing new state overwrites the old state, causing logic collapse

Symptoms: When resuming graph execution, the initial state was passed in, causing the already generated outline to disappear. Diagnosis: During the phase 2 call, you wrote graph.stream(initial_state, config). Prevention: If you simply want the graph to continue executing as-is, do not pass in any new state; pass None! That is, graph.stream(None, config). When LangGraph sees None, it knows: "Oh, the boss has no new instructions, I'll just keep working."

(Advanced Spoiler: If you want to modify the outline while it's paused, you can achieve this via graph.update_state(). We will dissect this advanced operation in detail in the next part, "Time Travel of States"!)

📝 Summary

Today, we put reins on our runaway AI. We learned:

  1. The essence of HITL: State snapshots and suspension based on a Checkpointer.
  2. Setting breakpoints: Building a human approval wall via interrupt_before=["writer"].
  3. Waking up the graph: Letting the workflow continue via the same thread_id and stream(None).

In our AI Content Agency, introducing a "Chief Editor Approval" node is a qualitative leap. It means we have officially moved from "roll-of-the-dice" AI generation to an industrial-grade workflow of "Human-AI Collaboration".

Homework / Thought Exercise: If the human editor reads the outline, is unsatisfied, and doesn't want to just exit, but instead wants to send it back to the Planner to rewrite, how should we modify our graph and code?

Unleash your geek spirit and give it a try! I'll see you in Part 14, where I'll unlock one of the coolest features in LangGraph: Graph State Modification and Time Travel. Class dismissed!