5.2 查询管道与工作流编排

5.2 查询管道与工作流编排

本节摘要：当检索流程需要条件分支、人工步骤或跨步骤传递状态时，用 Workflow 把流程写成显式的步骤与事件。本节把一条"先判断问题类型、再走不同检索路线、最后合并合成"的流程改造成 Workflow，体会"步骤即函数、传递靠事件、并发自然获得"三种编程感受，并比较它与智能体的分工。

为什么需要显式编排

Router（3.5 节）解决了"选一条路"，但真实流程常是"选路 + 并行 + 汇合 + 条件回退"的图。用 if-else 硬写在脚本里，流程逻辑埋进代码细节，改流程等于改代码；Workflow 的做法是把每一步写成独立的 Step，步骤间用类型化事件传递，流程结构一目了然，且天然支持并发。

from llama_index.core.workflow import (
    Workflow, Step, Event, Context, step

class RoutedEvent(Event):
    query: str
    route: str

class CollectedEvent(Event):
    query: str
    context: str

class ConditionalRAGWorkflow(Workflow):
    """先判问题类型：条款类走向量检索，总结类走向量+摘要并行"""

    @step
    async def classify(self, ctx: Context, ev: StartEvent) -> RoutedEvent:
        # 第一步：用便宜模型判断问题形态
        label = await cheap_llm.acomplete(
            f"判断问题类型，只回答条款或总结：{ev.query}")
        return RoutedEvent(query=ev.query, route=str(label).strip())

    @step
    async def retrieve_clause(self, ctx: Context, ev: RoutedEvent) -> CollectedEvent | None:
        if ev.route != "条款":
            return None                          # 不归我管的直接放行
        nodes = vec_retriever.retrieve(ev.query)
        return CollectedEvent(query=ev.query, context="\n".join(n.text for n in nodes))

    @step
    async def retrieve_summary(self, ctx: Context, ev: RoutedEvent) -> CollectedEvent | None:
        if ev.route != "总结":
            return None
        nodes = summary_retriever.retrieve(ev.query)
        return CollectedEvent(query=ev.query, context="\n".join(n.text for n in nodes))

    @step
    async def synthesize(self, ctx: Context, ev: CollectedEvent) -> StopEvent:
        answer = await strong_llm.acomplete(
            f"依据材料回答：{ev.context}\n问题：{ev.query}")
        return StopEvent(result=str(answer))

wf = ConditionalRAGWorkflow(timeout=60, verbose=True)
result = await wf.run(query="报销审批要经过哪些环节")
​

读这段代码的三个观察点。步骤即函数：每个 @step 只干一件事，单元测试可以对单步直接做。传递靠事件：步骤间不共享可变全局状态，靠事件的字段携带数据，谁消费什么一清二楚。条件即返回：retrieve_clause 对不属于自己的路由返回 None，流程引擎自动把事件投给感兴趣的步骤——分支逻辑不需要显式 if-else 树。

并发与人工环节

Workflow 的两个生产级福利。并发：同一事件类型的多个步骤自动并行，比如"多索引同时检索"只需把多个检索步骤都声明为消费同一事件，无需手写线程池。人工环节：某个 Step 里发一个等待事件，流程挂起，外部系统（审批回调、用户确认） resume 后继续——把"检索到一半等用户确认再合成"这类交互式流程写成了声明式代码。

与智能体的混用

成熟系统的常见形态是"外层 Workflow 定骨架、局部智能体做探索"：骨架步骤（分类、路由、合成）用 Workflow 写死保证可控，探索性的某一步（比如"自由检索多轮")内嵌一个受限智能体。全智能体系统延迟与成本失控，全 Workflow 系统不够灵活——混搭取中。

本节要点回顾

• 三感受：步骤即函数、传递靠事件、条件即返回——流程结构从代码细节里浮出来。
• 并发免费：同事件多消费者自动并行，多索引检索不用手写线程。
• 人工环节：挂起-恢复事件支持交互式流程，声明式地写人机协作。
• 分工口诀：流程已知用 Workflow，流程未知用智能体，生产系统常混搭。
• 测试友好：单步可单测，配合 5.4 节的观测事件流，全链路可验证。

常见问题

Workflow 和智能体到底怎么选？ 判断依据是"流程是否已知"：你能画出流程图（分类、路由、并行、汇总）就用 Workflow，画不出来、要靠模型临场判断的才用智能体。两者也能嵌套——Workflow 的某个步骤里跑一个受限智能体，兼得可控与灵活。

Workflow 的步骤抛异常怎么办？ 异常会中断该次运行并向上传播，生产上在每个关键步骤内部做 try-except，失败时返回带标记的降级事件（比如"该路检索失败，用空上下文继续"），让流程带着残缺信息走完而不是整体崩掉。降级路径的设计与 6.4 节部署的思路一致。

Workflow 能做定时批处理吗？ 能，但它不是调度器——把 Workflow 当"单次流程的描述"，调度交给外部的定时任务系统。职责分离：cron 或调度平台决定"什么时候跑"，Workflow 决定"跑的时候流程长什么样"。