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