Semantic memory backed by embeddings and cosine similarity.
| 148 | |
| 149 | |
| 150 | class VectorMemory(Memory): |
| 151 | """Semantic memory backed by embeddings and cosine similarity.""" |
| 152 | |
| 153 | def __init__( |
| 154 | self, |
| 155 | embedding_function: Callable[[str], np.ndarray], |
| 156 | top_k: int = 10, |
| 157 | query_threshold: float = 0.8, |
| 158 | memory_size: int = 20, |
| 159 | context_label: str = "SEMANTIC_KNOWLEDGE", |
| 160 | *, |
| 161 | passive: bool = True, |
| 162 | active: bool = False, |
| 163 | ): |
| 164 | super().__init__(context_label, passive=passive, active=active) |
| 165 | self._embedding_function = embedding_function |
| 166 | self._memory_size = int(memory_size) |
| 167 | self._top_k = int(top_k) |
| 168 | self._query_threshold = float(query_threshold) |
| 169 | self._memory: dict[str, str] = {} |
| 170 | self._embeddings: dict[str, np.ndarray] = {} |
| 171 | |
| 172 | def insert(self, key: str, value: object): |
| 173 | if self._memory_size > 0 and len(self._memory) >= self._memory_size: |
| 174 | oldest_key = next(iter(self._memory)) |
| 175 | self._memory.pop(oldest_key, None) |
| 176 | self._embeddings.pop(oldest_key, None) |
| 177 | |
| 178 | text_value = str(value) |
| 179 | self._memory[key] = text_value |
| 180 | content_for_embedding = f"{key}: {text_value}" |
| 181 | self._embeddings[key] = self._embedding_function(content_for_embedding) |
| 182 | |
| 183 | def update(self, key: str, value: object): |
| 184 | if key not in self._memory: |
| 185 | return |
| 186 | text_value = str(value) |
| 187 | self._memory[key] = text_value |
| 188 | content_for_embedding = f"{key}: {text_value}" |
| 189 | self._embeddings[key] = self._embedding_function(content_for_embedding) |
| 190 | |
| 191 | def delete(self, key: str, index: int = -1): |
| 192 | self._memory.pop(key, None) |
| 193 | self._embeddings.pop(key, None) |
| 194 | |
| 195 | def reset(self): |
| 196 | self._memory = {} |
| 197 | self._embeddings = {} |
| 198 | |
| 199 | def get_blocks(self, query: ContextQuery, *, top_k: int = 8) -> list[ContextBlock]: |
| 200 | if not self._memory: |
| 201 | return [] |
| 202 | |
| 203 | query_text = (query.query_text or "").strip() |
| 204 | if not query_text: |
| 205 | return [] |
| 206 | |
| 207 | query_embedding = self._embedding_function(query_text) |