所有结构化状态的基类，只强制一件事：有唯一 id（runtime/__init__.py:360）：

# runtime/__init__.py:360
class FlowState(BaseModel):
    """Base model for all flow states, ensuring each state has a unique ID."""
    id: str = Field(
        default_factory=lambda: str(uuid4()),   # 不给就自动生成一个 UUID
        description="Unique identifier for the flow state")

# runtime/__init__.py:1659  你写 self.state 时读到的
@property
def state(self) -> T:
    return cast(T, self._state)

状态初始化要兼顾多种写法，核心分支（runtime/__init__.py:1531）：

# runtime/__init__.py:1588（节选）
if isinstance(init_state, dict):
    new_state = dict(init_state)          # 拷贝一份，避免改到类属性
    if "id" not in new_state:
        new_state["id"] = str(uuid4())    # dict 也要补 id
    return cast(T, new_state)

if isinstance(init_state, BaseModel):
    model = init_state
    if hasattr(model, "id"):
        state_dict = model.model_dump()
        if not state_dict.get("id"):
            state_dict["id"] = str(uuid4())
        return cast(T, type(model)(**state_dict))
    # 模型没 id → 动态混入 FlowState 补上 id
    class StateWithId(FlowState, type(model)):
        pass
    state_dict = model.model_dump(); state_dict["id"] = str(uuid4())
    return cast(T, StateWithId(**state_dict))

@persist 本身不干活，只往类/方法上盖个持久化配置的章（flow/persistence/decorators.py:147）：

# flow/persistence/decorators.py:183
def decorator(target: type | Callable[..., T]) -> type | Callable[..., T]:
    actual_persistence = (
        persistence if persistence is not None else default_flow_persistence()  # 默认 SQLite
    )
    _stamp_persistence_metadata(target, actual_persistence, verbose)  # 盖章
    return target

# :56  盖的章长这样
def _stamp_persistence_metadata(target, persistence, verbose) -> None:
    target.__flow_persistence_config__ = SimpleNamespace(
        persistence=persistence, verbose=verbose)

D41 的 _execute_method 里，方法一完成就落盘（runtime/__init__.py:2610）：

# runtime/__init__.py:2608
self._completed_methods.add(method_name)
await asyncio.to_thread(self._persist_method_completion, method_name)  # ★存盘（放线程池，不阻塞事件循环）

# runtime/__init__.py:2668
def _persist_method_completion(self, method_name) -> None:
    method_definition = self._definition.methods[str(method_name)]
    persist_definition = (
        method_definition.persist                # 方法级优先
        if method_definition.persist is not None
        else self._definition.persist)           # 否则用流程级
    if persist_definition is None or not persist_definition.enabled:
        return                                   # 没配持久化 → 直接返回，什么都不存
    from crewai.flow.persistence.decorators import PersistenceDecorator
    backend = (self.persistence
        if self._instance_persistence and self.persistence is not None
        else self._persist_backend_for(persist_definition))
    PersistenceDecorator.persist_state(self, method_name, backend, verbose=persist_definition.verbose)

持久化后端是抽象基类，SQLite / Postgres / 内存都实现它（flow/persistence/base.py:18）：

# flow/persistence/base.py:18
class FlowPersistence(BaseModel, ABC):
    persistence_type: str = Field(default="base")

    def __init_subclass__(cls, **kwargs) -> None:      # 子类自动注册
        super().__init_subclass__(**kwargs)
        if not getattr(cls, "__abstractmethods__", set()):
            _persistence_registry[cls.__name__] = cls

    @abstractmethod
    def init_db(self) -> None: ...                     # 建表/连库
    @abstractmethod
    def save_state(self, flow_uuid, method_name, state_data) -> None: ...  # 存
    @abstractmethod
    def load_state(self, flow_uuid) -> dict[str, Any] | None: ...          # 取最新

    # 非抽象：async 人在环用（默认实现 = 直接存/不支持）
    def save_pending_feedback(self, flow_uuid, context, state_data) -> None:
        self.save_state(flow_uuid, context.method_name, state_data)

建表（flow/persistence/sqlite.py:71）——注意它是追加式历史表：

# flow/persistence/sqlite.py:78
conn.execute("PRAGMA journal_mode=WAL")               # 并发友好
conn.execute("""
CREATE TABLE IF NOT EXISTS flow_states (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    flow_uuid TEXT NOT NULL,      -- = state.id
    method_name TEXT NOT NULL,    -- 哪个方法完成时存的
    timestamp DATETIME NOT NULL,
    state_json TEXT NOT NULL      -- 整个 state 序列化成 JSON
)""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_flow_states_uuid ON flow_states(flow_uuid)")

取状态时取同 uuid 最新的一行（flow/persistence/sqlite.py:178）：

-- flow/persistence/sqlite.py:189
SELECT state_json FROM flow_states
WHERE flow_uuid = ?
ORDER BY id DESC        -- 自增 id 最大 = 最近一次
LIMIT 1

恢复流程：kickoff(inputs={"id": 已有uuid}) 时，引擎 load_state 回来、进入"恢复模式"，跳过已完成方法（runtime/__init__.py:2066）：

# runtime/__init__.py:2066
is_restoring = (inputs and "id" in inputs and self.persistence is not None) \
               or self._restored_from_checkpoint
if not is_restoring:
    self._completed_methods.clear()      # 全新执行：清空进度
    self._method_outputs.clear(); ...
else:
    if self._completed_methods:
        self._is_execution_resuming = True   # ★恢复模式：已完成的方法会被跳过（D44 见过这个标志）