人机交互

为了审查、编辑并批准代理中的工具调用，您可以使用LangGraph内置的人机交互(HIL)功能，特别是interrupt()原语。

LangGraph允许您无限期地暂停执行——从几分钟到几小时甚至几天——直到收到人类输入为止。

这是因为代理状态被**检查点化并存储在数据库中**，这使得系统能够持久化执行上下文，并稍后从中断的地方恢复工作流。

有关**人机交互**概念的更深入探讨，请参阅概念指南。

一个人可以在继续之前审查并编辑代理的输出。这在应用程序中尤为重要，其中请求的工具调用可能是敏感的或需要人工监督。

审查工具调用

要向工具添加一个人工审批步骤：

1. 在工具中使用 interrupt() 暂停执行。
2. 使用 Command(resume=...) 继续执行，基于人类输入。

API Reference: InMemorySaver | interrupt | create_react_agent

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt
from langgraph.prebuilt import create_react_agent

# 一个需要人工审查/批准的敏感工具示例
def book_hotel(hotel_name: str):
    """预订酒店"""
    response = interrupt(  # (1)!
        f"尝试调用 `book_hotel` 函数，参数为 {{'hotel_name': {hotel_name}}}. "
        "请确认或提出修改意见。"
    )
    if response["type"] == "accept":
        pass
    elif response["type"] == "edit":
        hotel_name = response["args"]["hotel_name"]
    else:
        raise ValueError(f"未知响应类型：{response['type']}")
    return f"成功在 {hotel_name} 预订了住宿。"

checkpointer = InMemorySaver() # (2)!

agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_hotel],
    checkpointer=checkpointer, # (3)!
)

1. 中断函数会在特定节点暂停代理图。在这种情况下，我们在工具函数的开头调用 interrupt()，这会暂停代理图中的该节点，该节点正在执行工具。interrupt() 中的信息（例如工具调用）可以呈现给一个人类用户，并且可以通过用户输入（工具调用批准、编辑或反馈）恢复图。
2. InMemorySaver 用于在工具调用循环的每一步存储代理状态。这使短期记忆和人机交互功能得以实现。在此示例中，我们使用 InMemorySaver 将代理状态存储在内存中。在生产应用中，代理状态将存储在数据库中。
3. 使用 checkpointer 初始化代理。

使用 stream() 方法运行代理，并传递 config 对象以指定线程ID。这允许代理在未来的调用中恢复相同的对话。

config = {
   "configurable": {
      "thread_id": "1"
   }
}

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "预订麦克基特里克酒店的住宿"}]},
    config
):
    print(chunk)
    print("\n")

您应该看到代理运行到 interrupt() 调用处时暂停并等待人类输入。

使用 Command(resume=...) 继续执行，基于人类输入。

API Reference: Command

from langgraph.types import Command

for chunk in agent.stream(
    Command(resume={"type": "accept"}),  # (1)!
    # Command(resume={"type": "edit", "args": {"hotel_name": "McKittrick Hotel"}}),
    config
):
    print(chunk)
    print("\n")

1. 中断函数与命令对象一起使用，以便使用人类提供的值恢复图。

使用代理收件箱

你可以创建一个包装器来向任何工具添加中断。

以下示例提供了一个与Agent Inbox UI和Agent Chat UI兼容的参考实现。

添加人机交互到任何工具的包装器

from typing import Callable
from langchain_core.tools import BaseTool, tool as create_tool
from langchain_core.runnables import RunnableConfig
from langgraph.types import interrupt 
from langgraph.prebuilt.interrupt import HumanInterruptConfig, HumanInterrupt

def add_human_in_the_loop(
    tool: Callable | BaseTool,
    *,
    interrupt_config: HumanInterruptConfig = None,
) -> BaseTool:
   """为工具添加人机交互支持的包装器。"""
    if not isinstance(tool, BaseTool):
        tool = create_tool(tool)

    if interrupt_config is None:
        interrupt_config = {
            "allow_accept": True,
            "allow_edit": True,
            "allow_respond": True,
        }

    @create_tool(  # (1)!
        tool.name,
        description=tool.description,
        args_schema=tool.args_schema
    )
    def call_tool_with_interrupt(config: RunnableConfig, **tool_input):
        request: HumanInterrupt = {
            "action_request": {
                "action": tool.name,
                "args": tool_input
            },
            "config": interrupt_config,
            "description": "请审核工具调用"
        }
        response = interrupt([request])[0]  # (2)!
        # 批准工具调用
        if response["type"] == "accept":
            tool_response = tool.invoke(tool_input, config)
        # 更新工具调用参数
        elif response["type"] == "edit":
            tool_input = response["args"]["args"]
            tool_response = tool.invoke(tool_input, config)
        # 以用户反馈回应LLM
        elif response["type"] == "response":
            user_feedback = response["args"]
            tool_response = user_feedback
        else:
            raise ValueError(f"不支持的中断响应类型：{response['type']}")

        return tool_response

    return call_tool_with_interrupt

1. 此包装器创建一个新的工具，在执行被包装的工具之前调用interrupt()。
2. interrupt()使用由Agent Inbox UI期望的特殊输入和输出格式：
   • 将包含HumanInterrupt对象的列表发送给AgentInbox，以便渲染中断信息供最终用户查看
   • AgentInbox提供恢复值作为列表（即，Command(resume=[...])）

你可以使用add_human_in_the_loop包装器在任何工具中添加interrupt()，而无需在工具内部添加：

API Reference: InMemorySaver | create_react_agent

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

checkpointer = InMemorySaver()

def book_hotel(hotel_name: str):
   """预订酒店"""
   return f"成功预订了{hotel_name}酒店的住宿。"

agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[
        add_human_in_the_loop(book_hotel), # (1)!
    ],
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "1"}}

# 运行代理
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "预订McKittrick酒店的住宿"}]},
    config
):
    print(chunk)
    print("\n")

1. 使用add_human_in_the_loop包装器在工具中添加interrupt()。这允许代理暂停执行并等待人类输入再继续执行工具调用。

你应该看到代理运行直到遇到interrupt()调用，此时它会暂停并等待人类输入。

使用Command(resume=...)恢复代理，根据人类输入继续执行。

API Reference: Command

from langgraph.types import Command 

for chunk in agent.stream(
    Command(resume=[{"type": "accept"}]),
    # Command(resume=[{"type": "edit", "args": {"args": {"hotel_name": "McKittrick Hotel"}}}]),
    config
):
    print(chunk)
    print("\n")

额外资源

• LangGraph 中的人工干预

Comments