Agent 跑起来之后，大多数时候我们希望它一气呵成把活干完。但总有些场景不太一样——比如 Agent 要调用一个会删文件的工具，你总得让人确认一下再动手吧？LangGraph 的 interrupt 机制就是干这个的：在指定节点前或节点后暂停执行，等人给了信号再接着跑。
动画视频在《28. Agent 执行到一半想暂停？用 interrupt 给它设个"关卡"！》。
interrupt_before：工具执行前拦截
最常用的场景是 interrupt_before。把它设为 ["tools"]，意思是 Agent 调用完模型、决定要用哪个工具之后，先别急着执行，停下来等人发话。配置方式很简单，在 create_agent 里加一个参数就行：
agent = create_agent( model=model, tools=[calculate, write_file, read_file, list_dir], system_prompt="你是一个助手，会用工具计算、读写文件、列出目录。", debug=True, checkpointer=checkpointer, interrupt_before=["tools"] )
这里有个前提——必须同时传入 checkpointer。原因很直接：暂停之后要恢复，就得靠 checkpointer 把当前状态存下来，不然中断 了就回不来了。
发送消息，观察暂停效果
配好之后，用 agent.invoke() 发一条消息试试：
config = {"configurable": {"thread_id": "session-2"}}
q = "计算 2024*12+500"
result = agent.invoke({"messages": [{"role": "user", "content": q}]}, config=config)
运行之后你会发现，Agent 调用模型产生了工具调用请求，但并没有真正执行工具——它停在了 tools 节点之前。result["messages"] 里的最后一条消息，就是 AI 生成的工具调用请求，你可以把它打印出来让用户看看 Agent 接下来打算干什么：
print(f"Agent 已暂停，最后消息: {result['messages'][-1].content[:100]}...")
用 Command(resume=...) 恢复执行
暂停不是目的，关键是暂停之后怎么继续。LangGraph 提供了一个 Command 对象，通过它的 resume 参数告诉 Agent"可以走了"：
from langgraph.types import Command # 恢复执行 result = agent.invoke(Command(resume="approved"), config=config) print(f"答：{result['messages'][-1].content}")
注意 thread_id 不用换——因为 checkpointer 靠它找到上次暂停时的状态，从断点 处继续往下走。
如果把这个过程做成交互式的，就是让用户自己决定要不要放行：
# 暂停后等待用户输入 user_input = input("确认执行？(yes/no): ") if user_input.lower() == "yes": result = agent.invoke(Command(resume=user_input), config=config) # 找到最后一条有内容的消息 for msg in reversed(result['messages']): if isinstance(msg, ToolMessage): print(f"答：{msg.content}") break else: print("已取消")
用户输入 yes，Agent 就继续调用工具、拿结果、生成回答；输入别的内容，这次工具调用就直接丢弃，流程到此结束。
interrupt_after：工具执行后再拦截
刚才说的是"事前审批"——工具还没跑，先问问人。那如果想在工具跑完之后再加一道关卡呢？把参数从 interrupt_before 换成 interrupt_after 就行，其他代码完全不用动：
agent = create_agent( model=model, tools=[calculate, write_file, read_file, list_dir], system_prompt="你是一个助手，会用工具计算、读写文件、列出目录。", debug=True, checkpointer=checkpointer, interrupt_after =["tools"] )
一字之差，执行时机完全不同。interrupt_before 是 Agent 想好要调用什么工具了、但还没动手的时候拦截；interrupt_after 是工具已经跑完了、结果也拿到了，但 Agent 还没把结果送回模型做下一步推理的时候拦截。前者适合"你确定要这么做吗"，后者适合"做完了，你看看结果对不对"。
恢复的方式一模一样：
user_input = input("工具已执行完毕，确认继续？(yes/no): ") if user_input.lower() == "yes": result = agent.invoke(Command(resume=user_input), config=config) # 找到最后一条有内容的消息 for msg in reversed(result['messages']): if isinstance(msg, ToolMessage): print(f"答：{msg.content}") break else: print("已取消")
把上面的内容串起来，下面是一份可以直接运行的完整代码：
import os import sqlite3 from dotenv import load_dotenv from langchain.agents import create_agent, AgentState from langchain.agents.middleware import AgentMiddleware from langchain.chat_models import init_chat_model from langchain_classic.agents import Agent from langchain_community.tools import WriteFileTool, ReadFileTool, ListDirectoryTool from langchain_core.messages import AIMessage, ToolMessage from langchain_core.tools import tool, BaseTool from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.runtime import Runtime from langgraph.types import Command load_dotenv() prefix = "QWEN" model = init_chat_model( model_provider="openai", configurable_fields=["model", "api_key", "base_url"], config_prefix=prefix ).with_config({ "configurable": { f"{prefix}_model": os.getenv(f"{prefix}_MODEL"), f"{prefix}_api_key": os.getenv(f"{prefix}_API_KEY"), f"{prefix}_base_url": os.getenv(f"{prefix}_BASE_URL") } }) class CalculateTool(BaseTool): name: str = "calculate" description: str = "计算数学表达式的值" def _run(self, expression: str) -> str: try: return f"计算结果：{eval(expression)}" except Exception as e: return f"计算错误：{str(e)}" async def _arun(self, expression: str) -> str: return self._run(expression) calculate = CalculateTool() write_file = WriteFileTool() read_file = ReadFileTool() list_dir = ListDirectoryTool() checkpoint_conn = sqlite3.connect("agent.db", check_same_thread=False, isolation_level=None) checkpointer = SqliteSaver(checkpoint_conn) agent = create_agent( model=model, tools=[calculate, write_file, read_file, list_dir], system_prompt="你是一个助手，会用工具计算、读写文件、列出目录。", debug=True, checkpointer=checkpointer, interrupt_before=["tools"] ) config = {"configurable": {"thread_id": "session-2"}} q = "计算 2024*12+500" result = agent.invoke({"messages": [{"role": "user", "content": q}]}, config=config) print(f"Agent 已暂停，最后消息: {result['messages'][-1].content[:100]}...") # 暂停后等待用户输入 user_input = input("确认执行？(yes/no): ") if user_input.lower() == "yes": result = agent.invoke(Command(resume=user_input), config=config) # 找到最后一条有内容的消息 for msg in reversed(result['messages']): if isinstance(msg, ToolMessage): print(f"答：{msg.content}") break else: print("已取消") checkpoint_conn.close()
运行这份代码，你会看到这样的流程：发送问题后，Agent 调用模型、决定使用 calculate 工具，但在执行前暂停；控制台打印出 Agent 打算执行的操作，等待你输入；输入 yes，Agent 继续执行工具调用，计算结果、写文件，最后返回完整回答；输入其他内容，这次调用直接取消。
如果要把暂停点改到工具执行之后，只需要把 interrupt_before=["tools"] 改成 interrupt_after=["tools"]，其他代码一行不用动。
interrupt_before 适合"事前审批"，工具还没跑，先看看 Agent 想干嘛；interrupt_after 适合"事后审查"，工具跑完了，检查一下结果再继续。两种机制配合使用，就能把 Agent 的执行流程牢牢攥在手里。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。