AI Agent 在执行任务时可能并不可靠，有时需要人类的输入才能成功完成任务。同样，对于某些敏感或关键操作，我们可能希望在执行前获得人工批准，以确保一切按预期进行。
LangGraph 的持久化 (persistence) 层原生支持人工干预 (human-in-the-loop) 工作流，允许执行过程根据用户反馈暂停和恢复。实现这一功能的核心是 interrupt 函数。在一个节点 (node) 内部调用 interrupt 将会暂停图 (graph) 的执行。之后，我们可以通过传入一个 Command 对象，并携带来自人类的新输入来恢复执行。
interrupt 的使用方式与 Python 内置的 input() 函数类似，但也有一些注意事项。
注意：本教程基于前一篇《为聊天机器人添加记忆》构建。
步骤一：添加 human_assistance 工具
我们从上一教程的代码开始，为聊天机器人添加一个名为 human_assistance 的新工具。这个工具将使用 interrupt 函数来从人类用户那里获取信息。
首先，我们选择并初始化一个聊天模型 (Chat Model)：
OpenAI
pip install -U "langchain[openai]"
import os
from langchain.chat_models import init_chat_model
os.environ["OPENAI_API_KEY"] = "sk-..."
llm = init_chat_model("openai:gpt-4o")
👉 阅读 OpenAI 集成文档
Anthropic
pip install -U "langchain[anthropic]"
import os
from langchain.chat_models import init_chat_model
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
👉 阅读 [Anthropic 集成文档](https://python.langchain.com/docs/integrations/chat/anthropic/)
Azure
pip install -U "langchain[openai]"
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"] = "2024-05-01-preview"
llm = init_chat_model(
"azure_openai:gpt-4o",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)
👉 阅读 [Azure 集成文档](https://python.langchain.com/docs/integrations/chat/azure_chat_openai/)
Google Gemini
pip install -U "langchain[google-genai]"
import os
from langchain.chat_models import init_chat_model
os.environ["GOOGLE_API_KEY"] = "..."
llm = init_chat_model("google_genai:gemini-1.5-flash")
👉 阅读 Google GenAI 集成文档
现在，我们将这个新工具整合到 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 InMemorySaver
from langgraph.graph import StateGraph, START
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"]
# 初始化现有的搜索工具
tavily_tool = TavilySearch(max_results=2)
# 将新工具和现有工具组合成一个列表
tools = [tavily_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")
提示：想了解更多关于人工干预工作流的信息和示例，请参阅人工干预 (Human-in-the-loop) 概念文档。
步骤二：编译 Graph
和之前一样，我们使用一个检查点工具 (checkpointer) 来编译图。检查点工具对于暂停和恢复至关重要，因为它负责保存图的当前状态。
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)
步骤三：可视化 Graph（可选）
可视化编译后的图，你会发现其布局与之前完全相同，只是现在多了一个可用的工具。
from IPython.display import Image, display
try:
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
# 这需要一些额外的依赖，是可选步骤
pass
步骤四：触发人工干预
现在，我们向聊天机器人提出一个问题，该问题会触发新添加的 human_assistance 工具：
user_input = "我需要一些构建 AI Agent 的专家建议。你能帮我请求协助吗？"
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 =================================
我需要一些构建 AI Agent 的专家建议。你能帮我请求协助吗？
================================== Ai Message ==================================
Tool Calls:
human_assistance (toolu_01AB...)
Call ID: toolu_01AB...
Args:
query: 一位用户正在请求关于构建 AI Agent 的专家指导。您能就此主题提供一些专家建议或资源吗？
聊天机器人正确地生成了一个工具调用，但随后执行就被中断了。此时，如果你检查图的状态，会发现它停在了 tools 节点：
snapshot = graph.get_state(config)
snapshot.next
('tools',)
深入理解： 让我们再看一下
human_assistance工具的实现：@tool def human_assistance(query: str) -> str: """当需要人类协助时调用此工具。""" human_response = interrupt({"query": query}) return human_response["data"]与 Python 的
input()函数类似，在工具内部调用interrupt会暂停执行流程。图的当前进展会通过我们配置的检查点工具 (checkpointer) 被持久化。如果使用 Postgres 作为后端，只要数据库在线，就可以随时恢复。在本例中，我们使用内存检查点，只要 Python 内核仍在运行，就可以随时恢复。
步骤五：提供人工输入并恢复执行
要恢复执行，我们需要传入一个 Command 对象，其中包含工具所期望的数据。这个数据的格式可以根据你的需求自定义。
在本例中，我们使用一个包含 data 键的字典作为人类的响应：
human_response = (
"我们专家随时为您提供帮助！我们建议您了解一下 LangGraph 来构建您的 Agent。"
"它比简单的自主 Agent 更可靠、更具可扩展性。"
)
# 创建一个 Command 对象，将人类的响应作为 resume 的一部分传入
human_command = Command(resume={"data": human_response})
# 将 Command 对象流式传输回图中，使用相同的 config
events = graph.stream(human_command, config, stream_mode="values")
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
执行恢复后，你将看到如下输出：
================================= Tool Message =================================
Name: human_assistance
我们专家随时为您提供帮助！我们建议您了解一下 LangGraph 来构建您的 Agent。它比简单的自主 Agent 更可靠、更具可扩展性。
================================== Ai Message ==================================
感谢您的耐心等待。我已收到关于您构建 AI Agent 指导请求的专家建议。以下是专家的建议：
专家建议您研究 LangGraph 来构建您的 AI Agent。他们提到，与简单的自主 Agent 相比，LangGraph 是一个更可靠、更具可扩展性的选择。
LangGraph 很可能是一个专门用于创建具有高级功能的 AI Agent 的框架或库。根据此建议，以下几点值得考虑：
1. **可靠性**：专家强调 LangGraph 比简单的自主 Agent 方法更可靠。这可能意味着它具有更好的稳定性、错误处理能力或一致的性能。
2. **可扩展性**：LangGraph 被描述为更具可扩展性，这表明它可能提供了一个灵活的架构，让您能够随着 Agent 需求的演变轻松添加新功能或修改现有功能。
...
可以看到，我们提供的人工输入被成功接收并作为工具消息处理。图从中断处继续执行，LLM 根据专家的建议生成了最终的回复。你可以查看此调用的 LangSmith 链路，观察状态是如何在第一步被加载，从而让聊天机器人从上次中断的地方继续执行的。
恭喜！ 你已经成功使用 interrupt 为你的聊天机器人添加了人工干预功能，实现了在需要时进行人工监督和干预。这为你使用 AI 系统创建更丰富的用户界面开辟了可能性。由于你已经配置了检查点 (checkpointer)，只要底层的持久化层在运行，图就可以被无限期地暂停，并在任何时候恢复，就像什么都没发生过一样。
总结与后续步骤
到目前为止，本系列教程中的示例都依赖于一个只包含消息列表的简单状态。虽然这种简单状态已经非常强大，但如果你想定义更复杂的行为而不完全依赖于消息列表，你可以向状态中添加额外的字段。
关于
关注我获取更多资讯
