Skip to content
LlamaIndex 文档版本 跟随上游 源快照 b1a1ce8f 更新于 2026-10-15

多数工作流可以直接用快速上手一节里的默认 StartEvent 与 StopEvent。当工作流的边界本身有有用的 schema 时才定义自定义起始与结束事件:开头是一个带类型的请求对象,结尾是一个带类型的结果对象。

使用自定义 StartEvent

当你用关键字参数调用 run() 时,Workflows 会依据这些参数构造工作流的起始事件。使用默认的 StartEvent 时,任何额外字段都会被接受:

result = await workflow.run(topic="pirates")

对小输入来说这很方便。但在生产代码里,自定义 StartEvent 能给入口一个真正的 schema,并让 Pydantic 在第一个步骤运行前就校验缺失或畸形的输入。

创建一个继承自 StartEvent 的自定义类:

from workflows.events import StartEvent

class JokeStartEvent(StartEvent):
    topic: str
    tone: str = "funny"

然后在启动工作流的那个步骤里使用该事件类型:

class JokeFlow(Workflow):
    @step
    async def generate_joke(self, ev: JokeStartEvent) -> JokeEvent:
        prompt = f"Write a {ev.tone} joke about {ev.topic}."
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

你仍然可以把各字段作为关键字参数传入:

w = JokeFlow(timeout=60)
result = await w.run(topic="pirates", tone="dry")

对于更大的输入对象,请通过 start_event 传入事件实例:

start_event = JokeStartEvent(topic="pirates", tone="dry")
w = JokeFlow(timeout=60)
result = await w.run(start_event=start_event)

事件只用于可序列化的数据。如果工作流需要 LLM 客户端、索引、数据库连接或文件句柄,请把它作为资源注入,而不是放到起始事件上。起始事件在你做快照或对外提供服务时会被序列化,而重量级的运行时对象通常无法序列化。

使用自定义 StopEvent

内置的 StopEvent 会把放进 result 的东西原样返回:

return StopEvent(result={"critique": critique, "score": score})

对快速上手的工作流来说这没问题,但结果的类型是 Any。自定义结束事件能让输出的形状变得明确。

创建一个 StopEvent 的子类:

from workflows.events import StopEvent

class JokeResult(StopEvent):
    joke: str
    critique: str

现在就可以在工作流里把 StopEvent 换成 JokeResult:

class JokeFlow(Workflow):
    ...

    @step
    async def critique_joke(self, ev: JokeEvent) -> JokeResult:
        prompt = f"Give a thorough analysis and critique of the following joke: {ev.joke}"
        response = await self.llm.acomplete(prompt)
        return JokeResult(joke=ev.joke, critique=str(response))

当某个步骤返回基类 StopEvent 时,await workflow.run(...) 返回 stop_event.result。而当某个步骤返回自定义的 StopEvent 子类时,await workflow.run(...) 返回的就是那个事件实例:

w = JokeFlow(timeout=60)
result = await w.run(topic="pirates")
print(result.joke)
print(result.critique)

这让结果对类型检查器、编辑器自动补全,以及会内省工作流 schema 的调用方都很友好。

BytesRouter 是社区翻译站点,所有译文仅供参考,一切以官方英文原文为准。