From d8c5c23e2a8457db453cf3ec6280a2c1ac5f44b0 Mon Sep 17 00:00:00 2001 From: Juan Date: Sun, 10 May 2026 15:15:35 +0200 Subject: [PATCH] =?UTF-8?q?fix(runtime):=20build=5Fcheckpointer=20as=C3=AD?= =?UTF-8?q?ncrono=20con=20AsyncSqliteSaver?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Los nodos del grafo son async, así que se invoca con `await graph.ainvoke`, que el SqliteSaver síncrono no soporta (NotImplementedError). Se cambia a AsyncSqliteSaver expuesto como context manager async. Requiere aiosqlite<0.21 (la 0.21 eliminó Connection.is_alive(), usado por langgraph-checkpoint-sqlite 2.x). Co-Authored-By: Claude Opus 4.7 --- core/requirements.txt | 3 ++ .../agentforge_core/runtime/checkpointer.py | 38 +++++++++++++------ 2 files changed, 29 insertions(+), 12 deletions(-) diff --git a/core/requirements.txt b/core/requirements.txt index 6a6395f..d24fd71 100644 --- a/core/requirements.txt +++ b/core/requirements.txt @@ -6,6 +6,9 @@ structlog>=24.4,<25.0 httpx>=0.27,<0.28 langgraph>=0.2.50,<0.3 langgraph-checkpoint-sqlite>=2.0,<3.0 +# AsyncSqliteSaver (langgraph-checkpoint-sqlite 2.x) usa Connection.is_alive(), +# eliminado en aiosqlite 0.21. Fijado a la última serie compatible. +aiosqlite>=0.20,<0.21 guardrails-ai>=0.5,<0.6 presidio-analyzer>=2.2,<3.0 presidio-anonymizer>=2.2,<3.0 diff --git a/core/src/agentforge_core/runtime/checkpointer.py b/core/src/agentforge_core/runtime/checkpointer.py index f23733d..051c57a 100644 --- a/core/src/agentforge_core/runtime/checkpointer.py +++ b/core/src/agentforge_core/runtime/checkpointer.py @@ -1,20 +1,34 @@ -"""Wrapper sobre SqliteSaver de LangGraph. Centraliza el path y el ciclo de vida.""" +"""Wrapper sobre ``AsyncSqliteSaver`` de LangGraph. Centraliza el path y el ciclo de vida. + +El grafo se invoca con ``await graph.ainvoke(...)`` (los nodos son ``async def``), así +que necesitamos el checkpointer **asíncrono** de LangGraph. ``AsyncSqliteSaver`` liga su +conexión ``aiosqlite`` al event loop activo, por lo que se expone como context manager +asíncrono:: + + async with build_checkpointer(data_dir) as cp: + graph = build_graph(..., checkpointer=cp) + await graph.ainvoke(...) + +La conexión se abre al entrar y se cierra al salir; LangGraph crea las tablas +(``setup()``) de forma perezosa en el primer acceso. El estado se persiste en +``data_dir/checkpoints.sqlite``, por lo que un ``awaiting_approval`` sobrevive a un +reinicio del proceso. +""" from __future__ import annotations -import sqlite3 +from contextlib import AbstractAsyncContextManager from pathlib import Path -from langgraph.checkpoint.sqlite import SqliteSaver +from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver -def build_checkpointer(data_dir: Path) -> SqliteSaver: - """Devuelve un SqliteSaver apuntando a ``data_dir/checkpoints.sqlite``. - - La base se crea si no existe. La conexión usa ``check_same_thread=False`` - porque LangGraph ejecuta nodos en hilos del executor. - """ +def checkpointer_path(data_dir: Path) -> Path: + """Ruta al fichero SQLite de checkpoints, creando ``data_dir`` si no existe.""" data_dir.mkdir(parents=True, exist_ok=True) - db_path = data_dir / "checkpoints.sqlite" - conn = sqlite3.connect(str(db_path), check_same_thread=False) - return SqliteSaver(conn) + return data_dir / "checkpoints.sqlite" + + +def build_checkpointer(data_dir: Path) -> AbstractAsyncContextManager[AsyncSqliteSaver]: + """Context manager asíncrono con un ``AsyncSqliteSaver`` sobre ``data_dir/checkpoints.sqlite``.""" + return AsyncSqliteSaver.from_conn_string(str(checkpointer_path(data_dir)))