Fixed freeze implementation (#1983)

* chore: Update utils imports and add cn to nodeToolbarComponent

* fix(utils.py): change key parameter name from 'flow_id' to 'key' for consistency
fix(chat.py): remove unused import 'functools.partial' to improve code readability
refactor(chat.py): remove 'set_cache_coro' partial function and pass 'chat_service' directly to 'build_vertex' method for better code organization
feat(schemas.py): add 'used_frozen_result' field to ResultDataResponse and ResultData classes with default value of False for better tracking of frozen result usage
feat(base.py): add 'chat_service' parameter to 'build_vertex' method in Graph class to allow passing ChatService instance for cache operations
feat(base.py): update 'build_vertex' method in Graph class to handle caching of frozen vertices and set 'used_frozen_result' flag in ResultData class
feat(cache/service.py): change parameter name from 'flow_id' to 'key' in 'set_cache' and 'get_cache' methods for consistency
feat(cache/utils.py): add 'CacheMiss' class to represent cache miss situations for better error handling

* feat: Add check for None before setting 'used_frozen_result' flag in Graph class

* feat: Add frozen effect to buttons and improve code organization

The code changes introduce a frozen effect to buttons by adding new CSS classes and styles. This effect is achieved by applying borders, shadows, and background colors. Additionally, the code is refactored to improve code organization and remove unused imports.

Note: This commit message follows the convention used in the recent user commits.

* feat: Add frozen effect to buttons and improve code organization

* style(applies.css): Update border styles for frozen state to improve visual appearance and consistency
style(applies.css): Adjust opacity of frosted background for better readability
style(tailwind.config.js): Increase opacity of frozen-ring shadow for better visual effect
style(tailwind.config.js): Increase opacity of frosted-ring shadow for better visual effect

* feat(parameterComponent): add snowflake icon to ParameterComponent when node is frozen and not aligned left

* style(applies.css): Update border styles for frozen state and add border to improve visual appearance and consistency

src/backend/base/langflow/api/utils.py

@@ -286,7 +286,7 @@ async def get_next_runnable_vertices(
         for v_id in set(next_runnable_vertices):  # Use set to avoid duplicates
             graph.vertices_to_run.remove(v_id)
             graph.remove_from_predecessors(v_id)
-        await chat_service.set_cache(flow_id=flow_id, data=graph, lock=lock)
+        await chat_service.set_cache(key=flow_id, data=graph, lock=lock)
     return next_runnable_vertices
 
 

src/backend/base/langflow/api/v1/chat.py

@@ -1,6 +1,5 @@
 import time
 import uuid
-from functools import partial
 from typing import TYPE_CHECKING, Annotated, Optional
 
 from fastapi import APIRouter, BackgroundTasks, Body, Depends, HTTPException
@@ -162,7 +161,6 @@ async def build_vertex(
         vertex = graph.get_vertex(vertex_id)
         try:
             lock = chat_service._cache_locks[flow_id_str]
-            set_cache_coro = partial(chat_service.set_cache, flow_id=flow_id_str)
             (
                 next_runnable_vertices,
                 top_level_vertices,
@@ -173,7 +171,7 @@ async def build_vertex(
                 vertex,
             ) = await graph.build_vertex(
                 lock=lock,
-                set_cache_coro=set_cache_coro,
+                chat_service=chat_service,
                 vertex_id=vertex_id,
                 user_id=current_user.id,
                 inputs_dict=inputs.model_dump() if inputs else {},

src/backend/base/langflow/api/v1/schemas.py

@@ -248,6 +248,7 @@ class ResultDataResponse(BaseModel):
     artifacts: Optional[Any] = Field(default_factory=dict)
     timedelta: Optional[float] = None
     duration: Optional[str] = None
+    used_frozen_result: Optional[bool] = False
 
 
 class VertexBuildResponse(BaseModel):

src/backend/base/langflow/graph/graph/base.py

@@ -17,6 +17,8 @@ from langflow.graph.vertex.base import Vertex
 from langflow.graph.vertex.types import InterfaceVertex, StateVertex
 from langflow.schema import Record
 from langflow.schema.schema import INPUT_FIELD_NAME, InputType
+from langflow.services.cache.utils import CacheMiss
+from langflow.services.chat.service import ChatService
 from langflow.services.deps import get_chat_service
 
 if TYPE_CHECKING:
@@ -704,7 +706,7 @@ class Graph:
     async def build_vertex(
         self,
         lock: asyncio.Lock,
-        set_cache_coro: Callable[["Graph", asyncio.Lock], Coroutine],
+        chat_service: ChatService,
         vertex_id: str,
         inputs_dict: Optional[Dict[str, str]] = None,
         user_id: Optional[str] = None,
@@ -729,17 +731,35 @@ class Graph:
         """
         vertex = self.get_vertex(vertex_id)
         try:
-            if not vertex.frozen or not vertex._built:
+            params = ""
+            if vertex.frozen:
+                # Check the cache for the vertex
+                cached_result = await chat_service.get_cache(key=vertex.id)
+                if isinstance(cached_result, CacheMiss):
+                    await vertex.build(user_id=user_id, inputs=inputs_dict, fallback_to_env_vars=fallback_to_env_vars)
+                    await chat_service.set_cache(key=vertex.id, data=vertex)
+                else:
+                    cached_vertex = cached_result["result"]
+                    # Now set update the vertex with the cached vertex
+                    vertex._built = cached_vertex._built
+                    vertex.result = cached_vertex.result
+                    vertex.artifacts = cached_vertex.artifacts
+                    vertex._built_object = cached_vertex._built_object
+                    vertex._custom_component = cached_vertex._custom_component
+                    if vertex.result is not None:
+                        vertex.result.used_frozen_result = True
+
+            else:
                 await vertex.build(user_id=user_id, inputs=inputs_dict, fallback_to_env_vars=fallback_to_env_vars)
 
             if vertex.result is not None:
-                params = vertex._built_object_repr()
+                params = f"{vertex._built_object_repr()}{params}"
                 valid = True
                 result_dict = vertex.result
                 artifacts = vertex.artifacts
             else:
                 raise ValueError(f"No result found for vertex {vertex_id}")
-
+            set_cache_coro = partial(chat_service.set_cache, key=self.flow_id)
             next_runnable_vertices, top_level_vertices = await self.get_next_and_top_level_vertices(
                 lock, set_cache_coro, vertex
             )
@@ -810,11 +830,10 @@ class Graph:
             for vertex_id in current_batch:
                 vertex = self.get_vertex(vertex_id)
                 lock = chat_service._cache_locks[self.run_id]
-                set_cache_coro = partial(chat_service.set_cache, flow_id=self.run_id)
                 task = asyncio.create_task(
                     self.build_vertex(
                         lock=lock,
-                        set_cache_coro=set_cache_coro,
+                        chat_service=chat_service,
                         vertex_id=vertex_id,
                         user_id=self.user_id,
                         inputs_dict={},

src/backend/base/langflow/graph/schema.py

@@ -15,6 +15,7 @@ class ResultData(BaseModel):
     duration: Optional[str] = None
     component_display_name: Optional[str] = None
     component_id: Optional[str] = None
+    used_frozen_result: Optional[bool] = False
 
     @field_serializer("results")
     def serialize_results(self, value):

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

@@ -9,6 +9,9 @@ from loguru import logger
 
 from langflow.services.base import Service
 from langflow.services.cache.base import AsyncBaseCacheService, CacheService
+from langflow.services.cache.utils import CacheMiss
+
+CACHE_MISS = CacheMiss()
 
 
 class ThreadingInMemoryCache(CacheService, Service):
@@ -341,12 +344,14 @@ class AsyncInMemoryCache(AsyncBaseCacheService, Service):
 
     async def _get(self, key):
         item = self.cache.get(key, None)
-        if item and (time.time() - item["time"] < self.expiration_time):
-            self.cache.move_to_end(key)
-            return pickle.loads(item["value"]) if isinstance(item["value"], bytes) else item["value"]
         if item:
-            await self.delete(key)
-        return None
+            if time.time() - item["time"] < self.expiration_time:
+                self.cache.move_to_end(key)
+                return pickle.loads(item["value"]) if isinstance(item["value"], bytes) else item["value"]
+            else:
+                logger.info(f"Cache item for key '{key}' has expired and will be deleted.")
+                await self.delete(key)  # Log before deleting the expired item
+        return CACHE_MISS
 
     async def set(self, key, value, lock: Optional[asyncio.Lock] = None):
         if not lock:

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

@@ -19,6 +19,11 @@ CACHE_DIR = user_cache_dir("langflow", "langflow")
 PREFIX = "langflow_cache"
 
 
+class CacheMiss:
+    def __repr__(self):
+        return "<CACHE_MISS>"
+
+
 def create_cache_folder(func):
     def wrapper(*args, **kwargs):
         # Get the destination folder

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

@@ -13,7 +13,7 @@ class ChatService(Service):
         self._cache_locks = defaultdict(asyncio.Lock)
         self.cache_service = get_cache_service()
 
-    async def set_cache(self, flow_id: str, data: Any, lock: Optional[asyncio.Lock] = None) -> bool:
+    async def set_cache(self, key: str, data: Any, lock: Optional[asyncio.Lock] = None) -> bool:
         """
         Set the cache for a client.
         """
@@ -23,17 +23,17 @@ class ChatService(Service):
             "result": data,
             "type": type(data),
         }
-        await self.cache_service.upsert(flow_id, result_dict, lock=lock or self._cache_locks[flow_id])
-        return flow_id in self.cache_service
+        await self.cache_service.upsert(key, result_dict, lock=lock or self._cache_locks[key])
+        return key in self.cache_service
 
-    async def get_cache(self, flow_id: str, lock: Optional[asyncio.Lock] = None) -> Any:
+    async def get_cache(self, key: str, lock: Optional[asyncio.Lock] = None) -> Any:
         """
         Get the cache for a client.
         """
-        return await self.cache_service.get(flow_id, lock=lock or self._cache_locks[flow_id])
+        return await self.cache_service.get(key, lock=lock or self._cache_locks[key])
 
-    async def clear_cache(self, flow_id: str, lock: Optional[asyncio.Lock] = None):
+    async def clear_cache(self, key: str, lock: Optional[asyncio.Lock] = None):
         """
         Clear the cache for a client.
         """
-        await self.cache_service.delete(flow_id, lock=lock or self._cache_locks[flow_id])
+        await self.cache_service.delete(key, lock=lock or self._cache_locks[key])