LangGraph人工干预（Human-in-the-Loop）
本文最后更新于 2026年7月26日
未完待续
一些节点的操作比较敏感高危，需要人工进行二次确认，此时就用到了interrupt，interrupt可以在任何节点内以一种特殊的异常的方式，主动暂停图的执行，冻结当前图的状态（依赖checkpointer持久化），向外抛出自定义数据（给前端/控制台展示待审核内容），外部接收用户输入后，通过Command把用户决策回填，继续运行。
interrupt依赖三大机制:
checkpointer
必须配置才能保存中断前状态，否则无法恢复，默认
InMemorySaver保存在内存，生产环境可以换成redis等持久化存储。thread_id
会话唯一标识，同一个对话/任务共用一个thread，中断恢复靠它匹配状态
Command
恢复图执行的专用对象，可以通过
resume进行恢复，也能goto到其他节点
例：输入用户名
第一次发起后，进入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=数据不对，改掉！'}"如果文章对您有帮助，可以请作者喝杯咖啡吗？"
微信支付
支付宝
LangGraph人工干预（Human-in-the-Loop）
https://blog.liuzijian.com/post/2026/07/26/langgraph-human-in-the-loop/