From d8d776636f9a1492e4df4d4d9eed594801202783 Mon Sep 17 00:00:00 2001 From: soumil-rathi Date: Wed, 15 Apr 2026 02:16:28 -0700 Subject: [PATCH] fix(v3): migration crashes + entity linking on OSS (#4836) Co-authored-by: Soumil Rathi Co-authored-by: Claude Opus 4.6 (1M context) --- mem0-ts/src/oss/src/memory/index.ts | 9 ++--- mem0-ts/src/oss/src/utils/scoring.ts | 10 ----- mem0/memory/main.py | 23 +++++++---- mem0/utils/scoring.py | 5 --- mem0/vector_stores/milvus.py | 57 +++++++++++++++++++--------- mem0/vector_stores/qdrant.py | 44 ++++++++++++++------- tests/test_main.py | 1 - tests/utils/test_scoring.py | 10 ----- 8 files changed, 90 insertions(+), 69 deletions(-) diff --git a/mem0-ts/src/oss/src/memory/index.ts b/mem0-ts/src/oss/src/memory/index.ts index 71a10758d..302c54891 100644 --- a/mem0-ts/src/oss/src/memory/index.ts +++ b/mem0-ts/src/oss/src/memory/index.ts @@ -993,12 +993,9 @@ export class Memory { createdAt: payload.createdAt, updatedAt: payload.updatedAt, score: scored.score, - metadata: { - ...Object.entries(payload) - .filter(([key]) => !excludedKeys.has(key)) - .reduce((acc, [key, value]) => ({ ...acc, [key]: value }), {}), - scoreBreakdown: scored.scoreBreakdown, - }, + metadata: Object.entries(payload) + .filter(([key]) => !excludedKeys.has(key)) + .reduce((acc, [key, value]) => ({ ...acc, [key]: value }), {}), ...(payload.user_id && { user_id: payload.user_id }), ...(payload.agent_id && { agent_id: payload.agent_id }), ...(payload.run_id && { run_id: payload.run_id }), diff --git a/mem0-ts/src/oss/src/utils/scoring.ts b/mem0-ts/src/oss/src/utils/scoring.ts index 6c2f5b1b1..13d0933ce 100644 --- a/mem0-ts/src/oss/src/utils/scoring.ts +++ b/mem0-ts/src/oss/src/utils/scoring.ts @@ -59,11 +59,6 @@ export function normalizeBm25( export interface ScoredResult { id: string; score: number; - scoreBreakdown: { - semantic: number; - bm25: number; - entityBoost: number; - }; payload: Record; } @@ -134,11 +129,6 @@ export function scoreAndRank( scored.push({ id: memIdStr, score: combined, - scoreBreakdown: { - semantic: semanticScore, - bm25: bm25Score, - entityBoost: entityBoost, - }, payload: result.payload, }); } diff --git a/mem0/memory/main.py b/mem0/memory/main.py index fb38de35b..78847bb04 100644 --- a/mem0/memory/main.py +++ b/mem0/memory/main.py @@ -334,6 +334,14 @@ class Memory(MemoryBase): entity_config.collection_name = entity_collection elif isinstance(entity_config, dict): entity_config['collection_name'] = entity_collection + # For Qdrant, share the existing client to avoid RocksDB lock contention + # when using embedded mode (path=...). QdrantConfig.client takes precedence + # over host/port/path. + if self.config.vector_store.provider == "qdrant" and hasattr(self.vector_store, "client"): + if hasattr(entity_config, "client"): + entity_config.client = self.vector_store.client + elif isinstance(entity_config, dict): + entity_config["client"] = self.vector_store.client self._entity_store = VectorStoreFactory.create( self.config.vector_store.provider, entity_config ) @@ -1188,8 +1196,6 @@ class Memory(MemoryBase): entity_boosts = self._compute_entity_boosts(query_entities, filters) # Step 7: Build candidate set from semantic results - # BM25 acts as a boost signal only (not recall-expanding) -- candidates must - # pass the semantic threshold gate, so only semantic results are candidates. candidates = [] for mem in semantic_results: mem_id = str(mem.id) @@ -1234,9 +1240,6 @@ class Memory(MemoryBase): score=scored["score"], ).model_dump() - # Add score breakdown to metadata - memory_item_dict["score_breakdown"] = scored.get("score_breakdown", {}) - for key in promoted_payload_keys: if key in payload: memory_item_dict[key] = payload[key] @@ -1640,6 +1643,14 @@ class AsyncMemory(MemoryBase): entity_config.collection_name = entity_collection elif isinstance(entity_config, dict): entity_config['collection_name'] = entity_collection + # For Qdrant, share the existing client to avoid RocksDB lock contention + # when using embedded mode (path=...). QdrantConfig.client takes precedence + # over host/port/path. + if self.config.vector_store.provider == "qdrant" and hasattr(self.vector_store, "client"): + if hasattr(entity_config, "client"): + entity_config.client = self.vector_store.client + elif isinstance(entity_config, dict): + entity_config["client"] = self.vector_store.client self._entity_store = VectorStoreFactory.create( self.config.vector_store.provider, entity_config ) @@ -2480,8 +2491,6 @@ class AsyncMemory(MemoryBase): score=scored["score"], ).model_dump() - memory_item_dict["score_breakdown"] = scored.get("score_breakdown", {}) - for key in promoted_payload_keys: if key in payload: memory_item_dict[key] = payload[key] diff --git a/mem0/utils/scoring.py b/mem0/utils/scoring.py index 1954f4484..ae0944f0b 100644 --- a/mem0/utils/scoring.py +++ b/mem0/utils/scoring.py @@ -113,11 +113,6 @@ def score_and_rank( { "id": mem_id_str, "score": combined, - "score_breakdown": { - "semantic": semantic_score, - "bm25": bm25_score, - "entity_boost": entity_boost, - }, "payload": result.get("payload"), } ) diff --git a/mem0/vector_stores/milvus.py b/mem0/vector_stores/milvus.py index 30e04940c..6ed99ad75 100644 --- a/mem0/vector_stores/milvus.py +++ b/mem0/vector_stores/milvus.py @@ -46,6 +46,9 @@ class MilvusDB(VectorStoreBase): self.embedding_model_dims = embedding_model_dims self.metric_type = metric_type self.client = MilvusClient(uri=url, token=token, db_name=db_name) + # Whether this collection has the `text` + `sparse` fields for v3 BM25. + # Pre-v3 collections lack them; writing a top-level `text` field is rejected. + self._has_bm25_schema = False self.create_col( collection_name=self.collection_name, vector_size=self.embedding_model_dims, @@ -68,6 +71,15 @@ class MilvusDB(VectorStoreBase): if self.client.has_collection(collection_name): logger.info(f"Collection {collection_name} already exists. Skipping creation.") + desc = self.client.describe_collection(collection_name=collection_name) + field_names = {f.get("name") for f in desc.get("fields", [])} + self._has_bm25_schema = "text" in field_names and "sparse" in field_names + if not self._has_bm25_schema: + logger.warning( + f"Collection '{collection_name}' predates v3 hybrid search (no 'text'/'sparse' fields). " + "BM25 keyword scoring will be disabled for this collection; semantic search works normally. " + "To enable hybrid search, use a fresh collection." + ) else: fields = [ FieldSchema(name="id", dtype=DataType.VARCHAR, is_primary=True, max_length=512), @@ -101,6 +113,7 @@ class MilvusDB(VectorStoreBase): index_name="sparse_index", ) self.client.create_collection(collection_name=collection_name, schema=schema, index_params=index_params) + self._has_bm25_schema = True def insert(self, ids, vectors, payloads, **kwargs: Optional[dict[str, any]]): """Insert vectors into a collection. @@ -110,17 +123,17 @@ class MilvusDB(VectorStoreBase): payloads (List[Dict], optional): List of payloads corresponding to vectors. ids (List[str], optional): List of IDs corresponding to vectors. """ - # Batch insert all records at once for better performance and consistency - data = [ - { - "id": idx, - "vectors": embedding, - "metadata": metadata, + # Batch insert all records at once for better performance and consistency. + # Only include the `text` field when the collection's schema has it — legacy + # collections created pre-v3 reject unknown top-level fields. + def _build_record(idx, embedding, metadata): + record = {"id": idx, "vectors": embedding, "metadata": metadata} + if self._has_bm25_schema: # Populate the text field for BM25 sparse search; prefer lemmatized text, fall back to raw data - "text": (metadata.get("text_lemmatized") or metadata.get("data", ""))[:65535] if metadata else "", - } - for idx, embedding, metadata in zip(ids, vectors, payloads) - ] + record["text"] = (metadata.get("text_lemmatized") or metadata.get("data", ""))[:65535] if metadata else "" + return record + + data = [_build_record(idx, embedding, metadata) for idx, embedding, metadata in zip(ids, vectors, payloads)] self.client.insert(collection_name=self.collection_name, data=data, **kwargs) def _create_filter(self, filters: dict): @@ -179,13 +192,21 @@ class MilvusDB(VectorStoreBase): list: Search results. """ query_filter = self._create_filter(filters) if filters else None - hits = self.client.search( - collection_name=self.collection_name, - data=[vectors], - limit=top_k, - filter=query_filter, - output_fields=["*"], - ) + # v3 collections carry both a dense `vectors` field and a sparse `sparse` + # field (for BM25), which makes anns_field ambiguous — Milvus rejects the + # query otherwise with "multiple anns_fields exist". Legacy single-vector + # collections don't need the hint, so only pass it when the hybrid schema + # is present. + search_kwargs = { + "collection_name": self.collection_name, + "data": [vectors], + "limit": top_k, + "filter": query_filter, + "output_fields": ["*"], + } + if self._has_bm25_schema: + search_kwargs["anns_field"] = "vectors" + hits = self.client.search(**search_kwargs) result = self._parse_output(data=hits[0]) return result @@ -206,6 +227,8 @@ class MilvusDB(VectorStoreBase): list: Search results in the same format as search(), or None if sparse search is not supported on this collection. """ + if not self._has_bm25_schema: + return None try: query_filter = self._create_filter(filters) if filters else None hits = self.client.search( diff --git a/mem0/vector_stores/qdrant.py b/mem0/vector_stores/qdrant.py index ff7486696..a42962782 100644 --- a/mem0/vector_stores/qdrant.py +++ b/mem0/vector_stores/qdrant.py @@ -79,6 +79,10 @@ class Qdrant(VectorStoreBase): self.embedding_model_dims = embedding_model_dims self.on_disk = on_disk self._bm25_encoder = None + # Whether this collection has the `bm25` named sparse vector slot. + # Pre-v3 collections lack it; writing a `bm25` sparse vector into such a + # collection is rejected by Qdrant ("Not existing vector name error: bm25"). + self._has_bm25_slot = False self.create_col(embedding_model_dims, on_disk) def _get_bm25_encoder(self): @@ -127,6 +131,15 @@ class Qdrant(VectorStoreBase): for collection in response.collections: if collection.name == self.collection_name: logger.debug(f"Collection {self.collection_name} already exists. Skipping creation.") + info = self.client.get_collection(self.collection_name) + sparse_cfg = info.config.params.sparse_vectors + self._has_bm25_slot = bool(sparse_cfg and "bm25" in sparse_cfg) + if not self._has_bm25_slot: + logger.warning( + f"Collection '{self.collection_name}' predates v3 hybrid search (no 'bm25' sparse slot). " + "BM25 keyword scoring will be disabled for this collection; semantic search works normally. " + "To enable hybrid search, use a fresh collection." + ) self._create_filter_indexes() return @@ -139,6 +152,7 @@ class Qdrant(VectorStoreBase): ), }, ) + self._has_bm25_slot = True self._create_filter_indexes() def _create_filter_indexes(self): @@ -177,13 +191,14 @@ class Qdrant(VectorStoreBase): payload = payloads[idx] if payloads else {} point_id = idx if ids is None else ids[idx] - # Build named vectors: dense + optional BM25 sparse + # Build named vectors: dense + optional BM25 sparse (only if collection has the slot). named_vectors = {"": vector} - text_for_bm25 = payload.get("text_lemmatized") or payload.get("data", "") - if text_for_bm25: - sparse = self._encode_bm25(text_for_bm25) - if sparse is not None: - named_vectors["bm25"] = sparse + if self._has_bm25_slot: + text_for_bm25 = payload.get("text_lemmatized") or payload.get("data", "") + if text_for_bm25: + sparse = self._encode_bm25(text_for_bm25) + if sparse is not None: + named_vectors["bm25"] = sparse points.append(PointStruct(id=point_id, vector=named_vectors, payload=payload)) @@ -382,7 +397,7 @@ class Qdrant(VectorStoreBase): """Batch search using Qdrant's query_batch_points for efficiency.""" query_filter = self._create_filter(filters) if filters else None requests = [ - models.QueryRequest(query=vec, filter=query_filter, limit=top_k) + models.QueryRequest(query=vec, filter=query_filter, limit=top_k, with_payload=True) for vec in vectors_list ] try: @@ -407,6 +422,8 @@ class Qdrant(VectorStoreBase): Returns: list: Search results, or None if BM25 is not available. """ + if not self._has_bm25_slot: + return None sparse_query = self._encode_bm25(query) if sparse_query is None: return None @@ -449,13 +466,14 @@ class Qdrant(VectorStoreBase): payload (dict, optional): Updated payload. Defaults to None. """ if vector is not None and payload is not None: - # Full update: attach BM25 sparse vector alongside dense vector + # Full update: attach BM25 sparse vector alongside dense vector (only if slot exists). named_vectors = {"": vector} - text_for_bm25 = payload.get("text_lemmatized") or payload.get("data", "") - if text_for_bm25: - sparse = self._encode_bm25(text_for_bm25) - if sparse is not None: - named_vectors["bm25"] = sparse + if self._has_bm25_slot: + text_for_bm25 = payload.get("text_lemmatized") or payload.get("data", "") + if text_for_bm25: + sparse = self._encode_bm25(text_for_bm25) + if sparse is not None: + named_vectors["bm25"] = sparse point = PointStruct(id=vector_id, vector=named_vectors, payload=payload) self.client.upsert(collection_name=self.collection_name, points=[point]) else: diff --git a/tests/test_main.py b/tests/test_main.py index e5459ddd1..2d8d32126 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -110,7 +110,6 @@ def test_search(memory_instance): assert result["results"][0]["user_id"] == "test_user" # Score is now combined score (semantic only since no BM25/entity), still 0.9 assert result["results"][0]["score"] == pytest.approx(0.9) - assert "score_breakdown" in result["results"][0] # Hybrid pipeline over-fetches: max(100*4, 60) = 400 memory_instance.vector_store.search.assert_called_once_with( diff --git a/tests/utils/test_scoring.py b/tests/utils/test_scoring.py index edc381100..062d55809 100644 --- a/tests/utils/test_scoring.py +++ b/tests/utils/test_scoring.py @@ -101,16 +101,6 @@ class TestScoreAndRank: scored = score_and_rank(results, {}, {}, threshold=0.1, top_k=5) assert len(scored) == 5 - def test_score_breakdown_present(self): - results = [{"id": "a", "score": 0.8, "payload": {"data": "x"}}] - bm25 = {"a": 0.4} - entity = {"a": 0.2} - scored = score_and_rank(results, bm25, entity, threshold=0.1, top_k=10) - breakdown = scored[0]["score_breakdown"] - assert breakdown["semantic"] == 0.8 - assert breakdown["bm25"] == 0.4 - assert breakdown["entity_boost"] == 0.2 - def test_adaptive_divisor_semantic_only(self): results = [{"id": "a", "score": 0.8, "payload": {}}] scored = score_and_rank(results, {}, {}, threshold=0.1, top_k=10)