How to save Flow state and restart from checkpoint?

Crews
,
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