GeneratorStep

GeneratorStep 是 Step 的子类，旨在用作 Pipeline 中的第一步，因为它不需要输入，并生成可供其他 step 使用的数据。或者，它也可以单独使用。

from typing import List, TYPE_CHECKING
from typing_extensions import override

from distilabel.steps import GeneratorStep

if TYPE_CHECKING:
    from distilabel.typing import StepColumns, GeneratorStepOutput

class MyGeneratorStep(GeneratorStep):
    instructions: List[str]

    @override
    def process(self, offset: int = 0) -> "GeneratorStepOutput":
        if offset:
            self.instructions = self.instructions[offset:]

        while self.instructions:
            batch = [
                {
                    "instruction": instruction
                } for instruction in self.instructions[: self.batch_size]
            ]
            self.instructions = self.instructions[self.batch_size :]
            yield (
                batch,
                True if len(self.instructions) == 0 else False,
            )

    @property
    def outputs(self) -> "StepColumns":
        return ["instruction"]

然后我们可以按如下方式使用它

step = MyGeneratorStep(
    name="my-generator-step",
    instructions=["Tell me a joke.", "Tell me a story."],
    batch_size=1,
)
step.load()

next(step.process(offset=0))
# ([{'instruction': 'Tell me a joke.'}], False)
next(step.process(offset=1))
# ([{'instruction': 'Tell me a story.'}], True)

注意

当作为独立 step 使用时，始终需要执行 Step.load()。在 pipeline 中，这将在 pipeline 执行期间自动完成。

定义自定义 GeneratorStep

我们可以通过创建 GeneratorStep 的新子类并定义以下内容来定义自定义生成器 step

• outputs：是一个属性，返回包含输出字段名称的字符串列表，或一个字典，其中键是列名，值是布尔值，指示列是否是必需的。

• process：是一个方法，用于生成输出数据和一个布尔标志，指示这是否是要生成的最后一批数据。

注意

process 方法的默认签名是 process(self, offset: int = 0) -> GeneratorStepOutput。应遵守参数 offset，不能提供更多参数，并且应遵守类型提示和返回类型提示，因为它应该能够默认接收任意数量的输入，即一次可以有多个 Step 连接到当前 step。

警告

为了使自定义 Step 子类与 distilabel 以及默认在 Pipeline 中的每个 Step 上执行的验证和序列化正常工作，StepInput 和 StepOutput 的类型提示应使用，并且不应被双引号包围或在 typing.TYPE_CHECKING 下导入，否则，验证和/或序列化将失败。

从 GeneratorStep 继承使用 @step 装饰器

我们可以从 GeneratorStep 类继承，并按如下方式定义 outputs 和 process 方法

from typing import List, TYPE_CHECKING
from typing_extensions import override

from distilabel.steps import GeneratorStep

if TYPE_CHECKING:
    from distilabel.typing import StepColumns, GeneratorStepOutput

class MyGeneratorStep(GeneratorStep):
    instructions: List[str]

    @override
    def process(self, offset: int = 0) -> "GeneratorStepOutput":
        ...

    @property
    def outputs(self) -> "StepColumns":
        ...

@step 装饰器将处理样板代码，并允许以更直接的方式定义 outputs 和 process 方法。一个缺点是它不允许您访问 self 属性（如果有），也不允许设置这些属性，因此如果您需要访问或设置任何属性，则应采用第一种方法来定义自定义 GeneratorStep 子类。

from typing import TYPE_CHECKING
from distilabel.steps import step

if TYPE_CHECKING:
    from distilabel.typing import GeneratorStepOutput

@step(outputs=[...], step_type="generator")
def CustomGeneratorStep(offset: int = 0) -> "GeneratorStepOutput":
    yield (
        ...,
        True if offset == 10 else False,
    )

step = CustomGeneratorStep(name="my-step")