mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 18:41:59 +00:00
ThearchyHelios
0e997eb968
- R-a1 老库 NOT NULL 列为空：第一版补 schedules.trigger 时没带缺省值，存量定时任务这一列是 NULL，导入 PostgreSQL 撞 NOT NULL，连带触发记录、通知撞外键。迁移表里这一列改成带 DEFAULT '{}'；新增 _NOT_NULL_BACKFILLS，启动时按模型的缺省值把已经补过、留着 NULL 的填上（SQLite 这边也填，两种库存的一样；只更新这一列，不碰 updated_at）；导入时 _Fixer 遇到 NOT NULL、有缺省值的列的 NULL 也填缺省值，计入「转换」。迁移表里的 NOT NULL 列都过了一遍，只有这一列没带缺省值
- R-a2 孤儿行：导入前在快照上按模型外键扫（settle_orphans）：外键是 SET NULL、列可空的置空，计入转换；CASCADE / 不可空的从快照里拿掉、不导入，反复扫到没有为止，报告里单列「孤儿行（SQLite 当时没强制外键）」的表和条数，不算核对失败
- R-a3 停服务后 agentlab.db-wal 还在：lifespan 收尾 dispose 应用库引擎（只加了这几行），SQLite 最后一个连接关掉时并回、删掉 -wal，PostgreSQL 的池子也正常关
- R-a8 自增主键的序列取「目标表 MAX、快照 sqlite_sequence 的计数器、序列现值」三者最大，删过的号不复用（召回按 source_changes.seq 当游标）
- R-a13 插不进的行的原因不再带「<class 'asyncpg.exceptions.…'>:」前缀（errors.first_line 两种写法都去）；设了 AGENTLAB_DATABASE_URL 却没装 [postgres] 时，应用库引擎和断点库都直接说要装什么
- 测试：DROP COLUMN 再按第一版迁移补回造老表、关着外键写孤儿行，迁移填值、快照上置空和拿掉、导入 PostgreSQL 核对通过且重跑一样；NULL 落回模型缺省值；端到端里删掉最后一条变化，导过去新变化从 3 编起；起停一次服务后数据目录里没有 -wal / -shm（去掉 dispose 这条测试就挂）
- PostgreSQL 引擎上另挂了去掉文本参数里 NUL 的钩子（db/postgres.strip_nul，见下一个提交）
86 lines
4.0 KiB
Python
86 lines
4.0 KiB
Python
"""LangGraph 的断点库：应用库是 SQLite 时用 AsyncSqliteSaver（数据目录里的 checkpoints.db，和业务表分开免得抢写锁），
|
||
是 PostgreSQL 时用官方的 AsyncPostgresSaver（psycopg 3 + 连接池），放在同一个库里，AGENTLAB_CHECKPOINT_URL 另给就放那里。
|
||
|
||
它建的四张表（checkpoints、checkpoint_blobs、checkpoint_writes、checkpoint_migrations）和应用的表不重名
|
||
（test_db_portable 守着），所以同一个库、同一个 schema 就行。
|
||
|
||
应用里读断点一律走 saver 的公开接口（aget_state / aget_state_history / aget_tuple / alist），不直接查这几张表：
|
||
两种库的表结构完全不一样（SQLite 整份断点是一个 msgpack，PostgreSQL 拆成 JSONB 加 blob）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import contextlib
|
||
from collections.abc import AsyncIterator
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from langgraph.checkpoint.base import BaseCheckpointSaver
|
||
|
||
from app.core.config import settings
|
||
|
||
#: 建断点表时拿的会话级咨询锁的号：几个进程同时启动，各自的 setup（建表、记迁移版本）一个一个来
|
||
_SETUP_LOCK = 0x4167656E74436B70 # "AgentCkp"
|
||
|
||
|
||
def _pool_size() -> int:
|
||
# 运行里每个超步写一次断点、每个任务写一次 pending writes，都是一来一回的短操作。并发运行多时多给几条
|
||
return max(4, min(settings.max_concurrent_runs, 20))
|
||
|
||
|
||
@contextlib.asynccontextmanager
|
||
async def open_postgres_saver(conninfo: str, *, max_size: int | None = None,
|
||
application_name: str = "agentlab") -> AsyncIterator[BaseCheckpointSaver]:
|
||
"""PostgreSQL 的断点库：开连接池、建表（幂等），退出时关池子。application_name 同应用库（db/base.APPLICATION_NAME）。"""
|
||
from app.db.base import _require_postgres_driver
|
||
|
||
_require_postgres_driver("psycopg", "psycopg_pool", "langgraph.checkpoint.postgres")
|
||
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
|
||
from psycopg.rows import dict_row
|
||
from psycopg_pool import AsyncConnectionPool
|
||
|
||
# 连接参数照 AsyncPostgresSaver.from_conn_string：自动提交（它不自己管事务）、语句第一次执行就预备、行按 dict 取
|
||
pool = AsyncConnectionPool(
|
||
conninfo, min_size=1, max_size=max_size or _pool_size(), open=False,
|
||
kwargs={"autocommit": True, "prepare_threshold": 0, "row_factory": dict_row,
|
||
"application_name": application_name},
|
||
check=AsyncConnectionPool.check_connection, name="agentlab-checkpoints")
|
||
await pool.open(wait=True)
|
||
try:
|
||
saver = AsyncPostgresSaver(pool) # type: ignore[arg-type]
|
||
async with pool.connection() as conn:
|
||
await conn.execute("SELECT pg_advisory_lock(%s)", (_SETUP_LOCK,))
|
||
try:
|
||
await saver.setup()
|
||
finally:
|
||
await conn.execute("SELECT pg_advisory_unlock(%s)", (_SETUP_LOCK,))
|
||
yield saver
|
||
finally:
|
||
await pool.close()
|
||
|
||
|
||
@contextlib.asynccontextmanager
|
||
async def open_sqlite_saver(path: Path | str) -> AsyncIterator[BaseCheckpointSaver]:
|
||
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
|
||
|
||
async with AsyncSqliteSaver.from_conn_string(str(path)) as saver:
|
||
yield saver
|
||
|
||
|
||
def open_saver() -> contextlib.AbstractAsyncContextManager[BaseCheckpointSaver]:
|
||
"""按配置开这个进程的断点库。"""
|
||
conninfo = settings.checkpoint_conninfo
|
||
if conninfo:
|
||
return open_postgres_saver(conninfo)
|
||
return open_sqlite_saver(settings.checkpoint_path)
|
||
|
||
|
||
async def count(saver: Any, thread_id: str | None = None) -> tuple[int, int]:
|
||
"""(断点条数, pending writes 条数)。thread_id 为空数全部。copy-to-postgres 核对用，走公开接口。"""
|
||
config = {"configurable": {"thread_id": thread_id}} if thread_id else None
|
||
checkpoints = writes = 0
|
||
async for tup in saver.alist(config):
|
||
checkpoints += 1
|
||
writes += len(tup.pending_writes or ())
|
||
return checkpoints, writes
|