cfc/cfc-langgraph/app/checkpointer.py

"""
共享的 LangGraph checkpointer 工厂。

原实现使用 MemorySaver（纯内存），服务重启即丢失所有对话状态。
这里改用 SQLite 持久化（SqliteSaver），对话状态落盘，重启不丢。

说明：LangGraph 官方 RedisSaver 依赖 RedisSearch 模块（FT.* 命令），
当前 Redis 未安装该模块，故采用 SQLite 持久化作为务实替代。
"""
import logging
import os
import sqlite3
import threading

from langgraph.checkpoint.sqlite import SqliteSaver

logger = logging.getLogger(__name__)

# SQLite 数据库文件路径：
# - Docker 容器内 /data 为持久化卷（与 Chroma 同级）
# - 本地开发 fallback 到项目 data/ 目录
# - 可用环境变量 CHECKPOINT_DB_PATH 覆盖
_DEFAULT_DB_PATH = os.path.join(
    os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
    "data",
    "checkpoints.db",
)

# 线程本地存储：每个线程维护独立的 SqliteSaver 实例
# check_same_thread=False 允许跨线程共享 sqlite3 连接
# SqliteSaver 内部有自己的 threading.Lock() 保证序列化写入
_local = threading.local()


def get_checkpointer() -> SqliteSaver:
    """获取当前线程的 SqliteSaver 实例（线程安全）。"""
    db_path = os.environ.get(
        "CHECKPOINT_DB_PATH",
        "/data/checkpoints.db" if os.path.isdir("/data") else _DEFAULT_DB_PATH,
    )
    os.makedirs(os.path.dirname(db_path), exist_ok=True)

    if not hasattr(_local, "saver"):
        conn = sqlite3.connect(db_path, check_same_thread=False)
        saver = SqliteSaver(conn)
        saver.setup()
        _local.saver = saver
        logger.info("LangGraph checkpointer 已初始化: %s", db_path)

    return _local.saver