主题
大多数示例把步骤定义为 Workflow 子类上的方法。你也可以用 @step(workflow=...) 把独立函数挂到工作流类上。当工作流由可复用的函数组装而成,或者某个集成想在不继承工作流类的情况下添加一个步骤时,这很有用。
首先创建一个空类来承载这些步骤:
from workflows import Workflow
class TestWorkflow(Workflow):
pass现在可以通过定义函数并用 @step() 装饰器装饰它们,把步骤加入工作流:
from workflows import step
from workflows.events import StartEvent, StopEvent
@step(workflow=TestWorkflow)
async def some_step(ev: StartEvent) -> StopEvent:
return StopEvent()在这个例子里,装饰器把 some_step 注册到了 TestWorkflow 类上。函数签名与作为方法的步骤相同,只是没有 self 参数。
注册发生在导入时。请确保在实例化或校验工作流之前,先导入定义该函数的模块。步骤名仍须唯一:如果以已有方法步骤或自由函数步骤的同名注册新的自由函数,会抛出校验错误。