失业一年了，天天想着怎么翻身。前五课我们已经学会了让Agent记住状态和多轮协作，但真实项目中，Agent不能完全“自己跑到底”。很多关键决策（比如生成重要代码、发送邮件、执行敏感操作）都需要人工审核或介入。这节课我们就来学习 Human-in-the-loop（人机交互） 与中断机制，让人类可以随时“插手”Agent的工作流。

本课目标

• 理解LangGraph的中断（interrupt）机制
• 学会使用 interrupt_before 精准控制中断时机
• 掌握中断后如何人工输入信息并恢复图的执行
• 构建一个带人工审核的编程Agent，在生成关键代码后等待人工确认

环境准备

pip install -U langgraph langchain langchain-deepseek langchain-core

核心代码实战

代码段1：定义State和编程Agent节点

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langchain_deepseek import ChatDeepSeek

# ==================== State ====================
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    project_requirements: Annotated[str, "项目需求"]
    current_code: Annotated[str, "当前生成的代码"]
    review_status: Annotated[str, "人工审核状态：pending / approved / rejected"]

# ==================== 编程Agent节点 ====================
llm = ChatDeepSeek(model="deepseek-chat", temperature=0.6)

def coding_agent_node(state: State):
    system_prompt = f"""
    你是一个专业的Python全栈工程师。
    项目需求：{state.get('project_requirements', '暂无')}
    当前已有代码：{state.get('current_code', '尚未开始')}
    
    请根据用户最新要求生成或修改代码。
    当生成重要功能模块（如认证、数据库操作等）后，请在消息末尾加上 [NEED_REVIEW] 标记。
    """
    messages = [AIMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

代码段2：构建带Human-in-the-loop的工作流（修正版）

# 创建Checkpointer（用于记忆 + 中断恢复）
checkpointer = MemorySaver()

workflow = StateGraph(State)

workflow.add_node("agent", coding_agent_node)

# 定义人工审核节点（占位节点，实际会在这里中断）
def human_review_node(state: State):
    print("\n=== 🔥 人工审核触发 ===")
    print("Agent生成了关键代码，请人工审核后输入反馈：")
    # 此节点仅作为中断点，实际反馈通过 graph.invoke() 传入
    return {"review_status": "pending"}

workflow.add_node("human_review_node", human_review_node)

# ==================== 条件路由（推荐写法） ====================
# 先让agent正常执行，执行完后根据内容决定是否需要人工审核
workflow.add_edge(START, "agent")

workflow.add_conditional_edges(
    "agent",
    lambda state: "human_review" if "[NEED_REVIEW]" in state["messages"][-1].content else END,
    {
        "human_review": "human_review_node",   # 需要审核 → 进入人工审核节点
        END: END                               # 不需要审核 → 直接结束
    }
)

# 人工审核完成后，回到agent继续工作
workflow.add_edge("human_review_node", "agent")

# ==================== 编译图并设置中断（关键修正） ====================
# 正确做法：只在进入人工审核节点前中断，而不是对所有agent都中断
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review_node"]   # 推荐写法：精确中断
)

# 注意：interrupt_after=["agent"] 与上面的条件边会冲突，不推荐同时使用

代码段3：多轮交互演示（包含人工介入）

config = {"configurable": {"thread_id": "coding_project_001"}}

# 第一轮：用户提出需求
initial_input = {
    "messages": [HumanMessage(content="帮我用FastAPI写一个用户注册接口，要求支持邮箱验证和密码加密")],
    "project_requirements": "FastAPI用户注册接口，支持邮箱验证和密码加密",
    "current_code": "",
    "review_status": "pending"
}

# 执行第一轮（Agent生成代码后，因含有[NEED_REVIEW]会进入human_review_node并中断）
result1 = graph.invoke(initial_input, config=config)

print("\n=== 图已中断，等待人工审核 ===")

# 模拟人工反馈（实际项目中可通过Web界面、命令行等输入）
human_feedback = "代码整体结构不错，但密码加密请使用bcrypt库，并增加请求体验证。批准继续。"

# 恢复执行，传入人工反馈
result2 = graph.invoke(
    {"messages": [HumanMessage(content=human_feedback)]},
    config=config
)

print("\n=== 人工介入后最终输出 ===")
print(result2["messages"][-1].content)

运行结果说明

1. Agent 生成代码后，如果消息包含 [NEED_REVIEW]，则跳转到 human_review_node
2. 因为设置了 interrupt_before=["human_review_node"]，图在进入人工审核节点前自动中断
3. 人工输入反馈后，调用 graph.invoke() 恢复执行，Agent 根据反馈继续工作

小练习（正好2道）

• 练习1（基础）
  将 interrupt_before=["human_review_node"] 改成 interrupt_after=["agent"]，运行后观察中断时机和行为有什么不同。

• 练习2（进阶）
  修改 human_review_node，如果人工反馈中包含“批准”或“通过”，则自动把 review_status 设为 “approved” 并让Agent继续；否则设为 “rejected” 并要求Agent重新生成。

本课小结
本课重点学习了LangGraph的Human-in-the-loop机制。核心是使用 interrupt_before / interrupt_after 配合条件路由，在需要人工介入的节点精准暂停。正确做法是：先用条件边决定是否需要人工审核，再对审核节点设置中断，而不是对所有agent节点都中断。这样既灵活又清晰，是生产环境中实现人机协作的关键技术。

下节预告
第7课：生产级特性：持久化、部署、监控
我们将把记忆、中断、持久化等特性整合起来，学习如何将LangGraph应用真正部署到生产环境并添加监控。下节见！

如果觉得这篇有用，欢迎点赞和关注，一起玩转 LangGraph！