工作流运行检查点¶

在本 Notebook 中，我们将演示如何通过 WorkflowCheckpointer 对象对 Workflow 运行进行检查点。我们还将展示如何查看此对象中存储的所有检查点，以及如何使用检查点作为新运行的起点。

定义一个工作流¶

In [ ]

import os

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

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

In [ ]

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 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))

定义一个 WorkflowCheckpointer 对象¶

In [ ]

from llama_index.core.workflow.checkpointer import WorkflowCheckpointer

from llama_index.core.workflow.checkpointer import WorkflowCheckpointer

In [ ]

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

# 实例化 Jokeflow workflow = JokeFlow() wflow_ckptr = WorkflowCheckpointer(workflow=workflow)

从 WorkflowCheckpointer 运行工作流¶

WorkflowCheckpointer.run() 方法是 Workflow.run() 方法的包装器，它注入了一个检查点回调，用于创建和存储检查点。请注意，检查点是在步骤完成后创建的，并且存储在检查点中的数据包括：

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

In [ ]

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

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

Out [ ]

'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，值是为该运行存储的检查点列表。

In [ ]

wflow_ckptr.checkpoints

wflow_ckptr.checkpoints

Out [ ]

{'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})]}

In [ ]

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

for run_id, ckpts in wflow_ckptr.checkpoints.items(): print(f"运行: {run_id} 存储了 {len(ckpts)} 个检查点")

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

过滤检查点¶

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

• 通过指定参数 last_completed_step 来按最后一个已完成步骤的名称过滤
• 通过指定 output_event_type 来按最后一个已完成步骤的输出事件类型过滤
• 类似地，通过指定 input_event_type 来按最后一个已完成步骤的输入事件类型过滤

指定多个过滤器将通过“AND”运算符组合。

让我们测试此功能，但首先我们将通过运行几次 Workflow 来让事情变得更有趣一些。

In [ ]

additional_topics = ["biology", "history"]

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

additional_topics = ["biology", "history"] for topic in additional_topics: handler = wflow_ckptr.run(topic=topic) await handler

In [ ]

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

for run_id, ckpts in wflow_ckptr.checkpoints.items(): print(f"运行: {run_id} 存储了 {len(ckpts)} 个检查点")

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

In [ ]

# 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]

# 按最后一个已完成步骤的名称过滤 checkpoints_right_after_generate_joke_step = wflow_ckptr.filter_checkpoints( last_completed_step="generate_joke", ) # 检查点 ID [ckpt for ckpt in checkpoints_right_after_generate_joke_step]

Out [ ]

[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 下。

In [ ]

# 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

# 可以使用新的实例 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

Out [ ]

'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.'

In [ ]

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

for run_id, ckpts in wflow_ckptr.checkpoints.items(): print(f"运行: {run_id} 存储了 {len(ckpts)} 个检查点")

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

由于我们从代表“generate_joke”步骤结束的检查点开始执行，因此在上次部分运行中仅存储了一个额外的检查点（即用于“critique_joke”步骤完成的检查点）。

指定要创建检查点的步骤¶

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

In [ ]

wflow_ckptr.enabled_checkpoints

wflow_ckptr.enabled_checkpoints

Out [ ]

{'critique_joke', 'generate_joke'}

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

In [ ]

wflow_ckptr.disable_checkpoint(step="critique_joke")

wflow_ckptr.disable_checkpoint(step="critique_joke")

In [ ]

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

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

Out [ ]

"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."

In [ ]

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]}"
    )

for run_id, ckpts in wflow_ckptr.checkpoints.items(): print( f"运行: {run_id} 存储了步骤 {[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() 方法重新启用检查点功能

In [ ]

wflow_ckptr.enable_checkpoint(step="critique_joke")

wflow_ckptr.enable_checkpoint(step="critique_joke")

In [ ]

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

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

Out [ ]

'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.'

In [ ]

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]}"
    )

for run_id, ckpts in wflow_ckptr.checkpoints.items(): print( f"运行: {run_id} 存储了步骤 {[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']