为什么需要人工干预？

先说个真实案例。前段时间做了个自动化运维工具，本来想着能自动处理服务器故障多好啊。结果有一次，AI判断某个服务需要重启，二话不说就执行了。问题是，那是个正在跑批的数据库服务…

从那之后我就明白了，AI再聪明也需要人类把关，特别是：

• 涉及金钱的操作（下单、转账、交易）
• 数据的增删改
• 发送消息给客户
• 调用外部API

这不是说AI不行，而是有些决策需要人类的经验和直觉。比如AI可能不知道"虽然服务器CPU占用率高，但现在是月底结算日不能重启"这种业务潜规则。

LangGraph是怎么做的

LangGraph的思路很简单：让流程在某个地方暂停，等人确认后再继续。有点像你在玩游戏时按暂停键，处理完事情再继续玩。

核心就两个概念：

1. interrupt - 暂停并等待人工输入
2. Command - 恢复执行

中断实现方式

动态中断

最灵活的方式是动态中断，就是在代码执行过程中根据情况决定要不要暂停

from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import InMemorySaver

def check_with_human(state):
    # 这里会暂停，等待人工输入, 人工输入的值会赋值给human_input
    human_input = interrupt({
        "question": f"要删除{state['count']}条数据，确定吗？",
        "data": state['data']
    })

    # 拿到人工输入后继续执行
    if human_input == "yes":
        return {"confirmed": True}
    return {"confirmed": False}

注意，你必须要有个checkpointer来保存状态，不然中断后就没法恢复了。我一开始就是忘了这个，调试了半天：

# 千万别忘了这个
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 还需要指定thread_id
config = {"configurable": {"thread_id": "some-unique-id"}}

静态中断

如果你知道某些节点肯定需要人工确认，可以直接在编译时指定, 相当度提前设置关卡。

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["dangerous_node"],  # 这个节点执行前必须人工确认
    # interrupt_after=["node1"],  执行node1后，中断!
)

说实话，静态中断我用得不多，主要用来调试。生产环境还是动态中断灵活。

典型场景

节点人工路由

这个最常见了，比如等待用户的选择继续走哪个节点。

from typing import Literal, TypedDict
import uuid

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver

# Define the shared graph state
class State(TypedDict):
    llm_output: str
    decision: str

# Simulate an LLM output node
def generate_llm_output(state: State):
    return {"llm_output": "This is the generated output."}

# 人工中断审核节点
def human_approval(state: State) -> Command[Literal["approved_path", "rejected_path"]]:

    decision = interrupt({
        "question": "Do you approve the following output?",
        "llm_output": state["llm_output"]
    })

# 人工审核通过
    if decision == "approve":
        return Command(goto="approved_path", update={"decision": "approved"})
# 人工审核拒绝
    else:
        return Command(goto="rejected_path", update={"decision": "rejected"})

# Next steps after approval
def approved_node(state: State) -> State:
    print("✅ Approved path taken.")
    return state

# Alternative path after rejection
def rejected_node(state: State) -> State:
    print("❌ Rejected path taken.")
    return state

# Build the graph
builder = StateGraph(State)
builder.add_node("generate_llm_output", generate_llm_output)
builder.add_node("human_approval", human_approval)
builder.add_node("approved_path", approved_node)
builder.add_node("rejected_path", rejected_node)

builder.set_entry_point("generate_llm_output")
builder.add_edge("generate_llm_output", "human_approval")
builder.add_edge("approved_path", END)
builder.add_edge("rejected_path", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# Run until interrupt
config = {"configurable": {"thread_id": uuid.uuid4()}}
result = graph.invoke({}, config=config)
print(result["__interrupt__"])
# Output:
# [Interrupt(value={'question': 'Do you approve the following output?', 'llm_output': 'This is the generated output.'}, resumable=True, ns=['human_approval:71743c02-e389-910f-7f94-21a6db33d824'])]

# 模拟人审核通过
# 测试拒绝, 则替换 resume="approve" with resume="reject"
final_result = graph.invoke(Command(resume="approve"), config=config)
print(final_result)

工具调用拦截

这个最重要！你绝对不想让AI随便调用危险的API。

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt
from langgraph.prebuilt import create_react_agent

from dotenv import load_dotenv  # 用于加载环境变量
load_dotenv()  # 加载 .env 文件中的环境变量

# An example of a sensitive tool that requires human review / approval
def book_hotel(hotel_name: str):
    """Book a hotel"""

    # 调用预订酒店API时中断点，等待用户确认
    response = interrupt(
        f"Trying to call `book_hotel` with args {{'hotel_name': {hotel_name}}}. "
        "Please approve or suggest edits."
    )
    if response["type"] == "accept":
        pass
    elif response["type"] == "edit":
        hotel_name = response["args"]["hotel_name"]
    else:
        raise ValueError(f"Unknown response type: {response['type']}")
    return f"Successfully booked a stay at {hotel_name}."

checkpointer = InMemorySaver()

model = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(
    model=model,
    tools=[book_hotel],
    checkpointer=checkpointer,
)

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

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "book a stay at Jinling hotel"}]},
    config
):
    print(chunk)

from langgraph.types import Command

for chunk in agent.stream(
    Command(resume={"type": "accept"}),
    # Command(resume={"type": "edit", "args": {"hotel_name": "Jinling Hotel"}}),
    config
):
    print(chunk)

以上在工具实现函数调用前进行中断，实现起来比较侵入性比较, 我的做法是包装一下工具函数。

from typing import Callable
from langchain_core.tools import BaseTool, tool as create_tool
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langgraph.types import interrupt
from langgraph.prebuilt.interrupt import HumanInterruptConfig, HumanInterrupt

from dotenv import load_dotenv  # 用于加载环境变量

load_dotenv()  # 加载 .env 文件中的环境变量


# 提供一个函数，用于包装工具，以便工具在调用前中断

def add_human_in_the_loop(
        tool: Callable | BaseTool,
        *,
        interrupt_config: HumanInterruptConfig = None,
):
    """Wrap a tool to support human-in-the-loop review."""
    if not isinstance(tool, BaseTool):
        tool = create_tool(tool)

    if interrupt_config is None:
        interrupt_config = {
            "allow_accept": True,
            "allow_edit": True,
            "allow_respond": True,
        }

    @create_tool(
        tool.name,
        args_schema=tool.args_schema
    )
    def call_tool_with_interrupt(config: RunnableConfig, **tool_input):
        request: HumanInterrupt = {
            "action_request": {
                "action": tool.name,
                "args": tool_input
            },
            "config": interrupt_config,
            "description": "Please review the tool call"
        }
        response = interrupt([request])
        response = response[0]
        # approve the tool call
        if response["type"] == "accept":
            tool_response = tool.invoke(tool_input, config)
        # update tool call args
        elif response["type"] == "edit":
            tool_input = response["args"]
            tool_response = tool.invoke(tool_input, config)
        # respond to the LLM with user feedback
        elif response["type"] == "response":
            user_feedback = response["args"]
            tool_response = user_feedback
        else:
            raise ValueError(f"Unsupported interrupt response type: {response['type']}")

        return tool_response

    return call_tool_with_interrupt


from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

checkpointer = InMemorySaver()


def book_hotel(hotel_name: str):
    """Book a hotel"""
    return f"Successfully booked a stay at {hotel_name}."

model = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(
    model=model,
    tools=[
        add_human_in_the_loop(book_hotel),  # 包装需要人工确认的工具
    ],
    checkpointer=checkpointer,
)

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

# Run the agent
"""
{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_suoSSBNFuBKyosHvyp4Un0GL', 'function': {'arguments': '{"hotel_name":"McKittrick hotel"}', 'name': 'book_hotel'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_46bff0e0c8'}, id='run-12084617-ec66-4acf-b996-5c759faf5b3c-0', tool_calls=[{'name': 'book_hotel', 'args': {'hotel_name': 'McKittrick hotel'}, 'id': 'call_suoSSBNFuBKyosHvyp4Un0GL', 'type': 'tool_call'}])]}}
{'__interrupt__': (Interrupt(value=[{'action_request': {'action': 'book_hotel', 'args': {'hotel_name': 'McKittrick hotel'}}, 'config': {'allow_accept': True, 'allow_edit': True, 'allow_respond': True}, 'description': 'Please review the tool call'}], resumable=True, ns=['tools:61613f8d-70ba-c104-7a36-abfbb6e97aa0']),)}
"""
for chunk in agent.stream(
        {"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
        config
):
    print(chunk)

from langgraph.types import Command
#
for chunk in agent.stream(
        # Command(resume=[{"type": "accept"}]),
        Command(resume=[{"type": "edit", "args": {"hotel_name": "Jinling Hotel"}}]),
        config
):
    print(chunk)

输入验证

有时候需要反复确认用户输入，直到符合要求：

def get_validated_input(state):
    prompt = "请输入手机号"

    while True:
        user_input = interrupt(prompt)

        # 验证格式
        if not user_input.startswith("1") or len(user_input) != 11:
            prompt = f"{user_input} 不是有效手机号，请重新输入"
            continue

        # 二次确认
        confirm = interrupt(f"确认手机号是 {user_input} 吗？")
        if confirm == "yes":
            break
        else:
            prompt = "请重新输入手机号"

    return {"phone": user_input}

审查和修改状态

用户可以查看和编辑图的状态。比如AI生成的内容经常需要人工润色，特别是对外的文案：有些营销部门特别喜欢这个功能，AI负责生成初稿，人工负责把关品牌调性。

from typing import TypedDict
import uuid

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver

from typing import TypedDict
import uuid

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver

# Define the graph state
class State(TypedDict):
    summary: str

# Simulate an LLM summary generation
def generate_summary(state: State) -> State:
    return {
        "summary": "The cat sat on the mat and looked at the stars."
    }

# 用户审查和修改节点函数
def human_review_edit(state: State) -> State:
    result = interrupt({
        "task": "Please review and edit the generated summary if necessary.",
        "generated_summary": state["summary"]
    })
    # 返回人工重新编辑修改summary
    return {
        "summary": result["edited_summary"]
    }

# Simulate downstream use of the edited summary
def downstream_use(state: State) -> State:
    print(f"✅ Using edited summary: {state['summary']}")
    return state

# Build the graph
builder = StateGraph(State)
builder.add_node("generate_summary", generate_summary)
builder.add_node("human_review_edit", human_review_edit)
builder.add_node("downstream_use", downstream_use)

builder.set_entry_point("generate_summary")
builder.add_edge("generate_summary", "human_review_edit")
builder.add_edge("human_review_edit", "downstream_use")
builder.add_edge("downstream_use", END)

# Set up in-memory checkpointing for interrupt support
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# Invoke the graph until it hits the interrupt
config = {"configurable": {"thread_id": uuid.uuid4()}}
result = graph.invoke({}, config=config)

# Output interrupt payload
print(result["__interrupt__"])
# Example output:
# Interrupt(
#   value={
#     'task': 'Please review and edit the generated summary if necessary.',
#     'generated_summary': 'The cat sat on the mat and looked at the stars.'
#   },
#   resumable=True,
#   ...
# )

# Resume the graph with human-edited input
edited_summary = "The cat lay on the rug, gazing peacefully at the night sky."
resumed_result = graph.invoke(
    Command(resume={"edited_summary": edited_summary}),
    config=config
)
print(resumed_result)

踩过的坑

1. 忘记设置checkpointer - 这个最常见，没有checkpointer就无法恢复
2. thread_id管理混乱 - 每个会话要用独立的thread_id，不然状态会串
3. 超时处理 - 人工可能不及时响应，要有超时机制和默认处理
4. 状态持久化 - InMemorySaver只适合测试，生产环境要用数据库

一些建议

经过这段时间的使用，我总结了几点：

1. 不要过度使用 - 不是所有地方都需要人工确认，只在真正重要的地方用，不然用户会疯的。
2. 提供足够的上下文 - 中断时要给用户足够的信息做决策，别就问"确认吗？"，要说清楚确认什么。
3. 有降级方案 - 如果人工长时间不响应怎么办？要有个兜底方案。
4. 记录审计日志 - 谁在什么时候批准了什么操作，这些都要记录下来

最后

HITL不是什么高大上的概念，就是在合适的地方让人参与进来。LangGraph的实现方式挺优雅的，通过interrupt和Command就能实现复杂的人机协作流程。

现在我们的系统里，所有涉及数据修改和外部API调用的地方都加了HITL机制。虽然牺牲了一点自动化程度，但睡得踏实多了。

如果你也在用LangGraph，强烈建议从一开始就把HITL机制设计进去，后期再加会很麻烦，别问我怎么知道的…

大模型算是目前当之无愧最火的一个方向了，算是新时代的风口！有小伙伴觉得，作为新领域、新方向人才需求必然相当大，与之相应的人才缺乏、人才竞争自然也会更少，那转行去做大模型是不是一个更好的选择呢？是不是更好就业呢？是不是就暂时能抵抗35岁中年危机呢？

答案当然是这样，大模型必然是新风口！

那如何学习大模型 ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。但是具体到个人，只能说是：

最先掌握AI的人，将会比较晚掌握AI的人有竞争优势。
这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

但现在很多想入行大模型的人苦于现在网上的大模型老课程老教材，学也不是不学也不是，基于此我用做产品的心态来打磨这份大模型教程，深挖痛点并持续修改了近100余次后，终于把整个AI大模型的学习路线完善出来！

在这个版本当中：

您只需要听我讲，跟着我做即可，为了让学习的道路变得更简单，这份大模型路线+学习教程已经给大家整理并打包分享出来, 😝有需要的小伙伴，可以 扫描下方二维码领取🆓↓↓↓

👉CSDN大礼包🎁：全网最全《LLM大模型学习资源包》免费分享（安全咨料，放心领取）👈

​

一、大模型经典书籍（免费分享）

AI大模型已经成为了当今科技领域的一大热点，那以下这些大模型书籍就是非常不错的学习资源。

二、640套大模型报告（免费分享）

这套包含640份报告的合集，涵盖了大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师，还是对AI大模型感兴趣的爱好者，这套报告合集都将为您提供宝贵的信息和启示。(几乎涵盖所有行业)

三、大模型系列视频教程（免费分享）

四、2025最新大模型学习路线（免费分享）

我们把学习路线分成L1到L4四个阶段，一步步带你从入门到进阶，从理论到实战。

L1阶段:启航篇丨极速破界AI新时代

L1阶段：了解大模型的基础知识，以及大模型在各个行业的应用和分析，学习理解大模型的核心原理、关键技术以及大模型应用场景。

L2阶段：攻坚篇丨RAG开发实战工坊

L2阶段：AI大模型RAG应用开发工程，主要学习RAG检索增强生成：包括Naive RAG、Advanced-RAG以及RAG性能评估，还有GraphRAG在内的多个RAG热门项目的分析。

L3阶段：跃迁篇丨Agent智能体架构设计

L3阶段：大模型Agent应用架构进阶实现，主要学习LangChain、 LIamaIndex框架，也会学习到AutoGPT、 MetaGPT等多Agent系统，打造Agent智能体。

L4阶段：精进篇丨模型微调与私有化部署

L4阶段：大模型的微调和私有化部署，更加深入的探讨Transformer架构，学习大模型的微调技术，利用DeepSpeed、Lamam Factory等工具快速进行模型微调，并通过Ollama、vLLM等推理部署框架，实现模型的快速部署。

L5阶段：专题集丨特训篇 【录播课】

全套的AI大模型学习资源已经整理打包，有需要的小伙伴可以微信扫描下方二维码，免费领取

👉CSDN大礼包🎁：全网最全《LLM大模型学习资源包》免费分享（安全资料，放心领取）👈

​