Flow 持久化与 HITL
学习目标
理解 CrewAI Flow 如何通过 @persist 装饰器实现方法级状态快照，SQLite 追加写入保留完整审计链，以及人机交互（HITL）的异步暂停/恢复机制。
项目实践
@persist 装饰器
@persist 可应用于类级别或方法级别：
- 类级别：自动包装所有 Flow 方法，在每个方法完成后保存状态
- 方法级别：仅在被装饰的方法完成后保存
装饰器工作原理：
- 创建
method_sync_wrapper/method_async_wrapper - 在原始方法执行之后调用
PersistenceDecorator.persist_state() - 保留所有原始装饰器属性（
__is_start_method__、__trigger_methods__等）
SQLite 持久化
SQLiteFlowPersistence 使用两个表：
| 表 | 用途 |
|---|---|
flow_states | 主状态表，每个方法执行后 INSERT 一条记录 |
pending_feedback | 异步 HITL 专用表，存储暂停时的状态和上下文 |
设计决策：
- 追加写入：不更新行，而是 INSERT 新行，保留完整的历史审计链
- WAL 模式：
PRAGMA journal_mode=WAL提高并发读性能 - 文件锁：通过
store_lock避免多进程竞争
Human-in-the-Loop（HITL）
同步模式：_request_human_feedback() 通过 input() 阻塞等待用户输入。
异步模式：抛出 HumanFeedbackPending 异常，状态保存到持久化层，后续通过 Flow.from_pending(flow_id) + flow.resume(feedback) 恢复。
LLM 结果折叠：_collapse_to_outcome() 使用 LLM 将自由文本反馈映射到预定义的 outcome 列表（如 “approved”/“rejected”）。
用户输入
self.ask(message, timeout) 方法：
- 支持超时，返回
None - 自动 checkpoint 状态
- 可配置
InputProvider（console、Slack 等）
问题与规避
持久化要求 state 有 id 字段
- 如果 state 没有
id字段，@persist会抛ValueError - 对策：在 State 类中定义
id: str字段
SQLite 文件大小增长
- 追加写入策略导致文件随方法调用次数线性增长
- 对策：定期清理旧的 flow_states 记录，或设置
max_checkpoints自动修剪
HITL 异步恢复的竞态
- 如果多个进程同时尝试恢复同一个 pending flow
- 对策：
pending_feedback表的claim()操作使用 SQLite 事务保证原子性
设计取舍
追加写入 vs 覆盖写入
- 追加写入保留完整历史，支持审计和调试
- 代价是存储成本随方法调用次数线性增长
- 对于生产环境，建议配合定期清理策略
同步 input() vs 异步异常
- 同步模式适合 CLI 场景，用户直接在终端输入
- 异步异常模式适合 Web/Slack 等需要外部回调的场景
- CrewAI 同时支持两种模式
参考来源
- 源码验证:
lib/crewai/src/crewai/flow/persistence/decorators.py - 源码验证:
lib/crewai/src/crewai/flow/persistence/sqlite.py - 源码验证:
lib/crewai/src/crewai/flow/persistence/base.py