|
|
|
from functools import lru_cache |
|
from typing import Tuple, Any, Optional,AsyncGenerator, Dict, List |
|
|
|
from config.config import GenerationConfig, ModelConfig |
|
|
|
|
|
|
|
class ResponseCache: |
|
def __init__(self, cache_size: int = 1000): |
|
self.cache_size = cache_size |
|
self._initialize_cache() |
|
|
|
def _initialize_cache(self): |
|
@lru_cache(maxsize=self.cache_size) |
|
def cached_response(prompt: str, config_hash: str) -> Tuple[str, float]: |
|
pass |
|
self.get_cached_response = cached_response |
|
|
|
def cache_response(self, prompt: str, config: GenerationConfig, response: str, score: float) -> None: |
|
config_hash = hash(str(config.__dict__)) |
|
self.get_cached_response(prompt, str(config_hash)) |
|
|
|
def get_response(self, prompt: str, config: GenerationConfig) -> Optional[Tuple[str, float]]: |
|
config_hash = hash(str(config.__dict__)) |
|
return self.get_cached_response(prompt, str(config_hash)) |
|
|
|
|
