How to save Flow state and restart from checkpoint?

Crews
,
1

I’m currently developing a flow in CrewAI composed of 4 sequential crews, each performing significant work that takes considerable time and tokens. To speed up my development, I’m trying to persist the state at certain checkpoints, allowing me to restart execution from these saved states instead of rerunning the entire flow every time I make a change.

I’m aware from the documentation (Flow State Lifecycle) that each state gets a unique ID, and there’s a @persist decorator available for saving states. However, I’m unclear about:

  1. How can I access or retrieve these specific flow state IDs after the flow has executed? (it seems there’s a CLI command ‘crewai log-tasks-outputs’ that shows IDs but only works for crews, not for flows).
  2. What’s the correct way to use the @persist decorator to store and later restore the flow state from the database? This documentation section ‘Mastering Flow State Management - CrewAI’ details how to persist, but not how to load back.

Any guidance, examples, or best practices on managing persisted states would be greatly appreciated! I’m guessing restoring a state is something we will need to implement independently for now (or am I missing something?)

2

If you try to run the example code that’s right here with version 0.105.0 of CrewAI, it won’t even work. I feel like the CrewAI dev team is so dedicated to implementing improvements that things can change very quickly, and the documentation doesn’t keep up with the pace of the changes. In this case, it’s best to dive into the source. Be brave! :flexed_biceps:

Looking at some things here in crewai/flow (version 0.105.0), I understood that we need to tie the identity of the flow that we are interested in keeping track of. In particular, the state must have the id field defined by you (if you don’t, the library will assign a new id each time it runs), and you must also pass the same id in the inputs dictionary of your flow. By doing this, your state is automatically saved and recovered each time it runs.

Here’s a fully functional code snippet, I hope it helps:

from crewai.flow.flow import Flow, listen, start
from crewai.flow.persistence import persist
from pydantic import BaseModel

from crewai.utilities.paths import db_storage_path
from pathlib import Path

print(
    '! DB default location is: '
    f'{str(Path(db_storage_path()) / "flow_states.db")}'
)

class CounterState(BaseModel):
    id: str = 'my-unique-id'
    value: int = 0

@persist(verbose=True)
class PersistentCounterFlow(Flow[CounterState]):
    @start()
    def increment(self):
        self.state.value += 1
        print(f"+ Incremented to {self.state.value}")
        return self.state.value

    @listen(increment)
    def double(self, value):
        self.state.value = value * 2
        print(f"x Doubled to {self.state.value}")
        return self.state.value

flow = PersistentCounterFlow()
result = flow.kickoff(
    inputs={
        'id': 'my-unique-id'
    }
)
print(f"= This run result: {result}")
1 Like
3

As @Max_Moura has clearly stated in his example code above you can control the flow state id by passing one in to your inputs.

This gives you control over how you can start a flow with the same state again. I usually like to save the whole state so I use the class @persist decorator which saves everything,

To start your flow on different stages depending on the flow, you could have checks that say, verify the flow method doesn’t have a method result assigned first before executing it. If it does skip and go to the next method. If done correctly this should work.

1 Like