找到一个静态的人机断点示例，测试通过，记录一下。

代码

from IPython.display import Image, display
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import START, StateGraph, MessagesState
from langgraph.prebuilt import tools_condition, ToolNode
from langchain_openai import ChatOpenAI
import os
from dotenv import load_dotenv

# 生产级配置
load_dotenv()
siliconflow_api_key = os.getenv("SILICONFLOWS_API_KEY")
llm = ChatOpenAI(
    model_name="deepseek-ai/DeepSeek-V2.5",
    openai_api_key=siliconflow_api_key,
    openai_api_base="https://api.siliconflow.cn/v1",
    temperature=0.3
)

def multiply(a: int, b: int) -> int:
    """Multiply a and b.

    Args:
        a: first int
        b: second int
    """
    return a * b

def add(a: int, b: int) -> int:
    """Adds a and b.

    Args:
        a: first int
        b: second int
    """
    return a + b

def divide(a: int, b: int) -> float:
    """Divide a by b.

    Args:
        a: first int
        b: second int
    """
    return a / b

tools = [add, multiply, divide]
llm_with_tools = llm.bind_tools(tools)

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

# System message
sys_msg = SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")

# Node
def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}


# Graph
builder = StateGraph(MessagesState)

# Define nodes: these do the work
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))

# Define edges: these determine the control flow
builder.add_edge(START, "assistant")
builder.add_conditional_edges(
    "assistant",
    tools_condition,
)
builder.add_edge("tools", "assistant")

memory = MemorySaver()
graph = builder.compile(interrupt_before=["tools"], checkpointer=memory)

# 🎨 可视化图结构
display(Image(graph.get_graph().draw_mermaid_png()))

"""
# Input
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}

# Thread
thread = {"configurable": {"thread_id": "1"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

state = graph.get_state(thread)
print(state.next)

for event in graph.stream(None, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()
"""

# Input
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}

# Thread
thread = {"configurable": {"thread_id": "2"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

# Get user feedback
user_approval = input("Do you want to call the tool? (yes/no): ")

# Check approval
if user_approval.lower() == "yes":
    # If approved, continue the graph execution
    for event in graph.stream(None, thread, stream_mode="values"):
        event['messages'][-1].pretty_print()
else:
    print("Operation cancelled by user.")

输出：

D:\book\PythonAI\Code\.venv\Scripts\python.exe D:\book\PythonAI\Code\ch12\langgraph_interrupt.py 
<IPython.core.display.Image object>
================================ Human Message =================================

Multiply 2 and 3
================================== Ai Message ==================================
Tool Calls:
  multiply (019c0d609d19f284f7d4fa384b3b5cae)
 Call ID: 019c0d609d19f284f7d4fa384b3b5cae
  Args:
    a: 2
    b: 3
Do you want to call the tool? (yes/no): yes
================================== Ai Message ==================================
Tool Calls:
  multiply (019c0d609d19f284f7d4fa384b3b5cae)
 Call ID: 019c0d609d19f284f7d4fa384b3b5cae
  Args:
    a: 2
    b: 3
================================= Tool Message =================================
Name: multiply

6
================================== Ai Message ==================================
Tool Calls:
  multiply (019c0d60c9bedc026a02fab707bc846d)
 Call ID: 019c0d60c9bedc026a02fab707bc846d
  Args:
    a: 2
    b: 3

进程已结束，退出代码为 0