当前位置：首页 > ai >正文

LangGraph-2-interrupt 流程中断

ai 2025/7/13 9:44:47

interrupt 是 LangGraph 中一个强大的流程控制机制，允许在状态机执行过程中根据特定条件中断当前流程并跳转到其他节点。这种机制特别适用于处理异常情况、用户中断或特定业务规则的触发。

在 LangGraph 中，interrupt_before 和 interrupt_after 是两个强大的流程控制机制，允许在节点执行前或执行后插入中断检查。这些机制提供了更精细的流程控制能力，特别适合需要精确管理执行顺序和条件的复杂工作流。

interrupt_before 一般用于指定在执行某个特定操作或者步骤之前中断流程。这一机制可以让开发者在关键操作执行前对条件进行检查，若不满足条件，就提前终止流程，避免不必要的计算或者操作。

#在tools节点执行前中断
graph = graph.compile(checkpointer=memory_checkpointer, interrupt_before=["tools"])

流程恢复执行：

graph.invoke(None, config=config)

案例，在调用外部工具前，进行人工确认，如果输入y或者yes流程恢复执行：

import os
from typing import Annotated, TypedDict
from langchain_tavily import TavilySearch
from langgraph.checkpoint.memory import MemorySaver
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from model.deepseek import deepseek_llmclass ChatSate(TypedDict):# messages:状态中保存数据的keymessages: Annotated[list, add_messages]graph = StateGraph(ChatSate)# 定义一个互联网搜索工具
os.environ["TAVILY_API_KEY"] = ""
search_tool = TavilySearch(max_result=5)
runnable = deepseek_llm.bind_tools([search_tool])# 定义第一个节点
def chatbot(state: ChatSate) -> ChatSate:resp = runnable.invoke(state["messages"])return {"messages": resp}# 定义第二个节点工具节点
too_node = ToolNode([search_tool])# 添加边
graph.add_node("agent", chatbot)
graph.add_node("tools", too_node)
# 根据智能体自主决定是否调用工具
graph.add_conditional_edges("agent",#条件变量，根据智能体的回答决定是否调用工具tools_condition
)
graph.add_edge("tools", "agent")  # 流程从tools 到 chatbot
graph.add_edge(START, "agent")  # 流程从start 到 chatbot
graph.add_edge("agent", END)  # 流程从chatbot 到 END
graph.set_entry_point("agent")  # 设置入口节点
#保存对话记录到内存中
memory_checkpointer = MemorySaver()
graph = graph.compile(checkpointer=memory_checkpointer, interrupt_before=["tools"])# png = graph.get_graph().draw_mermaid_png()
# with open("graph2.png", "wb") as f:
#    f.write(png)
config={"configurable": {"thread_id": "1234"}}
def loop_graph_invoke(user_input: str):if user_input:result = graph.invoke({'messages': [('user', user_input)]}, config=config)if type(result['messages']) is list:print('AI:', result['messages'][-1].content)else:print('AI:', result['messages'].content)else:result = graph.invoke(None, config=config)if type(result['messages']) is list:print('AI:', result['messages'][-1].content)else:print('AI:', result['messages'].content)while True:user_input = input('User:')if user_input.lower() in ['q', 'exit', 'bye', 'quit']:print('AI:', 'Bye')breakelse:loop_graph_invoke(user_input)state = graph.get_state(config)if "tools" in state.next:an = input('是否允许调用外部工具？(y/n)')if an.lower() in ['y', 'yes']:#继续执行流程loop_graph_invoke(None)else:#退出当前流程pass

测试验证：