使用langgraph创建工作流系列4：人机回环
到目前为止chatbot的工作流都是用户输入后等待应答，但有些情况下需要人工干预：
1）因为大模型本身不是可靠的，基于大模型的智能体也是不可靠的，需要人工输入以完成符合要求的任务
2）在工作流执行过程中，需要人工审核、修改等操作
在智能体语境下的人工干预也成为人机回环（human-in-the-loop）。langgraph的持久化层借助检查点支持人机回环的工作流。
在工作流执行过程中，在节点内部调用interrupt将暂停工作流的执行，用户完成处理后，可以从暂停点回复执行，并且可以获得用户输入的数据。
本文仍基于使用langgraph创建工作流系列3：增加记忆-CSDN博客中的chatbot构建人机回环的chatbot。
1.增加工具
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAIfrom langgraph.prebuilt import ToolNode, tools_condition#支持工具调用
from langgraph.types import Command, interrupt#引入Command和interrupte支持人机回环
class State(TypedDict):
messages: Annotated[list, add_messages]graph_builder = StateGraph(State)
llm = ChatOpenAI(
model = 'qwen-plus',
api_key = "sk-27cd7fd64e5a4c82a71f879efaebc40e",
base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1")#新增加支持人机回环的工具
@tool def human_assistance(query: str) -> str:
"""请求用户输入"""
human_response = interrupt({"query": query})
return human_response["data"]
tools = [human_assistance]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools=tools) #生成工具节点
graph_builder.add_node("tools", tools) #把工具节点增加到工作流图中
def chatbot(state: State):
return {"messages": [llm.invoke(state["messages"])]}graph_builder.add_node("chatbot", chatbot)
graph_builder.add_conditional_edges( "chatbot", tools_condition, )
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile(checkpointer=memory)
2.触发用户介入
输入一个问题，触发用户介入，也就是调用新增的human_assistant方法。
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_assiatant，调用参数为原始问题。
================================ Human Message =================================
I need some expert guidance for building an AI agent. Could you request assistance for me?
================================== Ai Message ==================================
Tool Calls:
human_assistance (call_44b1fdebb102451faf8151)
Call ID: call_44b1fdebb102451faf8151
Args:
query: I need some expert guidance for building an AI agent. Could you request assistance for me?
查看工作流下一个节点：
snapshot = graph.get_state(config)
snapshot.next
输出为('tools",)，可见工作流暂停在tools节点，也就是要调用human_assistant。等待用户介入：
（'tools',)
3.模拟用户输入，恢复工作流执行
用用户输入构造一个Command对象，然后用Command作为参数调用graph.stream
#这里的Command对象很简单，可以根据实际需要构造
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." )
#用用户输入包装一个Command对象，设置resume
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()
基于用户输出chatbot调用大模型后的输出如下：
================================== Ai Message ==================================
Tool Calls:
human_assistance (call_44b1fdebb102451faf8151)
Call ID: call_44b1fdebb102451faf8151
Args:
query: I need some expert guidance for building an AI agent. Could you request assistance for me?
================================= 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.