Added feature: Eviction logic and new memory cache types.

connected the memory caching logic to the main logic because it was not connected up until now

Co-authored-by: omerdor001 <omerdo@post.bgu.ac.il>
Co-authored-by: adiaybgu <adiay@post.bgu.ac.il>

Author: 2 people

flask4modelcache.py

@@ -84,7 +84,9 @@ def response_hitquery(cache_resp):
                     "ANNOY": {"metric_type": "COSINE", "params": {"search_k": 10}},
                     "AUTOINDEX": {"metric_type": "COSINE", "params": {}},
                 } if manager.MPNet_base else None
-    )
+    ),
+    eviction='WTINYLFU',
+    max_size=100000
 )
 
 

modelcache/adapter/adapter.py

@@ -30,6 +30,7 @@ def create_insert(cls, *args, **kwargs):
                 **kwargs
             )
         except Exception as e:
+            print(e)
             return str(e)
 
     @classmethod

modelcache/adapter/adapter_query.py

@@ -91,7 +91,7 @@ def adapt_query(cache_data_convert, *args, **kwargs):
             for cache_data in cache_data_list:
                 primary_id = cache_data[1]
                 ret = chat_cache.data_manager.get_scalar_data(
-                    cache_data, extra_param=context.get("get_scalar_data", None)
+                    cache_data, extra_param=context.get("get_scalar_data", None),model=model
                 )
                 if ret is None:
                     continue
@@ -124,27 +124,27 @@ def adapt_query(cache_data_convert, *args, **kwargs):
 
                 if len(pre_embedding_data) <= 256:
                     if rank_threshold <= rank:
-                        cache_answers.append((rank, ret[1]))
-                        cache_questions.append((rank, ret[0]))
+                        cache_answers.append((rank, ret[0]))
+                        cache_questions.append((rank, ret[1]))
                         cache_ids.append((rank, primary_id))
                 else:
                     if rank_threshold_long <= rank:
-                        cache_answers.append((rank, ret[1]))
-                        cache_questions.append((rank, ret[0]))
+                        cache_answers.append((rank, ret[0]))
+                        cache_questions.append((rank, ret[1]))
                         cache_ids.append((rank, primary_id))
         else:
             # 不使用 reranker 时，走原来的逻辑
             for cache_data in cache_data_list:
                 primary_id = cache_data[1]
                 ret = chat_cache.data_manager.get_scalar_data(
-                    cache_data, extra_param=context.get("get_scalar_data", None)
+                    cache_data, extra_param=context.get("get_scalar_data", None),model=model
                 )
                 if ret is None:
                     continue
 
                 if manager.MPNet_base:
-                    cache_answers.append((cosine_similarity, ret[1]))
-                    cache_questions.append((cosine_similarity, ret[0]))
+                    cache_answers.append((cosine_similarity, ret[0]))
+                    cache_questions.append((cosine_similarity, ret[1]))
                     cache_ids.append((cosine_similarity, primary_id))
                 else:
                     if "deps" in context and hasattr(ret.question, "deps"):
@@ -178,13 +178,13 @@ def adapt_query(cache_data_convert, *args, **kwargs):
 
                     if len(pre_embedding_data) <= 256:
                         if rank_threshold <= rank:
-                            cache_answers.append((rank, ret[1]))
-                            cache_questions.append((rank, ret[0]))
+                            cache_answers.append((rank, ret[0]))
+                            cache_questions.append((rank, ret[1]))
                             cache_ids.append((rank, primary_id))
                     else:
                         if rank_threshold_long <= rank:
-                            cache_answers.append((rank, ret[1]))
-                            cache_questions.append((rank, ret[0]))
+                            cache_answers.append((rank, ret[0]))
+                            cache_questions.append((rank, ret[1]))
                             cache_ids.append((rank, primary_id))
 
         cache_answers = sorted(cache_answers, key=lambda x: x[0], reverse=True)

modelcache/embedding/mpnet_base.py

@@ -7,7 +7,7 @@ def __init__(self):
 
     def to_embeddings(self, *args, **kwargs):
         if not args:
-            raise ValueError("No word provided for embedding.")
+            raise ValueError("No data provided for embedding.")
         embeddings = self.model.encode(args)
         return embeddings[0] if len(args) == 1 else embeddings
 

modelcache/manager/data_manager.py

@@ -22,6 +22,7 @@
 from modelcache.manager.object_data.base import ObjectBase
 from modelcache.manager.eviction import EvictionBase
 from modelcache.manager.eviction_manager import EvictionManager
+from modelcache.manager.eviction.memory_cache import MemoryCacheEviction
 from modelcache.utils.log import modelcache_log
 
 NORMALIZE = True
@@ -38,9 +39,7 @@ def save_query_resp(self, query_resp_dict, **kwargs):
         pass
 
     @abstractmethod
-    def import_data(
-        self, questions: List[Any], answers: List[Any], embedding_datas: List[Any], model:Any
-    ):
+    def import_data(self, questions: List[Any], answers: List[Any], embedding_datas: List[Any], model:Any):
         pass
 
     @abstractmethod
@@ -162,10 +161,18 @@ def __init__(
         self.v = v
         self.o = o
 
+        # added
+        self.eviction_base = MemoryCacheEviction(
+            policy=policy,
+            maxsize=max_size,
+            clean_size=clean_size,
+            on_evict=self._evict_ids)
+
     def save(self, questions: List[any], answers: List[any], embedding_datas: List[any], **kwargs):
         model = kwargs.pop("model", None)
         self.import_data(questions, answers, embedding_datas, model)
 
+
     def save_query_resp(self, query_resp_dict, **kwargs):
         save_query_start_time = time.time()
         self.s.insert_query_resp(query_resp_dict, **kwargs)
@@ -217,14 +224,20 @@ def import_data(
             cache_datas.append([ans, question, embedding_data, model])
 
         ids = self.s.batch_insert(cache_datas)
-        datas_ = [VectorData(id=ids[i], data=embedding_data.astype("float32")) for i, embedding_data in enumerate(embedding_datas)]
-        self.v.mul_add(
-            datas_,
-            model
-
-        )
+        datas = []
+        for i, embedding_data in enumerate(embedding_datas):
+            _id = ids[i]
+            datas.append(VectorData(id=_id, data=embedding_data.astype("float32")))
+            self.eviction_base.put([(_id, cache_datas[i])],model=model)
+        self.v.mul_add(datas,model)
 
     def get_scalar_data(self, res_data, **kwargs) -> Optional[CacheData]:
+        model = kwargs.pop("model")
+        #Get Data from RAM Cache
+        _id = res_data[1]
+        cache_hit = self.eviction_base.get(_id, model=model)
+        if cache_hit is not None:
+            return cache_hit
         cache_data = self.s.get_data_by_id(res_data[1])
         if cache_data is None:
             return None
@@ -244,8 +257,10 @@ def search(self, embedding_data, **kwargs):
         return self.v.search(data=embedding_data, top_k=top_k, model=model)
 
     def delete(self, id_list, **kwargs):
-        model = kwargs.pop("model", None)
+        model = kwargs.pop("model")
         try:
+            for id in id_list:
+                self.eviction_base.get_cache(model).pop(id, None)  # Remove from in-memory LRU too
             v_delete_count = self.v.delete(ids=id_list, model=model)
         except Exception as e:
             return {'status': 'failed', 'milvus': 'delete milvus data failed, please check! e: {}'.format(e),
@@ -262,23 +277,51 @@ def delete(self, id_list, **kwargs):
     def create_index(self, model, **kwargs):
         return self.v.create(model)
 
-    def truncate(self, model_name):
+    def truncate(self, model):
+        # drop memory cache data
+        self.eviction_base.clear(model)
+
         # drop vector base data
         try:
-            vector_resp = self.v.rebuild_col(model_name)
+            vector_resp = self.v.rebuild_col(model)
         except Exception as e:
             return {'status': 'failed', 'VectorDB': 'truncate VectorDB data failed, please check! e: {}'.format(e),
                     'ScalarDB': 'unexecuted'}
         if vector_resp:
             return {'status': 'failed', 'VectorDB': vector_resp, 'ScalarDB': 'unexecuted'}
         # drop scalar base data
         try:
-            delete_count = self.s.model_deleted(model_name)
+            delete_count = self.s.model_deleted(model)
         except Exception as e:
             return {'status': 'failed', 'VectorDB': 'rebuild',
                     'ScalarDB': 'truncate scalar data failed, please check! e: {}'.format(e)}
         return {'status': 'success', 'VectorDB': 'rebuild', 'ScalarDB': 'delete_count: ' + str(delete_count)}
 
+    # added
+    def _evict_ids(self, ids, **kwargs):
+        model = kwargs.get("model")
+        if not ids or any(i is None for i in ids):
+            modelcache_log.warning("Skipping eviction for invalid IDs: %s", ids)
+            return
+
+        if isinstance(ids,str):
+            ids = [ids]
+
+        for _id in ids:
+            self.eviction_base.get_cache(model).pop(_id, None)
+
+        try:
+            self.s.mark_deleted(ids)
+            modelcache_log.info("Evicted from scalar storage: %s", ids)
+        except Exception as e:
+            modelcache_log.error("Failed to delete from scalar storage: %s", str(e))
+
+        try:
+            self.v.delete(ids, model=model)
+            modelcache_log.info("Evicted from vector storage (model=%s): %s", model, ids)
+        except Exception as e:
+            modelcache_log.error("Failed to delete from vector storage (model=%s): %s", model, str(e))
+
     def flush(self):
         self.s.flush()
         self.v.flush()

modelcache/manager/eviction/arc_cache.py

@@ -0,0 +1,131 @@
+from cachetools import Cache
+from collections import OrderedDict
+
+class ARC(Cache):
+    """
+    Adaptive Replacement Cache (ARC) implementation with on_evict callback.
+    Balances recency and frequency via two active lists (T1, T2) and two ghost lists (B1, B2).
+    Calls on_evict([key]) whenever an item is evicted from the active cache.
+    """
+
+    def __init__(self, maxsize, getsizeof=None, on_evict=None):
+        """
+        Args:
+            maxsize (int): Maximum cache size.
+            getsizeof (callable, optional): Sizing function for items.
+            on_evict (callable, optional): Callback called as on_evict([key]) when a key is evicted.
+        """
+        super().__init__(maxsize, getsizeof)
+        self.t1 = OrderedDict()
+        self.t2 = OrderedDict()
+        self.b1 = OrderedDict()
+        self.b2 = OrderedDict()
+        self.p = 0  # Adaptive target for T1 size.
+        self.on_evict = on_evict
+
+    def __len__(self):
+        return len(self.t1) + len(self.t2)
+
+    def __contains__(self, key):
+        return key in self.t1 or key in self.t2
+
+    def _evict_internal(self):
+        """
+        Evicts items from T1 or T2 if cache is over capacity, and prunes ghost lists.
+        Calls on_evict for each evicted key.
+        """
+        # Evict from T1 or T2 if active cache > maxsize
+        while len(self.t1) + len(self.t2) > self.maxsize:
+            if len(self.t1) > self.p or (len(self.t1) == 0 and len(self.t2) > 0):
+                key, value = self.t1.popitem(last=False)
+                self.b1[key] = value
+                if self.on_evict:
+                    self.on_evict([key])
+            else:
+                key, value = self.t2.popitem(last=False)
+                self.b2[key] = value
+                if self.on_evict:
+                    self.on_evict([key])
+        # Prune ghost lists to their max lengths
+        while len(self.b1) > (self.maxsize - self.p):
+            self.b1.popitem(last=False)
+        while len(self.b2) > self.p:
+            self.b2.popitem(last=False)
+
+    def __setitem__(self, key, value):
+        # Remove from all lists before re-inserting
+        for l in (self.t1, self.t2, self.b1, self.b2):
+            l.pop(key, None)
+        self.t1[key] = value
+        self.t1.move_to_end(key)
+        self._evict_internal()
+
+    def __getitem__(self, key):
+        # Case 1: Hit in T1 → promote to T2
+        if key in self.t1:
+            value = self.t1.pop(key)
+            self.t2[key] = value
+            self.t2.move_to_end(key)
+            self.p = max(0, self.p - 1)
+            self._evict_internal()
+            return value
+        # Case 2: Hit in T2 → refresh in T2
+        if key in self.t2:
+            value = self.t2.pop(key)
+            self.t2[key] = value
+            self.t2.move_to_end(key)
+            self.p = min(self.maxsize, self.p + 1)
+            self._evict_internal()
+            return value
+        # Case 3: Hit in B1 (ghost) → fetch and promote to T2
+        if key in self.b1:
+            self.b1.pop(key)
+            self.p = min(self.maxsize, self.p + 1)
+            self._evict_internal()
+            value = super().__missing__(key)
+            self.t2[key] = value
+            self.t2.move_to_end(key)
+            return value
+        # Case 4: Hit in B2 (ghost) → fetch and promote to T2
+        if key in self.b2:
+            self.b2.pop(key)
+            self.p = max(0, self.p - 1)
+            self._evict_internal()
+            value = super().__missing__(key)
+            self.t2[key] = value
+            self.t2.move_to_end(key)
+            return value
+        # Case 5: Cold miss → handled by Cache base class (calls __setitem__ after __missing__)
+        return super().__getitem__(key)
+
+    def __missing__(self, key):
+        """
+        Override this in a subclass, or rely on direct assignment (cache[key] = value).
+        """
+        raise KeyError(key)
+
+    def pop(self, key, default=None):
+        """
+        Remove key from all lists.
+        """
+        for l in (self.t1, self.t2, self.b1, self.b2):
+            if key in l:
+                return l.pop(key)
+        return default
+
+    def clear(self):
+        self.t1.clear()
+        self.t2.clear()
+        self.b1.clear()
+        self.b2.clear()
+        self.p = 0
+        super().clear()
+
+    def __iter__(self):
+        yield from self.t1
+        yield from self.t2
+
+    def __repr__(self):
+        return (f"ARC(maxsize={self.maxsize}, p={self.p}, len={len(self)}, "
+                f"t1_len={len(self.t1)}, t2_len={len(self.t2)}, "
+                f"b1_len={len(self.b1)}, b2_len={len(self.b2)})")

modelcache/manager/eviction/base.py

@@ -9,11 +9,11 @@ class EvictionBase(metaclass=ABCMeta):
     """
 
     @abstractmethod
-    def put(self, objs: List[Any]):
+    def put(self, objs: List[Any], model:str):
         pass
 
     @abstractmethod
-    def get(self, obj: Any):
+    def get(self, obj: Any, model:str):
         pass
 
     @property