Agent 执行到一半想暂停？用 interrupt 给它设个“关卡“！

Agent 跑起来之后大多数时候我们希望它一气呵成把活干完。但总有些场景不太一样——比如 Agent 要调用一个会删文件的工具你总得让人确认一下再动手吧LangGraph 的 interrupt 机制就是干这个的在指定节点前或节点后暂停执行等人给了信号再接着跑。动画视频在《28. Agent 执行到一半想暂停用 interrupt 给它设个关卡》。interrupt_before工具执行前拦截最常用的场景是 interrupt_before。把它设为 [tools]意思是 Agent 调用完模型、决定要用哪个工具之后先别急着执行停下来等人发话。配置方式很简单在 create_agent 里加一个参数就行agent create_agent( modelmodel, tools[calculate, write_file, read_file, list_dir], system_prompt你是一个助手会用工具计算、读写文件、列出目录。, debugTrue, checkpointercheckpointer, interrupt_before[tools] )这里有个前提——必须同时传入 checkpointer。原因很直接暂停之后要恢复就得靠 checkpointer 把当前状态存下来不然中断 了就回不来了。发送消息观察暂停效果配好之后用 agent.invoke() 发一条消息试试config {configurable: {thread_id: session-2}} q 计算 2024*12500 result agent.invoke({messages: [{role: user, content: q}]}, configconfig)运行之后你会发现Agent 调用模型产生了工具调用请求但并没有真正执行工具——它停在了 tools 节点之前。result[messages] 里的最后一条消息就是 AI 生成的工具调用请求你可以把它打印出来让用户看看 Agent 接下来打算干什么print(fAgent 已暂停最后消息: {result[messages][-1].content[:100]}...)用 Command(resume...) 恢复执行暂停不是目的关键是暂停之后怎么继续。LangGraph 提供了一个 Command 对象通过它的 resume 参数告诉 Agent可以走了from langgraph.types import Command # 恢复执行 result agent.invoke(Command(resumeapproved), configconfig) print(f答{result[messages][-1].content})注意 thread_id 不用换——因为 checkpointer 靠它找到上次暂停时的状态从断点 处继续往下走。如果把这个过程做成交互式的就是让用户自己决定要不要放行# 暂停后等待用户输入 user_input input(确认执行(yes/no): ) if user_input.lower() yes: result agent.invoke(Command(resumeuser_input), configconfig) # 找到最后一条有内容的消息 for msg in reversed(result[messages]): if isinstance(msg, ToolMessage): print(f答{msg.content}) break else: print(已取消)用户输入 yesAgent 就继续调用工具、拿结果、生成回答输入别的内容这次工具调用就直接丢弃流程到此结束。interrupt_after工具执行后再拦截刚才说的是事前审批——工具还没跑先问问人。那如果想在工具跑完之后再加一道关卡呢把参数从 interrupt_before 换成 interrupt_after 就行其他代码完全不用动agent create_agent( modelmodel, tools[calculate, write_file, read_file, list_dir], system_prompt你是一个助手会用工具计算、读写文件、列出目录。, debugTrue, checkpointercheckpointer, interrupt_after [tools] )一字之差执行时机完全不同。interrupt_before 是 Agent 想好要调用什么工具了、但还没动手的时候拦截interrupt_after 是工具已经跑完了、结果也拿到了但 Agent 还没把结果送回模型做下一步推理的时候拦截。前者适合你确定要这么做吗后者适合做完了你看看结果对不对。恢复的方式一模一样user_input input(工具已执行完毕确认继续(yes/no): ) if user_input.lower() yes: result agent.invoke(Command(resumeuser_input), configconfig) # 找到最后一条有内容的消息 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_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix ).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_threadFalse, isolation_levelNone) checkpointer SqliteSaver(checkpoint_conn) agent create_agent( modelmodel, tools[calculate, write_file, read_file, list_dir], system_prompt你是一个助手会用工具计算、读写文件、列出目录。, debugTrue, checkpointercheckpointer, interrupt_before[tools] ) config {configurable: {thread_id: session-2}} q 计算 2024*12500 result agent.invoke({messages: [{role: user, content: q}]}, configconfig) print(fAgent 已暂停最后消息: {result[messages][-1].content[:100]}...) # 暂停后等待用户输入 user_input input(确认执行(yes/no): ) if user_input.lower() yes: result agent.invoke(Command(resumeuser_input), configconfig) # 找到最后一条有内容的消息 for msg in reversed(result[messages]): if isinstance(msg, ToolMessage): print(f答{msg.content}) break else: print(已取消) checkpoint_conn.close()运行这份代码你会看到这样的流程发送问题后Agent 调用模型、决定使用 calculate 工具但在执行前暂停控制台打印出 Agent 打算执行的操作等待你输入输入 yesAgent 继续执行工具调用计算结果、写文件最后返回完整回答输入其他内容这次调用直接取消。如果要把暂停点改到工具执行之后只需要把 interrupt_before[tools] 改成 interrupt_after[tools]其他代码一行不用动。interrupt_before 适合事前审批工具还没跑先看看 Agent 想干嘛interrupt_after