主题
多数工作流可以直接用快速上手一节里的默认 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 的调用方都很友好。