我们之前搭建了许多不同类型的智能体，其中最受欢迎的就是在智能体中搭建各种工作流——这也是我最喜欢探索和玩的领域。那么，究竟什么是工作流？如何在后端实现一个工作流呢？今天我们就先简单了解下。

什么是工作流？

工作流是一种事件驱动、基于步骤的方法，用于控制和管理应用程序的执行流程。在这种方法中，应用程序被拆分为多个互相独立但又相互关联的部分，每个部分都被称为“步骤”。这些步骤通常由特定的事件触发，而每个步骤在执行时也可能会触发其他事件，从而激活更多后续的步骤。

通过精心设计和组合这些步骤与事件，你可以创建出复杂的流程，进而封装业务逻辑。

简单入门

首先，在开始之前，你需要确保你的电脑上已经安装了Python环境。我这里使用的是Python 3.10版本，但你可以根据自己的需要选择任何其他版本，前提是它符合项目的需求。接下来，我们将从创建一个新的Python项目开始，带你一步步入门。为了确保项目的依赖不与你电脑中其他Python项目的环境冲突，我们将使用虚拟环境。

今天的教程将聚焦于Python项目的基础入门部分，我们会一步步进行讲解。在后续的章节中，我们会深入探讨更复杂的技术和方法，话不多说，我们正式开始吧！

项目创建

虚拟环境搭建

• 在你的项目目录中打开终端（命令提示符），然后使用以下命令创建虚拟环境： python -m venv venv
• 在 Windows 上，使用以下命令激活虚拟环境： .\venv\Scripts\activate
• 保存依赖库列表: pip freeze > requirements.txt
• 退出虚拟环境: deactivate

VSCode配置

我使用的IDE是VSCode，当然你也可以根据个人喜好选择其他开发工具。虽然我们已经启动了Python虚拟环境，但需要注意的是，默认情况下，当你点击运行时，代码依旧是通过原始的Python路径来执行，而非虚拟环境中的Python解释器。因此，我们需要进行一些简单的配置来确保代码在正确的环境下运行。

• 按下 Ctrl + Shift + P (Windows/Linux) 或 Cmd + Shift + P (Mac) 打开命令面板。
• 输入并选择 Python: Select Interpreter。
• 在弹出的列表中，选择你虚拟环境中的 Python 解释器。

在弹出的窗口中，选择我们的项目路径中的venv目录即可。然后每次再点击运行就OK了。

Hello World

首先，为了确保后续操作顺利进行，需要提前安装以下两个依赖包。请使用以下命令进行安装：

pip install llama-index-core

pip install llama-index-utils-workflow

然后，直接粘贴以下代码，我们就可以顺利运行这个简单的程序。这个工作流的主要功能是打印“Hello, World!”消息，是一个非常基础的工作流示例：

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

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


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

在这段代码中，整体逻辑其实相当简单易懂。主要通过定义带有 step 注解的方法，并通过执行这些方法来实现工作流的流程控制。不过，代码中有两个关键的节点事件需要特别注意，它们分别是 StartEvent 和 StopEvent。工作流的执行通常从 StartEvent 开始，并且必须以 StopEvent 结束，确保工作流的完整性和正确性。

需要特别注意的是，如果工作流在执行过程中存在其他事件或步骤没有被正确触发或执行，系统会抛出错误。下面看下。

多个事件

我们依然可以进行简单的测试，毕竟这是流数据处理的基本操作。在此，我们定义了两个自定义事件，如下所示：

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

class FirstEvent(Event):
    first_output: str

class SecondEvent(Event):
    second_output: str

然后我们在工作流中引用下：

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

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

这样，我们就可以逐步执行各个事件。如果在执行过程中某一个步骤未能顺利完成，例如我们直接修改了第二个事件并且导致返回结束，那么第三个步骤将无法执行，这样就会引发错误。

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

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

在构建智能体工作流的过程中，如果没有进行连线操作，系统通常会自动检测并提示错误。这种情况下，错误提示会提醒用户必须完成连线，以确保工作流的正常运行和功能的完整性。报错信息如下：

可视化分析

工作流可视化功能在这里显得相对简单，它主要用于查看整个调用链路的结构和流程，而不像其他平台那样具备强大的交互操作性。我们一起看下：

from llama_index.utils.workflow import draw_all_possible_flows
#省略重复代码

if __name__ == "__main__":
    import asyncio

    asyncio.run(main())
    draw_all_possible_flows(MyWorkflow, filename="multi_step_workflow.html")

执行完成后，将会生成一个可正常打开的HTML文件。下面请看下图所示的内容：

整个工作流程非常清晰明了。在每一次执行当前事件时，首先会打印出所调用的方法名称，接着会显示正在处理的事件名称，最后以“done”字样结束该过程。

总结

通过今天的简单介绍，我们初步了解了工作流的基本概念及实现方法。工作流作为一种事件驱动的执行机制，为我们提供了更灵活的方式来管理和控制应用程序的流程。我们通过创建和配置虚拟环境，逐步实现了基础的工作流示例，体验了如何使用Python构建多步骤的业务逻辑。

在后续的学习中，我们还将探索更复杂的工作流构建方法。希望通过这一系列的实践，能够激发大家在智能体领域的创新思维，共同探讨更高效的解决方案。

我是努力的小雨，一名 Java 服务端码农，潜心研究着 AI 技术的奥秘。我热爱技术交流与分享，对开源社区充满热情。同时也是一位腾讯云创作之星、阿里云专家博主、华为云云享专家、掘金优秀作者。