diff --git a/mem0-ts/src/oss/src/memory/index.ts b/mem0-ts/src/oss/src/memory/index.ts index 1fd7bf88a..f2856c493 100644 --- a/mem0-ts/src/oss/src/memory/index.ts +++ b/mem0-ts/src/oss/src/memory/index.ts @@ -262,6 +262,181 @@ export class Memory { return this._entityStore; } + /** + * Normalize a filters object for entity-store scoping: keeps only + * user_id/agent_id/run_id keys whose values are defined. + */ + private _sessionFiltersFromPayload( + payload: Record, + ): Record { + const filters: Record = {}; + if (payload.user_id) filters.user_id = payload.user_id; + if (payload.agent_id) filters.agent_id = payload.agent_id; + if (payload.run_id) filters.run_id = payload.run_id; + return filters; + } + + /** + * Remove `memoryId` from every entity record scoped to `filters`. + * If an entity's `linkedMemoryIds` becomes empty after removal, the + * entity record itself is deleted. Errors on individual entities are + * swallowed so one bad record does not break the whole operation. + * + * No-op if the entity store has not been initialized yet. + */ + private async _removeMemoryFromEntityStore( + memoryId: string, + filters: Record, + ): Promise { + let entityStore: VectorStore; + try { + entityStore = await this.getEntityStore(); + } catch (e) { + console.debug(`Entity store unavailable during cleanup: ${e}`); + return; + } + + let rows: Array<{ id: string; payload: Record }> = []; + try { + const listed = await entityStore.list(filters, 10000); + rows = ( + Array.isArray(listed) && Array.isArray(listed[0]) + ? listed[0] + : (listed as any) + ) as Array<{ id: string; payload: Record }>; + } catch (e) { + console.debug(`Entity store list failed during cleanup: ${e}`); + return; + } + + for (const row of rows) { + try { + const payload = row.payload || {}; + const linked: string[] = Array.isArray(payload.linkedMemoryIds) + ? payload.linkedMemoryIds + : []; + if (!linked.includes(memoryId)) continue; + + const remaining = linked.filter((id) => id !== memoryId); + if (remaining.length === 0) { + try { + await entityStore.delete(row.id); + } catch (e) { + console.debug(`Entity delete failed for id=${row.id}: ${e}`); + } + } else { + const newPayload = { ...payload, linkedMemoryIds: remaining }; + // entityStore.update requires a vector — re-embed entity text. + const entityText = + typeof payload.data === "string" ? payload.data : ""; + if (!entityText) { + // Can't re-embed without text; skip gracefully. + console.debug( + `Entity id=${row.id} missing 'data'; skipping update during cleanup`, + ); + continue; + } + let vec: number[]; + try { + vec = await this.embedder.embed(entityText); + } catch (e) { + console.debug(`Entity re-embed failed for '${entityText}': ${e}`); + continue; + } + try { + await entityStore.update(row.id, vec, newPayload); + } catch (e) { + console.debug(`Entity update failed for id=${row.id}: ${e}`); + } + } + } catch (e) { + console.debug(`Entity cleanup error for id=${row?.id}: ${e}`); + } + } + } + + /** + * Extract entities from `text` and link them to `memoryId` in the + * entity store, scoped to `filters` (user_id / agent_id / run_id). + * + * Simpler single-memory variant of Phase 7 in add(): no cross-memory + * dedup, but still does per-entity "search for existing, update if + * match >= 0.95 else insert new". Non-fatal errors are swallowed. + */ + private async _linkEntitiesForMemory( + memoryId: string, + text: string, + filters: Record, + ): Promise { + try { + const entities = extractEntities(text); + if (entities.length === 0) return; + + const entityStore = await this.getEntityStore(); + + for (const entity of entities) { + try { + let entityVec: number[]; + try { + entityVec = await this.embedder.embed(entity.text); + } catch (e) { + console.debug(`Entity embed failed for '${entity.text}': ${e}`); + continue; + } + + let matches: Array<{ + id: string; + score?: number; + payload: Record; + }> = []; + try { + matches = await entityStore.search(entityVec, 1, filters); + } catch {} + + if (matches.length > 0 && (matches[0].score ?? 0) >= 0.95) { + const match = matches[0]; + const payload = match.payload || {}; + const linked = new Set( + Array.isArray(payload.linkedMemoryIds) + ? payload.linkedMemoryIds + : [], + ); + linked.add(memoryId); + payload.linkedMemoryIds = Array.from(linked).sort(); + try { + await entityStore.update(match.id, entityVec, payload); + } catch (e) { + console.debug(`Entity update failed for '${entity.text}': ${e}`); + } + } else { + const entityPayload: Record = { + data: entity.text, + entityType: entity.type, + linkedMemoryIds: [memoryId], + }; + if (filters.user_id) entityPayload.user_id = filters.user_id; + if (filters.agent_id) entityPayload.agent_id = filters.agent_id; + if (filters.run_id) entityPayload.run_id = filters.run_id; + + try { + await entityStore.insert( + [entityVec], + [uuidv4()], + [entityPayload], + ); + } catch (e) { + console.debug(`Entity insert failed for '${entity.text}': ${e}`); + } + } + } catch (e) { + console.debug(`Entity link error for '${entity.text}': ${e}`); + } + } + } catch (e) { + console.warn(`Entity linking failed during update: ${e}`); + } + } + private buildSessionScope(filters: SearchFilters): string { const parts: string[] = []; for (const key of ["agent_id", "run_id", "user_id"].sort()) { @@ -863,17 +1038,23 @@ export class Memory { // Validate search parameters (before applying defaults) validateSearchParams(config.threshold, config.topK); - // Validate and trim entity IDs in filters - const normalizedFilters = config.filters - ? { - ...config.filters, - user_id: validateAndTrimEntityId(config.filters.user_id, "user_id"), - agent_id: validateAndTrimEntityId( - config.filters.agent_id, - "agent_id", - ), - run_id: validateAndTrimEntityId(config.filters.run_id, "run_id"), - } + // Validate and trim entity IDs in filters. Only include keys whose + // validated value is defined — otherwise downstream vector stores + // receive `agent_id: undefined` / `run_id: undefined` and fail + // (Qdrant rejects the malformed match, pgvector binds NULL, Redis + // emits a literal "undefined" string in TAG filters). + const normalizedFilters: Record = config.filters + ? Object.fromEntries( + Object.entries({ + ...config.filters, + user_id: validateAndTrimEntityId(config.filters.user_id, "user_id"), + agent_id: validateAndTrimEntityId( + config.filters.agent_id, + "agent_id", + ), + run_id: validateAndTrimEntityId(config.filters.run_id, "run_id"), + }).filter(([, v]) => v !== undefined), + ) : {}; await this._ensureInitialized(); @@ -1194,13 +1375,17 @@ export class Memory { const { topK = 20 } = config; - // Validate and trim entity IDs in filters - const filters = { - ...(config.filters || {}), - user_id: validateAndTrimEntityId(config.filters?.user_id, "user_id"), - agent_id: validateAndTrimEntityId(config.filters?.agent_id, "agent_id"), - run_id: validateAndTrimEntityId(config.filters?.run_id, "run_id"), - }; + // Validate and trim entity IDs in filters. Drop keys that resolve to + // undefined so downstream vector stores don't receive + // `agent_id: undefined` / `run_id: undefined` and fail. + const filters: Record = Object.fromEntries( + Object.entries({ + ...(config.filters || {}), + user_id: validateAndTrimEntityId(config.filters?.user_id, "user_id"), + agent_id: validateAndTrimEntityId(config.filters?.agent_id, "agent_id"), + run_id: validateAndTrimEntityId(config.filters?.run_id, "run_id"), + }).filter(([, v]) => v !== undefined), + ); await this._captureEvent("get_all", { topK, @@ -1260,6 +1445,7 @@ export class Memory { ...metadata, data, hash: createHash("md5").update(data).digest("hex"), + textLemmatized: lemmatizeForBm25(data), createdAt: new Date().toISOString(), }; @@ -1317,6 +1503,16 @@ export class Memory { newMetadata.updatedAt, ); + // Entity-store cleanup: strip this memory's id from old-text entities, + // then re-extract entities from the new text and link them back. + try { + const sessionFilters = this._sessionFiltersFromPayload(newMetadata); + await this._removeMemoryFromEntityStore(memoryId, sessionFilters); + await this._linkEntitiesForMemory(memoryId, data, sessionFilters); + } catch (e) { + console.warn(`Entity store cleanup/link failed during update: ${e}`); + } + return memoryId; } @@ -1327,6 +1523,9 @@ export class Memory { } const prevValue = existingMemory.payload.data; + const sessionFilters = this._sessionFiltersFromPayload( + existingMemory.payload || {}, + ); await this.vectorStore.delete(memoryId); await this.db.addHistory( memoryId, @@ -1338,6 +1537,14 @@ export class Memory { 1, ); + // Entity-store cleanup: strip this memory's id from any entity records + // that linked to it. Non-fatal — log and continue on error. + try { + await this._removeMemoryFromEntityStore(memoryId, sessionFilters); + } catch (e) { + console.warn(`Entity store cleanup failed during delete: ${e}`); + } + return memoryId; } diff --git a/mem0-ts/src/oss/src/utils/entity_extraction.ts b/mem0-ts/src/oss/src/utils/entity_extraction.ts index 7b50aaf3f..d6f6d2700 100644 --- a/mem0-ts/src/oss/src/utils/entity_extraction.ts +++ b/mem0-ts/src/oss/src/utils/entity_extraction.ts @@ -660,6 +660,9 @@ export function extractEntities(text: string): ExtractedEntity[] { txt = txt.replace(/\s*:+$/, ""); // Strip leading numbered list markers txt = txt.replace(/^\d+\s*\.\s*/, ""); + // Strip trailing sentence punctuation (".", ",", ";", "!", "?") — otherwise + // "Paris." and "Paris" produce different embeddings and break entity dedup. + txt = txt.replace(/[.,;!?]+$/, "").trim(); if (!txt || txt.length <= 2 || hasArtifacts(txt)) { continue; diff --git a/mem0-ts/src/oss/src/vector_stores/redis.ts b/mem0-ts/src/oss/src/vector_stores/redis.ts index 0fa52488f..f305eefc5 100644 --- a/mem0-ts/src/oss/src/vector_stores/redis.ts +++ b/mem0-ts/src/oss/src/vector_stores/redis.ts @@ -9,6 +9,19 @@ import type { import { VectorStore } from "./base"; import { SearchFilters, VectorStoreConfig, VectorStoreResult } from "../types"; +/** + * Escape RediSearch TAG filter special characters. Any punctuation in the + * value (including `-`, which appears in every UUID) must be backslash- + * escaped, otherwise RediSearch either parses it as an operator (`-` is + * minus, `|` is OR) or rejects the whole expression as a syntax error. + */ +function escapeRedisTagValue(value: unknown): string { + return String(value).replace( + /([,.<>{}\[\]"':;!@#$%^&*()\-+=~|/\\\s])/g, + "\\$1", + ); +} + interface RedisConfig extends VectorStoreConfig { redisUrl: string; collectionName: string; @@ -377,8 +390,8 @@ export class RedisDB implements VectorStore { const snakeFilters = filters ? toSnakeCase(filters) : undefined; const filterExpr = snakeFilters ? Object.entries(snakeFilters) - .filter(([_, value]) => value !== null) - .map(([key, value]) => `@${key}:{${value}}`) + .filter(([_, value]) => value !== null && value !== undefined) + .map(([key, value]) => `@${key}:{${escapeRedisTagValue(value)}}`) .join(" ") : "*"; @@ -615,8 +628,8 @@ export class RedisDB implements VectorStore { const snakeFilters = filters ? toSnakeCase(filters) : undefined; const filterExpr = snakeFilters ? Object.entries(snakeFilters) - .filter(([_, value]) => value !== null) - .map(([key, value]) => `@${key}:{${value}}`) + .filter(([_, value]) => value !== null && value !== undefined) + .map(([key, value]) => `@${key}:{${escapeRedisTagValue(value)}}`) .join(" ") : "*"; diff --git a/mem0/memory/main.py b/mem0/memory/main.py index 33783c237..28095de9b 100644 --- a/mem0/memory/main.py +++ b/mem0/memory/main.py @@ -453,6 +453,84 @@ class Memory(MemoryBase): except Exception as e: logger.warning(f"Entity upsert failed for '{entity_text}': {e}") + def _remove_memory_from_entity_store(self, memory_id, filters): + """Strip `memory_id` from every entity record scoped to `filters`. + + For each entity whose `linked_memory_ids` contains `memory_id`: + - remove the id; if the list becomes empty, delete the entity record. + - otherwise re-embed the entity text and update the payload + (the vector store's update() requires a vector). + + No-op if the entity store has never been initialized in this process. + Errors on individual entities are swallowed at debug level; outer + failures are swallowed at warning level so the primary delete/update + path is never broken by entity cleanup. + """ + if self._entity_store is None: + return + search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v} + try: + listed = self.entity_store.list(filters=search_filters, top_k=10000) + rows = listed[0] if isinstance(listed, (list, tuple)) and listed and isinstance(listed[0], list) else listed + for row in rows or []: + try: + payload = getattr(row, "payload", None) or {} + linked = payload.get("linked_memory_ids", []) + if not isinstance(linked, list) or memory_id not in linked: + continue + remaining = [mid for mid in linked if mid != memory_id] + if not remaining: + try: + self.entity_store.delete(vector_id=row.id) + except Exception as e: + logger.debug(f"Entity delete failed for id={row.id}: {e}") + else: + entity_text = payload.get("data") + if not isinstance(entity_text, str) or not entity_text: + logger.debug(f"Entity id={row.id} missing 'data'; skipping update during cleanup") + continue + try: + vec = self.embedding_model.embed(entity_text, "update") + except Exception as e: + logger.debug(f"Entity re-embed failed for '{entity_text}': {e}") + continue + new_payload = {**payload, "linked_memory_ids": remaining} + try: + self.entity_store.update( + vector_id=row.id, + vector=vec, + payload=new_payload, + ) + except Exception as e: + logger.debug(f"Entity update failed for id={row.id}: {e}") + except Exception as e: + logger.debug(f"Entity cleanup error: {e}") + except Exception as e: + logger.warning(f"Entity store cleanup failed for memory_id={memory_id}: {e}") + + def _link_entities_for_memory(self, memory_id, text, filters): + """Extract entities from `text` and link them to `memory_id` in the + entity store, scoped to `filters`. Simpler single-memory variant of + Phase 7 in add(): per-entity search-then-update-or-insert via the + existing `_upsert_entity` helper. Non-fatal on any failure. + """ + try: + entities = extract_entities(text) + if not entities: + return + seen = set() + for entity_type, entity_text in entities: + key = entity_text.strip().lower() + if not key or key in seen: + continue + seen.add(key) + try: + self._upsert_entity(entity_text, entity_type, memory_id, filters) + except Exception as e: + logger.debug(f"Entity link failed for '{entity_text}': {e}") + except Exception as e: + logger.warning(f"Entity linking failed for memory_id={memory_id}: {e}") + @classmethod def from_config(cls, config_dict: Dict[str, Any]): try: @@ -1624,6 +1702,13 @@ class Memory(MemoryBase): actor_id=new_metadata.get("actor_id"), role=new_metadata.get("role"), ) + + # Entity-store cleanup: strip this memory's id from old-text entities, + # then re-extract entities from the new text and link them back. + session_filters = {k: new_metadata[k] for k in ("user_id", "agent_id", "run_id") if new_metadata.get(k)} + self._remove_memory_from_entity_store(memory_id, session_filters) + self._link_entities_for_memory(memory_id, data, session_filters) + return memory_id def _delete_memory(self, memory_id, existing_memory=None): @@ -1635,6 +1720,8 @@ class Memory(MemoryBase): prev_value = existing_memory.payload.get("data", "") created_at = _normalize_iso_timestamp_to_utc(existing_memory.payload.get("created_at")) updated_at = datetime.now(timezone.utc).isoformat() + payload = existing_memory.payload or {} + session_filters = {k: payload[k] for k in ("user_id", "agent_id", "run_id") if payload.get(k)} self.vector_store.delete(vector_id=memory_id) self.db.add_history( memory_id, @@ -1647,6 +1734,11 @@ class Memory(MemoryBase): role=existing_memory.payload.get("role"), is_deleted=1, ) + + # Entity-store cleanup: strip this memory's id from any entity records + # that linked to it. Non-fatal — the helper swallows errors. + self._remove_memory_from_entity_store(memory_id, session_filters) + return memory_id def reset(self): @@ -1753,6 +1845,114 @@ class AsyncMemory(MemoryBase): ) return self._entity_store + async def _upsert_entity_async(self, entity_text, entity_type, memory_id, filters): + """Async variant of `_upsert_entity` — per-entity search-then-update-or-insert.""" + try: + entity_embedding = await asyncio.to_thread(self.embedding_model.embed, entity_text, "add") + search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v} + + existing = await asyncio.to_thread( + self.entity_store.search, + query=entity_text, + vectors=entity_embedding, + top_k=1, + filters=search_filters, + ) + + if existing and existing[0].score >= 0.95: + match = existing[0] + payload = match.payload or {} + linked_ids = payload.get("linked_memory_ids", []) + if memory_id not in linked_ids: + linked_ids.append(memory_id) + payload["linked_memory_ids"] = linked_ids + await asyncio.to_thread( + self.entity_store.update, + vector_id=match.id, + vector=None, + payload=payload, + ) + else: + entity_id = str(uuid.uuid4()) + entity_payload = { + "data": entity_text, + "entity_type": entity_type, + "linked_memory_ids": [memory_id], + **{k: v for k, v in search_filters.items()}, + } + await asyncio.to_thread( + self.entity_store.insert, + vectors=[entity_embedding], + ids=[entity_id], + payloads=[entity_payload], + ) + except Exception as e: + logger.warning(f"Entity upsert failed for '{entity_text}' (async): {e}") + + async def _remove_memory_from_entity_store(self, memory_id, filters): + """Async variant of `Memory._remove_memory_from_entity_store`.""" + if self._entity_store is None: + return + search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v} + try: + listed = await asyncio.to_thread(self.entity_store.list, filters=search_filters, top_k=10000) + rows = listed[0] if isinstance(listed, (list, tuple)) and listed and isinstance(listed[0], list) else listed + for row in rows or []: + try: + payload = getattr(row, "payload", None) or {} + linked = payload.get("linked_memory_ids", []) + if not isinstance(linked, list) or memory_id not in linked: + continue + remaining = [mid for mid in linked if mid != memory_id] + if not remaining: + try: + await asyncio.to_thread(self.entity_store.delete, vector_id=row.id) + except Exception as e: + logger.debug(f"Entity delete failed for id={row.id} (async): {e}") + else: + entity_text = payload.get("data") + if not isinstance(entity_text, str) or not entity_text: + logger.debug(f"Entity id={row.id} missing 'data'; skipping update during cleanup (async)") + continue + try: + vec = await asyncio.to_thread(self.embedding_model.embed, entity_text, "update") + except Exception as e: + logger.debug(f"Entity re-embed failed for '{entity_text}' (async): {e}") + continue + new_payload = {**payload, "linked_memory_ids": remaining} + try: + await asyncio.to_thread( + self.entity_store.update, + vector_id=row.id, + vector=vec, + payload=new_payload, + ) + except Exception as e: + logger.debug(f"Entity update failed for id={row.id} (async): {e}") + except Exception as e: + logger.debug(f"Entity cleanup error (async): {e}") + except Exception as e: + logger.warning(f"Entity store cleanup failed for memory_id={memory_id} (async): {e}") + + async def _link_entities_for_memory(self, memory_id, text, filters): + """Async variant of `Memory._link_entities_for_memory`.""" + try: + entities = await asyncio.to_thread(extract_entities, text) + if not entities: + return + seen = set() + for entity_type, entity_text in entities: + key = entity_text.strip().lower() + if not key or key in seen: + continue + seen.add(key) + try: + await self._upsert_entity_async(entity_text, entity_type, memory_id, filters) + except Exception as e: + logger.debug(f"Entity link failed for '{entity_text}' (async): {e}") + except Exception as e: + logger.warning(f"Entity linking failed for memory_id={memory_id} (async): {e}") + @classmethod def from_config(cls, config_dict: Dict[str, Any]): try: @@ -2926,6 +3126,13 @@ class AsyncMemory(MemoryBase): actor_id=new_metadata.get("actor_id"), role=new_metadata.get("role"), ) + + # Entity-store cleanup: strip this memory's id from old-text entities, + # then re-extract entities from the new text and link them back. + session_filters = {k: new_metadata[k] for k in ("user_id", "agent_id", "run_id") if new_metadata.get(k)} + await self._remove_memory_from_entity_store(memory_id, session_filters) + await self._link_entities_for_memory(memory_id, data, session_filters) + return memory_id async def _delete_memory(self, memory_id, existing_memory=None): @@ -2937,6 +3144,8 @@ class AsyncMemory(MemoryBase): prev_value = existing_memory.payload.get("data", "") created_at = _normalize_iso_timestamp_to_utc(existing_memory.payload.get("created_at")) updated_at = datetime.now(timezone.utc).isoformat() + payload = existing_memory.payload or {} + session_filters = {k: payload[k] for k in ("user_id", "agent_id", "run_id") if payload.get(k)} await asyncio.to_thread(self.vector_store.delete, vector_id=memory_id) await asyncio.to_thread( @@ -2952,6 +3161,10 @@ class AsyncMemory(MemoryBase): is_deleted=1, ) + # Entity-store cleanup: strip this memory's id from any entity records + # that linked to it. Non-fatal — the helper swallows errors. + await self._remove_memory_from_entity_store(memory_id, session_filters) + return memory_id async def reset(self):