Merge branch 'release-1.10.0' into feat/kb-v1-db-connectors

Author: erichare

docs/docs/Develop/memory.mdx

@@ -119,7 +119,7 @@ If you want to use a non-default cache setting, you can use the following enviro
 
 | Variable | Type | Default | Description |
 |----------|------|---------|-------------|
-| `LANGFLOW_CACHE_TYPE` | String | `async` | Set the cache type for Langflow's internal caching system. Possible values: `async`, `redis`, `memory`, `disk`. If you set the type to `redis`, then you must also set the `LANGFLOW_REDIS_*` environment variables. |
+| `LANGFLOW_CACHE_TYPE` | String | `async` | Set the cache type for Langflow's internal caching system. Possible values: `async`, `redis`, `memory`. If you set the type to `redis`, then you must also set the `LANGFLOW_REDIS_*` environment variables. The `disk` backend was removed in 1.10 — switch to `async` for in-memory or `redis` for cross-worker. |
 | `LANGFLOW_LANGCHAIN_CACHE` | String | `InMemoryCache` | Set the cache storage type for the LangChain caching system (a Langflow dependency), either `InMemoryCache` or `SQLiteCache`. |
 | `LANGFLOW_REDIS_HOST` | String | `localhost` | Redis server hostname if `LANGFLOW_CACHE_TYPE=redis`. |
 | `LANGFLOW_REDIS_PORT` | Integer | `6379` | Redis server port if `LANGFLOW_CACHE_TYPE=redis`. |

pyproject.toml

@@ -114,7 +114,6 @@ cassio = [
     "cassio>=0.1.7"
 ]
 local = [
-    "llama-cpp-python~=0.2.0",
     "sentence-transformers>=2.3.1",
     "ctransformers>=0.2.10"
 ]

src/backend/base/langflow/services/cache/disk.py

@@ -5,7 +5,6 @@
 from lfx.log.logger import logger
 from typing_extensions import override
 
-from langflow.services.cache.disk import AsyncDiskCache
 from langflow.services.cache.service import AsyncInMemoryCache, CacheService, RedisCache, ThreadingInMemoryCache
 from langflow.services.factory import ServiceFactory
 
@@ -36,10 +35,4 @@ def create(self, settings_service: SettingsService):
             return ThreadingInMemoryCache(expiration_time=settings_service.settings.cache_expire)
         if settings_service.settings.cache_type == "async":
             return AsyncInMemoryCache(expiration_time=settings_service.settings.cache_expire)
-        if settings_service.settings.cache_type == "disk":
-            cache_dir = settings_service.settings.cache_dir or settings_service.settings.config_dir
-            return AsyncDiskCache(
-                cache_dir=cache_dir,
-                expiration_time=settings_service.settings.cache_expire,
-            )
         return None

src/backend/base/langflow/services/cache/factory.py

@@ -1,13 +1,20 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from typing_extensions import override
 
 from langflow.services.factory import ServiceFactory
 from langflow.services.flow_events.service import FlowEventsService
 
+if TYPE_CHECKING:
+    from lfx.services.settings.service import SettingsService
+
 
 class FlowEventsServiceFactory(ServiceFactory):
     def __init__(self) -> None:
         super().__init__(FlowEventsService)
 
     @override
-    def create(self):
-        return FlowEventsService()
+    def create(self, settings_service: SettingsService):
+        return FlowEventsService(cache_dir=settings_service.settings.cache_dir)

src/backend/base/langflow/services/flow_events/factory.py

@@ -1,14 +1,13 @@
 from __future__ import annotations
 
-import json
+import sqlite3
 import tempfile
+import threading
 import time
-from dataclasses import asdict, dataclass
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Literal, get_args
 
-from diskcache import Cache
-
 from langflow.services.base import Service
 
 FLOW_EVENT_TYPES = Literal[
@@ -29,12 +28,25 @@ class FlowEvent:
     summary: str = ""
 
 
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS flow_events (
+    flow_id    TEXT NOT NULL,
+    ts         REAL NOT NULL,
+    type       TEXT NOT NULL,
+    summary    TEXT NOT NULL DEFAULT '',
+    expires_at REAL NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_flow_events_flow_ts ON flow_events(flow_id, ts);
+CREATE INDEX IF NOT EXISTS idx_flow_events_expires ON flow_events(expires_at);
+"""
+
+
 class FlowEventsService(Service):
-    """Disk-backed event queue keyed by flow_id.
+    """SQLite-backed event queue keyed by flow_id.
 
-    Uses diskcache for cross-worker visibility (multiple uvicorn/gunicorn workers
-    share the same SQLite-backed cache directory). TTL-based cleanup is handled
-    by diskcache's built-in expiry.
+    Uses Python's stdlib sqlite3 in WAL mode for cross-worker visibility (multiple
+    uvicorn/gunicorn workers share the same on-disk database file). TTL-based
+    cleanup is performed lazily on each write and read.
 
     Limitations:
     - Events are ephemeral: lost on disk cleanup or container restart.
@@ -49,28 +61,68 @@ class FlowEventsService(Service):
     SETTLE_TIMEOUT: float = 10.0
     MAX_EVENTS_PER_FLOW: int = 1000
 
+    _VALID_EVENT_TYPES: frozenset[str] = frozenset(get_args(FLOW_EVENT_TYPES))
+
     def __init__(self, cache_dir: str | Path | None = None) -> None:
         if cache_dir is None:
             cache_dir = Path(tempfile.gettempdir()) / "langflow_flow_events"
-        self._cache = Cache(str(cache_dir))
-
-    _VALID_EVENT_TYPES: frozenset[str] = frozenset(get_args(FLOW_EVENT_TYPES))
+        cache_dir = Path(cache_dir)
+        cache_dir.mkdir(parents=True, exist_ok=True)
+        self._db_path = cache_dir / "flow_events.sqlite"
+        # isolation_level=None puts pysqlite in autocommit mode so we manage
+        # transactions explicitly via BEGIN/COMMIT.
+        self._conn = sqlite3.connect(
+            str(self._db_path),
+            isolation_level=None,
+            check_same_thread=False,
+            timeout=5.0,
+        )
+        # Serialize use of the single connection across asyncio tasks / FastAPI threads
+        # in this worker. WAL handles cross-worker concurrency.
+        self._lock = threading.Lock()
+        with self._lock:
+            self._conn.execute("PRAGMA journal_mode=WAL")
+            self._conn.execute("PRAGMA synchronous=NORMAL")
+            self._conn.execute("PRAGMA busy_timeout=5000")
+            self._conn.executescript(_SCHEMA)
 
     def append(self, flow_id: str, event_type: str, summary: str = "") -> FlowEvent:
         if event_type not in self._VALID_EVENT_TYPES:
             msg = f"Invalid event type: {event_type!r}. Must be one of {sorted(self._VALID_EVENT_TYPES)}"
             raise ValueError(msg)
-        event = FlowEvent(type=event_type, timestamp=time.time(), summary=summary)
-        key = f"flow_events:{flow_id}"
-
-        with self._cache.transact():
-            raw = self._cache.get(key, default=None)
-            events: list[dict] = json.loads(raw) if raw else []
-            events.append(asdict(event))
-            # Trim to max size
-            if len(events) > self.MAX_EVENTS_PER_FLOW:
-                events = events[-self.MAX_EVENTS_PER_FLOW :]
-            self._cache.set(key, json.dumps(events), expire=self.TTL_SECONDS)
+
+        now = time.time()
+        event = FlowEvent(type=event_type, timestamp=now, summary=summary)
+        expires_at = now + self.TTL_SECONDS
+
+        with self._lock:
+            self._conn.execute("BEGIN IMMEDIATE")
+            try:
+                # Opportunistic TTL cleanup across all flows.
+                self._conn.execute("DELETE FROM flow_events WHERE expires_at < ?", (now,))
+                self._conn.execute(
+                    "INSERT INTO flow_events (flow_id, ts, type, summary, expires_at) VALUES (?, ?, ?, ?, ?)",
+                    (flow_id, now, event_type, summary, expires_at),
+                )
+                # Bound per-flow size: keep only the most recent MAX_EVENTS_PER_FLOW rows.
+                # Order by (ts, rowid) so events appended in the same microsecond have a
+                # stable, insertion-aware ordering when picking which rows to drop.
+                self._conn.execute(
+                    """
+                    DELETE FROM flow_events
+                    WHERE rowid IN (
+                        SELECT rowid FROM flow_events
+                        WHERE flow_id = ?
+                        ORDER BY ts DESC, rowid DESC
+                        LIMIT -1 OFFSET ?
+                    )
+                    """,
+                    (flow_id, self.MAX_EVENTS_PER_FLOW),
+                )
+                self._conn.execute("COMMIT")
+            except Exception:
+                self._conn.execute("ROLLBACK")
+                raise
 
         return event
 
@@ -82,10 +134,19 @@ def get_since(self, flow_id: str, since: float) -> tuple[list[FlowEvent], bool]:
         - A flow_settled event exists after `since`, OR
         - The most recent event is older than SETTLE_TIMEOUT seconds.
         """
-        key = f"flow_events:{flow_id}"
-        raw = self._cache.get(key, default=None)
-        all_events = [FlowEvent(**e) for e in json.loads(raw)] if raw else []
-
+        now = time.time()
+        with self._lock:
+            rows = self._conn.execute(
+                """
+                SELECT type, ts, summary
+                FROM flow_events
+                WHERE flow_id = ? AND expires_at >= ?
+                ORDER BY ts ASC, rowid ASC
+                """,
+                (flow_id, now),
+            ).fetchall()
+
+        all_events = [FlowEvent(type=r[0], timestamp=r[1], summary=r[2]) for r in rows]
         after = [e for e in all_events if e.timestamp > since]
 
         if not after and not all_events:
@@ -100,4 +161,5 @@ def get_since(self, flow_id: str, since: float) -> tuple[list[FlowEvent], bool]:
         return after, settled
 
     async def teardown(self) -> None:
-        self._cache.close()
+        with self._lock:
+            self._conn.close()

src/backend/base/langflow/services/flow_events/service.py

@@ -73,7 +73,6 @@ dependencies = [
     "filelock>=3.20.1,<4.0.0",
     "grandalf>=0.8.0,<1.0.0",
     "spider-client>=0.0.27,<1.0.0",
-    "diskcache>=5.6.3,<6.0.0",
     "clickhouse-connect==0.7.19",
     "assemblyai>=0.33.0,<1.0.0",
     "fastapi-pagination>=0.13.1,<1.0.0",
@@ -176,7 +175,6 @@ postgresql = [
 ]
 
 local = [
-    "llama-cpp-python>=0.2.0",
     "sentence-transformers>=2.0.0",
     "ctransformers>=0.2"
 ]
@@ -224,7 +222,6 @@ sambanova = ["langchain-sambanova~=1.0.0"]
 groq = ["langchain-groq~=1.1.1"]
 
 # Individual local LLM providers
-llama-cpp = ["llama-cpp-python~=0.2.0"]
 sentence-transformers = ["sentence-transformers>=2.3.1"]
 ctransformers = ["ctransformers>=0.2.10"]
 
@@ -242,7 +239,6 @@ pytube = ["pytube==15.0.0"]
 # Individual agent frameworks
 # crewai is commented out due to httpx version conflict
 # crewai = ["crewai>=0.126.0"]
-dspy = ["dspy-ai==2.5.41"]
 smolagents = ["smolagents>=1.8.0"]
 
 # Individual tools
@@ -399,7 +395,6 @@ complete = [
     "langflow-base[docx]",
     "langflow-base[pytube]",
     # "langflow-base[crewai]",  # commented out due to httpx version conflict
-    "langflow-base[dspy]",
     "langflow-base[smolagents]",
     "langflow-base[beautifulsoup]",
     "langflow-base[serpapi]",