跳到内容

人在回路中#

工具也可以被定义为引入人在回路中。这对于需要人工输入的任务很有用，例如确认工具调用或提供反馈。

正如我们在工作流教程中将看到的那样，AgentWorkflow 的底层工作方式是通过运行既发出又接收事件的步骤来实现。以下是构成 AgentWorkflow 的步骤（蓝色）以及在它们之间传递数据的事件（绿色）的图示。您会认出这些事件，它们与我们之前在输出流中处理的事件相同。

Workflows diagram

为了引入人在回路中，我们将让工具发出一个工作流中其他任何步骤都不接收的事件。然后，我们将告诉工具等待，直到接收到特定的“回复”事件。

我们有内置的 InputRequiredEvent 和 HumanResponseEvent 事件可用于此目的。如果您想捕获不同形式的人工输入，可以对这些事件进行子类化以满足您的偏好。我们来导入它们。

from llama_index.core.workflow import (
    InputRequiredEvent,
    HumanResponseEvent,
)

接下来，我们将创建一个执行假定危险任务的工具。这里有几点新内容：

  • wait_for_event 用于等待 HumanResponseEvent。
  • waiter_event 是写入事件流的事件，用于告知调用者我们正在等待响应。
  • waiter_id 是此特定等待调用的唯一标识符。它有助于确保我们对于每个 waiter_id 只发送一个 waiter_event。
  • requirements 参数用于指定我们要等待具有特定 user_name 的 HumanResponseEvent。
async def dangerous_task(ctx: Context) -> str:
    """A dangerous task that requires human confirmation."""

    # emit a waiter event (InputRequiredEvent here)
    # and wait until we see a HumanResponseEvent
    question = "Are you sure you want to proceed? "
    response = await ctx.wait_for_event(
        HumanResponseEvent,
        waiter_id=question,
        waiter_event=InputRequiredEvent(
            prefix=question,
            user_name="Laurie",
        ),
        requirements={"user_name": "Laurie"},
    )

    # act on the input from the event
    if response.response.strip().lower() == "yes":
        return "Dangerous task completed successfully."
    else:
        return "Dangerous task aborted."

我们像往常一样创建 Agent，并将刚刚定义的工具传递给它。

workflow = FunctionAgent(
    tools=[dangerous_task],
    llm=llm,
    system_prompt="You are a helpful assistant that can perform dangerous tasks.",
)

现在我们可以运行工作流，像处理任何其他流式事件一样处理 InputRequiredEvent，并使用 send_event 方法传入一个 HumanResponseEvent 进行响应。

handler = workflow.run(user_msg="I want to proceed with the dangerous task.")

async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        # capture keyboard input
        response = input(event.prefix)
        # send our response back
        handler.ctx.send_event(
            HumanResponseEvent(
                response=response,
                user_name=event.user_name,
            )
        )

response = await handler
print(str(response))

像往常一样，您可以查看此示例的完整代码。

您可以使用任何方式来捕获输入；可以使用 GUI、音频输入，甚至可以引入另一个独立的 Agent。如果您的输入需要一段时间或在另一个进程中发生，您可能希望序列化上下文并将其保存到数据库或文件中，以便之后可以恢复工作流。

谈到引入其他 Agent，这将我们带到下一节：多 Agent 系统。