在 LlamaIndex 中创建 Agentic 工作流

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

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

Workflow Schematic

工作流提供几个关键优势

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

正如您可能已经猜到的，工作流在 Agent 的自主性与保持对整个工作流的控制之间取得了很好的平衡。

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

创建工作流

您可以按照此 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() 来运行工作流。

连接多个步骤

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

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.set("query", "What is the capital of France?")

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

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

    return StopEvent(result=val)

太棒了！现在您知道如何在 LlamaIndex 中创建基本工作流了！

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

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

使用多 Agent 工作流自动化工作流

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

然后每个 Agent 可以

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

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

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

Agent 工具还可以修改我们之前提到的工作流状态。在开始工作流之前，我们可以提供一个初始状态字典，该字典将可供所有 Agent 使用。状态存储在工作流上下文的状态键中。它将被注入到 state_prompt 中，state_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.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.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.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.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.get("state")
print(state["num_fn_calls"])

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

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

< > 在 GitHub 上更新

Agents 课程