工作流往往需要一些时间才能完成：可能调用较慢的服务商、发生分支、对一批数据扇出，或等待人工回应。流式能力让你在运行仍在推进时就把进展呈现出来。

流式有两个侧面：

侧面	API
步骤内部	ctx.write_event_to_stream(...)
工作流外部	handler.stream_events()

workflow.run(...) 启动工作流并返回一个 WorkflowHandler。该 handler 可以被 await 以拿到最终结果，同时也拥有这次运行的事件流。

为了完成这件事，先把需要的依赖都引入：

import asyncio
from llama_index.llms.openai import OpenAI

from workflows import (
    Workflow,
    Context,
    step,
)
from workflows.events import (
    StartEvent,
    StopEvent,
    Event,
)

接着为一个简单的三步工作流准备若干事件，再加一个用于把进度流式送出的事件：

class FirstEvent(Event):
    first_output: str

class SecondEvent(Event):
    second_output: str
    response: str

class ProgressEvent(Event):
    msg: str

并定义一个会发送事件的工作流类：

class MyWorkflow(Workflow):
    @step
    async def step_one(self, ctx: Context, ev: StartEvent) -> FirstEvent:
        ctx.write_event_to_stream(ProgressEvent(msg="Step one is happening"))
        return FirstEvent(first_output="First step complete.")

    @step
    async def step_two(self, ctx: Context, ev: FirstEvent) -> SecondEvent:
        llm = OpenAI(model="gpt-4o-mini")
        generator = await llm.astream_complete(
            "Please give me the first 3 paragraphs of Moby Dick, a book in the public domain."
        )
        full_resp = ""
        async for response in generator:
            # Allow the workflow to stream this piece of response
            ctx.write_event_to_stream(ProgressEvent(msg=response.delta))
            full_resp += response.delta

        return SecondEvent(
            second_output="Second step complete, full response attached",
            response=full_resp,
        )

    @step
    async def step_three(self, ctx: Context, ev: SecondEvent) -> StopEvent:
        ctx.write_event_to_stream(ProgressEvent(msg="Step three is happening"))
        return StopEvent(result="Workflow complete.")

提示

这里的 OpenAI() 假定你的环境中已设置 OPENAI_API_KEY。也可以通过 api_key 参数传入。

在 step_one 与 step_three 中，我们各自向事件流写入一个事件。在 step_two 中，我们用 astream_complete 得到 LLM 回答的可迭代生成器，然后针对 LLM 返回的每一段数据产出一个事件，最后才把最终回答返回给 step_three。

要真正拿到这份输出，我们需要异步运行工作流并监听事件，像这样：

async def main():
    w = MyWorkflow(timeout=30, verbose=True)
    handler = w.run(first_input="Start the workflow.")

    async for ev in handler.stream_events():
        if isinstance(ev, ProgressEvent):
            print(ev.msg)

    final_result = await handler
    print("Final result", final_result)

if __name__ == "__main__":
    asyncio.run(main())

run 会在后台调度工作流。stream_events() 会产出写入事件流的每一个事件，并在流中投递出 StopEvent 时停止。此后 await 该 handler 即可拿到最终结果。

一个 handler 的事件流只能被消费一次。如果你要把工作流事件广播给多个客户端，就在自己的应用里消费一次该流，然后自行把这些事件扇出。

处理工作流的终止

当工作流异常结束时，会先向事件流发布一个特定的 StopEvent 子类，之后 await handler 才抛出异常：

• WorkflowTimedOutEvent - 工作流超出其超时时间时发布。包含 timeout（秒）与 active_steps（当时正在运行的步骤名列表）。
• WorkflowCancelledEvent - 工作流被用户取消时发布。
• WorkflowFailedEvent - 某个步骤在重试耗尽后永久失败时发布。包含 step_name、exception、attempts 与 elapsed_seconds。

from workflows.events import (
    WorkflowTimedOutEvent,
    WorkflowCancelledEvent,
    WorkflowFailedEvent,
)

async for ev in handler.stream_events():
    if isinstance(ev, WorkflowTimedOutEvent):
        print(f"Workflow timed out after {ev.timeout}s")
    elif isinstance(ev, WorkflowCancelledEvent):
        print("Workflow was cancelled")
    elif isinstance(ev, WorkflowFailedEvent):
        print(f"Step '{ev.step_name}' failed after {ev.attempts} attempts: {ev.exception}")