人机交互-interrupt

from typing import TypedDict
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
  # 用户输入
  user_feedback: str

# 第一步
def before(state: State):
  print("---before---")
  pass

# 第二步
def human_feedback(state: State):
  print("---human_feedback---")
  feedback = interrupt("请输入回调：")
  return { "user_feedback": feedback }

# 第三步
def after(state: State):
  print("---after---")
  print(state['user_feedback'])
  pass

graph = StateGraph(State).add_node(before).add_node(human_feedback).add_node(after).add_edge(START, "before").add_edge("before", "human_feedback").add_edge("human_feedback", "after").add_edge("after", END).compile(checkpointer=MemorySaver())

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

for event in graph.stream({}, config):
  print(event)
  print("\n")

for event in graph.stream(
  Command(resume="hello world!"),
  config
):
  print(event)
  print("\n")

from typing import Literal
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from my_openai.deepseekv3 import get_deepseek_v3_client
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import Command, interrupt

# 模拟一个天气搜索工具
@tool
def search_weather(city: str):
  """搜索一个城市的天气"""
  print(f"---搜索的城市：{city}---")
  return "晴朗"

tools = [search_weather]
# 创建工具节点
tool_node = ToolNode(tools)

model = get_deepseek_v3_client().bind_tools(tools)

def call_llm(state: MessagesState):
  return { "messages": [model.invoke(state["messages"])] }

# 定义用户反馈
def human_review_node(state: MessagesState) -> Command[Literal["call_llm", "tool_node"]]:
  last_message = state["messages"][-1]
  tool_call = last_message.tool_calls[0]

  # 把 AI 调用的工具函数的情况告诉用户，等待用户决定如何处理
  human_review = interrupt(
    {
      "question": "正确吗?",
      "tool_call": tool_call
    }
  )

  # 用户反馈意图
  review_action = human_review["action"]
  # 用户反馈数据（可能存在）
  review_data = human_review.get("data")

  # 如果用户返回同意执行
  if review_action == "continue":
    # 则继续正常往下执行，即把状态给到工具节点
    return Command(goto="tool_node")

  # 如果用户反馈，修改执行的入参
  elif review_action == "update":
    updated_message = AIMessage(
      content=last_message.content,
      tool_calls=[
        {
          "id": tool_call["id"],
          "name": tool_call["name"],
          # 使用用户传入的入参
          "args": review_data
        }
      ],
      id=last_message.id
    )
    return Command(goto="tool_node", update={ "messages": [updated_message] })

  # 用户输入自然语言，反馈给大模型
  elif review_action == "feedback":
    # 为了保证对话消息顺序是正确的，我们这里返回一个工具消息
    tool_message = ToolMessage(
      # 用户反馈消息，作为工具函数的结果
      content=review_data,
      tool_call_id=tool_call["id"],
      name=tool_call["name"]
    )
    return Command(goto="call_llm", update={ "messages": [tool_message] })

builder = StateGraph(MessagesState)
builder.add_node(call_llm).add_node("tool_node", tool_node).add_node(human_review_node)
builder.add_edge(START, "call_llm")
builder.add_conditional_edges(
  "call_llm", 
  tools_condition,
  { "tools": "human_review_node", END: END }
)
builder.add_edge("tool_node", "call_llm")

graph = builder.compile(checkpointer=MemorySaver())

for event in graph.stream(
  { "messages": HumanMessage(content="你好") },
  {"configurable": {"thread_id": "1"}},
  stream_mode="updates"
):
    print(event)

for event in graph.stream(
  { "messages": HumanMessage(content="今天北京天气如何？") },
  {"configurable": {"thread_id": "2"}},
  stream_mode="updates"
):
    print(event)
    print("\n")

for event in graph.stream(
    Command(resume={"action": "continue"}),
    {"configurable": {"thread_id": "2"}},
    stream_mode="updates",
):
    print(event)
    print("\n")

for event in graph.stream(
  { "messages": HumanMessage(content="今天北京天气如何？") },
  {"configurable": {"thread_id": "3"}},
  stream_mode="updates"
):
    print(event)
    print("\n")

for event in graph.stream(
    Command(resume={"action": "update", "data": {"city": "上海"}}),
    {"configurable": {"thread_id": "3"}},
    stream_mode="updates",
):
    print(event)
    print("\n")

for event in graph.stream(
  { "messages": HumanMessage(content="今天北京天气如何？") },
  {"configurable": {"thread_id": "4"}},
  stream_mode="updates"
):
    print(event)
    print("\n")

for event in graph.stream(
    Command(resume={"action": "feedback", "data": "问错了，我想问上海的天气如何？"}),
    {"configurable": {"thread_id": "4"}},
    stream_mode="updates",
):
    print(event)
    print("\n")

for event in graph.stream(
    Command(resume={"action": "continue"}),
    {"configurable": {"thread_id": "4"}},
    stream_mode="updates",
):
    print(event)
    print("\n")