Jelajahi Sumber

fix(langgraph): checkpointer 改用 AsyncSqliteSaver 修复 chat 端点异步调用报错

报告快速分析走 /api/v1/chat,其 chat_graph 节点为 async def 且用 graph.ainvoke(),
但 checkpointer 原用同步 SqliteSaver,LangGraph 抛 'SqliteSaver does not support
async methods',导致后端 AiGateway 收到 500 返回 null,前端提示 'AI 服务暂不可用'。

- checkpointer.py: 改用 AsyncSqliteSaver,新增 init_checkpointer() 在 startup 事件
  循环内初始化(AsyncSqliteSaver 绑定 get_running_loop,不能在临时 loop 创建)
- main.py: startup 调用 await init_checkpointer()
- requirements.txt / requirements_frozen.txt: 新增 aiosqlite 依赖
Sisyphus 1 bulan lalu
induk
melakukan
55c00ba68c

+ 46 - 22
cfc-langgraph/app/checkpointer.py

@@ -2,22 +2,25 @@
 共享的 LangGraph checkpointer 工厂。
 共享的 LangGraph checkpointer 工厂。
 
 
 原实现使用 MemorySaver(纯内存),服务重启即丢失所有对话状态。
 原实现使用 MemorySaver(纯内存),服务重启即丢失所有对话状态。
-这里改用 SQLite 持久化(SqliteSaver),对话状态落盘,重启不丢。
+这里改用 SQLite 持久化(AsyncSqliteSaver),支持异步调用,对话状态落盘,重启不丢。
 
 
 说明:LangGraph 官方 RedisSaver 依赖 RedisSearch 模块(FT.* 命令),
 说明:LangGraph 官方 RedisSaver 依赖 RedisSearch 模块(FT.* 命令),
 当前 Redis 未安装该模块,故采用 SQLite 持久化作为务实替代。
 当前 Redis 未安装该模块,故采用 SQLite 持久化作为务实替代。
+
+重要:AsyncSqliteSaver 内部绑定 asyncio.Lock + 当前事件循环(get_running_loop),
+必须在 FastAPI 的 running loop 内初始化(app startup 时调用 init_checkpointer),
+不能在临时 loop 中创建,否则后续 ainvoke 会报 "attached to a different loop"。
 """
 """
 import logging
 import logging
 import os
 import os
-import sqlite3
-import threading
 
 
-from langgraph.checkpoint.sqlite import SqliteSaver
+import aiosqlite
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
 
 
 logger = logging.getLogger(__name__)
 logger = logging.getLogger(__name__)
 
 
 # SQLite 数据库文件路径:
 # SQLite 数据库文件路径:
-# - Docker 容器内 /data 为持久化卷(与 Chroma 同级)
+# - Docker 容器内 /data 为持久化卷(与 Chroma 同级),但需检查写权限
 # - 本地开发 fallback 到项目 data/ 目录
 # - 本地开发 fallback 到项目 data/ 目录
 # - 可用环境变量 CHECKPOINT_DB_PATH 覆盖
 # - 可用环境变量 CHECKPOINT_DB_PATH 覆盖
 _DEFAULT_DB_PATH = os.path.join(
 _DEFAULT_DB_PATH = os.path.join(
@@ -26,25 +29,46 @@ _DEFAULT_DB_PATH = os.path.join(
     "checkpoints.db",
     "checkpoints.db",
 )
 )
 
 
-# 线程本地存储:每个线程维护独立的 SqliteSaver 实例
-# check_same_thread=False 允许跨线程共享 sqlite3 连接
-# SqliteSaver 内部有自己的 threading.Lock() 保证序列化写入
-_local = threading.local()
+# 全局单例(由 init_checkpointer 在 startup 时填充)
+_saver: AsyncSqliteSaver | None = None
+
+
+def _resolve_db_path() -> str:
+    """解析数据库路径:优先环境变量,其次有写权限的目录。"""
+    # 1. 环境变量覆盖
+    env_path = os.environ.get("CHECKPOINT_DB_PATH")
+    if env_path:
+        return env_path
+
+    # 2. 尝试 /data(容器卷),需检查写权限
+    if os.path.isdir("/data") and os.access("/data", os.W_OK):
+        return "/data/checkpoints.db"
 
 
+    # 3. fallback 到项目 data 目录
+    return _DEFAULT_DB_PATH
 
 
-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,
-    )
+
+async def init_checkpointer() -> AsyncSqliteSaver:
+    """在运行中的事件循环内初始化 AsyncSqliteSaver(app startup 时调用)。"""
+    global _saver
+    if _saver is not None:
+        return _saver
+
+    db_path = _resolve_db_path()
     os.makedirs(os.path.dirname(db_path), exist_ok=True)
     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)
+    conn = await aiosqlite.connect(db_path)
+    saver = AsyncSqliteSaver(conn)
+    await saver.setup()
+    _saver = saver
+    logger.info("LangGraph AsyncSqliteSaver 已初始化: %s", db_path)
+    return _saver
+
 
 
-    return _local.saver
+def get_checkpointer() -> AsyncSqliteSaver:
+    """获取全局 AsyncSqliteSaver(须已通过 init_checkpointer 在 startup 初始化)。"""
+    if _saver is None:
+        raise RuntimeError(
+            "checkpointer 未初始化,请确保 app startup 已调用 init_checkpointer()"
+        )
+    return _saver

+ 4 - 0
cfc-langgraph/app/main.py

@@ -57,6 +57,10 @@ async def startup():
     retriever = RagRetriever()
     retriever = RagRetriever()
     await retriever.initialize()
     await retriever.initialize()
 
 
+    # 初始化 LangGraph checkpointer(必须在运行中的事件循环内创建 AsyncSqliteSaver)
+    from app.checkpointer import init_checkpointer
+    await init_checkpointer()
+
     async def schedule_kb_sync():
     async def schedule_kb_sync():
         # 首次同步延迟 120s,让 healthcheck 先通过,避免 worker 刚启动就被阻塞
         # 首次同步延迟 120s,让 healthcheck 先通过,避免 worker 刚启动就被阻塞
         await asyncio.sleep(120)
         await asyncio.sleep(120)

+ 1 - 0
cfc-langgraph/requirements.txt

@@ -2,6 +2,7 @@ fastapi>=0.110,<1.0
 uvicorn[standard]>=0.29,<1.0
 uvicorn[standard]>=0.29,<1.0
 langgraph>=0.2,<1.0
 langgraph>=0.2,<1.0
 langgraph-checkpoint-sqlite==3.0.3
 langgraph-checkpoint-sqlite==3.0.3
+aiosqlite>=0.19,<1.0
 langchain-openai>=0.2,<1.0
 langchain-openai>=0.2,<1.0
 langchain-core>=0.3,<1.0
 langchain-core>=0.3,<1.0
 chromadb==0.5.23
 chromadb==0.5.23

+ 2 - 0
cfc-langgraph/requirements_frozen.txt

@@ -1,6 +1,7 @@
 aiohappyeyeballs==2.7.1
 aiohappyeyeballs==2.7.1
 aiohttp==3.14.2
 aiohttp==3.14.2
 aiosignal==1.4.0
 aiosignal==1.4.0
+aiosqlite==0.22.1
 annotated-doc==0.0.4
 annotated-doc==0.0.4
 annotated-types==0.7.0
 annotated-types==0.7.0
 anyio==4.14.2
 anyio==4.14.2
@@ -49,6 +50,7 @@ langchain-protocol==0.0.18
 langchain-text-splitters==0.3.11
 langchain-text-splitters==0.3.11
 langgraph==0.6.11
 langgraph==0.6.11
 langgraph-checkpoint==3.0.1
 langgraph-checkpoint==3.0.1
+langgraph-checkpoint-sqlite==3.0.3
 langgraph-prebuilt==0.6.5
 langgraph-prebuilt==0.6.5
 langgraph-sdk==0.2.15
 langgraph-sdk==0.2.15
 langsmith==0.10.9
 langsmith==0.10.9