基本工作流#

入门指南#

工作流内置于 LlamaIndex 核心中，因此要使用它们，您只需

pip install llama-index-core

在开发过程中，您可能会发现将工作流可视化很有帮助；您可以通过安装内置可视化工具来实现这一点

pip install llama-index-utils-workflow

依赖项#

工作流的最低依赖项为

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

单步工作流#

工作流通常实现为一个继承自 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()
print(result)

这将在控制台简单地打印 "Hello, World!"。

在这段代码中，我们

定义了一个继承自 Workflow 的类 MyWorkflow
使用 @step 装饰器定义了一个单步 my_step
该步骤接受一个参数 ev，它是 StartEvent 的一个实例
该步骤返回一个带有结果 "Hello, world!" 的 StopEvent
我们创建了一个 MyWorkflow 实例，设置超时时间为 10 秒，并关闭详细输出
我们运行工作流并打印结果

步骤的类型注解#

类型注解（例如 ev: StartEvent）和 -> StopEvent 对工作流的运行方式至关重要。期望的类型决定了哪些事件类型会触发一个步骤。像可视化工具（见下文）这样的工具也依赖这些注解来确定生成了哪些类型，从而决定控制流的下一个去向。

类型注解在编译时会进行验证，因此如果您发出的事件没有被其他步骤消费，您将收到错误消息。

Start 和 Stop 事件#

StartEvent 和 StopEvent 是用于启动和停止工作流的特殊事件。任何接受 StartEvent 的步骤都会由 run 命令触发。发出 StopEvent 将结束工作流的执行并返回最终结果，即使其他步骤尚未执行。

在普通 Python 中运行工作流#

工作流默认是异步的，因此您可以使用 await 获取 run 命令的结果。这在 Notebook 环境中运行良好；在普通的 Python 脚本中，您需要导入 asyncio 并将代码包装在一个异步函数中，像这样

async def main():
    w = MyWorkflow(timeout=10, verbose=False)
    result = await w.run()
    print(result)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

在本教程的其余示例中，为简单起见，我们将假设处于异步环境中。

可视化工作流#

工作流的一个很棒的功能是内置的可视化工具，我们已经安装了它。让我们可视化刚刚创建的简单工作流

from llama_index.utils.workflow import draw_all_possible_flows

draw_all_possible_flows(MyWorkflow, filename="basic_workflow.html")

这将在当前目录创建一个名为 basic_workflow.html 的文件。在浏览器中打开它，您将看到工作流的交互式可视化表示。它看起来像这样

Basic workflow

当然，只有一步的工作流用处不大！让我们定义一个多步工作流。

自定义事件#

通过定义可以由步骤发出并触发其他步骤的自定义事件来创建多个步骤。让我们定义一个简单的三步工作流。

我们像以前一样导入所需的模块，此外还新增了 Event 的导入

from llama_index.core.workflow import (
    StartEvent,
    StopEvent,
    Workflow,
    step,
    Event,
)
from llama_index.utils.workflow import draw_all_possible_flows

现在我们定义两个自定义事件 FirstEvent 和 SecondEvent。这些类可以有任意名称和属性，但必须继承自 Event

class FirstEvent(Event):
    first_output: str


class SecondEvent(Event):
    second_output: str

定义工作流#

现在我们定义工作流本身。我们通过在每个步骤上定义输入和输出类型来做到这一点。

step_one 接受一个 StartEvent 并返回一个 FirstEvent
step_two 接受一个 FirstEvent 并返回一个 SecondEvent
step_three 接受一个 SecondEvent 并返回一个 StopEvent

class MyWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent) -> FirstEvent:
        print(ev.first_input)
        return FirstEvent(first_output="First step complete.")

    @step
    async def step_two(self, ev: FirstEvent) -> SecondEvent:
        print(ev.first_output)
        return SecondEvent(second_output="Second step complete.")

    @step
    async def step_three(self, ev: SecondEvent) -> StopEvent:
        print(ev.second_output)
        return StopEvent(result="Workflow complete.")


w = MyWorkflow(timeout=10, verbose=False)
result = await w.run(first_input="Start the workflow.")
print(result)

完整输出将是

Start the workflow.
First step complete.
Second step complete.
Workflow complete.

我们可以使用可视化工具查看此工作流的所有可能流程

from llama_index.utils.workflow import draw_all_possible_flows

draw_all_possible_flows(MyWorkflow, filename="multi_step_workflow.html")

A simple multi-step workflow

当然，如果只是从头到尾运行，工作流的意义也不大！让我们进行一些分支和循环操作。