LangGraph Human-in-the-Loop 人工干预

本文最后更新于 2026年7月29日

概述

一些节点的操作比较敏感高危，需要人工进行二次确认，此时就用到了interrupt，interrupt可以在任何节点内以一种特殊的异常的方式，主动暂停图的执行，冻结当前图的状态（依赖checkpointer持久化），向外抛出自定义数据（给前端/控制台展示待审核内容），外部接收用户输入后，通过Command(resume=)把用户决策回填，继续运行。

interrupt依赖三大机制:

checkpointer

必须配置才能保存中断前状态，否则无法恢复，默认InMemorySaver保存在内存，生产环境可以换成redis等持久化存储。
thread_id

会话唯一标识，同一个对话/任务共用一个thread，中断恢复靠它匹配状态
Command(resume=)

恢复图的执行，通过resume属性进行恢复

1.interrupt

interrupt()处函数手动触发人工干预，通过.invoke(Command(resume=), )继续执行

例：输入用户名

第一次发起后，进入node()即在interrupt()处手动触发人工干预，流程暂时冻结，保存状态到checkpointer，返回供人工判断的信息，人工输入数据并通过Command(resume=)把用户决策回填二次发起后，会从node()节点重新开始执行而不是interrupt()处重新开始执行，节点收到用户编辑的信息后，决定下一步的路径。

❗ 因为是从node()节点重新开始执行，interrupt()处之前的逻辑必须做幂等处理，否则可能引发问题

from typing import TypedDict

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


class TransferState(TypedDict):
    username: str


def node(state: TransferState):

    print('即将中断')

    user_review = interrupt('请输入用户名：')

    print('恢复执行')

    return {
        'username': user_review
    }





if __name__ == '__main__':
    graph = StateGraph(TransferState)

    graph.add_node('node', node)

    graph.add_edge(START, 'node')
    graph.add_edge('node', END)

    app = graph.compile(checkpointer=InMemorySaver())

    cfg = {
        'configurable': {'thread_id': 'lzj-001'}
    }

    initial_state = {
        "username": ""
    }

    first_res = app.invoke(initial_state, config=cfg)
    print(first_res)

    resume_transfer = input( first_res['__interrupt__'][0].value )

    final = app.invoke(Command(resume=resume_transfer), config=cfg)

    print('最终结果：', final)

即将中断
{'username': '', '__interrupt__': [Interrupt(value='请输入用户名：', id='a0cce0c5af90188c07177e6387bdb4a3')]}
请输入用户名：嫱嫱
即将中断
恢复执行
最终结果： {'username': '嫱嫱'}

例：转账审核

第一次发起后，进入review_transfer()即在interrupt()处手动触发人工干预，流程暂时冻结，保存状态到checkpointer，返回供人工判断的信息，人工修改数据并修改approved标志位为通过后，review_transfer()重新执行，节点收到用户编辑的resume_transfer信息后，决定下一步的路径。

from typing import TypedDict

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


class TransferState(TypedDict):
    recipient: str #收款人
    amount: float
    memo: str
    approved: bool
    final_status: str


def review_transfer(state: TransferState):
    pending_transfer = {
        "recipient": state['recipient'],
        "amount": state['amount'],
        "memo": state['memo']
    }

    user_review = interrupt(
        {
            "title": "转账人工审核",
            "pending_transfer": pending_transfer,
            "instruction": "请返回bool或dict，并带有approved字段"
        }
    )

    print(type(user_review))

    # 复制原始数据作为默认值，再用审核员修改的数据覆盖
    update_transfer = dict(pending_transfer)
    print(f'修改前的数据 {update_transfer}' )

    approved = bool(user_review.get('approved', False))
    # 如果通过，则将新数据覆盖旧数据
    if approved:
        for k in ['recipient', 'amount', 'memo']:
            if k in user_review:
                update_transfer[k] = user_review[k]

    print(f'修改后的数据 {update_transfer}')

    return {
        'recipient': update_transfer['recipient'] ,
        'amount': update_transfer['amount'],
        'memo': update_transfer['memo'],
        'approved': approved
    }



def execute_transfer(state: TransferState):
    if state['approved']:
        recipient = state['recipient']
        amount = state['amount']
        memo = state['memo']
        print('审核通过，execute_transfer执行')
        return {
            'final_status': f'人工审核通过，recipient={recipient}, amount={amount}, memo={memo}'
        }
    else:
        print('审核通过，execute_transfer撤回')
        return {
            'final_status': f'人工审核驳回'
        }




if __name__ == '__main__':
    graph = StateGraph(TransferState)

    graph.add_node('review_transfer', review_transfer)
    graph.add_node('execute_transfer', execute_transfer)

    graph.add_edge(START, 'review_transfer')
    graph.add_edge('review_transfer', 'execute_transfer')
    graph.add_edge('execute_transfer', END)

    app = graph.compile(checkpointer=InMemorySaver())


    cfg = {
        'configurable': {'thread_id': 'lzj-001'}
    }

    initial_state = {
        "recipient": "转账收款人lzj",
        "amount": 123.4,
        "memo": "货款",
        "approved": False,
        "final_status": "",
    }

    first_res = app.invoke(initial_state, config=cfg)

    #获取中断时返回的信息
    interrupt_payload = first_res['__interrupt__'][0].value
    print(f' interrupt_payload: {interrupt_payload}')

    resume_transfer = {
        'amount': float(input('请输入修改后的金额：')),
        'memo': input('请输入修改后的备注：'),
        'approved': True,
    }
    # 人干预后的新数据传回
    final = app.invoke(Command(resume=resume_transfer), config=cfg)

    print('最终结果：', final)

 interrupt_payload: {'title': '转账人工审核', 'pending_transfer': {'recipient': '转账收款人lzj', 'amount': 123.4, 'memo': '货款'}, 'instruction': '请返回bool或dict，并带有approved字段'}
请输入修改后的金额：345.67
请输入修改后的备注：数据不对，改掉！
<class 'dict'>
修改前的数据 {'recipient': '转账收款人lzj', 'amount': 123.4, 'memo': '货款'}
修改后的数据 {'recipient': '转账收款人lzj', 'amount': 345.67, 'memo': '数据不对，改掉！'}
审核通过，execute_transfer执行
最终结果： {'recipient': '转账收款人lzj', 'amount': 345.67, 'memo': '数据不对，改掉！', 'approved': True, 'final_status': '人工审核通过，recipient=转账收款人lzj, amount=345.67, memo=数据不对，改掉！'}

例：大模型对话

是否调用大模型，interrupt放在第一行保证幂等

import os
from typing import TypedDict, Annotated, Literal

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START, END
from langgraph.graph import StateGraph, add_messages
from langgraph.types import interrupt, Command

class TransferState(TypedDict):
    messages: Annotated[list, add_messages]

llm = init_chat_model(
    model = 'deepseek-chat',
    model_provider = 'openai',
    api_key = os.getenv('DSKEY'),
    base_url = 'https://api.deepseek.com'
)

def chatbot(state: TransferState) -> dict:
    return {
        'messages': [llm.invoke(state['messages'])]
    }

def human(state: TransferState) -> Command:
    user_review = interrupt('是否同意调用大模型（y/n）：')
    if user_review in ('yes', 'y'):
        return Command(goto='chatbot')
    else:
        return Command(goto=END)


if __name__ == '__main__':
    graph = StateGraph(TransferState)

    graph.add_node('chatbot', chatbot)
    graph.add_node('human', human)

    graph.add_edge(START, 'human')
    graph.add_edge('chatbot', END)

    app = graph.compile(checkpointer=InMemorySaver())

    cfg = {
        'configurable': {'thread_id': 'lzj-001'}
    }

    msg = {
        "messages": ['在吗？']
    }

    first_resp = app.invoke(input=msg, config=cfg)

    user_input = input( first_resp['__interrupt__'][0].value )

    final_resp = app.invoke(Command(resume=user_input), config=cfg)

    for e in final_resp['messages']:
        print(e)

是否同意调用大模型（y/n）：y
content='在吗？' additional_kwargs={} response_metadata={} id='84de1961-f9b7-4d1b-ae9c-0f24dca70182'
content='在的！😊 有什么我可以帮你的吗？无论是聊天、解答问题，还是需要什么帮助，我都在这里等着你。请说吧！' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 32, 'prompt_tokens': 7, 'total_tokens': 39, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 7}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '697ea961-6f63-4cb5-9329-61f8a3e27d3a', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019f9e1f-4f47-76b0-9a52-275e94f51369-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 7, 'output_tokens': 32, 'total_tokens': 39, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}

是否同意调用大模型（y/n）：n
content='在吗？' additional_kwargs={} response_metadata={} id='98126191-2c11-4d58-9809-d08620cdabb3'

2.interrupt_before/after

interrupt_before=['']编译期间声明，在某个节点执行前中断，通过.invoke(input=None, )继续执行，同样支持interrupt_after=['']，原理类似

例：节点执行前中断

interrupt_before=['node2']在节点node2执行前中断，通过app.invoke(input=None, config=cfg)恢复，interrupt_after=['node1']能起到相同效果

import operator
from typing import TypedDict, Annotated

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph
from langgraph.constants import START
from langgraph.constants import END


class DemoState(TypedDict):
    count: Annotated[int, operator.add]

def node1(state: DemoState) -> dict:
    print('=>node1')
    return {'count': 1}

def node2(state: DemoState) -> dict:
    print('=>node2')
    return {'count': 1}



if __name__ == '__main__':
    graph = StateGraph(state_schema=DemoState)


    graph.add_node('node1', node1)
    graph.add_node('node2', node2)

    graph.add_edge(START, 'node1')
    graph.add_edge('node1', 'node2')
    graph.add_edge('node2', END)

    app = graph.compile(
        checkpointer=InMemorySaver(),
        interrupt_before=['node2']
    )

    app.get_graph().print_ascii()

    cfg = {
        'configurable': {'thread_id': 'lzj-001'}
    }

    res = app.invoke(
        input={'count': 0},
        config=cfg
    )

    history = app.get_state_history(cfg)

    for point in history :
        print('最新状态值', point.values)
        print('下一个要执行的节点', point.next)
        print(100 * '*')

    print('#'*35)

    res = app.invoke(input=None, config=cfg)

+-----------+  
| __start__ |  
+-----------+  
      *        
      *        
      *        
  +-------+    
  | node1 |    
  +-------+    
      *        
      *        
      *        
  +-------+    
  | node2 |    
  +-------+    
      *        
      *        
      *        
 +---------+   
 | __end__ |   
 +---------+   
=>node1
最新状态值 {'count': 1}
下一个要执行的节点 ('node2',)
****************************************************************************************************
最新状态值 {'count': 0}
下一个要执行的节点 ('node1',)
****************************************************************************************************
最新状态值 {'count': 0}
下一个要执行的节点 ('__start__',)
****************************************************************************************************
###################################
=>node2

例：调用工具前确认

import os
from typing import TypedDict, Annotated

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START, END
from langgraph.graph import StateGraph, add_messages
from langgraph.prebuilt import ToolNode
from pydantic import Field, BaseModel


class AgentState(TypedDict):
    messages: Annotated[list, add_messages]


llm = init_chat_model(
    model = 'deepseek-chat',
    model_provider = 'openai',
    api_key = os.getenv('DSKEY'),
    base_url = 'https://api.deepseek.com'
)


class FiledInfo(BaseModel):
    """
    定义参数信息
    """
    city: str = Field(description='城市')


@tool(args_schema=FiledInfo, description='根据城市名称获取温度')
def tp_tool(city: str):
    print('=======tp_tool=======')
    if city == '北京':
        return 12
    elif city == '武汉':
        return 23
    elif city == '沈阳':
        return -10
    elif city == '泉州':
        return 27
    else:
        return None

tools = [tp_tool]

llm_with_tool = llm.bind_tools(tools)

SYSTEM_PROMPT = SystemMessage(content='你是一个智能助手，能调用工具和回答问题')


def chat_node(state: AgentState) -> dict:
    res = llm_with_tool.invoke( [SYSTEM_PROMPT] + state['messages'] )

    return {
        'messages': [res]
    }



tools_node = ToolNode(tools=tools)


def route_after_chat(state: AgentState):
    last_message = state['messages'][-1]

    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return 'tools_node'

    return END


def ask_approval(prompt: str) -> bool:
    """循环读取 y/n 输入，直到合法为止，返回 True 表示批准"""
    while True:
        answer = input(prompt).strip().lower()
        if answer in ("y", "yes"):
            return True
        elif answer in ("n", "no"):
            return False
        print("⚠️ 输入非法，请输入 y 或 n")


if __name__ == '__main__':
    graph = StateGraph(AgentState)

    graph.add_node('chat_node', chat_node)
    graph.add_node('tools_node', tools_node)

    graph.add_edge(START, 'chat_node')
    graph.add_conditional_edges('chat_node', route_after_chat, ['tools_node', END])
    graph.add_edge('tools_node', 'chat_node')

    app = graph.compile(
        checkpointer=InMemorySaver(),
        interrupt_before=['tools_node']
    )

    cfg = {
        'configurable': {'thread_id': 'lzj-001'}
    }

    #app.get_graph().print_ascii()

    app.invoke(
        input={"messages": '在吗？沈阳的温度是？'},
        config=cfg
    )

    snapshot = app.get_state(cfg)

    if snapshot.next:
        #snapshot.next是一个元组，代表下一个节点是那个
        last_msg = snapshot.values['messages'][-1]
        total = len(last_msg.tool_calls)

        for idx, tool_call in enumerate(last_msg.tool_calls, 1):
            print(f'工具{idx}/{total} {tool_call['name']}  ')
            print(f'参数 {tool_call['args'] }  ')
            if not ask_approval('是否调用该工具:'):
                cancel_messages = [
                    ToolMessage(
                        content="工具调用被用户拒绝，请询问用户是否需要调整方案或提供更多信息。",
                        tool_call_id=tc["id"]  # ← 就是 'call_eab76da2cd5e4b5f8aee2d'
                    )
                    for tc in last_msg.tool_calls  # 注意是所有工具，不只是被拒绝的那个
                ]

                app.update_state(
                    config=cfg,
                    values={"messages": cancel_messages}, #注意是values
                    as_node="tools_node"
                )
                break

        final_resp = app.invoke(input=None, config=cfg)
        for e in final_resp['messages']:
            print(e)

工具1/1 tp_tool  
参数 {'city': '沈阳'}  
是否调用该工具:y
=======tp_tool=======
content='在吗？沈阳的温度是？' additional_kwargs={} response_metadata={} id='abf22ea5-9ed8-477e-aa37-1e9dcc51ef7a'
content='在的！我来查一下沈阳的温度。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 52, 'prompt_tokens': 296, 'total_tokens': 348, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 256}, 'prompt_cache_hit_tokens': 256, 'prompt_cache_miss_tokens': 40}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '78861c15-5c71-4f11-9374-0be2a77b1903', 'finish_reason': 'tool_calls', 'logprobs': None} id='lc_run--019fa93f-0e72-7ec1-9fdd-d8c9971b4080-0' tool_calls=[{'name': 'tp_tool', 'args': {'city': '沈阳'}, 'id': 'call_00_O4pTl36Im1yQVTajuqug4756', 'type': 'tool_call'}] invalid_tool_calls=[] usage_metadata={'input_tokens': 296, 'output_tokens': 52, 'total_tokens': 348, 'input_token_details': {'cache_read': 256}, 'output_token_details': {}}
content='-10' name='tp_tool' id='8fefccb2-7a2d-440a-9a91-4e4cd9d3a136' tool_call_id='call_00_O4pTl36Im1yQVTajuqug4756'
content='沈阳目前的温度是 **-10°C**。天气挺冷的，出门的话请注意保暖哦！🧣❄️' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 26, 'prompt_tokens': 361, 'total_tokens': 387, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 256}, 'prompt_cache_hit_tokens': 256, 'prompt_cache_miss_tokens': 105}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '47ca97cb-01f2-4d6b-9c0f-755017270ae2', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019fa93f-2a50-7a01-b1c5-74737cda1647-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 361, 'output_tokens': 26, 'total_tokens': 387, 'input_token_details': {'cache_read': 256}, 'output_token_details': {}}

工具1/1 tp_tool  
参数 {'city': '沈阳'}  
是否调用该工具:n
content='在吗？沈阳的温度是？' additional_kwargs={} response_metadata={} id='f53a70b3-1449-4c6b-a8a1-d0f50183dae9'
content='在的！让我查一下沈阳的温度。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 52, 'prompt_tokens': 296, 'total_tokens': 348, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 256}, 'prompt_cache_hit_tokens': 256, 'prompt_cache_miss_tokens': 40}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '5760c7c4-58e3-44fc-ae75-5ef10fe2ef94', 'finish_reason': 'tool_calls', 'logprobs': None} id='lc_run--019fa949-73ac-7a72-937c-7eed643d9860-0' tool_calls=[{'name': 'tp_tool', 'args': {'city': '沈阳'}, 'id': 'call_00_aZSmmZL3a3GWHJ3PTsJy7979', 'type': 'tool_call'}] invalid_tool_calls=[] usage_metadata={'input_tokens': 296, 'output_tokens': 52, 'total_tokens': 348, 'input_token_details': {'cache_read': 256}, 'output_token_details': {}}
content='工具调用被用户拒绝，请询问用户是否需要调整方案或提供更多信息。' id='d4c34efc-b22c-47c3-ae61-4f24e00ff6b5' tool_call_id='call_00_aZSmmZL3a3GWHJ3PTsJy7979'
content='看起来查询被限制了。请问您是否需要我帮忙做其他事情呢？或者您可以自己查一下沈阳当前的天气情况哦～有其他问题随时问我！😊' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 33, 'prompt_tokens': 376, 'total_tokens': 409, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 256}, 'prompt_cache_hit_tokens': 256, 'prompt_cache_miss_tokens': 120}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '1e2d8c89-42a4-459a-84f6-1d7596bcd098', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019fa949-816a-7a91-9716-d2392b2e705f-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 376, 'output_tokens': 33, 'total_tokens': 409, 'input_token_details': {'cache_read': 256}, 'output_token_details': {}}

"如果文章对您有帮助，可以请作者喝杯咖啡吗？"