feat: add sqlite checkpoint fallback switch
This commit is contained in:
1 parent
690d09a3d1
commit
f3def88342
4 files changed
+59
-9
No files matched your search
@@ -220,7 +220,11 @@ database:
|
|||||||
|
|
||||||
checkpointer:
|
checkpointer:
|
||||||
type: postgres
|
type: postgres
|
||||||
|
# true: 新 checkpoint 写入 Postgres;false: 保留 PG 配置但继续使用 SQLite。
|
||||||
|
# 也可临时用环境变量 DEER_FLOW_USE_PG_CHECKPOINTER=0 覆盖。
|
||||||
|
use_postgres: true
|
||||||
connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow
|
connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow
|
||||||
|
sqlite_connection_string: .deer-flow/data/deerflow.db
|
||||||
dual_read_enabled: true
|
dual_read_enabled: true
|
||||||
dual_read_sqlite_paths:
|
dual_read_sqlite_paths:
|
||||||
- .deer-flow/data/deerflow.db
|
- .deer-flow/data/deerflow.db
|
||||||
|
|||||||
+14
@@ -23,6 +23,20 @@ class CheckpointerConfig(BaseModel):
|
|||||||
"For sqlite, use a file path like '.deer-flow/checkpoints.db' or ':memory:' for in-memory. "
|
"For sqlite, use a file path like '.deer-flow/checkpoints.db' or ':memory:' for in-memory. "
|
||||||
"For postgres, use a DSN like 'postgresql://user:pass@localhost:5432/db'.",
|
"For postgres, use a DSN like 'postgresql://user:pass@localhost:5432/db'.",
|
||||||
)
|
)
|
||||||
|
use_postgres: bool = Field(
|
||||||
|
default=True,
|
||||||
|
description=(
|
||||||
|
"When type='postgres', set false to keep the Postgres DSN in config "
|
||||||
|
"but run the checkpointer on SQLite instead."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
sqlite_connection_string: str | None = Field(
|
||||||
|
default=None,
|
||||||
|
description=(
|
||||||
|
"SQLite file used when type='postgres' and use_postgres=false. "
|
||||||
|
"Defaults to the first dual_read_sqlite_paths entry or 'store.db'."
|
||||||
|
),
|
||||||
|
)
|
||||||
dual_read_enabled: bool = Field(
|
dual_read_enabled: bool = Field(
|
||||||
default=False,
|
default=False,
|
||||||
description=(
|
description=(
|
||||||
|
|||||||
+37
-9
@@ -48,6 +48,13 @@ def _dual_read_enabled(config) -> bool:
|
|||||||
return os.getenv("DEER_FLOW_CHECKPOINT_DUAL_READ", "").strip().lower() in {"1", "true", "yes", "on"}
|
return os.getenv("DEER_FLOW_CHECKPOINT_DUAL_READ", "").strip().lower() in {"1", "true", "yes", "on"}
|
||||||
|
|
||||||
|
|
||||||
|
def _use_postgres_enabled(config) -> bool:
|
||||||
|
raw = os.getenv("DEER_FLOW_USE_PG_CHECKPOINTER")
|
||||||
|
if raw:
|
||||||
|
return raw.strip().lower() not in {"0", "false", "no", "off"}
|
||||||
|
return bool(getattr(config, "use_postgres", True))
|
||||||
|
|
||||||
|
|
||||||
def _dual_read_paths(config) -> list[str]:
|
def _dual_read_paths(config) -> list[str]:
|
||||||
paths = list(getattr(config, "dual_read_sqlite_paths", []) or [])
|
paths = list(getattr(config, "dual_read_sqlite_paths", []) or [])
|
||||||
paths.extend(_csv_env("DEER_FLOW_CHECKPOINT_SQLITE_PATHS"))
|
paths.extend(_csv_env("DEER_FLOW_CHECKPOINT_SQLITE_PATHS"))
|
||||||
@@ -64,6 +71,28 @@ def _dual_read_paths(config) -> list[str]:
|
|||||||
return resolved
|
return resolved
|
||||||
|
|
||||||
|
|
||||||
|
def _sqlite_conn_str_for_postgres_fallback(config) -> str:
|
||||||
|
configured = getattr(config, "sqlite_connection_string", None)
|
||||||
|
if configured:
|
||||||
|
return configured
|
||||||
|
paths = _dual_read_paths(config)
|
||||||
|
return paths[0] if paths else "store.db"
|
||||||
|
|
||||||
|
|
||||||
|
@contextlib.asynccontextmanager
|
||||||
|
async def _sqlite_checkpointer_from_conn_str(raw: str) -> AsyncIterator[Checkpointer]:
|
||||||
|
try:
|
||||||
|
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
|
||||||
|
except ImportError as exc:
|
||||||
|
raise ImportError(SQLITE_INSTALL) from exc
|
||||||
|
|
||||||
|
conn_str = resolve_sqlite_conn_str(raw)
|
||||||
|
await asyncio.to_thread(ensure_sqlite_parent_dir, conn_str)
|
||||||
|
async with AsyncSqliteSaver.from_conn_string(conn_str) as saver:
|
||||||
|
await saver.setup()
|
||||||
|
yield saver
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Async factory
|
# Async factory
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -79,19 +108,18 @@ async def _async_checkpointer(config) -> AsyncIterator[Checkpointer]:
|
|||||||
return
|
return
|
||||||
|
|
||||||
if config.type == "sqlite":
|
if config.type == "sqlite":
|
||||||
try:
|
async with _sqlite_checkpointer_from_conn_str(config.connection_string or "store.db") as saver:
|
||||||
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
|
|
||||||
except ImportError as exc:
|
|
||||||
raise ImportError(SQLITE_INSTALL) from exc
|
|
||||||
|
|
||||||
conn_str = resolve_sqlite_conn_str(config.connection_string or "store.db")
|
|
||||||
await asyncio.to_thread(ensure_sqlite_parent_dir, conn_str)
|
|
||||||
async with AsyncSqliteSaver.from_conn_string(conn_str) as saver:
|
|
||||||
await saver.setup()
|
|
||||||
yield saver
|
yield saver
|
||||||
return
|
return
|
||||||
|
|
||||||
if config.type == "postgres":
|
if config.type == "postgres":
|
||||||
|
if not _use_postgres_enabled(config):
|
||||||
|
conn_str = _sqlite_conn_str_for_postgres_fallback(config)
|
||||||
|
logger.info("Checkpointer: use_postgres=false; using SQLite fallback (%s)", conn_str)
|
||||||
|
async with _sqlite_checkpointer_from_conn_str(conn_str) as saver:
|
||||||
|
yield saver
|
||||||
|
return
|
||||||
|
|
||||||
try:
|
try:
|
||||||
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
|
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
|
||||||
except ImportError as exc:
|
except ImportError as exc:
|
||||||
|
|||||||
@@ -188,7 +188,11 @@ checkpointer:
|
|||||||
# 这里同时被「主 agent」和「AI 写作 graph」共用 —— LangGraph 用
|
# 这里同时被「主 agent」和「AI 写作 graph」共用 —— LangGraph 用
|
||||||
# (thread_id, checkpoint_ns) 做隔离,session_id 是 uuid,不会撞。
|
# (thread_id, checkpoint_ns) 做隔离,session_id 是 uuid,不会撞。
|
||||||
type: postgres
|
type: postgres
|
||||||
|
# true: 新 checkpoint 写入 Postgres;false: 保留 PG 配置但继续使用 SQLite。
|
||||||
|
# 也可临时用环境变量 DEER_FLOW_USE_PG_CHECKPOINTER=0 覆盖。
|
||||||
|
use_postgres: true
|
||||||
connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow
|
connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow
|
||||||
|
sqlite_connection_string: backend/.deer-flow/data/deerflow.db
|
||||||
dual_read_enabled: true
|
dual_read_enabled: true
|
||||||
dual_read_sqlite_paths:
|
dual_read_sqlite_paths:
|
||||||
- backend/.deer-flow/data/deerflow.db
|
- backend/.deer-flow/data/deerflow.db
|
||||||
|
|||||||
Reference in new issue
Block a user