如何添加动态断点¶

人机交互 (HIL) 对代理系统至关重要。断点是常见的 HIL 交互模式，允许图在特定步骤停止并寻求人类批准后再继续（例如，对于敏感操作）。

在 LangGraph 中，您可以在节点执行之前或之后添加断点。但通常情况下，根据某些条件从给定节点内部动态中断图可能会有所帮助。这样做时，将有关原因的信息包含到中断中可能也很有帮助。

本指南展示了如何使用 NodeInterrupt 动态中断图，NodeInterrupt 是一种可以从节点内部引发的特殊异常。让我们看看它的实际操作吧！

设置¶

首先，让我们安装所需的软件包

In [1]

%%capture --no-stderr
%pip install -U langgraph

%%capture --no-stderr %pip install -U langgraph

为 LangGraph 开发设置 LangSmith

注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序 - 阅读有关如何开始的更多信息 此处.

定义图¶

In [1]

from typing_extensions import TypedDict
from IPython.display import Image, display

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.errors import NodeInterrupt


class State(TypedDict):
    input: str


def step_1(state: State) -> State:
    print("---Step 1---")
    return state


def step_2(state: State) -> State:
    # Let's optionally raise a NodeInterrupt
    # if the length of the input is longer than 5 characters
    if len(state["input"]) > 5:
        raise NodeInterrupt(
            f"Received input that is longer than 5 characters: {state['input']}"
        )

    print("---Step 2---")
    return state


def step_3(state: State) -> State:
    print("---Step 3---")
    return state


builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)

# Set up memory
memory = MemorySaver()

# Compile the graph with memory
graph = builder.compile(checkpointer=memory)

# View
display(Image(graph.get_graph().draw_mermaid_png()))

from typing_extensions import TypedDict from IPython.display import Image, display from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langgraph.errors import NodeInterrupt class State(TypedDict): input: str def step_1(state: State) -> State: print("---Step 1---") return state def step_2(state: State) -> State: # 让我们有条件地引发 NodeInterrupt # 如果输入的长度大于 5 个字符 if len(state["input"]) > 5: raise NodeInterrupt( f"Received input that is longer than 5 characters: {state['input']}" ) print("---Step 2---") return state def step_3(state: State) -> State: print("---Step 3---") return state builder = StateGraph(State) builder.add_node("step_1", step_1) builder.add_node("step_2", step_2) builder.add_node("step_3", step_3) builder.add_edge(START, "step_1") builder.add_edge("step_1", "step_2") builder.add_edge("step_2", "step_3") builder.add_edge("step_3", END) # 设置内存 memory = MemorySaver() # 使用内存编译图 graph = builder.compile(checkpointer=memory) # 查看 display(Image(graph.get_graph().draw_mermaid_png()))

使用动态中断运行图¶

首先，让我们使用长度 <= 5 个字符的输入运行图。这应该安全地忽略我们定义的中断条件，并在图执行结束时返回原始输入。

In [3]

initial_input = {"input": "hello"}
thread_config = {"configurable": {"thread_id": "1"}}

for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

initial_input = {"input": "hello"} thread_config = {"configurable": {"thread_id": "1"}} for event in graph.stream(initial_input, thread_config, stream_mode="values"): print(event)

{'input': 'hello'}
---Step 1---
{'input': 'hello'}
---Step 2---
{'input': 'hello'}
---Step 3---
{'input': 'hello'}

如果我们在此时检查图，我们可以看到没有更多任务要运行，并且图确实已完成执行。

In [4]

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

state = graph.get_state(thread_config) print(state.next) print(state.tasks)

()
()

现在，让我们使用长度超过 5 个字符的输入运行图。这应该会触发我们通过在 step_2 节点内部引发 NodeInterrupt 错误来定义的动态中断。

In [5]

initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "2"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

initial_input = {"input": "hello world"} thread_config = {"configurable": {"thread_id": "2"}} # 运行图直到第一次中断 for event in graph.stream(initial_input, thread_config, stream_mode="values"): print(event)

{'input': 'hello world'}
---Step 1---
{'input': 'hello world'}

我们可以看到图现在在执行 step_2 时停止了。如果我们在此时检查图状态，我们可以看到关于接下来要执行的节点（step_2）的信息，以及引发中断的节点（也是 step_2），以及有关中断的其他信息。

In [6]

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

state = graph.get_state(thread_config) print(state.next) print(state.tasks)

('step_2',)
(PregelTask(id='365d4518-bcff-5abd-8ef5-8a0de7f510b0', name='step_2', error=None, interrupts=(Interrupt(value='Received input that is longer than 5 characters: hello world', when='during'),)),)

如果我们尝试从断点恢复图，我们只会再次中断，因为我们的输入和图状态没有改变。

In [7]

# NOTE: to resume the graph from a dynamic interrupt we use the same syntax as with regular interrupts -- we pass None as the input
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

# 注意：要从动态中断恢复图，我们使用与常规中断相同的语法 - 我们将 None 作为输入传递 for event in graph.stream(None, thread_config, stream_mode="values"): print(event)

In [8]

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

state = graph.get_state(thread_config) print(state.next) print(state.tasks)

('step_2',)
(PregelTask(id='365d4518-bcff-5abd-8ef5-8a0de7f510b0', name='step_2', error=None, interrupts=(Interrupt(value='Received input that is longer than 5 characters: hello world', when='during'),)),)

更新图状态¶

为了解决这个问题，我们可以做几件事。

首先，我们可以简单地在另一个线程上使用更短的输入运行图，就像我们一开始做的那样。或者，如果我们想从断点恢复图执行，我们可以更新状态以包含一个长度小于 5 个字符的输入（我们的中断条件）。

In [9]

# NOTE: this update will be applied as of the last successful node before the interrupt, i.e. `step_1`, right before the node with an interrupt
graph.update_state(config=thread_config, values={"input": "foo"})
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print(state.next)
print(state.values)

# 注意：此更新将从中断之前最后一个成功节点（即 `step_1`）开始应用，即在有中断的节点之前 graph.update_state(config=thread_config, values={"input": "foo"}) for event in graph.stream(None, thread_config, stream_mode="values"): print(event) state = graph.get_state(thread_config) print(state.next) print(state.values)

---Step 2---
{'input': 'foo'}
---Step 3---
{'input': 'foo'}
()
{'input': 'foo'}

您还可以作为节点 step_2（中断节点）更新状态，这将完全跳过该节点

In [10]

initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "3"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

initial_input = {"input": "hello world"} thread_config = {"configurable": {"thread_id": "3"}} # 运行图直到第一次中断 for event in graph.stream(initial_input, thread_config, stream_mode="values"): print(event)

{'input': 'hello world'}
---Step 1---
{'input': 'hello world'}

In [11]

# NOTE: this update will skip the node `step_2` altogether
graph.update_state(config=thread_config, values=None, as_node="step_2")
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print(state.next)
print(state.values)

# 注意：此更新将完全跳过节点 `step_2` graph.update_state(config=thread_config, values=None, as_node="step_2") for event in graph.stream(None, thread_config, stream_mode="values"): print(event) state = graph.get_state(thread_config) print(state.next) print(state.values)

---Step 3---
{'input': 'hello world'}
()
{'input': 'hello world'}

评论