添加人机循环控制¶

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

LangGraph的持久化层支持**人机循环**工作流，允许根据用户反馈暂停和恢复执行。此功能的主要接口是interrupt函数。在节点内调用interrupt将暂停执行。可以通过传入Command来恢复执行，同时包含来自人类的新输入。

interrupt在人体工程学上类似于Python内置的input()函数，但有一些注意事项。

Note

本教程基于添加记忆构建。

1. 添加 `human_assistance` 工具¶

从为聊天机器人添加记忆教程的现有代码开始，为聊天机器人添加human_assistance工具。该工具使用interrupt接收来自人类的信息。

首先，让我们选择一个聊天模型：

OpenAIAnthropicAzureGoogle GeminiAWS Bedrock

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-4.1")

👉 Read the OpenAI integration docs

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")

👉 Read the Anthropic integration docs

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"] = "2025-03-01-preview"

llm = init_chat_model(
    "azure_openai:gpt-4.1",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)

👉 Read the Azure integration docs

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-2.0-flash")

👉 Read the Google GenAI integration docs

pip install -U "langchain[aws]"

from langchain.chat_models import init_chat_model

# Follow the steps here to configure your credentials:
# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

llm = init_chat_model(
    "anthropic.claude-3-5-sonnet-20240620-v1:0",
    model_provider="bedrock_converse",
)

👉 Read the AWS Bedrock integration docs

我们现在可以将其与额外的工具一起整合到我们的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, 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")

Tip

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

2. 编译图¶

我们像之前一样使用检查点编译图：

memory = InMemorySaver()

graph = graph_builder.compile(checkpointer=memory)

3. 可视化图形（可选）¶

可视化图形后，您将获得与之前相同的布局——只是添加了工具！

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # 这需要一些额外的依赖项，是可选的
    pass

chatbot-with-tools-diagram

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?

聊天机器人生成了一个工具调用，但执行被中断了。如果你检查图的状态，你会看到它在工具节点停止了：

snapshot = graph.get_state(config)
snapshot.next

('tools',)

Info

仔细看看 human_assistance 工具：

python @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 trace 以了解上述调用中完成的确切工作。请注意，状态在第一步被加载，以便我们的聊天机器人可以从中断处继续。

恭喜！ 您已经使用 interrupt 为您的聊天机器人添加了人机协同执行，从而在需要时允许人工监督和干预。这为您使用 AI 系统创建潜在的用户界面打开了可能性。由于您已经添加了 checkpointer，只要底层的持久化层正在运行，图就可以被 无限期 暂停，并在任何时间恢复，仿佛什么都没有发生过。

查看下面的代码片段以回顾本教程中的图：