[Feature] Checkpointing with Workflows

[nerdai]

Description

The additions/changes in this PR are made to enable a user to load and run a Workflow from a stored checkpoint from its past execution(s).

Round 4

Introduce WorkflowCheckpointer class that wraps a Workflow and creates and manages checkpoints.

• After some iterations on this, it seems that the best approach would be to have a dedicated class to handle checkpointing (i.e., outside of Workflow as well as Context)
• If in Context: the UX gets a bit awkward because you need to pass context of previous runs along to maintain the checkpoint history
• If in Workflow: then it becomes too attached to the instance -- you wouldn't be able to create/load checkpoints from different instances of the same Workflow

Sample Usage

from llama_index.core.workflow import WorkflowCheckpointer

wflow_ckptr = WorkflowCheckpointer(workflow=...)
handler = wflow_ckptr.run()
await handler

# see all checkpoints dict
print(wflow_ckptr.checkpoints)

# start from a given ckpt
ckpt = wflow_ckptr.filter_checkpoints(output_event=StartEvent)[0]
handler = wflow_ckptr.run_from(checkpoint=ckpt)
await handler

Enabling/Disabling Steps For Checkpointing

from llama_index.core.workflow import WorkflowCheckpointer

# by default all steps are enabled for checkpointing
wflow_ckptr = WorkflowCheckpointer(workflow=...)

# disable checkpointing for a step
step_name = ...
wflow_ckptr.disable_checkpoint(step=step_name)

# enable checkpointing for a step
step_name = ...
wflow_ckptr.enable_checkpoint(step=step_name)

Type of Change

Please delete options that are not relevant.

• New feature (non-breaking change which adds functionality)
• This change requires a documentation update

How Has This Been Tested?

Your pull-request will likely not be merged unless it is covered by some form of impactful unit testing.

• I added new unit tests to cover this change

[nerdai]

After discussing with @masci, we landed on the following:

• If we want mostly to run from a checkpoint, then we can (and probably should) just do this in Workflow itself
• have _broker_log maintain the Checkpoint -- ofc, the Checkpoint should contain all that it needs in order to be executed from it (events, Context states etc.)
• have a way to navigate/filter Checkpoint
• have a way to start from a Checkpoint

Will rework to this shape.

[review-notebook-app]

Check out this pull request on 

See visual diffs & provide feedback on Jupyter Notebooks.

---

Powered by ReviewNB

[masci]

LGTM overall, left a couple of comments!

[nerdai]

We need to keep track of steps that are in progress at the creation time of a checkpoint. Otherwise, since their input events have already been popped from their respective queues, we will lose these events if we attempt to load from a checkpoint.