人机交互-interrupt_before

graph = builder.compile(
  checkpointer=MemorySaver(),
  # 自定义在执行 step2 之前，中断
  interrupt_before=["step2"]
)

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
  input: str

def step1(state: State):
  print("step1")
  pass

def step2(state: State):
  print("step2")
  pass

def step3(state: State):
  print("step3")
  pass

builder = StateGraph(State).add_sequence([step1, step2, step3]).add_edge(START, "step1")

graph = builder.compile(
  checkpointer=MemorySaver(),
  # 自定义在执行 step2 之前，中断
  interrupt_before=["step2"]
)

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

for event in graph.stream({ "input": "hello" }, thread, stream_mode="values"):
  print(event)

print(graph.get_state(thread).values) # 可以获取到中断前的状态
graph.update_state(thread, { "input": "world" }) # 可以更新状态
print(graph.get_state(thread).values)

for event in graph.stream(None, thread, stream_mode="values"):
  print(event)