代理可能不可靠，可能需要人工输入才能成功完成任务。同样，对于某些操作，您可能需要在运行前获得人工批准，以确保一切按预期运行。

LangGraph 的持久层支持人机交互工作流，允许根据用户反馈暂停和恢复执行。此功能的主要接口是interrupt函数。在节点内部调用该函数将暂停执行。可以通过传入Commandinterrupt来恢复执行，并接收来自用户的新输入。

interrupt从人体工程学角度来看，它与 Python 的内置类似input()，但有一些注意事项。

本教程以“添加内存”为基础。

添加human_assistance工具

从“为聊天机器人添加内存”教程中的现有代码开始，将human_assistance工具添加到聊天机器人。此工具用于interrupt接收来自人类的信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
from langgraph.types import Command, interrupt
from langchain_core.tools import tool

@tool
def human_assistance(query: str) -> str:
    """Request assistance from a human."""
    human_response = interrupt({"query": query})
    return human_response["data"]
    
    
tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
llm_with_tools = llm.bind_tools(tools)

有关人机交互工作流程的更多信息和示例，请参阅人机交互。

提示聊天机器人

现在，向聊天机器人提出一个可以使用新工具的问题human_assistance：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
user_input = "I need some expert guidance for building an AI agent. Could you request assistance for me?"
config = {"configurable": {"thread_id": "1"}}

events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()
        
def chatbot(state: State):
    message = llm_with_tools.invoke(state["messages"])
    # Because we will be interrupting during tool execution,
    # we disable parallel tool calling to avoid repeating any
    # tool invocations when we resume.
    assert len(message.tool_calls) <= 1
    return {"messages": [message]}
    
graph_builder.add_node("chatbot", chatbot)


tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)


graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)

# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

对话记录：

1
2
3
4
5
6
7
8
9
================================ Human Message =================================

I need some expert guidance for building an AI agent. Could you request assistance for me?
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_ud2cy1vc)
 Call ID: call_ud2cy1vc
  Args:
    query: I need expert guidance for building an AI agent.

聊天机器人生成了一个工具调用，但随后执行被中断。如果您检查图形状态，您会发现它在工具节点处停止了：

1
2
snapshot = graph.get_state(config)
print(snapshot.next)

结果：

1
('tools',)

仔细看看这个human_assistance工具：

1
2
3
4
5
@tool
def human_assistance(query: str) -> str:
    """Request assistance from a human."""
    human_response = interrupt({"query": query})
    return human_response["data"]

与 Python 的内置input()函数类似，interrupt在工具内部调用该函数会暂停执行。进度会根据检查点进行持久化；因此，如果使用 Postgres 进行持久化，只要数据库处于活动状态，就可以随时恢复。在本例中，它使用内存中的检查点进行持久化，只要 Python 内核正在运行，就可以随时恢复。

恢复执行

要恢复执行，请传递一个Command包含工具所需数据的对象。此数据的格式可以根据需求自定义。

对于此示例，使用带有键的字典"data"：

1
2
3
4
5
6
7
8
9
10
11
human_response = (
    "We, the experts are here to help! We'd recommend you check out LangGraph to build your agent."
    " It's much more reliable and extensible than simple autonomous agents."
)

human_command = Command(resume={"data": human_response})

events = graph.stream(human_command, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

对话记录：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
================================ Human Message =================================

I need some expert guidance for building an AI agent. Could you request assistance for me?
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_gslu6y58)
 Call ID: call_gslu6y58
  Args:
    query: I need expert guidance for building an AI agent.
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_gslu6y58)
 Call ID: call_gslu6y58
  Args:
    query: I need expert guidance for building an AI agent.
================================= Tool Message =================================
Name: human_assistance

We, the experts are here to help! We'd recommend you check out LangGraph to build your agent. It's much more reliable and extensible than simple autonomous agents.
================================== Ai Message ==================================

Sure, I can provide some guidance based on the expert advice we received. According to our experts, you should consider using a platform like LangGraph for building your AI agent. They found it to be more reliable and easier to extend compared to simpler autonomous agents.

Would you like more detailed information or specific steps on how to get started with LangGraph?

恭喜！您已使用interrupt为聊天机器人添加了人机交互执行功能，以便在需要时进行人工监督和干预。这开启了您使用 AI 系统创建潜在 UI 的大门。

由于您已添加检查点，只要底层持久层正在运行，图表就可以无限期暂停并随时恢复，就像什么都没发生过一样。

完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
import os
from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
from langchain_tavily import TavilySearch
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt
from langchain_core.tools import tool

import json

from langchain_core.messages import ToolMessage

load_dotenv()

memory = InMemorySaver()

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)


llm = ChatOpenAI(
    model=os.getenv("LLM_MODEL_NAME"),
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL"),
    temperature=os.getenv("LLM_TEMPERATURE"),
    max_tokens=os.getenv("LLM_MAX_TOKENS")
)

@tool
def human_assistance(query: str) -> str:
    """Request assistance from a human."""
    human_response = interrupt({"query": query})
    return human_response["data"]

tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    message = llm_with_tools.invoke(state["messages"])
    # Because we will be interrupting during tool execution,
    # we disable parallel tool calling to avoid repeating any
    # tool invocations when we resume.
    assert len(message.tool_calls) <= 1
    return {"messages": [message]}

graph_builder.add_node("chatbot", chatbot)


tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)


graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)

# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile(checkpointer=memory)


# 画图
png_bytes = graph.get_graph().draw_mermaid_png()

with open("graph.png", "wb") as f:
    f.write(png_bytes)

# import os
# os.system("open graph.png")


user_input = "I need some expert guidance for building an AI agent. Could you request assistance for me?"
config = {"configurable": {"thread_id": "1"}}

events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

snapshot = graph.get_state(config)
# print(snapshot.next)

human_response = (
    "We, the experts are here to help! We'd recommend you check out LangGraph to build your agent."
    " It's much more reliable and extensible than simple autonomous agents."
)

human_command = Command(resume={"data": human_response})

events = graph.stream(human_command, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()