人机协作

要在 Agent 中审查、编辑和批准工具调用，您可以使用 LangGraph 内置的人机协作 (HIL) 功能，特别是 interrupt() 原语。

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

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

要更深入地了解人机协作概念，请参阅概念指南。

人类可以在继续之前审查和编辑 Agent 的输出。这在请求的工具调用可能很敏感或需要人工监督的应用程序中尤为重要。

审查工具调用

为工具添加人工批准步骤

1. 在工具中使用 interrupt() 暂停执行。
2. 使用 Command(resume=...) 恢复，以便根据人工输入继续。

API 参考: InMemorySaver | interrupt | create_react_agent

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

# An example of a sensitive tool that requires human review / approval
def book_hotel(hotel_name: str):
    """Book a hotel"""
    response = interrupt(  # (1)!
        f"Trying to call `book_hotel` with args {{'hotel_name': {hotel_name}}}. "
        "Please approve or suggest edits."
    )
    if response["type"] == "accept":
        pass
    elif response["type"] == "edit":
        hotel_name = response["args"]["hotel_name"]
    else:
        raise ValueError(f"Unknown response type: {response['type']}")
    return f"Successfully booked a stay at {hotel_name}."

checkpointer = InMemorySaver() # (2)!

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

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

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

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

您应该看到 Agent 运行直到到达 interrupt() 调用，此时它会暂停并等待人工输入。

使用 Command(resume=...) 恢复 Agent，以便根据人工输入继续。

API 参考: 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. 的 interrupt 函数 与 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:
    """Wrap a tool to support human-in-the-loop review.""" 
    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": "Please review the tool call"
        }
        response = interrupt([request])[0]  # (2)!
        # approve the tool call
        if response["type"] == "accept":
            tool_response = tool.invoke(tool_input, config)
        # update tool call args
        elif response["type"] == "edit":
            tool_input = response["args"]["args"]
            tool_response = tool.invoke(tool_input, config)
        # respond to the LLM with user feedback
        elif response["type"] == "response":
            user_feedback = response["args"]
            tool_response = user_feedback
        else:
            raise ValueError(f"Unsupported interrupt response type: {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 参考: 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):
   """Book a hotel"""
   return f"Successfully booked a stay at {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"}}

# Run the agent
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
    config
):
    print(chunk)
    print("\n")

1. add_human_in_the_loop 包装器用于为工具添加 interrupt()。这允许 Agent 在继续进行工具调用之前暂停执行并等待人工输入。

您应该看到 Agent 运行直到到达 interrupt() 调用，此时它会暂停并等待人工输入。

使用 Command(resume=...) 恢复 Agent，以便根据人工输入继续。

API 参考: 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 中的人机协作

评论