人机协作（Human-in-the-loop）¶

要审查、编辑并批准代理中的工具调用，可以使用 LangGraph 内置的人机协作（HIL）功能，特别是 [interrupt()][langgraph.types.interrupt] 原语。

LangGraph 允许你**无限期地暂停**执行——可以是几分钟、几小时，甚至是几天——直到接收到人工输入。

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

如需深入了解 人机协作 的概念，请参阅概念指南。

在继续之前，人类可以审查和编辑代理的输出。这对于工具调用可能敏感或需要人工监督的应用尤为重要。

审查工具调用¶

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

在工具中使用 interrupt() 暂停执行。
使用 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)!
)

[interrupt 函数][langgraph.types.interrupt] 会在特定节点暂停代理图。在此示例中，我们在工具函数开始处调用 interrupt()，这会暂停在执行该工具的节点上。interrupt() 中的信息（例如工具调用）可以展示给人类，然后可以通过用户输入（工具调用批准、编辑或反馈）恢复图的执行。
InMemorySaver 用于存储代理状态在每次工具调用循环中的每一步。这启用了短期记忆和人机协作功能。在此示例中，我们使用 InMemorySaver 将代理状态存储在内存中。在生产应用程序中，代理状态将存储在数据库中。
使用 checkpointer 初始化代理。

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

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

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
    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")

[interrupt 函数][langgraph.types.interrupt] 与 [Command][langgraph.types.Command] 对象结合使用，以根据人类提供的值恢复图的执行。

与 Agent Inbox 配合使用¶

你可以创建一个包装器，为任何工具添加中断功能。

下面的示例提供了一个参考实现，兼容 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

此包装器创建了一个新的工具，在执行被包装的工具之前会调用 interrupt()。
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": "book a stay at McKittrick hotel"}]},
    config
):
    print(chunk)
    print("\n")

使用 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 中的人机协作