如何等待用户输入¶

人机交互 (HIL) 对 代理系统 至关重要。等待用户输入是一种常见的人机交互模式，它允许代理向用户提出澄清问题并等待输入，然后再继续执行。

我们可以在 LangGraph 中使用 断点 来实现这一点：断点允许我们在图执行过程中的特定步骤停止执行。在这个断点处，我们可以等待用户输入。一旦我们获得用户的输入，我们就可以将它添加到图状态并继续执行。

设置¶

首先，我们需要安装所需的包。

在 [1]

%%capture --no-stderr
%pip install --quiet -U langgraph langchain_anthropic langchain_openai

%%capture --no-stderr %pip install --quiet -U langgraph langchain_anthropic langchain_openai

接下来，我们需要设置 Anthropic 和/或 OpenAI 的 API 密钥（我们将使用的 LLM）。

在 [ ]

import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("OPENAI_API_KEY")
_set_env("ANTHROPIC_API_KEY")

import getpass import os def _set_env(var: str): if not os.environ.get(var): os.environ[var] = getpass.getpass(f"{var}: ") _set_env("OPENAI_API_KEY") _set_env("ANTHROPIC_API_KEY")

设置 LangSmith 用于 LangGraph 开发

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

简单使用¶

让我们看一下这种方法的基本用法。一个直观的做法是简单地创建一个节点，human_feedback，来获取用户反馈。这允许我们在图中的特定点插入反馈收集操作。

1. 我们使用 interrupt_before 在 human_feedback 节点之前指定 断点。

2. 我们设置一个 检查点 来保存该节点之前的图状态。

3. 我们使用 .update_state 来使用我们收到的用户响应更新图的状态。

• 我们 使用 as_node 参数 将此状态更新应用于指定的节点 human_feedback。
• 然后，图将恢复执行，就好像 human_feedback 节点刚刚执行一样。

在 [5]

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from IPython.display import Image, display


class State(TypedDict):
    input: str
    user_feedback: str


def step_1(state):
    print("---Step 1---")
    pass


def human_feedback(state):
    print("---human_feedback---")
    pass


def step_3(state):
    print("---Step 3---")
    pass


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

# Set up memory
memory = MemorySaver()

# Add
graph = builder.compile(checkpointer=memory, interrupt_before=["human_feedback"])

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

from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from IPython.display import Image, display class State(TypedDict): input: str user_feedback: str def step_1(state): print("---Step 1---") pass def human_feedback(state): print("---human_feedback---") pass def step_3(state): print("---Step 3---") pass builder = StateGraph(State) builder.add_node("step_1", step_1) builder.add_node("human_feedback", human_feedback) builder.add_node("step_3", step_3) builder.add_edge(START, "step_1") builder.add_edge("step_1", "human_feedback") builder.add_edge("human_feedback", "step_3") builder.add_edge("step_3", END) # 设置内存 memory = MemorySaver() # 添加图 graph = builder.compile(checkpointer=memory, interrupt_before=["human_feedback"]) # 查看 display(Image(graph.get_graph().draw_mermaid_png()))