Chapter 16 · Flow Advanced: State Management and Persistence

Chapter goals: Master the trade-offs between structured and unstructured state, learn to use @persist for state persistence and recovery (including restore_from_state_id branching), orchestrate multi-listener merge timing with or_ / and_, and use @router for conditional branching.

16.1 Revisiting Two State Types: Implementation Details

Chapter 15 introduced dictionary-style (unstructured) and Pydantic-style (structured) state. This section fills in the engineering details:

# state_ops.py — complete state operations comparison
from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, start

class OrderState(BaseModel):
    order_id: str = ""
    amount: float = 0.0
    risk_level: str = ""

class UnstructuredFlow(Flow):
    @start()
    def load(self):
        # Dictionary style: keys freely added/deleted; auto carries an id key (UUID)
        print(f"State ID: {self.state['id']}")
        self.state["order_id"] = "A-1024"
        self.state["amount"] = 199.0

class StructuredFlow(Flow[OrderState]):
    @start()
    def load(self):
        # Pydantic style: attribute access; typos raise errors immediately
        self.state.order_id = "A-1024"
        self.state.amount = 199.0
        print(f"State ID: {self.state.id}")   # id is also auto-managed

Selection criteria (official recommendation): simple or highly dynamic structures, prototyping → dictionary; consistent structure, type safety, IDE validation → Pydantic. For team projects, default to Pydantic — it surfaces "typoed field name" bugs at edit time rather than deep in runtime.

16.2 @persist: Making Flows Survive Process Restart

If a long pipeline crashes or restarts mid-run, starting over is both expensive and time-consuming. The @persist decorator auto-stores state into a local SQLite database (default backend: SQLiteFlowPersistence), and restores it on the next startup:

# persist_flow.py — class-level persistence
from pydantic import BaseModel
from crewai.flow.flow import Flow, start, listen
from crewai.flow.persistence import persist

class CounterState(BaseModel):
    counter: int = 0

@persist                       # Class-level: all methods' state gets persisted
class CounterFlow(Flow[CounterState]):
    @start()
    def step(self):
        self.state.counter += 1
        print(f"[id={self.state.id}] counter={self.state.counter}")

# First run: counter 0 -> 1, snapshot written to SQLite
CounterFlow().kickoff()

You can also attach @persist only on key methods (method-level) for finer-grained control:

class AnotherFlow(Flow[dict]):
    @persist               # Only persist this method's state
    @start
    def begin(self):
        if "runs" not in self.state:
            self.state["runs"] = 0
        self.state["runs"] += 1
        print("Persisted run count:", self.state["runs"])

Recovery has two semantics (explicitly distinguished in the official docs):

• Resume: kickoff(inputs={"id": <uuid>}) — loads the latest snapshot for that UUID and continues appending history under the same flow_uuid;
• Fork: kickoff(restore_from_state_id=<uuid>) — initializes the new run's state from the old snapshot but assigns a new state.id; old and new histories remain independent.

# Fork example
flow_1 = CounterFlow()
flow_1.kickoff()                                   # counter -> 1

flow_2 = CounterFlow()
flow_2.kickoff(restore_from_state_id=flow_1.state.id)
# flow_2 starts from counter=1, then step() increments to 2;
# flow_2.state.id is a new ID; flow_1's history is unaffected.

Persistence caveats

• Both structured and dictionary state are supported; id field is auto-injected if missing;
• If restore_from_state_id can't find a matching snapshot, kickoff silently falls back to a fresh run — when troubleshooting "why it didn't restore," first confirm the UUID is correct;
• It cannot be used together with from_checkpoint (raises ValueError); pick one or the other.

16.3 or_ and and_: Merging Multi-Source Listeners

When multiple methods might trigger the same follow-up action, you need merge primitives:

# or_and_flow.py — or_ fires on any one, and_ waits for all
from crewai.flow.flow import Flow, listen, start, or_, and_

class AlertFlow(Flow):

    @start()
    def check_cpu(self):
        return "CPU OK"

    @listen(check_cpu)
    def check_disk(self):
        self.state["disk"] = "Disk alert"
        return "Disk alert"

    @listen(or_(check_cpu, check_disk))       # Fires when any source completes (may fire multiple times)
    def logger(self, result):
        print(f"[OR] log: {result}")

    @listen(and_(check_cpu, check_disk))      # Fires once after all sources complete
    def summary(self):
        print(f"[AND] summary: {self.state['disk']} / CPU checked")
        return "Inspection complete"

flow = AlertFlow()
print(flow.kickoff())

The output shows [OR] log printed twice (each time a source method completes), while [AND] summary runs only once after both sources finish. Rule of thumb:

• Log/audit side-channel actions → use or_ (every event must be recorded);
• Summary/decision actions → use and_ (must wait for all inputs);
• and_ listener methods do not accept a single return-value parameter (multiple sources can't map to one arg); pass data through state.

16.4 @router: Conditional Routing Branches

@router() turns a method's return value into a "signpost" that directs execution down different branches:

# router_flow.py — route approval flow by risk level
import random
from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, router, start

class RiskState(BaseModel):
    amount: float = 0.0
    success_flag: bool = False

class ApprovalFlow(Flow[RiskState]):

    @start()
    def submit_order(self):
        self.state.amount = random.choice([99.0, 99000.0])
        # Large orders flagged as high risk (in production, call a risk service here)
        self.state.success_flag = self.state.amount < 10000

    @router(submit_order)                 # Return value is the route label
    def route_by_risk(self):
        if self.state.success_flag:
            return "auto_approve"
        return "manual_review"

    @listen("auto_approve")
    def approve(self):
        print(f"Order {self.state.amount} auto-approved")
        return "approved"

    @listen("manual_review")
    def review(self):
        print(f"Order {self.state.amount} enters manual review queue")
        return "needs_human"

if __name__ == "__main__":
    ApprovalFlow().plot("approval_plot")   # plot() shows the branch diagram
    print(ApprovalFlow().kickoff())

Key point: the function body of @router(listened_method) is plain Python conditional logic; the returned string determines which branch to take; downstream uses @listen("label") to catch it. Paired with plot()-generated HTML diagrams, complex branching becomes instantly readable. Multi-level routers can be chained: a branch method can itself be listened to by another @router, forming a decision tree.

16.5 Checkpoint Resume Combination Play

Combining this chapter's capabilities gives you a production-grade fault-tolerance scheme:

1. Mark key-stage methods with @persist (or class-level persistence);
2. Wrap external calls in try/except; on failure, use @router to redirect the flow into a "degraded branch" instead of crashing outright;
3. After process restart, use inputs={"id": ...} to resume the same instance, or restore_from_state_id to spawn a replay from a snapshot;
4. Use the Chapter 14 event listeners (MethodExecutionStartedEvent / MethodExecutionFinishedEvent / FlowFailedEvent) to log each step's landing point, making it easy to determine which snapshot to recover from.

# resume_demo.py — failure degradation + resumable minimal skeleton
import os
from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, router, start
from crewai.flow.persistence import persist

class PipelineState(BaseModel):
    step_done: int = 0

@persist
class RobustFlow(Flow[PipelineState]):
    @start()
    def stage1(self):
        self.state.step_done = 1          # Persist as soon as complete
        return "s1_ok"

    @listen(stage1)
    def stage2(self):
        try:
            raise TimeoutError("Simulated external service timeout")
        except TimeoutError:
            self.state.step_done = 2
            return "s2_failed"            # Return failure label instead of raising

    @router(stage2)
    def route(self):
        return "retry_stage2"             # In production, consult state to decide retry or skip

    @listen("retry_stage2")
    def on_retry(self):
        print(f"Can retry after snapshot restore; current progress: step {self.state.step_done} completed")

RobustFlow().kickoff()
# After process crash:
# RobustFlow().kickoff(inputs={"id": "<state.id printed last run>"})

16.6 Chapter Summary

• Dictionary state wins on flexibility; Pydantic state wins on type safety and validation — default to the latter for team projects; both auto-carry a unique id;
• @persist supports class-level and method-level; default SQLite backend; inputs={"id":...} resumes the same instance, restore_from_state_id forks a new ID replay;
• or_ fires when any source completes (may fire multiple times), and_ fires once after all complete and data must go through state;
• @router turns a method's returned string into a route label; @listen("label") catches the branch; plot() visualizes the decision tree;
• Production fault tolerance = persistence + degraded branches + snapshot recovery + event logging combined.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. Regarding the trigger behavior of and_(a, b) listener methods, which is correct?

2. What is the semantics of kickoff(restore_from_state_id=<uuid>)?

3. What determines which branch @router sends execution down?

4. What happens if restore_from_state_id is given a non-existent uuid?

🛠️ Hands-On Exercises

1. Add @persist to the writing pipeline from Chapter 15; kill the process mid-run with Ctrl-C, then resume with inputs={"id": ...} to verify state continues.
2. Implement a three-way risk-control router: amount < 1000 auto-approve, < 50000 AI preliminary review, rest human; chain two-level routers with @router and output the flow diagram with plot().
3. Build a Flow that "parallels fetches two data sources + merges with and_"; intentionally make one source raise an exception, observe the flow hang, then refactor to a degraded version that returns a label even on failure.

The last piece of engineering is missing — scaffolding and configuration separation: Chapter 17 · CLI Engineering and YAML Configuration Projects.