Finds a semantically similar query in the cache.
(self, query_embedding: np.ndarray, session_id: Optional[str] = None)
| 123 | return dot_product / (norm_v1 * norm_v2) |
| 124 | |
| 125 | def _find_in_semantic_cache(self, query_embedding: np.ndarray, session_id: Optional[str] = None) -> Optional[Dict[str, Any]]: |
| 126 | """Finds a semantically similar query in the cache.""" |
| 127 | if not self._query_cache or query_embedding is None: |
| 128 | return None |
| 129 | |
| 130 | for key, cached_item in self._query_cache.items(): |
| 131 | cached_embedding = cached_item.get('embedding') |
| 132 | if cached_embedding is None: |
| 133 | continue |
| 134 | |
| 135 | # Respect cache scoping: if scope is session-level, skip results from other sessions |
| 136 | if self.cache_scope == "session" and session_id is not None: |
| 137 | if cached_item.get("session_id") != session_id: |
| 138 | continue |
| 139 | |
| 140 | try: |
| 141 | similarity = self._cosine_similarity(query_embedding, cached_embedding) |
| 142 | |
| 143 | if similarity >= self.semantic_cache_threshold: |
| 144 | print(f"🚀 Semantic cache hit! Similarity: {similarity:.3f} with cached query '{key}'") |
| 145 | return cached_item.get('result') |
| 146 | except ValueError: |
| 147 | # In case of shape mismatch, just skip |
| 148 | continue |
| 149 | |
| 150 | return None |
| 151 | |
| 152 | def _format_query_with_history(self, query: str, history: list) -> str: |
| 153 | """Formats the user query with conversation history for context.""" |
no test coverage detected