LlamaIndex Workflow 实战：事件驱动的 RAG 流水线

一句话结论：LlamaIndex Workflow 把"链式 RAG"升级为"事件驱动 pipeline"。复杂场景可读性和可维护性大幅提升。

核心概念

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

# Workflow = 一组 step + event
# 每个 step 接收 event，处理后发出新 event

实战：多步 RAG Pipeline

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


class QueryEvent(Event):
    query: str


class RetrievalEvent(Event):
    query: str
    nodes: list


class RerankEvent(Event):
    query: str
    nodes: list


class GenerationEvent(Event):
    query: str
    nodes: list


class RAGWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> QueryEvent:
        return QueryEvent(query=ev.query)

    @step
    async def retrieve(self, ev: QueryEvent) -> RetrievalEvent:
        retriever = VectorIndexRetriever(index=index, top_k=10)
        nodes = retriever.retrieve(ev.query)
        return RetrievalEvent(query=ev.query, nodes=nodes)

    @step
    async def rerank(self, ev: RetrievalEvent) -> RerankEvent:
        from sentence_transformers import CrossEncoder
        reranker = CrossEncoder("BAAI/bge-reranker-base")
        pairs = [[ev.query, n.text] for n in ev.nodes]
        scores = reranker.predict(pairs)
        ranked = sorted(zip(ev.nodes, scores), key=lambda x: -x[1])[:3]
        return RerankEvent(query=ev.query, nodes=[n for n, _ in ranked])

    @step
    async def generate(self, ev: RerankEvent) -> StopEvent:
        context = "\n\n".join([n.text for n in ev.nodes])
        response = llm.invoke(
            f"基于以下上下文回答：\n{context}\n\n问题：{ev.query}"
        )
        return StopEvent(result=response)


# 启动
workflow = RAGWorkflow()
result = await workflow.run(query="什么是 RAG？")
print(result.result)

5 个核心优势

1. 可视化

# LlamaIndex 提供 Workflow 可视化
from llama_index.core.workflow import draw_all_possible_flows
draw_all_possible_flows(workflow, filename="workflow.html")

2. 事件驱动

# 多个 step 可以并行
@step
async def parallel_step(self, ev: Event) -> List[Event]:
    # 触发多个并行任务
    return [EventA(...), EventB(...)]

3. 错误处理

from llama_index.core.workflow import WorkflowTimeoutError

try:
    result = await workflow.run(query=query, timeout=30)
except WorkflowTimeoutError:
    # 处理超时
    pass

4. 状态持久化

# Workflow 内置 context
@step
async def step_with_state(self, ev: StartEvent) -> StopEvent:
    self.context.state["key"] = "value"
    return StopEvent(result=...)

5. 流式输出

async for event in workflow.stream(query="..."):
    print(event)

何时用 Workflow

✅ 适合

• 复杂 RAG（多步检索 / 多 Agent）
• 需要状态管理的 pipeline
• 多模态处理

❌ 不适合

• 简单 RAG（直接用 query_engine）
• 一次性脚本

参考

• LlamaIndex Workflow 文档

本文基于 LlamaIndex 0.12。