1. 引言

??智能體可能不可靠，甚至需要人工輸入才能完成任務。同樣，對于某些操作，你可能需要在運行前獲得人工批準，以保證一切按預期運行。
??LangGraph的持久層支持人機交互工作流，允許根據用戶反饋暫停和恢復執行。此功能的主要接口是interrupt函數。在節點內部調用Interrupt將暫停執行。可以通過傳入command來interrupt執行，并接收新的人工輸入。interrupt在人機工程學上類似于Python的內置input()，但也有一些注意事項。

2. 添加Human Assistance工具

??初始化聊天模型：

from langchain.chat_models import init_chat_modelllm = init_chat_model("deepseek:deepseek-chat"
)

??使用附加工具將human assistance附加到狀態圖中：

from typing import Annotatedfrom langchain_tavily import TavilySearch
from langchain_core.tools import tool
from typing_extensions import TypedDictfrom 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_conditionfrom langgraph.types import Command, interruptclass 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"])# 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) <= 1return {"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")

3. 編譯狀態圖

??使用檢查點編譯狀態圖：

memory = MemorySaver()graph = graph_builder.compile(checkpointer=memory)

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

??運行結果為：

??聊天機器人生成了一個工具調用，但隨后執行被中斷。如果你檢查狀態圖，會發現它在工具節點處停止了：

snapshot = graph.get_state(config)
snapshot.next

??運行結果為：

('tools',)

5. 恢復執行

??要恢復執行需要傳遞一個包含工具所需數據的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 ==================================
Tool Calls:human_assistance (call_0_cee258cf-15db-49d4-8495-46761c7ddc65)Call ID: call_0_cee258cf-15db-49d4-8495-46761c7ddc65Args:query: I need expert guidance for building an AI agent.
================================= 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 ==================================Great! It seems the experts recommend using **LangGraph** for building your AI agent, as it is more reliable and extensible compared to simple autonomous agents. If you'd like, I can provide more details about LangGraph or assist you with specific steps to get started. Let me know how you'd like to proceed!

參考

https://langchain-ai.github.io/langgraph/tutorials/get-started/4-human-in-the-loop/