PythonAI算法
深入解析 LlamaIndex Workflows：构建复杂 RAG 与智能体工作流
LlamaIndex Workflows 是一种事件驱动的框架，用于编排复杂的 RAG 与 Agent 应用。相比传统的链式或 DAG 结构，它支持循环、并行及灵活分支。核心概念包括 Workflow、Step、Event 和 Context。本文介绍了其设计思想、关键组件及基于 RAG 场景的代码实现示例，展示了如何通过事件传递驱动工作流执行，并对比了其与 LangGraph 的差异。
LlamaIndex Workflows 是一种事件驱动的框架，用于编排复杂的 RAG 与 Agent 应用。相比传统的链式或 DAG 结构，它支持循环、并行及灵活分支。核心概念包括 Workflow、Step、Event 和 Context。本文介绍了其设计思想、关键组件及基于 RAG 场景的代码实现示例，展示了如何通过事件传递驱动工作流执行，并对比了其与 LangGraph 的差异。
面对多样的 LLM 应用需求，无论是主流如 LangChain 这样的基础开发框架，还是以云服务提供的低代码开发平台，都试图在提供基础组件抽象的基础上，提供更便捷的组装与编排这些组件的方式，以帮助构建更高级的 AI 工作流应用。比如年初发布的 LangChain 新版本中就引入了 LangGraph，以帮助构建 Graph 表示的 AI 工作流。
最近，另一家主流 LLM 框架 LlamaIndex 推出了一项新 beta 特性：Workflows。提供了一种与 LangGraph 不同、事件驱动的框架，以编排复杂的 RAG 或 Agent 应用工作流，支持灵活的分支、并行与循环等处理。
我们将分成两篇来给大家深入解析与介绍 LlamaIndex Workflows：
以两种主流 LLM 应用开发框架 LangChain 与 LlamaIndex 来看，它们都提供了大量的基础组件封装，将常见的 RAG 或 Agent 应用开发中常用到的流程环节进行了抽象，简化了开发并避免锁定到特定的底层设施（如大模型、向量库）。利用它们提供的 API 可以在极短的时间内开发出大模型应用，比如在 LlamaIndex 中，几行代码可以快速完成一个 RAG 原型：
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
documents = SimpleDirectoryReader("../data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query('你的问题')
print(response)
但这里的问题是：当面对复杂的业务与优化需求时，不断出现更多的 RAG 工作模块与算法（重写、路由、排序、图索引等），如何用最佳的方法来组合与协调这些组件，实现更复杂的工作流程？
在框架的早期，可以总结成两种主要方式：
比如简单的向量检索器是 VectorIndexRetriever，但现在需要增加路由功能，就封装一个带有路由的 RouterRetriever；简单的查询引擎是 QueryEngine，但现在需要增加重排功能，就增加一个可以传入 Reranker 模块的参数。
最具代表性的是 LangChain 框架中的 Chain（链）与 LCEL（LangChain 表达语言）特性；以及 LlamaIndex 在年初推出的 Query Pipeline（查询管道）声明式 API 特性。它们都在一定程度上提供了编排能力。
尽管这种受限的工作流编排仍然面临较大的局限性：
基于这样一些原因，提供一种更强大的工作流定义与编排的支持，也成为了开发框架的发力重点，典型的就是之前介绍过的 LangGraph 以及本篇将一窥究竟的 LlamaIndex Workflows。
LlamaIndex Workflows 是 LlamaIndex 近期推出（仍然是测试版）的用来替代之前 Query Pipeline（查询管道）的新特性。与 LangGraph 不同的是，其没有采用类似 LangGraph 基于图结构的流程编排模式，而是采用了一种事件驱动的工作流编排方式：
工作流中的每个环节被作为 step（步骤，代表一个处理动作），每个 step 可以选择接收（类似订阅）一种或多种 event（事件）做处理，并同样可以发送一种或多种 event 给其他 step。通过这种方式，把多个 step 自然的连接起来形成完整的 Workflow。
在这种架构中，工作流的运行不是由框架来根据预先的定义（比如 Graph）来调度任务组件执行，而是由组件自身决定：你想接收什么 event，做什么处理，并发出什么 event。如果组件 B 接受了 A 发出的某种 event，那么 B 就会在 A 发出 event 后触发执行。
图中涉及到 LlamaIndex Workflows 中的一些关键概念：
工作流代表一个复杂的 RAG、Agent 或者任意复合形式的 LLM 应用的端到端流程。创建完工作流后，调用 run 方法，并输入任务即可启动。
Workflow 类似 LangGraph 中的 Graph。
步骤代表工作流中的单个任务，你可以自由的定义步骤内容。每个步骤接收输入事件（订阅），并返回输出事件。当输入事件发生时，这个步骤就会自动执行。步骤使用 Python 函数定义。
Step 类似 LangGraph 中的 Node。
事件是一个步骤的输入输出，也是工作流各个步骤之间的数据载体。当事件发生时，'订阅'该事件的步骤就会执行，同时从事件中取出必要数据。事件是一个 Pydantic 类型对象，可以自由定义结构。注意两个特殊事件：
Event 是 LlamaIndex 中连接 Step 的机制，作用类似 LangGraph 中的 Edge（边），但机制完全不同。
Context 是一个用来在整个工作流内自动流转的上下文状态对象，放入 Context 中的数据可以被所有步骤接收和读取到，可以理解为全局变量。
Context 类似 LangGraph 中的 State（状态），但 LlamaIndex 中各 Step 的数据交换更多通过 event 传递，而不是 Context。
最后总结：LlamaIndex Workflows 是通过一种事件驱动、自我管理的任务组件（step）来组装与编排复杂工作流的方式。此外还具有可视化（生成流程图）、可单步调试、可复用工作流等特性。
现在我们用 LlamaIndex 的 Workflows 来实现一个最简单的带 Rerank 重排功能的 RAG 工作流。其工作流程图如下：
这个简单的 Workflow 反映了一个经典 RAG 流程的两个工作阶段，索引（indexing）与生成（generate）阶段。这里把两个阶段做成不同的分支，是为了便于一次索引后做多次生成，也体现了 Workflows 的一个特征：一个 Workflow 的入口 step 可以有多个，只要这个 step 接收 StartEvent 事件。
看下这个 Workflow 的简单实现代码：
from llama_index.core.workflow import Event
from llama_index.core.schema import NodeWithScore
# 定义两个事件
# RetrieverEvent: 用于存储检索到的节点
# RerankEvent: 用于存储重排后的节点
class RetrieverEvent(Event):
nodes: list[NodeWithScore]
class RerankEvent(Event):
nodes: list[NodeWithScore]
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
from llama_index.core.response_synthesizers import CompactAndRefine
from llama_index.core.postprocessor.llm_rerank import LLMRerank
from llama_index.core.workflow import (
Context,
Workflow,
StartEvent,
StopEvent,
step,
)
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
# workflow 定义
class RAGWorkflow(Workflow):
@step
async def indexing(self, ctx: Context, ev: StartEvent) -> StopEvent | None:
# 不带 query 参数，表示索引阶段
if not ev.get("query"):
documents = SimpleDirectoryReader(input_files=['../../data/sales_tips1.txt']).load_data()
index = VectorStoreIndex.from_documents(documents=documents)
print("Indexing complete")
StopEvent(result=index)
:
() -> RetrieverEvent | :
ev.get():
query = ev.get()
index = ev.get()
ctx.(, query)
retriever = index.as_retriever(similarity_top_k=)
nodes = retriever.aretrieve(query)
RetrieverEvent(nodes=nodes)
() -> RerankEvent:
ranker = LLMRerank(
choice_batch_size=, top_n=, llm=OpenAI(model=)
)
new_nodes = ranker.postprocess_nodes(
ev.nodes, query_str= ctx.get(, default=)
)
RerankEvent(nodes=new_nodes)
() -> StopEvent:
llm = OpenAI(model=)
summarizer = CompactAndRefine(llm=llm, streaming=, verbose=)
query = ctx.get(, default=)
response = summarizer.asynthesize(query, nodes=ev.nodes)
StopEvent(result=response)
w = RAGWorkflow()
代码非常清晰易懂：其中四个函数定义了四个 step，各个 step 之间通过输入的 event 与返回的 event 进行连接，并驱动整个工作流的自动运行。这里也能体现 Workflows 与 LangGraph 的最大区别：LangGraph 通过定义图中的 edge 来体现流程节点之间的关系；而 Workflows 则是通过 step 之间的 event 传递来体现。
你可以通过如下代码运行工作流并测试：
w = RAGWorkflow()
# 索引阶段
index = await w.run()
# 生成阶段，需要带入问题和创建好的索引
result = await w.run(query="你的问题", index=index)
async for chunk in result.async_response_gen():
print(chunk, end="", flush=True)
此外，Workflows 提供了工具对定义的 Workflow 进行可视化，使用如下代码：
from llama_index.utils.workflow import draw_all_possible_flows
draw_all_possible_flows(
RAGWorkflow, filename="RAG.html"
)
可以生成可视化的工作流程图，方便开发者理解逻辑流向。
如果你使用过 LlamaIndex，可能会觉得这里的代码似乎更复杂化了，这并不意外：Workflows（以及 LangChain 的 LangGraph）显然都是为了更复杂与更灵活化的 LLM 应用场景而设计，而非简单应用。
LlamaIndex Workflows 为构建复杂的 RAG 和 Agent 应用提供了一种新的范式。它通过事件驱动机制解决了传统链式或 DAG 结构在循环支持和精细化控制上的不足。虽然学习曲线略高于简单的 Query Engine，但在处理多步骤、条件分支及状态流转复杂的业务场景时，其优势明显。
下一篇我们将分析如何使用 Workflows 实现一个复杂的 Agent 工作流，展示其在多智能体协作中的应用。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online