检查点工作流运行

在本笔记本中，我们将演示如何通过 WorkflowCheckpointer 对象对 Workflow 运行进行检查点保存。我们还将展示如何查看存储在此对象中的所有检查点，最后演示如何将检查点用作新运行的起点。

定义一个工作流

import os

api_key = os.environ.get("OPENAI_API_KEY")

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


class JokeEvent(Event):
    joke: str


class JokeFlow(Workflow):
    llm = OpenAI(api_key=api_key)

    @step
    async def generate_joke(self, ev: StartEvent) -> JokeEvent:
        topic = ev.topic

        prompt = f"Write your best joke about {topic}."
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

    @step
    async def critique_joke(self, ev: JokeEvent) -> StopEvent:
        joke = ev.joke

        prompt = f"Give a thorough analysis and critique of the following joke: {joke}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))

定义一个工作流检查点对象

from llama_index.core.workflow.checkpointer import WorkflowCheckpointer

# instantiate Jokeflow
workflow = JokeFlow()
wflow_ckptr = WorkflowCheckpointer(workflow=workflow)

从工作流检查点运行工作流程

WorkflowCheckpointer.run() 方法是 Workflow.run() 方法的一个封装器，它会注入一个检查点回调函数以创建和存储检查点。请注意，检查点是在步骤完成时创建的，存储在检查点中的数据包括：

last_completed_step: 最后完成的步骤名称
input_event: 此最后完成步骤的输入事件
output_event: 由最后完成的步骤输出的事件
ctx_state: 已附加 Context 的快照

handler = wflow_ckptr.run(
    topic="chemistry",
)
await handler

'This joke plays on the double meaning of the word "rates," which can refer to both the cost of something and the passage of time. The punchline suggests that chemists prefer nitrates because they are less expensive than day rates, implying that chemists are frugal or cost-conscious individuals.\n\nOverall, the joke is clever and plays on a pun that is likely to be appreciated by those familiar with chemistry and the concept of nitrates. However, the humor may be lost on those who are not well-versed in chemistry terminology. Additionally, the joke relies on a somewhat simplistic play on words, which may not be as engaging or humorous to some audiences.\n\nIn terms of structure, the joke follows a classic setup and punchline format, with the punchline providing a surprising twist on the initial premise. The delivery of the joke may also play a role in its effectiveness, as timing and tone can greatly impact the humor of a joke.\n\nOverall, while the joke may appeal to a specific audience and demonstrate some clever wordplay, it may not have universal appeal and may be considered somewhat niche in its humor.'

我们可以通过 .checkpoints 属性查看所有检查点，这是一个字典，其键代表运行的 run_id，其值是为该运行存储的检查点列表。

wflow_ckptr.checkpoints

{'483eccdd-a035-42cc-b596-cd33d42938a7': [Checkpoint(id_='d5acd098-47e2-4acf-9520-9ca06ee4e238', last_completed_step='generate_joke', input_event=StartEvent(), output_event=JokeEvent(joke="Why do chemists like nitrates so much?\n\nBecause they're cheaper than day rates!"), ctx_state={'globals': {}, 'streaming_queue': '[]', 'queues': {'_done': '[]', 'critique_joke': '[]', 'generate_joke': '[]'}, 'stepwise': False, 'events_buffer': {}, 'in_progress': {'generate_joke': []}, 'accepted_events': [('generate_joke', 'StartEvent'), ('critique_joke', 'JokeEvent')], 'broker_log': ['{"__is_pydantic": true, "value": {"_data": {"topic": "chemistry", "store_checkpoints": false}}, "qualified_name": "llama_index.core.workflow.events.StartEvent"}'], 'is_running': True}),
  Checkpoint(id_='288c3e54-292b-4c7e-aed8-662537508b46', last_completed_step='critique_joke', input_event=JokeEvent(joke="Why do chemists like nitrates so much?\n\nBecause they're cheaper than day rates!"), output_event=StopEvent(result='This joke plays on the double meaning of the word "rates," which can refer to both the cost of something and the passage of time. The punchline suggests that chemists prefer nitrates because they are less expensive than day rates, implying that chemists are frugal or cost-conscious individuals.\n\nOverall, the joke is clever and plays on a pun that is likely to be appreciated by those familiar with chemistry and the concept of nitrates. However, the humor may be lost on those who are not well-versed in chemistry terminology. Additionally, the joke relies on a somewhat simplistic play on words, which may not be as engaging or humorous to some audiences.\n\nIn terms of structure, the joke follows a classic setup and punchline format, with the punchline providing a surprising twist on the initial premise. The delivery of the joke may also play a role in its effectiveness, as timing and tone can greatly impact the humor of a joke.\n\nOverall, while the joke may appeal to a specific audience and demonstrate some clever wordplay, it may not have universal appeal and may be considered somewhat niche in its humor.'), ctx_state={'globals': {}, 'streaming_queue': '[]', 'queues': {'_done': '[]', 'critique_joke': '[]', 'generate_joke': '[]'}, 'stepwise': False, 'events_buffer': {}, 'in_progress': {'generate_joke': [], 'critique_joke': []}, 'accepted_events': [('generate_joke', 'StartEvent'), ('critique_joke', 'JokeEvent')], 'broker_log': ['{"__is_pydantic": true, "value": {"_data": {"topic": "chemistry", "store_checkpoints": false}}, "qualified_name": "llama_index.core.workflow.events.StartEvent"}', '{"__is_pydantic": true, "value": {"joke": "Why do chemists like nitrates so much?\\n\\nBecause they\'re cheaper than day rates!"}, "qualified_name": "__main__.JokeEvent"}'], 'is_running': True})]}

for run_id, ckpts in wflow_ckptr.checkpoints.items():
    print(f"Run: {run_id} has {len(ckpts)} stored checkpoints")

Run: 483eccdd-a035-42cc-b596-cd33d42938a7 has 2 stored checkpoints

筛选检查点

WorkflowCheckpointer 对象还具有一个 .filter_checkpoints() 方法，允许我们通过以下方式筛选：

通过指定参数 last_completed_step 来获取最后完成的步骤名称
通过指定 output_event_type 来获取最后完成步骤的输出事件类型
同样地，通过指定 input_event_type 来获取最后完成步骤的输入事件类型

指定多个此类筛选条件将通过“AND”运算符进行组合。

让我们测试一下这个功能，但首先我们会通过使用Workflow进行几次额外运行来让事情变得更有趣。

additional_topics = ["biology", "history"]

for topic in additional_topics:
    handler = wflow_ckptr.run(topic=topic)
    await handler

for run_id, ckpts in wflow_ckptr.checkpoints.items():
    print(f"Run: {run_id} has {len(ckpts)} stored checkpoints")

Run: 483eccdd-a035-42cc-b596-cd33d42938a7 has 2 stored checkpoints
Run: e112bca9-637c-4492-a8aa-926c302c99d4 has 2 stored checkpoints
Run: 7a59c918-90a3-47f8-a818-71e45897ae39 has 2 stored checkpoints

# Filter by the name of last completed step
checkpoints_right_after_generate_joke_step = wflow_ckptr.filter_checkpoints(
    last_completed_step="generate_joke",
)

# checkpoint ids
[ckpt for ckpt in checkpoints_right_after_generate_joke_step]

[Checkpoint(id_='d5acd098-47e2-4acf-9520-9ca06ee4e238', last_completed_step='generate_joke', input_event=StartEvent(), output_event=JokeEvent(joke="Why do chemists like nitrates so much?\n\nBecause they're cheaper than day rates!"), ctx_state={'globals': {}, 'streaming_queue': '[]', 'queues': {'_done': '[]', 'critique_joke': '[]', 'generate_joke': '[]'}, 'stepwise': False, 'events_buffer': {}, 'in_progress': {'generate_joke': []}, 'accepted_events': [('generate_joke', 'StartEvent'), ('critique_joke', 'JokeEvent')], 'broker_log': ['{"__is_pydantic": true, "value": {"_data": {"topic": "chemistry", "store_checkpoints": false}}, "qualified_name": "llama_index.core.workflow.events.StartEvent"}'], 'is_running': True}),
 Checkpoint(id_='87865641-14e7-4eb0-bb62-4c211567acfc', last_completed_step='generate_joke', input_event=StartEvent(), output_event=JokeEvent(joke="Why did the biologist break up with the mathematician?\n\nBecause they couldn't find a common denominator!"), ctx_state={'globals': {}, 'streaming_queue': '[]', 'queues': {'_done': '[]', 'critique_joke': '[]', 'generate_joke': '[]'}, 'stepwise': False, 'events_buffer': {}, 'in_progress': {'generate_joke': []}, 'accepted_events': [('generate_joke', 'StartEvent'), ('critique_joke', 'JokeEvent')], 'broker_log': ['{"__is_pydantic": true, "value": {"_data": {"topic": "biology"}}, "qualified_name": "llama_index.core.workflow.events.StartEvent"}'], 'is_running': True}),
 Checkpoint(id_='69a99535-d45c-46b4-a1f4-d3ecc128cb08', last_completed_step='generate_joke', input_event=StartEvent(), output_event=JokeEvent(joke='Why did the history teacher go to the beach?\n\nTo catch some waves of the past!'), ctx_state={'globals': {}, 'streaming_queue': '[]', 'queues': {'_done': '[]', 'critique_joke': '[]', 'generate_joke': '[]'}, 'stepwise': False, 'events_buffer': {}, 'in_progress': {'generate_joke': []}, 'accepted_events': [('generate_joke', 'StartEvent'), ('critique_joke', 'JokeEvent')], 'broker_log': ['{"__is_pydantic": true, "value": {"_data": {"topic": "history"}}, "qualified_name": "llama_index.core.workflow.events.StartEvent"}'], 'is_running': True})]

从特定检查点重新运行工作流

要从选定的 Checkpoint 运行，我们可以使用 WorkflowCheckpointer.run_from() 方法。请注意，这样做将创建一个新的 run，如果启用了检查点功能，其检查点将存储在新分配的 run_id 下。

# can work with a new instance
new_workflow_instance = JokeFlow()
wflow_ckptr.workflow = new_workflow_instance

ckpt = checkpoints_right_after_generate_joke_step[0]

handler = wflow_ckptr.run_from(checkpoint=ckpt)
await handler

'Analysis:\nThis joke plays on the double meaning of the word "rates," which can refer to both the cost of something and the passage of time. In this case, the joke suggests that chemists prefer nitrates because they are less expensive than day rates, implying that chemists are frugal or cost-conscious individuals.\n\nCritique:\n- Clever wordplay: The joke relies on a clever play on words, which can be entertaining for those who appreciate puns and linguistic humor.\n- Niche audience: The humor in this joke may be more appreciated by individuals with a background in chemistry or a specific interest in science, as the punchline relies on knowledge of chemical compounds.\n- Lack of universal appeal: The joke may not resonate with a general audience who may not understand the reference to nitrates or the significance of their cost compared to day rates.\n- Lack of depth: While the joke is amusing on a surface level, it may be considered somewhat shallow or simplistic compared to more nuanced or thought-provoking humor.\n\nOverall, the joke is a light-hearted play on words that may appeal to individuals with a specific interest in chemistry or wordplay. However, its niche appeal and lack of universal relevance may limit its effectiveness as a joke for a broader audience.'

for run_id, ckpts in wflow_ckptr.checkpoints.items():
    print(f"Run: {run_id} has {len(ckpts)} stored checkpoints")

Run: 483eccdd-a035-42cc-b596-cd33d42938a7 has 2 stored checkpoints
Run: e112bca9-637c-4492-a8aa-926c302c99d4 has 2 stored checkpoints
Run: 7a59c918-90a3-47f8-a818-71e45897ae39 has 2 stored checkpoints
Run: 9dccfda3-b5bf-4771-8293-efd3e3a275a6 has 1 stored checkpoints

由于我们是从代表“生成笑话”步骤结束的检查点开始执行的，因此只有一个额外的检查点（即完成“评论笑话”步骤的检查点）存储在最后一个部分运行中。

指定需要检查点的步骤

默认情况下，附加工作流的所有步骤（不包括“_done”步骤）都将被设置检查点。您可以通过 enabled_checkpoints 属性查看哪些步骤启用了检查点功能。

wflow_ckptr.enabled_checkpoints

{'critique_joke', 'generate_joke'}

要禁用某个步骤的检查点功能，我们可以使用 .disable_checkpoint() 方法

wflow_ckptr.disable_checkpoint(step="critique_joke")

handler = wflow_ckptr.run(topic="cars")
await handler

"Analysis:\nThis joke plays on the common stereotype that mechanics are overly critical and nitpicky when it comes to fixing cars. The humor comes from the unexpected twist of the car being the one to break up with the mechanic, rather than the other way around. The punchline is clever and plays on the idea of a relationship ending due to constant criticism.\n\nCritique:\nOverall, this joke is light-hearted and easy to understand. It relies on a simple pun and doesn't require much thought to appreciate. However, the humor may be seen as somewhat predictable and not particularly original. The joke also perpetuates the stereotype of mechanics being overly critical, which may not sit well with some people in the automotive industry. Additionally, the joke may not be as universally funny as some other jokes, as it relies on a specific understanding of the relationship between cars and mechanics. Overall, while the joke is amusing, it may not be considered particularly groundbreaking or memorable."

for run_id, ckpts in wflow_ckptr.checkpoints.items():
    print(
        f"Run: {run_id} has stored checkpoints for steps {[c.last_completed_step for c in ckpts]}"
    )

Run: 483eccdd-a035-42cc-b596-cd33d42938a7 has stored checkpoints for steps ['generate_joke', 'critique_joke']
Run: e112bca9-637c-4492-a8aa-926c302c99d4 has stored checkpoints for steps ['generate_joke', 'critique_joke']
Run: 7a59c918-90a3-47f8-a818-71e45897ae39 has stored checkpoints for steps ['generate_joke', 'critique_joke']
Run: 9dccfda3-b5bf-4771-8293-efd3e3a275a6 has stored checkpoints for steps ['critique_joke']
Run: 6390e2ce-63f4-44a3-8a75-64ccb765abfd has stored checkpoints for steps ['generate_joke']

我们可以通过使用 .enable_checkpoint() 方法重新启用检查点功能

wflow_ckptr.enable_checkpoint(step="critique_joke")

handler = wflow_ckptr.run(topic="cars")
await handler

'Analysis:\nThis joke plays on the common stereotype that mechanics are overly critical and nitpicky when it comes to fixing cars. The humor comes from the unexpected twist of the car being the one to break up with the mechanic, rather than the other way around. The joke also cleverly uses the term "nitpicking" in a literal sense, as in picking at the nit (small details) of the car.\n\nCritique:\nWhile the joke is clever and plays on a well-known stereotype, it may not be the most original or groundbreaking joke. The punchline is somewhat predictable and relies on a common trope about mechanics. Additionally, the joke may not be universally relatable or understood by all audiences, as it requires some knowledge of the stereotype about mechanics being nitpicky.\n\nOverall, the joke is light-hearted and humorous, but it may not be the most memorable or impactful joke due to its reliance on a common stereotype. It could be improved by adding a more unexpected or unique twist to the punchline.'

for run_id, ckpts in wflow_ckptr.checkpoints.items():
    print(
        f"Run: {run_id} has stored checkpoints for steps {[c.last_completed_step for c in ckpts]}"
    )

Run: 483eccdd-a035-42cc-b596-cd33d42938a7 has stored checkpoints for steps ['generate_joke', 'critique_joke']
Run: e112bca9-637c-4492-a8aa-926c302c99d4 has stored checkpoints for steps ['generate_joke', 'critique_joke']
Run: 7a59c918-90a3-47f8-a818-71e45897ae39 has stored checkpoints for steps ['generate_joke', 'critique_joke']
Run: 9dccfda3-b5bf-4771-8293-efd3e3a275a6 has stored checkpoints for steps ['critique_joke']
Run: 6390e2ce-63f4-44a3-8a75-64ccb765abfd has stored checkpoints for steps ['generate_joke']
Run: e3291623-6eb8-43c6-b102-dc6f88a42f4d has stored checkpoints for steps ['generate_joke', 'critique_joke']