??interrupt?是 LangGraph 中一個強大的流程控制機制，允許在狀態機執行過程中根據特定條件中斷當前流程并跳轉到其他節點。這種機制特別適用于處理異常情況、用戶中斷或特定業務規則的觸發。

????????在 LangGraph 中，interrupt_before?和?interrupt_after?是兩個強大的流程控制機制，允許在節點執行前或執行后插入中斷檢查。這些機制提供了更精細的流程控制能力，特別適合需要精確管理執行順序和條件的復雜工作流。

????????interrupt_before?一般用于指定在執行某個特定操作或者步驟之前中斷流程。這一機制可以讓開發者在關鍵操作執行前對條件進行檢查，若不滿足條件，就提前終止流程，避免不必要的計算或者操作。

????????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

測試驗證：