深入解析 LlamaIndex Workflows：构建复杂 RAG 与智能体工作流

工作流中的每个环节被作为 step（步骤，代表一个处理动作），每个 step 可以选择接收（类似订阅）一种或多种 event（事件）做处理，并同样可以发送一种或多种 event 给其他 step。通过这种方式，把多个 step 自然的连接起来形成完整的 Workflow。

在这种架构中，工作流的运行不是由框架来根据预先的定义（比如 Graph）来调度任务组件执行，而是由组件自身决定：你想接收什么 event，做什么处理，并发出什么 event。如果组件 B 接受了 A 发出的某种 event，那么 B 就会在 A 发出 event 后触发执行。

图中涉及到 LlamaIndex Workflows 中的一些关键概念：

1. Workflow（工作流）

工作流代表一个复杂的 RAG、Agent 或者任意复合形式的 LLM 应用的端到端流程。创建完工作流后，调用 run 方法，并输入任务即可启动。

Workflow 类似 LangGraph 中的 Graph。

2. Step（步骤）

步骤代表工作流中的单个任务，你可以自由的定义步骤内容。每个步骤接收输入事件（订阅），并返回输出事件。当输入事件发生时，这个步骤就会自动执行。步骤使用 Python 函数定义。

Step 类似 LangGraph 中的 Node。

3. Event（事件）

事件是一个步骤的输入输出，也是工作流各个步骤之间的数据载体。当事件发生时，'订阅'该事件的步骤就会执行，同时从事件中取出必要数据。事件是一个 Pydantic 类型对象，可以自由定义结构。注意两个特殊事件：

• StartEvent 与 StopEvent：是两个系统事件，代表工作流开始与结束。
  • StartEvent 由框架派发，接收 StartEvent 的步骤代表工作流的开始。
  • StopEvent 由框架接收，发送 StopEvent 的步骤代表没有后续步骤。

Event 是 LlamaIndex 中连接 Step 的机制，作用类似 LangGraph 中的 Edge（边），但机制完全不同。

4. Context（上下文）

Context 是一个用来在整个工作流内自动流转的上下文状态对象，放入 Context 中的数据可以被所有步骤接收和读取到，可以理解为全局变量。

Context 类似 LangGraph 中的 State（状态），但 LlamaIndex 中各 Step 的数据交换更多通过 event 传递，而不是 Context。

最后总结：LlamaIndex Workflows 是通过一种事件驱动、自我管理的任务组件（step）来组装与编排复杂工作流的方式。此外还具有可视化（生成流程图）、可单步调试、可复用工作流等特性。

Hello, Workflows! 实战示例

现在我们用 LlamaIndex 的 Workflows 来实现一个最简单的带 Rerank 重排功能的 RAG 工作流。其工作流程图如下：

1. 索引阶段 (Indexing)：加载文档并构建向量索引。
2. 检索阶段 (Retrieve)：根据问题检索相关节点。
3. 重排阶段 (Rerank)：对检索结果进行重排序。
4. 生成阶段 (Generate)：基于重排后的结果生成最终回答。

这个简单的 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")
            return StopEvent(result=index)
        else:
            return None

    @step
    async def retrieve(
        self, ctx: Context, ev: StartEvent) -> RetrieverEvent | None:
        # 带 query 参数，表示生成阶段
        if not ev.get("query"):
            return None
          
        query = ev.get("query")
        index = ev.get("index")

        await ctx.set("query", query)
        retriever = index.as_retriever(similarity_top_k=2)
        nodes = await retriever.aretrieve(query)
        return RetrieverEvent(nodes=nodes)

    @step
    async def rerank(self, ctx: Context, ev: RetrieverEvent) -> RerankEvent:
        ranker = LLMRerank(
            choice_batch_size=5, top_n=3, llm=OpenAI(model="gpt-4o-mini")
        )
        new_nodes = ranker.postprocess_nodes(
            ev.nodes, query_str=await ctx.get("query", default=None)
        )
        return RerankEvent(nodes=new_nodes)

    @step
    async def generate(self, ctx: Context, ev: RerankEvent) -> StopEvent:   
        llm = OpenAI(model="gpt-4o-mini")
        summarizer = CompactAndRefine(llm=llm, streaming=True, verbose=True)
        query = await ctx.get("query", default=None)

        response = await summarizer.asynthesize(query, nodes=ev.nodes)
        return 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 工作流，展示其在多智能体协作中的应用。