|
|
@ -1,8 +1,9 @@
|
|
|
|
"""Beta Feature: base interface for cache."""
|
|
|
|
"""Beta Feature: base interface for cache."""
|
|
|
|
import hashlib
|
|
|
|
import hashlib
|
|
|
|
|
|
|
|
import inspect
|
|
|
|
import json
|
|
|
|
import json
|
|
|
|
from abc import ABC, abstractmethod
|
|
|
|
from abc import ABC, abstractmethod
|
|
|
|
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, cast
|
|
|
|
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union, cast
|
|
|
|
|
|
|
|
|
|
|
|
from sqlalchemy import Column, Integer, String, create_engine, select
|
|
|
|
from sqlalchemy import Column, Integer, String, create_engine, select
|
|
|
|
from sqlalchemy.engine.base import Engine
|
|
|
|
from sqlalchemy.engine.base import Engine
|
|
|
@ -274,7 +275,12 @@ class RedisSemanticCache(BaseCache):
|
|
|
|
class GPTCache(BaseCache):
|
|
|
|
class GPTCache(BaseCache):
|
|
|
|
"""Cache that uses GPTCache as a backend."""
|
|
|
|
"""Cache that uses GPTCache as a backend."""
|
|
|
|
|
|
|
|
|
|
|
|
def __init__(self, init_func: Optional[Callable[[Any], None]] = None):
|
|
|
|
def __init__(
|
|
|
|
|
|
|
|
self,
|
|
|
|
|
|
|
|
init_func: Union[
|
|
|
|
|
|
|
|
Callable[[Any, str], None], Callable[[Any], None], None
|
|
|
|
|
|
|
|
] = None,
|
|
|
|
|
|
|
|
):
|
|
|
|
"""Initialize by passing in init function (default: `None`).
|
|
|
|
"""Initialize by passing in init function (default: `None`).
|
|
|
|
|
|
|
|
|
|
|
|
Args:
|
|
|
|
Args:
|
|
|
@ -291,19 +297,17 @@ class GPTCache(BaseCache):
|
|
|
|
|
|
|
|
|
|
|
|
# Avoid multiple caches using the same file,
|
|
|
|
# Avoid multiple caches using the same file,
|
|
|
|
causing different llm model caches to affect each other
|
|
|
|
causing different llm model caches to affect each other
|
|
|
|
i = 0
|
|
|
|
|
|
|
|
file_prefix = "data_map"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def init_gptcache_map(cache_obj: gptcache.Cache):
|
|
|
|
def init_gptcache(cache_obj: gptcache.Cache, llm str):
|
|
|
|
nonlocal i
|
|
|
|
|
|
|
|
cache_path = f'{file_prefix}_{i}.txt'
|
|
|
|
|
|
|
|
cache_obj.init(
|
|
|
|
cache_obj.init(
|
|
|
|
pre_embedding_func=get_prompt,
|
|
|
|
pre_embedding_func=get_prompt,
|
|
|
|
data_manager=get_data_manager(data_path=cache_path),
|
|
|
|
data_manager=manager_factory(
|
|
|
|
|
|
|
|
manager="map",
|
|
|
|
|
|
|
|
data_dir=f"map_cache_{llm}"
|
|
|
|
|
|
|
|
),
|
|
|
|
)
|
|
|
|
)
|
|
|
|
i += 1
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
langchain.llm_cache = GPTCache(init_gptcache_map)
|
|
|
|
langchain.llm_cache = GPTCache(init_gptcache)
|
|
|
|
|
|
|
|
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
try:
|
|
|
|
try:
|
|
|
@ -314,30 +318,38 @@ class GPTCache(BaseCache):
|
|
|
|
"Please install it with `pip install gptcache`."
|
|
|
|
"Please install it with `pip install gptcache`."
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
self.init_gptcache_func: Optional[Callable[[Any], None]] = init_func
|
|
|
|
self.init_gptcache_func: Union[
|
|
|
|
|
|
|
|
Callable[[Any, str], None], Callable[[Any], None], None
|
|
|
|
|
|
|
|
] = init_func
|
|
|
|
self.gptcache_dict: Dict[str, Any] = {}
|
|
|
|
self.gptcache_dict: Dict[str, Any] = {}
|
|
|
|
|
|
|
|
|
|
|
|
def _get_gptcache(self, llm_string: str) -> Any:
|
|
|
|
def _new_gptcache(self, llm_string: str) -> Any:
|
|
|
|
"""Get a cache object.
|
|
|
|
"""New gptcache object"""
|
|
|
|
|
|
|
|
|
|
|
|
When the corresponding llm model cache does not exist, it will be created."""
|
|
|
|
|
|
|
|
from gptcache import Cache
|
|
|
|
from gptcache import Cache
|
|
|
|
from gptcache.manager.factory import get_data_manager
|
|
|
|
from gptcache.manager.factory import get_data_manager
|
|
|
|
from gptcache.processor.pre import get_prompt
|
|
|
|
from gptcache.processor.pre import get_prompt
|
|
|
|
|
|
|
|
|
|
|
|
_gptcache = self.gptcache_dict.get(llm_string, None)
|
|
|
|
_gptcache = Cache()
|
|
|
|
if _gptcache is None:
|
|
|
|
if self.init_gptcache_func is not None:
|
|
|
|
_gptcache = Cache()
|
|
|
|
sig = inspect.signature(self.init_gptcache_func)
|
|
|
|
if self.init_gptcache_func is not None:
|
|
|
|
if len(sig.parameters) == 2:
|
|
|
|
self.init_gptcache_func(_gptcache)
|
|
|
|
self.init_gptcache_func(_gptcache, llm_string) # type: ignore[call-arg]
|
|
|
|
else:
|
|
|
|
else:
|
|
|
|
_gptcache.init(
|
|
|
|
self.init_gptcache_func(_gptcache) # type: ignore[call-arg]
|
|
|
|
pre_embedding_func=get_prompt,
|
|
|
|
else:
|
|
|
|
data_manager=get_data_manager(data_path=llm_string),
|
|
|
|
_gptcache.init(
|
|
|
|
)
|
|
|
|
pre_embedding_func=get_prompt,
|
|
|
|
self.gptcache_dict[llm_string] = _gptcache
|
|
|
|
data_manager=get_data_manager(data_path=llm_string),
|
|
|
|
|
|
|
|
)
|
|
|
|
return _gptcache
|
|
|
|
return _gptcache
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _get_gptcache(self, llm_string: str) -> Any:
|
|
|
|
|
|
|
|
"""Get a cache object.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
When the corresponding llm model cache does not exist, it will be created."""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
return self.gptcache_dict.get(llm_string, self._new_gptcache(llm_string))
|
|
|
|
|
|
|
|
|
|
|
|
def lookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
|
|
|
|
def lookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
|
|
|
|
"""Look up the cache data.
|
|
|
|
"""Look up the cache data.
|
|
|
|
First, retrieve the corresponding cache object using the `llm_string` parameter,
|
|
|
|
First, retrieve the corresponding cache object using the `llm_string` parameter,
|
|
|
|