工作流子类化

工作流的另一个强大特性是其可扩展性。您可以将他人编写的工作流或LlamaIndex内置的工作流进行扩展，以满足您的特定需求。我们将探讨两种实现方式。

第一种是子类化：工作流本质上是普通的Python类，这意味着您可以通过创建子类来添加新功能。例如，假设您有一个代理工作流，它执行某些处理后发送电子邮件。您可以通过子类化该工作流，额外添加发送短信的步骤。

以下是我们的基础工作流：

from llama_index.core.workflow import (
    StartEvent,
    StopEvent,
    Workflow,
    step,
    Event,
    Context,
)


class Step2Event(Event):
    query: str


class Step3Event(Event):
    query: str


class MainWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> Step2Event:
        print("Starting up")
        return Step2Event(query=ev.query)

    @step
    async def step_two(self, ev: Step2Event) -> Step3Event:
        print("Sending an email")
        return Step3Event(query=ev.query)

    @step
    async def step_three(self, ev: Step3Event) -> StopEvent:
        print("Finishing up")
        return StopEvent(result=ev.query)

运行此工作流：

w = MainWorkflow(timeout=10, verbose=False)
result = await w.run(query="Initial query")
print(result)

输出结果为：

Starting up
Sending an email
Finishing up
Initial query

现在让我们通过子类化扩展该工作流，添加短信发送功能：

class Step2BEvent(Event):
    query: str


class CustomWorkflow(MainWorkflow):
    @step
    async def step_two(self, ev: Step2Event) -> Step2BEvent:
        print("Sending an email")
        return Step2BEvent(query=ev.query)

    @step
    async def step_two_b(self, ev: Step2BEvent) -> Step3Event:
        print("Also sending a text message")
        return Step3Event(query=ev.query)

修改后的输出将变为：

Starting up
Sending an email
Also sending a text message
Finishing up
Initial query

我们可以可视化这个子类化的工作流，它将显示所有步骤，如下所示：

draw_all_possible_flows(CustomWorkflow, "custom_workflow.html")

接下来，我们将探讨另一种扩展工作流的方法：资源。