本文介绍了如何在LangGraph中构建一个可以与人类交互、实现人工协助的聊天机器人。通过使用interrupt()函数，可以在工具调用过程中暂停执行，等待人工输入，从而实现对AI代理的监督和干预。文章详细解释了如何定义人工协助工具、如何处理用户输入、以及如何恢复执行。最终实现了一个可以根据人类建议给出回复的聊天机器人，并强调了LangGraph的持久化功能，使其能够无限期地暂停和恢复。

💬 LangGraph的`interrupt()`函数允许在工具调用中暂停执行，等待人工干预，从而实现人在回路的工作流。

🛠️ 通过定义`human_assistance`工具，聊天机器人可以请求人工协助，接收来自人类的反馈，并根据反馈继续执行。

💡 使用Command对象恢复执行，传入人工提供的输入数据，使聊天机器人能够根据人类的建议做出回应。

🔄 LangGraph的持久化层支持无限期地暂停和恢复执行，确保了聊天机器人的稳定性和可靠性。

🤖 示例展示了如何构建一个能够响应人类专家建议的聊天机器人，并强调了其在构建AI系统中的潜力。

前言

Agent 可能不可靠，并且可能需要人工输入才能成功完成任务。因此，您可能希望在运行之前需要人工批准，以确保一切按预期运行。例如以下操作：

批准工具调用并继续手动修改工具调用然后继续给出自然语言反馈，然后将其传回给代理

我们可以使用 interrupt() 函数在 LangGraph 中实现这些功能。

LangGraph 的 持久化 层支持人在回路工作流，允许根据用户反馈暂停和恢复执行。此功能的主要接口是 interrupt 函数。

在节点内部调用 interrupt 将暂停执行。通过传入 Command，可以恢复执行，并接收来自人工的新输入。

官方资料

API 参考：interrupt

操作指南：人类在环

官方文档：提示聊天机器人

实践人工协助聊天机器人

创建可以人工协助的聊天机器人

在本节，我们在第三部分的现有代码基础上添加一个humna_assistance工具，此工具使用 interrupt 从人工接收信息。

import osfrom typing import Annotated# from langchain.chat_models import init_chat_modelfrom langchain_openai import ChatOpenAIfrom langchain_tavily import TavilySearchfrom langchain_core.tools import toolfrom typing_extensions import TypedDictfrom langgraph.checkpoint.memory import MemorySaver# 内存检查点，用于保存会话状态from langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langgraph.prebuilt import ToolNode, tools_condition# 预构建的工具节点和条件路由函数from langgraph.types import Command, interrupt# 命令类型和中断函数，用于与外部交互#环境变量设置os.environ['TAVILY_API_KEY'] = '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'os.environ['ARK_API_KEY'] = '8748a662-692f-4e88-a12e-53a6442b7518'class State(TypedDict):    messages: Annotated[list, add_messages]graph_builder = StateGraph(State)# 使用@tool装饰器定义自定义工具：请求人工协助@tooldef 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]# 定义工具列表，包含Tavily搜索和人工协助工具llm = ChatOpenAI(    base_url="https://ark.cn-beijing.volces.com/api/v3",    api_key=os.environ.get('ARK_API_KEY'),      model="doubao-1-5-pro-32k-250115"  # 根据实际模型名称修改)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# 限制每次最多生成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)

我们开始对话尝试让聊天机器人去请求人工协助：

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 ==================================用户需要关于构建AI代理的专家指导，调用human_assistance函数请求人工协助。Tool Calls:  human_assistance (call_2y67vyca6qdhn3egwq4xh5v6) Call ID: call_2y67vyca6qdhn3egwq4xh5v6  Args:    query: 需要关于构建AI代理的专家指导

聊天机器人生成了一个工具调用，但执行被中断了！请注意，如果我们检查图状态，会发现它停在了工具节点。

snapshot = graph.get_state(config)snapshot.next

返回的待处理节点：

('tools',)

要恢复执行，请传递一个包含工具预期数据的 Command 对象。此数据的格式可以根据需要进行自定义。在本示例中，使用一个带有键 "data" 的字典：

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

返回结果：

================================== Ai Message ==================================用户需要关于构建AI代理的专家指导，调用human_assistance函数请求人工协助。Tool Calls:  human_assistance (call_2y67vyca6qdhn3egwq4xh5v6) Call ID: call_2y67vyca6qdhn3egwq4xh5v6  Args:    query: 需要关于构建AI代理的专家指导================================= Tool Message =================================Name: human_assistanceWe, 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 ==================================专家推荐使用LangGraph来构建AI代理，它比简单的自主代理更可靠和可扩展。如果你想了解更多关于LangGraph的信息，可以继续向我提问。

可以看到我们冒充专家人工协助建议使用LangGraph来构建AI代理，AI也根据了我们的建议给出了回答。

恭喜!您已使用 interrupt 为您的聊天机器人添加了人在回路执行功能，从而在需要时进行人工监督和干预。这为您可以使用 AI 系统创建的潜在 UI 打开了大门。由于我们已经添加了检查点，只要底层的持久化层正在运行，图就可以无限期地暂停并在任何时候恢复，就像什么都没发生一样。

原文地址：