我要投稿

LlamaIndex 开发大模型 Agent Workflow攻略

发布日期：2025-07-04 08:50:19 浏览次数： 1551

作者：AI悠悠

微信搜一搜，关注“AI悠悠”

一、背景

开发AI大模型Agent一般有两个老牌开发框架 LangGraph和LlamaIndex 。俩者各有优缺点，我选择LlamaIndex框架,原因有以下几点：

RAG开发使用LlamaIndex，那么开发Agent可以很容易将RAG功能融合；
LlamaIndex开发Agent的门槛很低，可快速上手。
LlamaIndex的工作流足够的强大；
官方文档很详细；

二、基本信息

1、概述

AI Agent是LLM驱动的“智能体”，能自主决策、调用工具、分步完成任务，适合处理复杂推理和多轮交互；LlamaIndex使用Workflow来开发智能体Agent 。

Workflow是一种事件驱动的流程编排工具，用于将复杂的AI应用拆解为多个可控的步骤（step），每个步骤由事件（Event）触发并产生新事件。最基础的用法是：定义事件类，继承Workflow类并用@step装饰器声明步骤，然后run方法启动流程。这样可以实现流程。

2、用法介绍

应用程序由多个模块组成，每个模块被称为"步骤"(Step)。这些步骤通过"事件"(Event)相互触发：一个步骤完成时发出事件，该事件又会触发后续步骤的执行。通过灵活组合步骤与事件，您可以构建任意复杂度的逻辑流程，从而提升应用程序的可维护性和可读性。每个步骤可以是任何功能单元，从简单的单行代码到复杂的代理程序，其输入输出完全由事件机制进行传递。
支持复杂控制流处理，包括分支跳转（基于事件类型选择不同步骤）、循环执行以及多步骤并行处理等功能
Context对象支持在各步骤间共享和持久化状态，可用于实现全局变量和缓存功能。
提供可视化界面、流式处理、自动重试、断点续传、人工干预等高级功能，适用于构建多智能体系统、检索增强生成以及复杂数据处理等AI应用场景。

三、示例代码

1、基本工作流

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

pip install llama-index-core

为了优化开发效率，建议使用内置的可视化工具来直观展现工作流程

pip install llama-index-utils-workflow

工作流的最小依赖项包括：

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

单步工作流

工作流通常作为继承自 Workflow 的类实现。该类可以定义任意数量的步骤，每个步骤都是一个用 @step 修饰的方法。以下是最简单的工作流程：

from llama_index.core.workflow import (    StartEvent,    StopEvent,    Workflow,    step,    Event,)from llama_index.utils.workflow import draw_all_possible_flowsclass MyWorkflow(Workflow):    @step    async def one_step(self,ev: StartEvent) -> StopEvent:        print(f"first_input={ev.first_input}")        print(f"second_input={ev.second_input}")        return StopEvent(result="hello workflow")async  def run_workflow():    w= MyWorkflow(timeout=10, verbose=False)    result2= await  w.run(first_input="Start the workflow.",second_input=2)    print(result2)if __name__ == "__main__":    import asyncio    asyncio.run(run_workflow())    # draw_all_possible_flows(MyWorkflow, filename="multi_step_workflow.html")

输出结果

first_input=Start the workflow.second_input=2hello workflow

代码解析：

定义从 Workflow 继承的类 MyWorkflow ；
使用 @step 修饰器定义单个步骤 one_step ;
该步骤采用单个参数 ev，它是 StartEvent 的实例;
该步骤返回一个 StopEvent，其结果为 “Hello， world！”;
创建一个 MyWorkflow 实例，超时为 10 秒，详细程度关闭;
运行工作流并打印结果；

类型注释

类型注释（例如 ev： StartEvent）和 -> StopEvent 对于工作流的工作方式至关重要。预期类型确定哪些事件类型将触发步骤。可视化工具（见下文）等工具也依赖于这些注释来确定生成哪些类型，从而确定控制流的下一步发展方向。

类型注释在编译时进行验证，因此，如果您发出的事件从未被其他步骤使用，您将收到一条错误消息。

启动和停止事件

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

可视化工作流

工作流的一大特色是LlamaIndex已经安装了内置的可视化工具。让我们可视化我们刚刚创建的简单工作流：

from llama_index.utils.workflow import draw_all_possible_flowsdraw_all_possible_flows(    MyWorkflow,    filename="basic_workflow.html",    # Optional, can limit long event names in your workflow    # Can help with readability    # max_label_length=10,)

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

当然，单步的流程不是很有用！让我们定义一个多步骤工作流。

自定义事件

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

我们像以前一样导入 Event，并为 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: strclass 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_flowsdraw_all_possible_flows(MyWorkflow, filename="multi_step_workflow.html")

2、全局参数

使用自定义事件的属性逐步传递数据。这是一种传递数据的强大方法，但它有局限性。例如，如果要在未直接连接的步骤之间传递数据，则需要通过两者之间的所有步骤传递数据。这会使您的代码更难阅读和维护。

为避免这种情况，LlamaIndex有一个 Context 对象可用于工作流中的每个步骤。要使用它，请为您的步骤声明一个 Context 类型的参数。

需要一个新的导入，Context 类型：

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

我们定义一个 start 事件，用于检查数据是否已加载到上下文中。否则，它将返回一个 SetupEvent，该事件触发加载数据并循环回 start 的 setup。

class SetupEvent(Event):    query: strclass StepTwoEvent(Event):    query: strclass StatefulFlow(Workflow):    @step    async def start(        self, ctx: Context, ev: StartEvent    ) -> SetupEvent | StepTwoEvent:        db = await ctx.get("some_database", default=None)        if db is None:            print("Need to load data")            return SetupEvent(query=ev.query)        # do something with the query        return StepTwoEvent(query=ev.query)    @step    async def setup(self, ctx: Context, ev: SetupEvent) -> StartEvent:        # load data        await ctx.set("some_database", [1, 2, 3])        return StartEvent(query=ev.query)

然后，在 step_two 中，我们可以直接从上下文访问数据，而无需显式传递数据。在一代 AI 应用程序中，这对于加载索引和其他大数据作非常有用。

@stepasync def step_two(self, ctx: Context, ev: StepTwoEvent) -> StopEvent:    # do something with the data    print("Data is ", await ctx.get("some_database"))    return StopEvent(result=await ctx.get("some_database"))w = StatefulFlow(timeout=10, verbose=False)result = await w.run(query="Some query")print(result)

LlamaIndex的工作流还有分支循环、流式处理、并发执行等强大功能，以及可以观测整体流程等方便调试和跟踪。

53AI，企业落地大模型首选服务商

产品：场景落地咨询+大模型应用平台+行业解决方案

承诺：免费场景POC验证，效果验证后签署服务协议。零风险落地应用大模型，已交付160+中大型企业