Get prompts that are already cached.
(
params: Dict[str, Any],
prompts: List[str],
cache: Optional[Union[BaseCache, bool, None]] = None,
)
| 137 | |
| 138 | |
| 139 | def get_prompts( |
| 140 | params: Dict[str, Any], |
| 141 | prompts: List[str], |
| 142 | cache: Optional[Union[BaseCache, bool, None]] = None, |
| 143 | ) -> Tuple[Dict[int, List], str, List[int], List[str]]: |
| 144 | """Get prompts that are already cached.""" |
| 145 | llm_string = str(sorted([(k, v) for k, v in params.items()])) |
| 146 | missing_prompts = [] |
| 147 | missing_prompt_idxs = [] |
| 148 | existing_prompts = {} |
| 149 | |
| 150 | llm_cache = _resolve_cache(cache) |
| 151 | for i, prompt in enumerate(prompts): |
| 152 | if llm_cache: |
| 153 | cache_val = llm_cache.lookup(prompt, llm_string) |
| 154 | if isinstance(cache_val, list): |
| 155 | existing_prompts[i] = cache_val |
| 156 | else: |
| 157 | missing_prompts.append(prompt) |
| 158 | missing_prompt_idxs.append(i) |
| 159 | return existing_prompts, llm_string, missing_prompt_idxs, missing_prompts |
| 160 | |
| 161 | |
| 162 | async def aget_prompts( |
no test coverage detected