添加人机协作控制¶
代理可能不可靠，可能需要人工输入以成功完成任务。同样地，对于某些操作，您可能希望在执行前要求人工批准，以确保一切按预期进行。
LangGraph 的 持久化 层支持 人机协作 工作流，允许根据用户反馈暂停和恢复执行。此功能的主要接口是 interrupt 函数。在节点内部调用 interrupt 将暂停执行。通过传递一个 Command，可以结合来自人类的新输入来恢复执行。interrupt 在使用上类似于 Python 内置的 input()，有一些注意事项。
Note
本教程基于 添加记忆。
1. 添加 human_assistance 工具¶
从 为聊天机器人添加记忆 教程中的现有代码开始，向聊天机器人添加 human_assistance 工具。该工具使用 interrupt 来接收来自人类的信息。
首先，我们选择一个聊天模型：
{}¶
import os
from langchain.chat_models import init_chat_model
os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"
llm = init_chat_model(
"azure_openai:gpt-4.1",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)
现在我们可以将其与附加的工具一起集成到我们的 StateGraph 中：
from typing import Annotated
from langchain_tavily import TavilySearch
from langchain_core.tools import tool
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import Command, interrupt
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
@tool
def human_assistance(query: str) -> str:
"""请求人类的帮助。"""
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"])
# 因为我们将在工具执行期间中断，
# 我们禁用并行调用工具以避免在恢复时重复任何
# 工具调用。
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,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
2. 编译图¶
我们像之前一样使用检查点器来编译图：
3. 可视化图（可选）¶
可视化图，你将获得与之前相同的布局——只是增加了工具！
from IPython.display import Image, display
try:
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
# 这需要一些额外的依赖项，是可选的
pass
4. 提示聊天机器人¶
现在，使用一个问题来提示聊天机器人，该问题将使用新的 human_assistance 工具：
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()
================================ Human Message =================================
I need some expert guidance for building an AI agent. Could you request assistance for me?
================================== Ai Message ==================================
[{'text': "Certainly! I'd be happy to request expert assistance for you regarding building an AI agent. To do this, I'll use the human_assistance function to relay your request. Let me do that for you now.", 'type': 'text'}, {'id': 'toolu_01ABUqneqnuHNuo1vhfDFQCW', 'input': {'query': 'A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?'}, 'name': 'human_assistance', 'type': 'tool_use'}]
Tool Calls:
human_assistance (toolu_01ABUqneqnuHNuo1vhfDFQCW)
Call ID: toolu_01ABUqneqnuHNuo1vhfDFQCW
Args:
query: A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?
聊天机器人生成了一个工具调用，但执行随后被中断。如果你检查图状态，会发现它在工具节点处停止了：
Info
更仔细地看一下 human_assistance 工具：
@tool
def human_assistance(query: str) -> str:
"""Request assistance from a human."""
human_response = interrupt({"query": query})
return human_response["data"]
类似于 Python 内置的 input() 函数，在工具内部调用 interrupt 将暂停执行。进度是基于 checkpointer 持久化的；因此如果使用 Postgres 持久化，只要数据库运行，就可以随时恢复。在这个例子中，它是使用内存中的 checkpointer 进行持久化的，只要 Python 内核在运行，就可以随时恢复。
5. 恢复执行¶
要恢复执行，请传递一个包含工具预期数据的 Command 对象。此数据的格式可以根据需要进行自定义。在本示例中，使用一个带有键 "data" 的字典：
human_response = (
"我们，专家们在这里为您提供帮助！我们建议您查看 LangGraph 来构建您的代理程序。"
"与简单的自主代理相比，它更加可靠且可扩展。"
)
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()
================================== Ai Message ==================================
[{'text': "当然！我很乐意为您请求专家协助来构建AI代理。为此，我将使用human_assistance函数来传达您的请求。让我现在为您完成这个操作。", 'type': 'text'}, {'id': 'toolu_01ABUqneqnuHNuo1vhfDFQCW', 'input': {'query': '用户正在请求关于构建AI代理的专家指导。您能否请提供一些专家建议或资源？'}, 'name': 'human_assistance', 'type': 'tool_use'}]
Tool Calls:
human_assistance (toolu_01ABUqneqnuHNuo1vhfDFQCW)
Call ID: toolu_01ABUqneqnuHNuo1vhfDFQCW
Args:
query: 用户正在请求关于构建AI代理的专家指导。您能否请提供一些专家建议或资源？
================================= Tool Message =================================
Name: human_assistance
我们，专家们在这里为您提供帮助！我们建议您查看 LangGraph 来构建您的代理程序。与简单的自主代理相比，它更加可靠且可扩展。
================================== Ai Message ==================================
感谢您的耐心等待。我已经收到了有关构建AI代理的专家建议。以下是专家们的建议：
专家建议您查看 LangGraph 来构建您的AI代理。他们提到，LangGraph 相比于简单的自主代理来说更加可靠和可扩展。
LangGraph 很可能是一个专门用于创建具有高级功能的AI代理的框架或库。根据这一建议，以下是一些需要考虑的要点：
1. 可靠性：专家强调 LangGraph 比简单自主代理方法更可靠。这可能意味着它具有更好的稳定性、错误处理或一致的性能。
2. 可扩展性：LangGraph 被描述为更具可扩展性，这意味着它可能提供了一个灵活的架构，允许您轻松添加新功能或修改现有功能，以适应代理需求的变化。
3. 高级功能：鉴于其推荐程度高于“简单自主代理”，LangGraph 很可能提供了更多复杂工具和技术来构建复杂的AI代理。
...
2. 寻找专门针对使用 LangGraph 构建AI代理的教程或指南。
3. 检查是否存在任何社区论坛或讨论组，以便您可以提问并从其他使用 LangGraph 的开发者那里获得支持。
如果您想要了解更多关于 LangGraph 的具体信息，或者对这一建议有任何疑问，请随时提出，我可以请求专家提供进一步的帮助。
输出被截断。作为滚动元素查看或在文本编辑器中打开。调整单元格输出设置...
输入已被接收并作为工具消息进行处理。请查看此调用的 LangSmith 追踪，以了解上述调用中确切执行的内容。请注意，在第一步中加载了状态，这样我们的聊天机器人可以从中断的地方继续。
恭喜！ 您已使用 interrupt 将人类介入执行添加到您的聊天机器人中，从而在需要时允许人类监督和干预。这为您能创建的UI打开了更多可能性。由于您已经添加了 checkpointer，只要底层持久化层正在运行，该图就可以 无限期地 暂停，并随时恢复，仿佛什么都没有发生过。
查看下面的代码片段，回顾本教程中的图：
{}¶
import os
from langchain.chat_models import init_chat_model
os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"
llm = init_chat_model(
"azure_openai:gpt-4.1",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)
API Reference: TavilySearch | tool | MemorySaver | StateGraph | START | END | add_messages | ToolNode | tools_condition | Command | interrupt
from typing import Annotated
from langchain_tavily import TavilySearch
from langchain_core.tools import tool
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import Command, interrupt
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
@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"])
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,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
下一步¶
到目前为止，教程示例依赖于一个简单的状态，其中只包含一个条目：消息列表。你可以通过这个简单的状态实现很多功能，但如果你想在不依赖消息列表的情况下定义更复杂的行为，可以向状态中添加额外的字段。