添加人机协同控制¶
智能体可能不可靠,可能需要人工输入才能成功完成任务。类似地,对于某些操作,你可能希望在运行之前需要人工批准,以确保一切按预期运行。
LangGraph 的持久化层支持**人机协同**工作流,允许根据用户反馈暂停和恢复执行。此功能的主要接口是 interrupt 函数。在节点内调用 interrupt 将暂停执行。可以通过传入 Command 以及来自人类的新输入来恢复执行。
interrupt 在人体工程学上类似于 Python 的内置 input(),但有一些注意事项。
interrupt 在人体工程学上类似于 Node.js 的内置 readline.question() 函数,但有一些注意事项。
Note
本教程基于添加记忆。
1. 添加 human_assistance 工具¶
从为聊天机器人添加记忆教程的现有代码开始,将 human_assistance 工具添加到聊天机器人。此工具使用 interrupt 从人类接收信息。
让我们首先选择一个聊天模型:
我们现在可以将其与额外的工具合并到我们的 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")
import { interrupt, MessagesZodState } from "@langchain/langgraph";
import { ChatAnthropic } from "@langchain/anthropic";
import { TavilySearch } from "@langchain/tavily";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const humanAssistance = tool(
async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
},
{
name: "humanAssistance",
description: "向人类请求帮助。",
schema: z.object({
query: z.string().describe("给人类的可读问题"),
}),
}
);
const searchTool = new TavilySearch({ maxResults: 2 });
const tools = [searchTool, humanAssistance];
const llmWithTools = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
}).bindTools(tools);
async function chatbot(state: z.infer<typeof MessagesZodState>) {
const message = await llmWithTools.invoke(state.messages);
// 因为我们将在工具执行期间中断,
// 我们禁用并行工具调用以避免在恢复时重复任何
// 工具调用。
if (message.tool_calls && message.tool_calls.length > 1) {
throw new Error("Multiple tool calls not supported with interrupts");
}
return { messages: message };
}
Tip
有关人机协同工作流的更多信息和示例,请参阅人机协同。
2. 编译图¶
像以前一样,我们使用检查点器编译图:
import { StateGraph, MemorySaver, START, END } from "@langchain/langgraph";
const memory = new MemorySaver();
const graph = new StateGraph(MessagesZodState)
.addNode("chatbot", chatbot)
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
.compile({ checkpointer: memory });
3. 可视化图(可选)¶
可视化图,你会得到与之前相同的布局 - 只是添加了工具!
from IPython.display import Image, display
try:
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
# 这需要一些额外的依赖项并且是可选的
pass
import * as fs from "node:fs/promises";
const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());
await fs.writeFile("chatbot-with-tools.png", imageBuffer);
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?
import { isAIMessage } from "@langchain/core/messages";
const userInput =
"I need some expert guidance for building an AI agent. Could you request assistance for me?";
const events = await graph.stream(
{ messages: [{ role: "user", content: userInput }] },
{ configurable: { thread_id: "1" }, streamMode: "values" }
);
for await (const event of events) {
if ("messages" in event) {
const lastMessage = event.messages.at(-1);
console.log(`[${lastMessage?.getType()}]: ${lastMessage?.text}`);
if (
lastMessage &&
isAIMessage(lastMessage) &&
lastMessage.tool_calls?.length
) {
console.log("Tool calls:", lastMessage.tool_calls);
}
}
}
[human]: I need some expert guidance for building an AI agent. Could you request assistance for me?
[ai]: I'll help you request human assistance for guidance on building an AI agent.
Tool calls: [
{
name: 'humanAssistance',
args: {
query: 'I would like expert guidance on building an AI agent. Could you please provide assistance with this topic?'
},
id: 'toolu_01Bpxc8rFVMhSaRosS6b85Ts',
type: 'tool_call'
}
]
聊天机器人生成了一个工具调用,但随后执行已被中断。如果你检查图状态,你会看到它停在了 tools 节点:
Info
仔细看看 human_assistance 工具:
@tool
def human_assistance(query: str) -> str:
"""向人类请求帮助。"""
human_response = interrupt({"query": query})
return human_response["data"]
类似于 Python 的内置 input() 函数,在工具内调用 interrupt 将暂停执行。进度基于检查点器持久化;因此,如果使用 Postgres 持久化,只要数据库存活,就可以随时恢复。在此示例中,它使用内存检查点器持久化,只要 Python 内核正在运行,就可以随时恢复。
仔细看看 humanAssistance 工具:
const humanAssistance = tool(
async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
},
{
name: "humanAssistance",
description: "向人类请求帮助。",
schema: z.object({
query: z.string().describe("给人类的可读问题"),
}),
},
);
在工具内调用 interrupt 将暂停执行。进度基于检查点器持久化;因此,如果使用 Postgres 持久化,只要数据库存活,就可以随时恢复。在此示例中,它使用内存检查点器持久化,只要 JavaScript 运行时正在运行,就可以随时恢复。
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 ==================================
[{'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?
================================= Tool Message =================================
Name: human_assistance
我们专家在这里提供帮助！我们建议你查看 LangGraph 来构建你的智能体。它比简单的自主智能体更可靠和可扩展。
================================== Ai Message ==================================
Thank you for your patience. I've received some expert advice regarding your request for guidance on building an AI agent. Here's what the experts have suggested:
专家建议你研究 LangGraph 来构建你的 AI 智能体。他们提到 LangGraph 与简单的自主智能体相比是一个更可靠和可扩展的选择。
LangGraph is likely a framework or library designed specifically for creating AI agents with advanced capabilities. Here are a few points to consider based on this recommendation:
1. Reliability: The experts emphasize that LangGraph is more reliable than simpler autonomous agent approaches. This could mean it has better stability, error handling, or consistent performance.
2. Extensibility: LangGraph is described as more extensible, which suggests that it probably offers a flexible architecture that allows you to easily add new features or modify existing ones as your agent's requirements evolve.
3. Advanced capabilities: Given that it's recommended over "simple autonomous agents," LangGraph likely provides more sophisticated tools and techniques for building complex AI agents.
...
2. Look for tutorials or guides specifically focused on building AI agents with LangGraph.
3. Check if there are any community forums or discussion groups where you can ask questions and get support from other developers using LangGraph.
如果你想了解有关 LangGraph 的更具体信息或对此建议有任何疑问，请随时询问，我可以向专家请求进一步的帮助。
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
在此示例中,使用带有键 "data" 的对象:
import { Command } from "@langchain/langgraph";
const humanResponse =
"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.";
const humanCommand = new Command({ resume: { data: humanResponse } });
const resumeEvents = await graph.stream(humanCommand, {
configurable: { thread_id: "1" },
streamMode: "values",
});
for await (const event of resumeEvents) {
if ("messages" in event) {
const lastMessage = event.messages.at(-1);
console.log(`[${lastMessage?.getType()}]: ${lastMessage?.text}`);
}
}
[tool]: 我们专家在这里提供帮助！我们建议你查看 LangGraph 来构建你的智能体。它比简单的自主智能体更可靠和可扩展。
[ai]: Thank you for your patience. I've received some expert advice regarding your request for guidance on building an AI agent. Here's what the experts have suggested:
专家建议你研究 LangGraph 来构建你的 AI 智能体。他们提到 LangGraph 与简单的自主智能体相比是一个更可靠和可扩展的选择。
LangGraph is likely a framework or library designed specifically for creating AI agents with advanced capabilities. Here are a few points to consider based on this recommendation:
1. Reliability: The experts emphasize that LangGraph is more reliable than simpler autonomous agent approaches. This could mean it has better stability, error handling, or consistent performance.
2. Extensibility: LangGraph is described as more extensible, which suggests that it probably offers a flexible architecture that allows you to easily add new features or modify existing ones as your agent's requirements evolve.
3. Advanced capabilities: Given that it's recommended over "simple autonomous agents," LangGraph likely provides more sophisticated tools and techniques for building complex AI agents.
...
输入已被接收并作为工具消息处理。查看此调用的 LangSmith trace 以查看上述调用中完成的确切工作。注意,状态在第一步加载,以便我们的聊天机器人可以从中断的地方继续。
恭喜! 你已经使用 interrupt 向聊天机器人添加了人机协同执行,允许在需要时进行人工监督和干预。这为你可以用 AI 系统创建的潜在 UI 开辟了可能性。由于你已经添加了**检查点器**,只要底层持久化层正在运行,图就可以**无限期**暂停,并随时恢复,就好像什么都没发生过一样。
查看下面的代码片段以查看本教程中的图:
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:
"""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"])
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")
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)
import {
interrupt,
MessagesZodState,
StateGraph,
MemorySaver,
START,
END,
} from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { isAIMessage } from "@langchain/core/messages";
import { ChatAnthropic } from "@langchain/anthropic";
import { TavilySearch } from "@langchain/tavily";
import {
interrupt,
MessagesZodState,
StateGraph,
MemorySaver,
START,
END,
} from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { isAIMessage } from "@langchain/core/messages";
import { ChatAnthropic } from "@langchain/anthropic";
import { TavilySearch } from "@langchain/tavily";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const humanAssistance = tool(
async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
},
{
name: "humanAssistance",
description: "Request assistance from a human.",
schema: z.object({
query: z.string().describe("Human readable question for the human"),
}),
}
);
const humanAssistance = tool(
async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
},
{
name: "humanAssistance",
description: "Request assistance from a human.",
schema: z.object({
query: z.string().describe("Human readable question for the human"),
}),
}
);
const searchTool = new TavilySearch({ maxResults: 2 });
const searchTool = new TavilySearch({ maxResults: 2 });
const tools = [searchTool, humanAssistance];
const llmWithTools = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
}).bindTools(tools);
const llmWithTools = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
}).bindTools(tools);
const chatbot = async (state: z.infer<typeof MessagesZodState>) => {
const chatbot = async (state: z.infer<typeof MessagesZodState>) => {
const message = await llmWithTools.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.
// Because we will be interrupting during tool execution,
// we disable parallel tool calling to avoid repeating any
// tool invocations when we resume.
if (message.tool_calls && message.tool_calls.length > 1) {
throw new Error("Multiple tool calls not supported with interrupts");
}
return { messages: message };
return { messages: message };
};
const memory = new MemorySaver();
const graph = new StateGraph(MessagesZodState)
.addNode("chatbot", chatbot)
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
.compile({ checkpointer: memory });
const graph = new StateGraph(MessagesZodState)
.addNode("chatbot", chatbot)
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
.compile({ checkpointer: memory });
下一步¶
到目前为止,教程示例依赖于一个简单的状态,只有一个条目:消息列表。你可以使用这个简单的状态走得很远,但如果你想定义复杂的行为而不依赖于消息列表,你可以向状态添加额外的字段。