fix(checkpoint): streamline checkpointer creation to prioritize MemorySaver for in-memory paths

Author: BinglanLi

BaseAgent/base_agent.py

@@ -1028,21 +1028,22 @@ def _create_checkpointer(self):
         warning if langgraph-checkpoint-sqlite is not installed and a file path
         was requested.
         """
+        from langgraph.checkpoint.memory import MemorySaver
+
         db_path = self.checkpoint_db_path
+        if db_path == ":memory:":
+            return MemorySaver()
         if _HAS_SQLITE_SAVER:
             import sqlite3
             conn = sqlite3.connect(db_path, check_same_thread=False)
             return SqliteSaver(conn)
-        else:
-            from langgraph.checkpoint.memory import MemorySaver
-            if db_path != ":memory:":
-                warnings.warn(
-                    "langgraph-checkpoint-sqlite is not installed; falling back to "
-                    "in-memory checkpointer. State will NOT persist across sessions. "
-                    "Install with: pip install langgraph-checkpoint-sqlite",
-                    stacklevel=3,
-                )
-            return MemorySaver()
+        warnings.warn(
+            "langgraph-checkpoint-sqlite is not installed; falling back to "
+            "in-memory checkpointer. State will NOT persist across sessions. "
+            "Install with: pip install langgraph-checkpoint-sqlite",
+            stacklevel=3,
+        )
+        return MemorySaver()
 
     def close(self):
         """Release checkpointer resources (closes the SQLite connection if open)."""

BaseAgent/multi_agent/orchestrator.py

@@ -263,15 +263,18 @@ def _route(self, state: MultiAgentState) -> str:
     # ------------------------------------------------------------------
 
     def _create_checkpointer(self):
+        from langgraph.checkpoint.memory import MemorySaver
+
         db_path = default_config.checkpoint_db_path
-        if _HAS_SQLITE_SAVER:
-            return SqliteSaver.from_conn_string(db_path)
-        else:
-            from langgraph.checkpoint.memory import MemorySaver
-            if db_path != ":memory:":
-                warnings.warn(
-                    "langgraph-checkpoint-sqlite is not installed; falling back to "
-                    "in-memory checkpointer for AgentTeam.",
-                    stacklevel=3,
-                )
+        if db_path == ":memory:":
             return MemorySaver()
+        if _HAS_SQLITE_SAVER:
+            import sqlite3
+            conn = sqlite3.connect(db_path, check_same_thread=False)
+            return SqliteSaver(conn)
+        warnings.warn(
+            "langgraph-checkpoint-sqlite is not installed; falling back to "
+            "in-memory checkpointer for AgentTeam.",
+            stacklevel=3,
+        )
+        return MemorySaver()

BaseAgent/tests/test_checkpointing.py

@@ -79,13 +79,12 @@ def test_returns_sqlite_saver_when_available(self, base_agent, tmp_path):
         cp = base_agent._create_checkpointer()
         assert isinstance(cp, SqliteSaver)
 
-    def test_memory_path_uses_in_memory_sqlite(self, base_agent):
-        pytest.importorskip("langgraph.checkpoint.sqlite", reason="langgraph-checkpoint-sqlite not installed")
-        from langgraph.checkpoint.sqlite import SqliteSaver
+    def test_memory_path_uses_memory_saver(self, base_agent):
+        from langgraph.checkpoint.memory import MemorySaver
 
         base_agent.checkpoint_db_path = ":memory:"
         cp = base_agent._create_checkpointer()
-        assert isinstance(cp, SqliteSaver)
+        assert isinstance(cp, MemorySaver)
 
     def test_fallback_warns_when_sqlite_missing_and_file_path(self, base_agent):
         """When sqlite package absent AND a file path is given, a warning is emitted."""