在 LlamaIndex 中创建代理工作流

LlamaIndex 中的工作流提供了一种结构化的方式，可以将您的代码组织成顺序的、可管理的步骤。

这样的工作流是通过定义由 `Events` 触发的 `Steps` 来创建的，并且它们本身会发出 `Events` 来触发进一步的步骤。让我们看看 Alfred 展示一个用于 RAG 任务的 LlamaIndex 工作流。

Workflow Schematic

工作流提供以下几个主要优点

将代码清晰地组织成离散的步骤
用于灵活控制流的事件驱动架构
步骤之间类型安全的通信
内置状态管理
支持简单和复杂的代理交互

正如你可能已经猜到的那样，**工作流在代理的自主性与保持对整体工作流的控制之间取得了很好的平衡。**

那么，让我们学习如何自己创建一个工作流吧！

创建工作流

您可以按照此 Notebook 中的代码进行操作，您可以使用 Google Colab 运行它。

基本工作流创建

安装 Workflow 包

如LlamaHub 部分所述，我们可以使用以下命令安装 Workflow 包

pip install llama-index-utils-workflow

我们可以通过定义一个继承自 `Workflow` 的类，并用 `@step` 装饰你的函数来创建单步工作流。我们还需要添加 `StartEvent` 和 `StopEvent`，它们是用于指示工作流开始和结束的特殊事件。

from llama_index.core.workflow import StartEvent, StopEvent, Workflow, step

class MyWorkflow(Workflow):
    @step
    async def my_step(self, ev: StartEvent) -> StopEvent:
        # do something here
        return StopEvent(result="Hello, world!")


w = MyWorkflow(timeout=10, verbose=False)
result = await w.run()

如您所见，我们现在可以通过调用 `w.run()` 来运行工作流。

连接多个步骤

要连接多个步骤，我们**创建自定义事件以在步骤之间传输数据。**为此，我们需要添加一个在步骤之间传递的 `Event`，并将第一步的输出传输到第二步。

from llama_index.core.workflow import Event

class ProcessingEvent(Event):
    intermediate_result: str

class MultiStepWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent) -> ProcessingEvent:
        # Process initial data
        return ProcessingEvent(intermediate_result="Step 1 complete")

    @step
    async def step_two(self, ev: ProcessingEvent) -> StopEvent:
        # Use the intermediate result
        final_result = f"Finished processing: {ev.intermediate_result}"
        return StopEvent(result=final_result)

w = MultiStepWorkflow(timeout=10, verbose=False)
result = await w.run()
result

这里的类型提示很重要，因为它确保工作流正确执行。让我们把事情弄得更复杂一点！

循环和分支

类型提示是工作流中最强大的部分，因为它允许我们创建分支、循环和连接以促进更复杂的工作流。

让我们通过使用联合运算符 `|` 来展示一个**创建循环**的示例。在下面的示例中，我们看到 `LoopEvent` 被用作步骤的输入，也可以作为输出返回。

from llama_index.core.workflow import Event
import random


class ProcessingEvent(Event):
    intermediate_result: str


class LoopEvent(Event):
    loop_output: str


class MultiStepWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent | LoopEvent) -> ProcessingEvent | LoopEvent:
        if random.randint(0, 1) == 0:
            print("Bad thing happened")
            return LoopEvent(loop_output="Back to step one.")
        else:
            print("Good thing happened")
            return ProcessingEvent(intermediate_result="First step complete.")

    @step
    async def step_two(self, ev: ProcessingEvent) -> StopEvent:
        # Use the intermediate result
        final_result = f"Finished processing: {ev.intermediate_result}"
        return StopEvent(result=final_result)


w = MultiStepWorkflow(verbose=False)
result = await w.run()
result

绘制工作流

我们也可以绘制工作流。让我们使用 `draw_all_possible_flows` 函数来绘制工作流。这会将工作流存储在一个 HTML 文件中。

from llama_index.utils.workflow import draw_all_possible_flows

w = ... # as defined in the previous section
draw_all_possible_flows(w, "flow.html")

workflow drawing

本课程中我们将介绍最后一个很酷的技巧，即向工作流添加状态的能力。

状态管理

当您希望跟踪工作流状态，以便每个步骤都可以访问相同状态时，状态管理非常有用。我们可以通过在步骤函数的参数顶部使用 `Context` 类型提示来实现这一点。

from llama_index.core.workflow import Context, StartEvent, StopEvent


@step
async def query(self, ctx: Context, ev: StartEvent) -> StopEvent:
    # store query in the context
    await ctx.store.set("query", "What is the capital of France?")

    # do something with context and event
    val = ...

    # retrieve query from the context
    query = await ctx.store.get("query")

    return StopEvent(result=val)

太棒了！现在您已经掌握了 LlamaIndex 中代理的基础知识！

工作流还有一些更复杂的细微差别，您可以在LlamaIndex 文档中了解它们。

然而，还有另一种创建工作流的方法，它依赖于 `AgentWorkflow` 类。让我们看看如何使用它来创建多代理工作流。

使用多代理工作流实现工作流自动化

除了手动创建工作流之外，我们还可以使用 **`AgentWorkflow` 类来创建多代理工作流**。`AgentWorkflow` 使用工作流代理，允许您创建一个由一个或多个代理组成的系统，这些代理可以根据其专业能力相互协作和移交任务。这使得能够构建复杂的代理系统，其中不同的代理处理任务的不同方面。我们不是从 `llama_index.core.agent` 导入类，而是从 `llama_index.core.agent.workflow` 导入代理类。在 `AgentWorkflow` 构造函数中，必须将一个代理指定为根代理。当用户消息进来时，它首先被路由到根代理。

然后每个代理可以

直接使用其工具处理请求
将任务移交给更适合该任务的另一个代理
向用户返回响应

让我们看看如何创建一个多代理工作流。

from llama_index.core.agent.workflow import AgentWorkflow, ReActAgent
from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI

# Define some tools
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

llm = HuggingFaceInferenceAPI(model_name="Qwen/Qwen2.5-Coder-32B-Instruct")

# we can pass functions directly without FunctionTool -- the fn/docstring are parsed for the name/description
multiply_agent = ReActAgent(
    name="multiply_agent",
    description="Is able to multiply two integers",
    system_prompt="A helpful assistant that can use a tool to multiply numbers.",
    tools=[multiply],
    llm=llm,
)

addition_agent = ReActAgent(
    name="add_agent",
    description="Is able to add two integers",
    system_prompt="A helpful assistant that can use a tool to add numbers.",
    tools=[add],
    llm=llm,
)

# Create the workflow
workflow = AgentWorkflow(
    agents=[multiply_agent, addition_agent],
    root_agent="multiply_agent",
)

# Run the system
response = await workflow.run(user_msg="Can you add 5 and 3?")

代理工具还可以修改我们之前提到的工作流状态。在启动工作流之前，我们可以提供一个初始状态字典，所有代理都可以访问该字典。状态存储在工作流上下文的状态键中。它将被注入到 state_prompt 中，该 prompt 会增强每个新的用户消息。

让我们通过修改之前的示例来注入一个计数器以计数函数调用

from llama_index.core.workflow import Context

# Define some tools
async def add(ctx: Context, a: int, b: int) -> int:
    """Add two numbers."""
    # update our count
    cur_state = await ctx.store.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.store.set("state", cur_state)

    return a + b

async def multiply(ctx: Context, a: int, b: int) -> int:
    """Multiply two numbers."""
    # update our count
    cur_state = await ctx.store.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.store.set("state", cur_state)

    return a * b

...

workflow = AgentWorkflow(
    agents=[multiply_agent, addition_agent],
    root_agent="multiply_agent",
    initial_state={"num_fn_calls": 0},
    state_prompt="Current state: {state}. User message: {msg}",
)

# run the workflow with context
ctx = Context(workflow)
response = await workflow.run(user_msg="Can you add 5 and 3?", ctx=ctx)

# pull out and inspect the state
state = await ctx.store.get("state")
print(state["num_fn_calls"])

恭喜！您现在已经掌握了 LlamaIndex 中代理的基础知识！🎉

让我们继续进行最后一个测验，以巩固您的知识！🚀

< > 在 GitHub 上更新

Agents 课程