From 5f7d879178c50c224e5e100a65d13416b1a3aeb6 Mon Sep 17 00:00:00 2001 From: "xuyan.wxy" Date: Tue, 16 Jun 2026 20:33:05 +0800 Subject: [PATCH] fix memory search fallback without embeddings --- src/powermem/core/async_memory.py | 49 +++- src/powermem/core/base.py | 8 + src/powermem/core/memory.py | 67 ++++- src/powermem/storage/adapter.py | 259 ++++++++++++++---- src/powermem/storage/oceanbase/oceanbase.py | 139 ++++++++-- .../storage/sqlite/sqlite_vector_store.py | 231 ++++++++++++++-- src/server/api/v1/search.py | 47 ++++ src/server/models/request.py | 38 ++- src/server/services/search_service.py | 14 + tests/integration/test_noop_embedding_mode.py | 11 + .../test_sqlite_fts_integration.py | 15 + .../server/test_search_score_threshold.py | 78 +++++- tests/unit/test_list_memory_filters.py | 52 ++++ tests/unit/test_memory.py | 175 ++++++++++++ tests/unit/test_sqlite_fts.py | 253 +++++++++++++++++ 15 files changed, 1332 insertions(+), 104 deletions(-) diff --git a/src/powermem/core/async_memory.py b/src/powermem/core/async_memory.py index 176d19539..a6cf06ebc 100644 --- a/src/powermem/core/async_memory.py +++ b/src/powermem/core/async_memory.py @@ -253,6 +253,13 @@ def _is_llm_disabled(self) -> bool: """Return True when PowerMem is running without LLM-backed features.""" return self.llm_provider == "noop" or getattr(self.llm, "is_noop", False) is True + def _is_embedding_disabled(self) -> bool: + """Return True when embedding is explicitly disabled (EMBEDDING_PROVIDER=none).""" + return ( + getattr(self, "embedding_provider", None) == "none" + or getattr(getattr(self, "embedding", None), "is_noop", False) is True + ) + def _get_component_config(self, component: str) -> Dict[str, Any]: """ Helper method to get component configuration uniformly. @@ -1046,6 +1053,13 @@ async def search( filters: Optional[Dict[str, Any]] = None, limit: int = 30, threshold: Optional[float] = None, + retrieval_mode: str = "auto", + fusion: str = "rrf", + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + rrf_k: int = 60, + candidate_limit: Optional[int] = None, + include_explanation: bool = False, ) -> Dict[str, Any]: """Search for memories asynchronously. @@ -1073,8 +1087,23 @@ async def search( # Select embedding service based on filters (for sub-store routing) embedding_service = self._get_embedding_service(filters) - # Generate query embedding asynchronously - query_embedding = await asyncio.to_thread(embedding_service.embed, query, memory_action="search") + query_embedding = None + if retrieval_mode != "fts": + if self._is_embedding_disabled(): + return { + "results": [], + "relations": [] + } + try: + query_embedding = await asyncio.to_thread( + embedding_service.embed, query, memory_action="search" + ) + except Exception as exc: + logger.warning( + "Search embedding failed; falling back to text search " + "when available: %s", + exc, + ) # Search in storage asynchronously - pass query text to enable hybrid search @@ -1085,7 +1114,15 @@ async def search( run_id=run_id, filters=filters, limit=limit, - query=query # Pass query text for hybrid search (vector + full-text) + query=query, # Pass query text for hybrid search (vector + full-text) + threshold=threshold, + retrieval_mode=retrieval_mode, + fusion=fusion, + vector_weight=vector_weight, + fts_weight=fts_weight, + rrf_k=rrf_k, + candidate_limit=candidate_limit, + include_explanation=include_explanation, ) # Process results with intelligence manager (only if enabled to avoid unnecessary calls) @@ -1125,7 +1162,9 @@ async def search( # Quality score represents absolute similarity quality (0-1 range) # It's calculated from weighted average of all search paths' similarity scores metadata = result.get("metadata", {}) - quality_score = metadata.get("_quality_score") + quality_score = result.get("_quality_score") + if quality_score is None: + quality_score = metadata.get("_quality_score") # If quality_score is not available (e.g., from older data or non-hybrid search), # fall back to using the ranking score @@ -1139,7 +1178,7 @@ async def search( transformed_result = { "memory": result.get("memory", ""), - "metadata": metadata, # Keep metadata as-is from storage (includes debug info like _quality_score) + "metadata": metadata, "score": score, } # Preserve other fields if needed diff --git a/src/powermem/core/base.py b/src/powermem/core/base.py index cd4d84620..c89f59da9 100644 --- a/src/powermem/core/base.py +++ b/src/powermem/core/base.py @@ -51,6 +51,14 @@ def search( run_id: Optional[str] = None, filters: Optional[Dict[str, Any]] = None, limit: int = 30, + threshold: Optional[float] = None, + retrieval_mode: str = "auto", + fusion: str = "rrf", + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + rrf_k: int = 60, + candidate_limit: Optional[int] = None, + include_explanation: bool = False, ) -> Dict[str, Any]: """ Search for memories. diff --git a/src/powermem/core/memory.py b/src/powermem/core/memory.py index 28288e5dc..f76d74472 100644 --- a/src/powermem/core/memory.py +++ b/src/powermem/core/memory.py @@ -254,6 +254,14 @@ def search( run_id: Optional[str] = None, filters: Optional[Dict[str, Any]] = None, limit: int = 30, + threshold: Optional[float] = None, + retrieval_mode: str = "auto", + fusion: str = "rrf", + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + rrf_k: int = 60, + candidate_limit: Optional[int] = None, + include_explanation: bool = False, **_: Any, ) -> Dict[str, Any]: data = self._request( @@ -266,6 +274,14 @@ def search( "run_id": run_id, "filters": filters, "limit": limit, + "threshold": threshold, + "retrieval_mode": retrieval_mode, + "fusion": fusion, + "vector_weight": vector_weight, + "fts_weight": fts_weight, + "rrf_k": rrf_k, + "candidate_limit": candidate_limit, + "include_explanation": include_explanation, }, ) results = data.get("results", []) if isinstance(data, dict) else [] @@ -807,7 +823,10 @@ def _is_llm_disabled(self) -> bool: def _is_embedding_disabled(self) -> bool: """Return True when embedding is explicitly disabled (EMBEDDING_PROVIDER=none).""" - return self.embedding_provider == "none" or getattr(self.embedding, "is_noop", False) is True + return ( + getattr(self, "embedding_provider", None) == "none" + or getattr(getattr(self, "embedding", None), "is_noop", False) is True + ) def _embed(self, text: str) -> Optional[List[float]]: """Embed text, returning None when embedding is disabled or fails.""" @@ -1700,6 +1719,13 @@ def search( filters: Optional[Dict[str, Any]] = None, limit: int = 30, threshold: Optional[float] = None, + retrieval_mode: str = "auto", + fusion: str = "rrf", + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + rrf_k: int = 60, + candidate_limit: Optional[int] = None, + include_explanation: bool = False, ) -> Dict[str, Any]: """Search for memories. @@ -1727,6 +1753,13 @@ def search( filters=filters, limit=limit, threshold=threshold, + retrieval_mode=retrieval_mode, + fusion=fusion, + vector_weight=vector_weight, + fts_weight=fts_weight, + rrf_k=rrf_k, + candidate_limit=candidate_limit, + include_explanation=include_explanation, ) if not query or not query.strip(): @@ -1738,8 +1771,23 @@ def search( # Select embedding service based on filters (for sub-store routing) embedding_service = self._get_embedding_service(filters) - # Generate query embedding - query_embedding = embedding_service.embed(query, memory_action="search") + query_embedding = None + if retrieval_mode != "fts": + if self._is_embedding_disabled(): + return { + "results": [], + "relations": [] + } + try: + query_embedding = embedding_service.embed( + query, memory_action="search" + ) + except Exception as exc: + logger.warning( + "Search embedding failed; falling back to text search " + "when available: %s", + exc, + ) # Search in storage - pass query text to enable hybrid search @@ -1752,6 +1800,13 @@ def search( limit=limit, query=query, # Pass query text for hybrid search (vector + full-text + sparse vector) threshold=threshold, # Pass threshold to storage for native hybrid search condition check + retrieval_mode=retrieval_mode, + fusion=fusion, + vector_weight=vector_weight, + fts_weight=fts_weight, + rrf_k=rrf_k, + candidate_limit=candidate_limit, + include_explanation=include_explanation, ) # Process results with intelligence manager (only if enabled to avoid unnecessary calls) @@ -1805,7 +1860,9 @@ def search( # Quality score represents absolute similarity quality (0-1 range) # It's calculated from weighted average of all search paths' similarity scores metadata = result.get("metadata", {}) - quality_score = metadata.get("_quality_score") + quality_score = result.get("_quality_score") + if quality_score is None: + quality_score = metadata.get("_quality_score") # If quality_score is not available (e.g., from older data or non-hybrid search), # fall back to using the ranking score @@ -1819,7 +1876,7 @@ def search( transformed_result = { "memory": result.get("memory", ""), - "metadata": metadata, # Keep metadata as-is from storage (includes debug info like _quality_score) + "metadata": metadata, "score": score, } # Preserve other fields if needed diff --git a/src/powermem/storage/adapter.py b/src/powermem/storage/adapter.py index ded89141b..798083764 100644 --- a/src/powermem/storage/adapter.py +++ b/src/powermem/storage/adapter.py @@ -24,14 +24,20 @@ class StorageAdapter: "actor_id", "category", "created_at", - "data", - "fulltext_content", - "hash", "role", "sparse_embedding", "type", "updated_at", } + _INTERNAL_SEARCH_PAYLOAD_KEYS = { + "_vector_similarity", + "_fts_score", + "_sparse_similarity", + "_quality_score", + "_fusion_score", + "_fusion_info", + "_rerank_score", + } def __init__(self, vector_store: VectorStoreBase, embedding_service=None, sparse_embedder_service=None): """Initialize the adapter with a vector store and embedding service.""" @@ -68,15 +74,34 @@ def _generate_sparse_embedding(self, content: str, memory_action: str) -> Option logger.warning(f"Failed to generate sparse embedding ({memory_action}): {e}") return None - def _metadata_filter_key_for_store(self, key: str) -> str: + def _supports_text_search_without_vector(self, target_store: VectorStoreBase) -> bool: + """Return whether a backend can search by text when embeddings are absent.""" + store_module = target_store.__class__.__module__ + if store_module.endswith("sqlite.sqlite_vector_store"): + return True + if ".oceanbase." in store_module: + return bool(getattr(target_store, "hybrid_search", False)) + return False + + def _metadata_filter_key_for_store( + self, + key: str, + target_store: Optional[VectorStoreBase] = None, + ) -> str: """Translate logical metadata filters to backend-specific payload paths.""" - if key in self._PAYLOAD_FILTER_KEYS: - return key - store_module = self.vector_store.__class__.__module__ + store = target_store or self.vector_store + store_module = store.__class__.__module__ payload_nested_store = ( store_module.endswith("sqlite.sqlite_vector_store") or ".pgvector." in store_module ) + + if key in self._SYSTEM_FILTER_KEYS: + return key + if key.startswith("payload."): + return key[len("payload."):] + if key in self._PAYLOAD_FILTER_KEYS: + return key if key.startswith("metadata."): return key if payload_nested_store else key[len("metadata."):] if payload_nested_store: @@ -89,6 +114,7 @@ def _build_db_filters( agent_id: Optional[str] = None, run_id: Optional[str] = None, filters: Optional[Dict[str, Any]] = None, + target_store: Optional[VectorStoreBase] = None, ) -> Dict[str, Any]: """Build filters that can be executed by the vector store.""" db_filters: Dict[str, Any] = {} @@ -100,15 +126,15 @@ def _build_db_filters( db_filters["run_id"] = run_id if filters: for key, value in filters.items(): - if key in self._SYSTEM_FILTER_KEYS: - if key not in db_filters: - db_filters[key] = value - else: - db_filters[self._metadata_filter_key_for_store(key)] = value + db_key = self._metadata_filter_key_for_store(key, target_store) + if db_key not in db_filters: + db_filters[db_key] = value return db_filters def _memory_matches_filter(self, memory: Dict[str, Any], key: str, expected: Any) -> bool: """Match logical filters against normalized memory payloads.""" + if key.startswith("payload."): + key = key[len("payload."):] actual = memory.get(key) metadata = memory.get("metadata") if actual is None and isinstance(metadata, dict): @@ -193,7 +219,7 @@ def add_memory(self, memory_data: Dict[str, Any]) -> int: def search_memories( self, - query_embedding: List[float], + query_embedding: Optional[List[float]], user_id: Optional[str] = None, agent_id: Optional[str] = None, run_id: Optional[str] = None, @@ -201,38 +227,93 @@ def search_memories( limit: int = 30, query: Optional[str] = None, threshold: Optional[float] = None, + retrieval_mode: str = "auto", + fusion: str = "rrf", + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + rrf_k: int = 60, + candidate_limit: Optional[int] = None, + include_explanation: bool = False, ) -> List[Dict[str, Any]]: """Search for memories.""" - # Use the provided query embedding or generate one - if query_embedding: - query_vector = query_embedding - else: - # If no query embedding provided, we can't search meaningfully - logger.warning("No query embedding provided for search") + mode = (retrieval_mode or "auto").lower() + search_query = query if query else "" + has_query = bool(search_query.strip()) + query_vector = query_embedding if query_embedding else None + + if mode not in {"auto", "fts", "vector", "hybrid"}: + raise ValueError(f"Invalid retrieval mode: {retrieval_mode}") + fusion = (fusion or "rrf").lower() + if fusion not in {"rrf", "weighted"}: + raise ValueError(f"Invalid fusion method: {fusion}") + if vector_weight is not None and not 0.0 <= vector_weight <= 1.0: + raise ValueError("vector_weight must be between 0.0 and 1.0") + if fts_weight is not None and not 0.0 <= fts_weight <= 1.0: + raise ValueError("fts_weight must be between 0.0 and 1.0") + if rrf_k < 1: + raise ValueError("rrf_k must be >= 1") + if candidate_limit is not None and candidate_limit < 1: + raise ValueError("candidate_limit must be >= 1") + + if mode == "fts": + query_vector = None + elif mode == "vector": + search_query = "" + has_query = False + if query_vector is None: + logger.warning("Vector retrieval requested without query embedding") + return [] + elif mode == "hybrid" and query_vector is None: + if has_query: + logger.warning("Hybrid retrieval missing query embedding; falling back to FTS") + else: + logger.warning("Hybrid retrieval requested without query or query embedding") + return [] + elif mode == "auto" and query_vector is None and not has_query: + logger.warning("No query embedding or text query provided for search") return [] # Generate sparse embedding if sparse embedder service is available and query is provided - sparse_embedding = self._generate_sparse_embedding(query, "search") if query else None + sparse_embedding = ( + self._generate_sparse_embedding(search_query, "search") + if has_query + else None + ) - # Merge user_id/agent_id/run_id into logical filters for sub-store routing, - # then translate metadata filters into backend-specific payload paths for search. - effective_filters = filters.copy() if filters else {} + # Route with logical filters, then translate them for the selected store. + routing_filters = filters.copy() if filters else {} if user_id is not None: - effective_filters["user_id"] = user_id + routing_filters["user_id"] = user_id if agent_id is not None: - effective_filters["agent_id"] = agent_id + routing_filters["agent_id"] = agent_id if run_id is not None: - effective_filters["run_id"] = run_id - db_filters = self._build_db_filters(user_id, agent_id, run_id, filters) - - # Route to target store (main or sub store) - target_store = self._route_to_store(effective_filters) + routing_filters["run_id"] = run_id + + target_store = self._route_to_store(routing_filters) + effective_filters = self._build_db_filters( + user_id, + agent_id, + run_id, + filters, + target_store=target_store, + ) + if ( + query_vector is None + and has_query + and not self._supports_text_search_without_vector(target_store) + ): + logger.warning( + "No query embedding provided and target store does not support " + "text-only search fallback" + ) + return [] # Unified search method - try OceanBase format first, fallback to SQLite # Pass query text to enable hybrid search (vector + full-text search) + search_limit = candidate_limit if candidate_limit is not None else limit + search_vectors = query_vector try: # Try OceanBase format first - pass query text for hybrid search - search_query = query if query else "" # Check if target_store.search supports sparse_embedding and threshold parameters import inspect search_sig = inspect.signature(target_store.search) @@ -241,24 +322,36 @@ def search_memories( # Build search kwargs based on supported parameters search_kwargs = { "query": search_query, - "vectors": query_vector, - "limit": limit, - "filters": db_filters if db_filters else None, + "vectors": search_vectors, + "limit": search_limit, + "filters": effective_filters or None, } if 'sparse_embedding' in search_params: search_kwargs["sparse_embedding"] = sparse_embedding if 'threshold' in search_params: search_kwargs["threshold"] = threshold + optional_search_kwargs = { + "retrieval_mode": mode, + "fusion": fusion, + "vector_weight": vector_weight, + "fts_weight": fts_weight, + "rrf_k": rrf_k, + "candidate_limit": search_limit, + "include_explanation": include_explanation, + } + for key, value in optional_search_kwargs.items(): + if key in search_params: + search_kwargs[key] = value results = target_store.search(**search_kwargs) except TypeError: # Fallback to SQLite format (doesn't support query text parameter) # Pass filters to ensure filtering works correctly results = target_store.search( - search_query if query else "", - vectors=[query_vector], - limit=limit, - filters=db_filters if db_filters else None, + search_query, + vectors=search_vectors, + limit=search_limit, + filters=effective_filters or None, ) # Convert results to unified format @@ -298,8 +391,24 @@ def search_memories( # Extract unified fields # Core and promoted keys that should not be in metadata - promoted_payload_keys = ["user_id", "agent_id", "run_id", "actor_id", "role"] - core_and_promoted_keys = {"data", "hash", "created_at", "updated_at", "id", "metadata", *promoted_payload_keys} + promoted_payload_keys = [ + "user_id", + "agent_id", + "run_id", + "actor_id", + "role", + ] + core_and_promoted_keys = { + "data", + "hash", + "created_at", + "updated_at", + "fulltext_content", + "sparse_embedding", + "id", + "metadata", + *promoted_payload_keys, + } # Extract core fields content = payload.get("data", "") @@ -319,13 +428,26 @@ def search_memories( user_metadata = payload["metadata"].copy() if payload["metadata"] else {} else: # Extract additional metadata (all fields not in core_and_promoted_keys) - user_metadata = {k: v for k, v in payload.items() if k not in core_and_promoted_keys} + user_metadata = { + k: v + for k, v in payload.items() + if ( + k not in core_and_promoted_keys + and k not in self._INTERNAL_SEARCH_PAYLOAD_KEYS + and not k.startswith("_") + ) + } # Merge any user-defined fields from payload top-level into metadata # These fields (like "category") were extracted from metadata for filtering purposes # but should still be visible in the returned metadata for key, value in payload.items(): - if key not in core_and_promoted_keys and key not in user_metadata: + if ( + key not in core_and_promoted_keys + and key not in user_metadata + and key not in self._INTERNAL_SEARCH_PAYLOAD_KEYS + and not key.startswith("_") + ): if value is not None and (value or value == 0 or value == 0.0): user_metadata[key] = value @@ -338,13 +460,33 @@ def search_memories( **promoted_fields, # Add promoted fields at top level "metadata": user_metadata if user_metadata else {}, # Add user metadata } + if "_quality_score" in payload: + memory["_quality_score"] = payload.get("_quality_score") + if include_explanation: + explanation = { + "ranking_score": score, + "fusion_method": fusion, + "retrieval_mode": mode, + "vector_weight": vector_weight + if vector_weight is not None + else 0.5, + "fts_weight": fts_weight if fts_weight is not None else 0.5, + "rrf_k": rrf_k, + } + fusion_info = payload.get("_fusion_info") + if isinstance(fusion_info, dict): + explanation.update(fusion_info) + if "_vector_similarity" in payload: + explanation["vector_similarity"] = payload.get("_vector_similarity") + if "_fts_score" in payload: + explanation["fts_score"] = payload.get("_fts_score") + memory["metadata"]["search_explanation"] = explanation # No need to apply filters here - filters are already applied at the database level # in vector_store.search(), so all returned results should already match the filters memories.append(memory) - # Vector store already applied limit, no need to slice again - return memories + return memories[:limit] def get_memory( self, @@ -739,17 +881,42 @@ async def add_memory_async(self, memory_data: Dict[str, Any]) -> int: async def search_memories_async( self, - query_embedding: List[float], + query_embedding: Optional[List[float]], user_id: Optional[str] = None, agent_id: Optional[str] = None, run_id: Optional[str] = None, filters: Optional[Dict[str, Any]] = None, limit: int = 30, query: Optional[str] = None, + threshold: Optional[float] = None, + retrieval_mode: str = "auto", + fusion: str = "rrf", + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + rrf_k: int = 60, + candidate_limit: Optional[int] = None, + include_explanation: bool = False, ) -> List[Dict[str, Any]]: """Search for memories asynchronously.""" import asyncio - return await asyncio.to_thread(self.search_memories, query_embedding, user_id, agent_id, run_id, filters, limit, query) + return await asyncio.to_thread( + self.search_memories, + query_embedding=query_embedding, + user_id=user_id, + agent_id=agent_id, + run_id=run_id, + filters=filters, + limit=limit, + query=query, + threshold=threshold, + retrieval_mode=retrieval_mode, + fusion=fusion, + vector_weight=vector_weight, + fts_weight=fts_weight, + rrf_k=rrf_k, + candidate_limit=candidate_limit, + include_explanation=include_explanation, + ) async def get_memory_async( self, diff --git a/src/powermem/storage/oceanbase/oceanbase.py b/src/powermem/storage/oceanbase/oceanbase.py index 92b80fd42..358595420 100644 --- a/src/powermem/storage/oceanbase/oceanbase.py +++ b/src/powermem/storage/oceanbase/oceanbase.py @@ -889,13 +889,47 @@ def search(self, limit: int = 5, filters: Optional[Dict] = None, sparse_embedding: Optional[Dict[int, float]] = None, - threshold: Optional[float] = None) -> list[OutputData]: - # Check if hybrid search is enabled, and we have query text - # Full-text search is always enabled by default - if self.hybrid_search and query: - return self._hybrid_search(query, vectors, limit, filters, sparse_embedding, threshold=threshold) - else: - return self._vector_search(query, vectors, limit, filters) + threshold: Optional[float] = None, + retrieval_mode: str = "auto", + fusion: str = "rrf", + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + rrf_k: int = 60, + candidate_limit: Optional[int] = None, + include_explanation: bool = False) -> list[OutputData]: + mode = (retrieval_mode or "auto").lower() + fusion_method = (fusion or "rrf").lower() + search_limit = candidate_limit if candidate_limit is not None else limit + + if mode not in {"auto", "fts", "vector", "hybrid"}: + raise ValueError(f"Invalid retrieval mode: {retrieval_mode}") + if fusion_method not in {"rrf", "weighted"}: + raise ValueError(f"Invalid fusion method: {fusion}") + + if mode == "fts": + results = self._fulltext_search(query, search_limit, filters) + if threshold is not None: + results = [ + result for result in results + if result.payload.get("_quality_score", result.score) >= threshold + ] + return results + + if mode == "vector" or not self.hybrid_search or not query: + return self._vector_search(query, vectors, search_limit, filters) + + return self._hybrid_search( + query, + vectors, + search_limit, + filters, + sparse_embedding, + fusion_method=fusion_method, + k=rrf_k, + threshold=threshold, + vector_weight=vector_weight, + fts_weight=fts_weight, + ) def _vector_search(self, query: str, @@ -1084,6 +1118,7 @@ def _fulltext_search(self, query: str, limit: int = 5, filters: Optional[Dict] = # Store original similarity in metadata metadata = parsed["metadata"] metadata['_fts_score'] = fts_score + metadata['_quality_score'] = fts_score fts_results.append(self._create_output_data( parsed["vector_id"], @@ -1340,7 +1375,9 @@ def _native_hybrid_search( def _hybrid_search(self, query: str, vectors: List[List[float]], limit: int = 5, filters: Optional[Dict] = None, sparse_embedding: Optional[Dict[int, float]] = None, fusion_method: str = "rrf", k: int = 60, - threshold: Optional[float] = None): + threshold: Optional[float] = None, + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None): """Perform hybrid search combining vector, full-text, and sparse vector search with optional reranking. When enable_native_hybrid is True and conditions are met, uses OceanBase native @@ -1350,9 +1387,26 @@ def _hybrid_search(self, query: str, vectors: List[List[float]], limit: int = 5, # 1. enable_native_hybrid must be True # 2. threshold must be None (native search doesn't support threshold filtering) # 3. All filter fields must be in table columns + effective_vector_weight = ( + vector_weight + if vector_weight is not None + else self.vector_weight + ) + effective_fts_weight = ( + fts_weight + if fts_weight is not None + else self.fts_weight + ) + uses_configured_weights = ( + effective_vector_weight == self.vector_weight + and effective_fts_weight == self.fts_weight + ) + use_native = ( self.enable_native_hybrid and threshold is None + and fusion_method == "rrf" + and uses_configured_weights and OceanBaseUtil.check_filters_all_in_columns(filters, self.model_class) ) @@ -1454,7 +1508,15 @@ def _hybrid_search(self, query: str, vectors: List[List[float]], limit: int = 5, # Step 1: Coarse ranking - Combine results using RRF or weighted fusion coarse_ranked_results = self._combine_search_results( - vector_results, fts_results, sparse_results, candidate_limit, fusion_method, k, sparse_embedding + vector_results, + fts_results, + sparse_results, + candidate_limit, + fusion_method, + k, + sparse_embedding, + vector_weight=vector_weight, + fts_weight=fts_weight, ) logger.debug(f"Coarse ranking completed, candidates: {len(coarse_ranked_results)}") @@ -1592,15 +1654,34 @@ def _calculate_quality_score( def _combine_search_results(self, vector_results: List[OutputData], fts_results: List[OutputData], sparse_results: Optional[List[OutputData]], - limit: int, fusion_method: str = "rrf", k: int = 60, sparse_embedding: Optional[Dict[int, float]] = None): + limit: int, fusion_method: str = "rrf", k: int = 60, + sparse_embedding: Optional[Dict[int, float]] = None, + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None): """Combine and rerank vector, full-text, and sparse vector search results using RRF or weighted fusion.""" if sparse_results is None: sparse_results = [] if fusion_method == "rrf": - return self._rrf_fusion(vector_results, fts_results, sparse_results, limit, k, sparse_embedding) + return self._rrf_fusion( + vector_results, + fts_results, + sparse_results, + limit, + k, + sparse_embedding, + vector_weight=vector_weight, + fts_weight=fts_weight, + ) else: - return self._weighted_fusion(vector_results, fts_results, sparse_results, limit) + return self._weighted_fusion( + vector_results, + fts_results, + sparse_results, + limit, + vector_weight=vector_weight, + text_weight=fts_weight, + ) def _normalize_weights_adaptively( self, @@ -1656,7 +1737,9 @@ def _normalize_weights_adaptively( def _rrf_fusion(self, vector_results: List[OutputData], fts_results: List[OutputData], sparse_results: Optional[List[OutputData]], - limit: int, k: int = 60, sparse_embedding: Optional[Dict[int, float]] = None): + limit: int, k: int = 60, sparse_embedding: Optional[Dict[int, float]] = None, + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None): """ Reciprocal Rank Fusion (RRF) for combining search results from vector, FTS, and sparse vector searches. @@ -1665,8 +1748,16 @@ def _rrf_fusion(self, vector_results: List[OutputData], fts_results: List[Output if sparse_results is None: sparse_results = [] - vector_w = self.vector_weight if self.vector_weight is not None else 0 - fts_w = self.fts_weight if self.fts_weight is not None else 0 + vector_w = ( + vector_weight + if vector_weight is not None + else self.vector_weight if self.vector_weight is not None else 0 + ) + fts_w = ( + fts_weight + if fts_weight is not None + else self.fts_weight if self.fts_weight is not None else 0 + ) sparse_w = 0 if self.include_sparse and sparse_results and sparse_embedding: @@ -1783,7 +1874,9 @@ def _rrf_fusion(self, vector_results: List[OutputData], fts_results: List[Output def _weighted_fusion(self, vector_results: List[OutputData], fts_results: List[OutputData], sparse_results: Optional[List[OutputData]], - limit: int, vector_weight: float = 0.7, text_weight: float = 0.3, sparse_weight: float = 0.0): + limit: int, vector_weight: Optional[float] = 0.7, + text_weight: Optional[float] = 0.3, + sparse_weight: float = 0.0): """ Traditional weighted score fusion (fallback method). @@ -1794,8 +1887,16 @@ def _weighted_fusion(self, vector_results: List[OutputData], fts_results: List[O sparse_results = [] # Use instance weights if available - vector_w = self.vector_weight if self.vector_weight is not None else vector_weight - fts_w = self.fts_weight if self.fts_weight is not None else text_weight + vector_w = ( + vector_weight + if vector_weight is not None + else self.vector_weight if self.vector_weight is not None else 0.7 + ) + fts_w = ( + text_weight + if text_weight is not None + else self.fts_weight if self.fts_weight is not None else 0.3 + ) sparse_w = 0.0 if self.include_sparse and sparse_results: sparse_w = self.sparse_weight if self.sparse_weight is not None else sparse_weight @@ -2450,4 +2551,4 @@ def execute_sql(self, sql: str, params: Optional[Dict[str, Any]] = None) -> List except Exception as e: logger.error(f"Failed to execute SQL: {e}") logger.debug(f"SQL statement: {sql}") - raise \ No newline at end of file + raise diff --git a/src/powermem/storage/sqlite/sqlite_vector_store.py b/src/powermem/storage/sqlite/sqlite_vector_store.py index f4c4e361d..26b30e628 100644 --- a/src/powermem/storage/sqlite/sqlite_vector_store.py +++ b/src/powermem/storage/sqlite/sqlite_vector_store.py @@ -82,6 +82,11 @@ def _extract_fulltext_content(payload: dict) -> str: return "" +def _quote_fts_query(query: str) -> str: + escaped = query.replace('"', '""') + return f'"{escaped}"' + + class SQLiteVectorStore(VectorStoreBase): """Simple SQLite-based vector store implementation with FTS5 hybrid search.""" @@ -205,32 +210,96 @@ def insert(self, vectors: List[List[float]], payloads=None, ids=None) -> List[in return generated_ids - def search(self, query: str, vectors: List[List[float]] = None, limit: int = 5, filters=None) -> List[OutputData]: + def search( + self, + query: str, + vectors: List[List[float]] = None, + limit: int = 5, + filters=None, + retrieval_mode: str = "auto", + fusion: str = "rrf", + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + rrf_k: int = 60, + candidate_limit: Optional[int] = None, + include_explanation: bool = False, + ) -> List[OutputData]: """Search using vector similarity, fulltext, or hybrid (RRF fusion). - - Both ``query`` and ``vectors`` provided: hybrid search (vector + FTS5, RRF fusion) + - Both ``query`` and ``vectors`` provided: hybrid search + (vector + FTS5, RRF fusion) - Only ``vectors``: pure vector cosine similarity - Only ``query`` (non-empty string): pure FTS5 fulltext search - Neither: returns empty list """ + del include_explanation + mode = (retrieval_mode or "auto").lower() + if mode not in {"auto", "fts", "vector", "hybrid"}: + raise ValueError(f"Invalid retrieval mode: {retrieval_mode}") + fusion = (fusion or "rrf").lower() + if fusion not in {"rrf", "weighted"}: + raise ValueError(f"Invalid fusion method: {fusion}") + vector_weight = 0.5 if vector_weight is None else vector_weight + fts_weight = 0.5 if fts_weight is None else fts_weight + + if not 0.0 <= vector_weight <= 1.0: + raise ValueError("vector_weight must be between 0.0 and 1.0") + if not 0.0 <= fts_weight <= 1.0: + raise ValueError("fts_weight must be between 0.0 and 1.0") + if rrf_k < 1: + raise ValueError("rrf_k must be >= 1") + if candidate_limit is not None and candidate_limit < 1: + raise ValueError("candidate_limit must be >= 1") + + search_limit = candidate_limit if candidate_limit is not None else limit has_vectors = vectors is not None and len(vectors) > 0 has_query = isinstance(query, str) and query.strip() != "" + query_vector = None + + if has_vectors: + if isinstance(vectors[0], (int, float)): + query_vector = vectors + else: + query_vector = vectors[0] + + if mode == "fts": + query_vector = None + has_vectors = False + elif mode == "vector": + has_query = False + elif mode == "hybrid" and query_vector is None: + has_vectors = False if has_vectors and has_query: # Hybrid search: vector + FTS5, combined with RRF - vector_results = self._vector_search(vectors[0], limit, filters) - fts_results = self._fulltext_search(query, limit, filters) - return self._rrf_fusion(vector_results, fts_results, limit) + vector_results = self._vector_search(query_vector, search_limit, filters) + fts_results = self._fulltext_search(query, search_limit, filters) + if fusion == "weighted": + return self._weighted_fusion( + vector_results, + fts_results, + limit, + vector_weight=vector_weight, + fts_weight=fts_weight, + ) + return self._rrf_fusion( + vector_results, + fts_results, + limit, + k=rrf_k, + vector_weight=vector_weight, + fts_weight=fts_weight, + ) elif has_vectors: # Pure vector search (backward compatible) - return self._vector_search(vectors[0], limit, filters) + return self._vector_search(query_vector, search_limit, filters)[:limit] elif has_query: # Pure fulltext search - return self._fulltext_search(query, limit, filters) + return self._fulltext_search(query, search_limit, filters)[:limit] else: # Fallback: if query is a list (legacy), use as vector if isinstance(query, list): - return self._vector_search(query, limit, filters) + return self._vector_search(query, search_limit, filters)[:limit] return [] def _vector_search(self, query_vector: List[float], limit: int = 5, @@ -306,36 +375,50 @@ def _fulltext_search(self, query: str, limit: int = 5, sql += " ORDER BY rank LIMIT ?" params.append(limit) + queries = [query] + quoted_query = _quote_fts_query(query) + if quoted_query != query: + queries.append(quoted_query) + results = [] with self._lock: - try: - cursor = self.connection.execute(sql, params) - for row in cursor.fetchall(): - doc_id, payload_str, fts_score = row - payload = json.loads(payload_str) - payload['_fts_score'] = float(fts_score) - - results.append(OutputData( - id=doc_id, - score=float(fts_score), - payload=payload, - )) - except sqlite3.OperationalError as e: - logger.warning(f"FTS5 search failed, returning empty: {e}") - return [] + for fts_query in queries: + params[0] = fts_query + try: + cursor = self.connection.execute(sql, params) + for row in cursor.fetchall(): + doc_id, payload_str, fts_score = row + payload = json.loads(payload_str) + payload['_fts_score'] = float(fts_score) + payload['_quality_score'] = 1.0 + + results.append(OutputData( + id=doc_id, + score=float(fts_score), + payload=payload, + )) + return results + except sqlite3.OperationalError as e: + logger.warning( + f"FTS5 search failed for query {fts_query!r}: {e}" + ) + results = [] return results def _rrf_fusion(self, vector_results: List[OutputData], fts_results: List[OutputData], limit: int, k: int = 60, - vector_weight: float = 0.5, - fts_weight: float = 0.5) -> List[OutputData]: + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None) -> List[OutputData]: """Reciprocal Rank Fusion combining vector and FTS5 results. Simplified 2-way RRF (no sparse path) modeled after ``OceanBaseVectorStore._rrf_fusion``. """ + vector_weight = 0.5 if vector_weight is None else vector_weight + fts_weight = 0.5 if fts_weight is None else fts_weight + all_docs: Dict[int, dict] = {} # Process vector results @@ -356,7 +439,9 @@ def _rrf_fusion(self, vector_results: List[OutputData], all_docs[result.id]['fts_rank'] = rank all_docs[result.id]['rrf_score'] += fts_rrf # Merge FTS score into payload - all_docs[result.id]['result'].payload['_fts_score'] = result.payload.get('_fts_score') + all_docs[result.id]['result'].payload['_fts_score'] = ( + result.payload.get('_fts_score') + ) else: all_docs[result.id] = { 'result': result, @@ -366,15 +451,23 @@ def _rrf_fusion(self, vector_results: List[OutputData], } # Sort by RRF score descending, take top ``limit`` - sorted_docs = sorted(all_docs.values(), key=lambda d: d['rrf_score'], reverse=True) + sorted_docs = sorted( + all_docs.values(), key=lambda d: d['rrf_score'], reverse=True + ) final_results = [] for doc_data in sorted_docs[:limit]: result = doc_data['result'] score = doc_data['rrf_score'] + quality_score = ( + 1.0 + if doc_data['fts_rank'] is not None + else result.payload.get('_vector_similarity', score) + ) result.score = score result.payload['_fusion_score'] = score + result.payload['_quality_score'] = quality_score result.payload['_fusion_info'] = { 'vector_rank': doc_data['vector_rank'], 'fts_rank': doc_data['fts_rank'], @@ -382,6 +475,90 @@ def _rrf_fusion(self, vector_results: List[OutputData], 'fusion_method': 'rrf', 'vector_weight': vector_weight, 'fts_weight': fts_weight, + 'rrf_k': k, + } + final_results.append(result) + + return final_results + + def _weighted_fusion( + self, + vector_results: List[OutputData], + fts_results: List[OutputData], + limit: int, + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + ) -> List[OutputData]: + """Weighted fusion using per-path min-max normalized scores.""" + vector_weight = 0.5 if vector_weight is None else vector_weight + fts_weight = 0.5 if fts_weight is None else fts_weight + + def normalized_scores(results: List[OutputData]) -> Dict[int, float]: + if not results: + return {} + scores = [float(r.score or 0.0) for r in results] + min_score = min(scores) + max_score = max(scores) + if max_score == min_score: + return {r.id: 1.0 for r in results} + return { + r.id: (float(r.score or 0.0) - min_score) / (max_score - min_score) + for r in results + } + + vector_scores = normalized_scores(vector_results) + fts_scores = normalized_scores(fts_results) + all_docs: Dict[int, dict] = {} + + for rank, result in enumerate(vector_results, 1): + all_docs[result.id] = { + "result": result, + "vector_rank": rank, + "fts_rank": None, + "weighted_score": ( + vector_weight * vector_scores.get(result.id, 0.0) + ), + } + + for rank, result in enumerate(fts_results, 1): + weighted_score = fts_weight * fts_scores.get(result.id, 0.0) + if result.id in all_docs: + all_docs[result.id]["fts_rank"] = rank + all_docs[result.id]["weighted_score"] += weighted_score + all_docs[result.id]["result"].payload["_fts_score"] = ( + result.payload.get("_fts_score") + ) + else: + all_docs[result.id] = { + "result": result, + "vector_rank": None, + "fts_rank": rank, + "weighted_score": weighted_score, + } + + sorted_docs = sorted( + all_docs.values(), key=lambda d: d["weighted_score"], reverse=True + ) + + final_results = [] + for doc_data in sorted_docs[:limit]: + result = doc_data["result"] + score = doc_data["weighted_score"] + quality_score = ( + 1.0 + if doc_data["fts_rank"] is not None + else result.payload.get("_vector_similarity", score) + ) + result.score = score + result.payload["_fusion_score"] = score + result.payload["_quality_score"] = quality_score + result.payload["_fusion_info"] = { + "vector_rank": doc_data["vector_rank"], + "fts_rank": doc_data["fts_rank"], + "weighted_score": score, + "fusion_method": "weighted", + "vector_weight": vector_weight, + "fts_weight": fts_weight, } final_results.append(result) diff --git a/src/server/api/v1/search.py b/src/server/api/v1/search.py index 3803016fa..2bd4022ae 100644 --- a/src/server/api/v1/search.py +++ b/src/server/api/v1/search.py @@ -102,6 +102,13 @@ async def search_memories_post( filters=body.filters, limit=fetch_limit, threshold=body.threshold, + retrieval_mode=body.retrieval_mode, + fusion=body.fusion, + vector_weight=body.vector_weight, + fts_weight=body.fts_weight, + rrf_k=body.rrf_k, + candidate_limit=body.candidate_limit, + include_explanation=body.include_explanation, ) raw_items = results.get("results", []) @@ -144,6 +151,39 @@ async def search_memories_get( le=1.0, description="Minimum similarity score threshold", ), + retrieval_mode: str = Query( + "auto", + pattern="^(auto|fts|vector|hybrid)$", + description="Retrieval mode", + ), + fusion: str = Query( + "rrf", + pattern="^(rrf|weighted)$", + description="Hybrid fusion method", + ), + vector_weight: Optional[float] = Query( + None, + ge=0.0, + le=1.0, + description="Vector path weight for hybrid retrieval; omitted uses backend configuration", + ), + fts_weight: Optional[float] = Query( + None, + ge=0.0, + le=1.0, + description="Full-text path weight for hybrid retrieval; omitted uses backend configuration", + ), + rrf_k: int = Query(60, ge=1, le=1000, description="RRF rank constant"), + candidate_limit: Optional[int] = Query( + None, + ge=1, + le=1000, + description="Candidate count to retrieve before final limiting", + ), + include_explanation: bool = Query( + False, + description="Include retrieval path and fusion metadata in result metadata", + ), api_key: str = Depends(verify_api_key), service: SearchService = Depends(get_search_service), ): @@ -156,6 +196,13 @@ async def search_memories_get( filters=None, # GET method doesn't support complex filters limit=limit, threshold=threshold, + retrieval_mode=retrieval_mode, + fusion=fusion, + vector_weight=vector_weight, + fts_weight=fts_weight, + rrf_k=rrf_k, + candidate_limit=candidate_limit, + include_explanation=include_explanation, ) search_results = [ diff --git a/src/server/models/request.py b/src/server/models/request.py index be574127a..707f58829 100644 --- a/src/server/models/request.py +++ b/src/server/models/request.py @@ -2,7 +2,7 @@ Request models for PowerMem API """ -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Literal, Optional from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator @@ -139,6 +139,42 @@ class SearchRequest(BaseModel): le=1.0, description="Minimum similarity score threshold", ) + retrieval_mode: Literal["auto", "fts", "vector", "hybrid"] = Field( + default="auto", + description="Retrieval mode: auto, fts, vector, or hybrid", + ) + fusion: Literal["rrf", "weighted"] = Field( + default="rrf", + description="Fusion method for hybrid retrieval", + ) + vector_weight: Optional[float] = Field( + default=None, + ge=0.0, + le=1.0, + description="Vector path weight for hybrid retrieval; omitted uses backend configuration", + ) + fts_weight: Optional[float] = Field( + default=None, + ge=0.0, + le=1.0, + description="Full-text path weight for hybrid retrieval; omitted uses backend configuration", + ) + rrf_k: int = Field( + default=60, + ge=1, + le=1000, + description="RRF rank constant", + ) + candidate_limit: Optional[int] = Field( + default=None, + ge=1, + le=1000, + description="Candidate count to retrieve before final limiting", + ) + include_explanation: bool = Field( + default=False, + description="Include retrieval path and fusion metadata in result metadata", + ) time_range: Optional[str] = Field( default=None, description="Preset time window: '7d', '30d', '90d', or 'all'. When set, only memories created within the window are returned.", diff --git a/src/server/services/search_service.py b/src/server/services/search_service.py index 195217dae..f526ef972 100644 --- a/src/server/services/search_service.py +++ b/src/server/services/search_service.py @@ -36,6 +36,13 @@ def search_memories( filters: Optional[Dict[str, Any]] = None, limit: int = 30, threshold: Optional[float] = None, + retrieval_mode: str = "auto", + fusion: str = "rrf", + vector_weight: Optional[float] = None, + fts_weight: Optional[float] = None, + rrf_k: int = 60, + candidate_limit: Optional[int] = None, + include_explanation: bool = False, ) -> Dict[str, Any]: """ Search memories. @@ -71,6 +78,13 @@ def search_memories( filters=filters, limit=limit, threshold=threshold, + retrieval_mode=retrieval_mode, + fusion=fusion, + vector_weight=vector_weight, + fts_weight=fts_weight, + rrf_k=rrf_k, + candidate_limit=candidate_limit, + include_explanation=include_explanation, ) logger.info(f"Search completed: {len(results.get('results', []))} results") diff --git a/tests/integration/test_noop_embedding_mode.py b/tests/integration/test_noop_embedding_mode.py index e12dbbc41..7ea2043ca 100644 --- a/tests/integration/test_noop_embedding_mode.py +++ b/tests/integration/test_noop_embedding_mode.py @@ -94,3 +94,14 @@ async def test_noop_embedding_async_crud(tmp_path): assert await memory.delete(memory_id, user_id="async_noop") is True assert await memory.get(memory_id, user_id="async_noop") is None + + +@pytest.mark.asyncio +async def test_noop_embedding_async_search_returns_empty(tmp_path): + """Async vector search must return empty results when embedding is disabled.""" + memory = AsyncMemory(config=_sqlite_noop_embedding_config(tmp_path)) + + await memory.add("Async user loves hiking", user_id="async_search_noop") + + results = await memory.search("hiking", user_id="async_search_noop") + assert results["results"] == [] diff --git a/tests/integration/test_sqlite_fts_integration.py b/tests/integration/test_sqlite_fts_integration.py index e5d28608a..c0b7d16fc 100644 --- a/tests/integration/test_sqlite_fts_integration.py +++ b/tests/integration/test_sqlite_fts_integration.py @@ -133,6 +133,21 @@ def test_search_nonmatching_text_returns_vector_results(self): # so cosine sim = 1.0 for everything) assert len(hits) >= 1, "Vector results should still appear even when FTS has no match" + def test_embedding_failure_falls_back_to_fts_with_threshold(self): + """Threshold filtering should not discard valid FTS fallback results.""" + self._add("offline threshold fallback phrase") + self.memory.embedding.embed = MagicMock(side_effect=RuntimeError("offline")) + + results = self.memory.search( + "threshold fallback", + user_id="integ_user", + threshold=0.3, + ) + + hits = results["results"] + assert len(hits) >= 1 + assert "threshold fallback" in hits[0]["memory"] + # ------------------------------------------------------------------ # # 5. FTS table is populated after add (verify via raw store) # ------------------------------------------------------------------ # diff --git a/tests/unit/server/test_search_score_threshold.py b/tests/unit/server/test_search_score_threshold.py index 9d6a95268..8bc846f1d 100644 --- a/tests/unit/server/test_search_score_threshold.py +++ b/tests/unit/server/test_search_score_threshold.py @@ -54,10 +54,41 @@ def test_search_request_accepts_threshold(): assert request.threshold == 0.3 -def test_search_get_route_exposes_threshold_query_parameter(): +def test_search_request_accepts_retrieval_parameters(): + request = SearchRequest( + query="coffee", + retrieval_mode="hybrid", + fusion="weighted", + vector_weight=0.7, + fts_weight=0.3, + rrf_k=30, + candidate_limit=50, + include_explanation=True, + ) + + assert request.retrieval_mode == "hybrid" + assert request.fusion == "weighted" + assert request.vector_weight == 0.7 + assert request.fts_weight == 0.3 + assert request.rrf_k == 30 + assert request.candidate_limit == 50 + assert request.include_explanation is True + + +def test_search_request_omits_weights_by_default(): + request = SearchRequest(query="coffee") + + assert request.vector_weight is None + assert request.fts_weight is None + + +def test_search_get_route_exposes_retrieval_query_parameters(): signature = inspect.signature(search_api.search_memories_get) assert "threshold" in signature.parameters + assert "retrieval_mode" in signature.parameters + assert "fusion" in signature.parameters + assert "include_explanation" in signature.parameters @pytest.mark.parametrize("threshold", [-0.1, 1.1]) @@ -66,6 +97,22 @@ def test_search_request_rejects_invalid_threshold(threshold): SearchRequest(query="coffee", threshold=threshold) +@pytest.mark.parametrize( + "field,value", + [ + ("retrieval_mode", "semantic"), + ("fusion", "sum"), + ("vector_weight", 1.1), + ("fts_weight", -0.1), + ("rrf_k", 0), + ("candidate_limit", 0), + ], +) +def test_search_request_rejects_invalid_retrieval_parameters(field, value): + with pytest.raises(ValidationError): + SearchRequest(query="coffee", **{field: value}) + + def test_search_service_passes_threshold_to_memory_search(monkeypatch): fake_memory = FakeMemory() service = SearchService.__new__(SearchService) @@ -78,3 +125,32 @@ def test_search_service_passes_threshold_to_memory_search(monkeypatch): service.search_memories(query="coffee", threshold=0.3) assert fake_memory.search_kwargs["threshold"] == 0.3 + + +def test_search_service_passes_retrieval_parameters_to_memory_search(monkeypatch): + fake_memory = FakeMemory() + service = SearchService.__new__(SearchService) + service.memory = fake_memory + monkeypatch.setattr( + "server.services.search_service.get_metrics_collector", + lambda: FakeMetricsCollector(), + ) + + service.search_memories( + query="coffee", + retrieval_mode="fts", + fusion="weighted", + vector_weight=0.2, + fts_weight=0.8, + rrf_k=25, + candidate_limit=40, + include_explanation=True, + ) + + assert fake_memory.search_kwargs["retrieval_mode"] == "fts" + assert fake_memory.search_kwargs["fusion"] == "weighted" + assert fake_memory.search_kwargs["vector_weight"] == 0.2 + assert fake_memory.search_kwargs["fts_weight"] == 0.8 + assert fake_memory.search_kwargs["rrf_k"] == 25 + assert fake_memory.search_kwargs["candidate_limit"] == 40 + assert fake_memory.search_kwargs["include_explanation"] is True diff --git a/tests/unit/test_list_memory_filters.py b/tests/unit/test_list_memory_filters.py index 93ce06287..1e691cb12 100644 --- a/tests/unit/test_list_memory_filters.py +++ b/tests/unit/test_list_memory_filters.py @@ -95,6 +95,25 @@ def test_storage_adapter_preserves_sqlite_payload_and_dotted_filter_keys(): } +def test_storage_adapter_sqlite_collision_keys_default_to_metadata(): + store = SQLiteVectorStore(database_path=":memory:") + adapter = StorageAdapter(store) + + assert adapter._build_db_filters( + filters={ + "hash": "metadata-hash", + "data": "metadata-data", + "payload.hash": "payload-hash", + "payload.data": "payload-data", + }, + ) == { + "metadata.hash": "metadata-hash", + "metadata.data": "metadata-data", + "hash": "payload-hash", + "data": "payload-data", + } + + def test_storage_adapter_sqlite_filters_payload_and_metadata_keys(): store = SQLiteVectorStore(database_path=":memory:") adapter = StorageAdapter(store) @@ -121,6 +140,39 @@ def test_storage_adapter_sqlite_filters_payload_and_metadata_keys(): assert [memory["memory"] for memory in priority_results] == ["python"] +def test_storage_adapter_sqlite_search_filters_collision_metadata_keys(): + store = SQLiteVectorStore(database_path=":memory:") + adapter = StorageAdapter(store) + + adapter.add_memory( + { + "content": "alpha collision content", + "user_id": "u01", + "metadata": {"hash": "user-hash", "data": "user-data"}, + } + ) + adapter.add_memory( + { + "content": "alpha other content", + "user_id": "u01", + "metadata": {"hash": "other-hash", "data": "other-data"}, + } + ) + + assert adapter.count_all_memories(filters={"hash": "user-hash"}) == 1 + assert adapter.count_all_memories(filters={"data": "user-data"}) == 1 + assert adapter.count_all_memories(filters={"payload.data": "alpha collision content"}) == 1 + + results = adapter.search_memories( + query_embedding=None, + query="alpha", + retrieval_mode="fts", + filters={"hash": "user-hash", "data": "user-data"}, + ) + + assert [memory["memory"] for memory in results] == ["alpha collision content"] + + def test_storage_adapter_keeps_oceanbase_metadata_filter_key(): class OceanBaseLikeStore: collection_name = "memories" diff --git a/tests/unit/test_memory.py b/tests/unit/test_memory.py index 531869ce7..008915f7f 100644 --- a/tests/unit/test_memory.py +++ b/tests/unit/test_memory.py @@ -7,7 +7,9 @@ import pytest from unittest.mock import MagicMock, patch, Mock from powermem import Memory +from powermem.core.async_memory import AsyncMemory from powermem.core.base import MemoryBase +from powermem.core.memory import _HTTPMemoryClient class TestMemory: @@ -91,6 +93,179 @@ def test_search_memories(self, mock_intelligence_manager, mock_embedder_factory, assert isinstance(results, dict) assert "results" in results + + @patch('powermem.core.memory.VectorStoreFactory') + @patch('powermem.core.memory.LLMFactory') + @patch('powermem.core.memory.EmbedderFactory') + @patch('powermem.core.memory.IntelligenceManager') + def test_search_falls_back_to_fts_when_embedding_fails( + self, + mock_intelligence_manager, + mock_embedder_factory, + mock_llm_factory, + mock_vector_factory, + ): + """Embedding failure should not force Memory.search() to return empty.""" + mock_vector_store = MagicMock() + mock_vector_factory.create.return_value = mock_vector_store + + mock_llm = MagicMock() + mock_llm_factory.create.return_value = mock_llm + + mock_embedder = MagicMock() + mock_embedder.embed.side_effect = RuntimeError("embedder offline") + mock_embedder_factory.create.return_value = mock_embedder + + mock_intelligence = MagicMock() + mock_intelligence.enabled = False + mock_intelligence.plugin = None + mock_intelligence_manager.return_value = mock_intelligence + + memory = Memory() + storage_results = [ + { + "id": 1, + "memory": "offline fallback result", + "metadata": {}, + "score": 0.2, + "user_id": "test_user", + } + ] + + with patch.object( + memory.storage, 'search_memories', return_value=storage_results + ) as mock_search: + results = memory.search("offline fallback", user_id="test_user") + + assert results["results"][0]["memory"] == "offline fallback result" + call_kwargs = mock_search.call_args.kwargs + assert call_kwargs["query_embedding"] is None + assert call_kwargs["query"] == "offline fallback" + assert call_kwargs["retrieval_mode"] == "auto" + + def test_memory_search_passes_retrieval_parameters_to_http_client(self): + """HTTP-backed SDK search should forward retrieval controls.""" + memory = Memory.__new__(Memory) + fake_http_client = MagicMock() + fake_http_client.search.return_value = {"results": [], "relations": []} + memory._http_client = fake_http_client + memory.agent_id = None + + memory.search( + "coffee", + retrieval_mode="fts", + fusion="weighted", + vector_weight=0.2, + fts_weight=0.8, + rrf_k=25, + candidate_limit=40, + include_explanation=True, + threshold=0.3, + ) + + call_kwargs = fake_http_client.search.call_args.kwargs + assert call_kwargs["retrieval_mode"] == "fts" + assert call_kwargs["fusion"] == "weighted" + assert call_kwargs["vector_weight"] == 0.2 + assert call_kwargs["fts_weight"] == 0.8 + assert call_kwargs["rrf_k"] == 25 + assert call_kwargs["candidate_limit"] == 40 + assert call_kwargs["include_explanation"] is True + assert call_kwargs["threshold"] == 0.3 + + def test_http_memory_client_search_sends_retrieval_parameters(self, monkeypatch): + """HTTP client payload should include retrieval controls accepted by API.""" + captured = {} + + class FakeResponse: + def raise_for_status(self): + pass + + def json(self): + return {"success": True, "data": {"results": []}} + + def fake_request(method, url, **kwargs): + captured["method"] = method + captured["url"] = url + captured["json"] = kwargs["json"] + return FakeResponse() + + monkeypatch.setattr("httpx.request", fake_request) + + client = _HTTPMemoryClient("http://example.test") + client.search( + "coffee", + retrieval_mode="fts", + fusion="weighted", + vector_weight=0.2, + fts_weight=0.8, + rrf_k=25, + candidate_limit=40, + include_explanation=True, + threshold=0.3, + ) + + assert captured["method"] == "POST" + assert captured["url"] == "http://example.test/api/v1/memories/search" + payload = captured["json"] + assert payload["retrieval_mode"] == "fts" + assert payload["fusion"] == "weighted" + assert payload["vector_weight"] == 0.2 + assert payload["fts_weight"] == 0.8 + assert payload["rrf_k"] == 25 + assert payload["candidate_limit"] == 40 + assert payload["include_explanation"] is True + assert payload["threshold"] == 0.3 + + @pytest.mark.asyncio + @patch('powermem.core.async_memory.VectorStoreFactory') + @patch('powermem.core.async_memory.LLMFactory') + @patch('powermem.core.async_memory.EmbedderFactory') + @patch('powermem.core.async_memory.IntelligenceManager') + async def test_async_search_falls_back_to_fts_when_embedding_fails( + self, + mock_intelligence_manager, + mock_embedder_factory, + mock_llm_factory, + mock_vector_factory, + ): + """AsyncMemory.search should mirror sync embedding fallback behavior.""" + mock_vector_store = MagicMock() + mock_vector_factory.create.return_value = mock_vector_store + + mock_llm = MagicMock() + mock_llm_factory.create.return_value = mock_llm + + mock_embedder = MagicMock() + mock_embedder.embed.side_effect = RuntimeError("embedder offline") + mock_embedder_factory.create.return_value = mock_embedder + + mock_intelligence = MagicMock() + mock_intelligence.enabled = False + mock_intelligence.plugin = None + mock_intelligence_manager.return_value = mock_intelligence + + memory = AsyncMemory() + storage_results = [ + { + "id": 1, + "memory": "async offline fallback result", + "metadata": {}, + "score": 0.2, + "user_id": "test_user", + } + ] + + with patch.object( + memory.storage, 'search_memories_async', return_value=storage_results + ) as mock_search: + results = await memory.search("async fallback", user_id="test_user") + + assert results["results"][0]["memory"] == "async offline fallback result" + call_kwargs = mock_search.call_args.kwargs + assert call_kwargs["query_embedding"] is None + assert call_kwargs["query"] == "async fallback" + assert call_kwargs["retrieval_mode"] == "auto" @patch('powermem.core.memory.VectorStoreFactory') @patch('powermem.core.memory.LLMFactory') diff --git a/tests/unit/test_sqlite_fts.py b/tests/unit/test_sqlite_fts.py index ef524589a..f42c65e04 100644 --- a/tests/unit/test_sqlite_fts.py +++ b/tests/unit/test_sqlite_fts.py @@ -9,6 +9,7 @@ import pytest from powermem.storage.sqlite.sqlite_vector_store import SQLiteVectorStore +from powermem.storage.adapter import StorageAdapter from powermem.storage.base import OutputData @@ -45,9 +46,83 @@ def _insert_docs(store, docs: list[dict]) -> list[int]: "fulltext_content": doc["content"], "user_id": doc.get("user_id", "u1"), }) + if "metadata" in doc: + payloads[-1]["metadata"] = doc["metadata"] + if "category" in doc: + payloads[-1]["category"] = doc["category"] return store.insert(vectors, payloads) +class FlatVectorStore: + """Minimal pgvector-like store that expects a flat query vector.""" + + collection_name = "flat_vectors" + + def __init__(self): + self.seen_vectors = None + self.called = False + + def search(self, query, vectors, limit=5, filters=None): + self.called = True + self.seen_vectors = vectors + return [ + OutputData( + id=1, + score=0.9, + payload={"data": "flat vector result"}, + ) + ] + + +class OceanBaseLikeHybridStore: + """Minimal OceanBase-like store exposing the public hybrid search controls.""" + + __module__ = "powermem.storage.oceanbase.oceanbase" + collection_name = "oceanbase_like" + hybrid_search = True + + def __init__(self): + self.seen_kwargs = None + + def search( + self, + query, + vectors, + limit=5, + filters=None, + sparse_embedding=None, + threshold=None, + retrieval_mode="auto", + fusion="rrf", + vector_weight=0.5, + fts_weight=0.5, + rrf_k=60, + candidate_limit=None, + include_explanation=False, + ): + self.seen_kwargs = { + "query": query, + "vectors": vectors, + "limit": limit, + "filters": filters, + "threshold": threshold, + "retrieval_mode": retrieval_mode, + "fusion": fusion, + "vector_weight": vector_weight, + "fts_weight": fts_weight, + "rrf_k": rrf_k, + "candidate_limit": candidate_limit, + "include_explanation": include_explanation, + } + return [ + OutputData( + id=1, + score=0.9, + payload={"data": "oceanbase hybrid result", "metadata": {}}, + ) + ] + + # --------------------------------------------------------------------------- # 1. FTS5 table creation # --------------------------------------------------------------------------- @@ -135,6 +210,23 @@ def test_pure_text_search_scores_positive(self, store): for r in results: assert r.score > 0 + def test_pure_text_search_handles_file_paths(self, store): + """FTS-only search should safely handle punctuation-heavy file paths.""" + _insert_docs(store, [ + {"content": "look in src/powermem/core/memory.py for search logic"}, + {"content": "unrelated release notes"}, + ]) + + results = store.search( + query="src/powermem/core/memory.py", + vectors=None, + limit=5, + retrieval_mode="fts", + ) + + assert len(results) == 1 + assert "memory.py" in results[0].payload["data"] + # --------------------------------------------------------------------------- # 4. Hybrid search (vector + FTS, RRF fusion) @@ -163,6 +255,24 @@ def test_hybrid_fusion_metadata(self, store): assert "_fusion_score" in r.payload assert "_fusion_info" in r.payload + def test_hybrid_weighted_fusion(self, store): + """Hybrid search can use weighted fusion when explicitly requested.""" + _insert_docs(store, [ + {"content": "database search fallback", "vector_seed": 0.5}, + {"content": "unrelated cooking note", "vector_seed": 0.2}, + ]) + query_vec = _make_vector(seed=0.5) + + results = store.search( + query="database", + vectors=[query_vec], + limit=5, + fusion="weighted", + ) + + assert len(results) >= 1 + assert results[0].payload["_fusion_info"]["fusion_method"] == "weighted" + def test_hybrid_doc_in_both_paths_ranks_higher(self, store): """A doc matching both vector and text should rank above one matching only one.""" # Doc A: matches text "database" AND has similar vector (seed=0.7) @@ -239,6 +349,149 @@ def test_fts_filter_no_match(self, store): ) assert results == [] + def test_adapter_allows_query_embedding_none_for_fts(self, store): + """StorageAdapter.search_memories(query_embedding=None) should run FTS-only.""" + _insert_docs(store, [ + {"content": "offline fallback should find this", "user_id": "alice"}, + {"content": "different user offline fallback", "user_id": "bob"}, + ]) + adapter = StorageAdapter(store) + + results = adapter.search_memories( + query_embedding=None, + query="offline", + user_id="alice", + limit=5, + include_explanation=True, + ) + + assert len(results) == 1 + assert results[0]["user_id"] == "alice" + explanation = results[0]["metadata"]["search_explanation"] + assert explanation["retrieval_mode"] == "auto" + assert explanation["fts_score"] > 0 + + def test_adapter_fts_maps_user_metadata_filters(self, store): + """Logical metadata filters should apply to nested payload metadata.""" + _insert_docs(store, [ + { + "content": "metadata fallback should find this", + "user_id": "alice", + "metadata": {"scope": "personal"}, + }, + { + "content": "metadata fallback should not match", + "user_id": "alice", + "metadata": {"scope": "group"}, + }, + ]) + adapter = StorageAdapter(store) + + results = adapter.search_memories( + query_embedding=None, + query="metadata", + user_id="alice", + filters={"scope": "personal"}, + limit=5, + ) + + assert len(results) == 1 + assert results[0]["metadata"] == {"scope": "personal"} + assert "_fts_score" not in results[0]["metadata"] + assert "_quality_score" not in results[0]["metadata"] + + def test_adapter_forwards_retrieval_controls_to_oceanbase_like_store(self): + """OceanBase-capable stores should receive per-query retrieval controls.""" + store = OceanBaseLikeHybridStore() + adapter = StorageAdapter(store) + + adapter.search_memories( + query_embedding=[0.1, 0.2, 0.3], + query="hybrid", + filters={"scope": "personal"}, + limit=5, + ) + + assert store.seen_kwargs["vector_weight"] is None + assert store.seen_kwargs["fts_weight"] is None + + adapter.search_memories( + query_embedding=[0.1, 0.2, 0.3], + query="hybrid", + filters={"scope": "personal"}, + limit=5, + threshold=0.2, + retrieval_mode="hybrid", + fusion="weighted", + vector_weight=0.25, + fts_weight=0.75, + rrf_k=24, + candidate_limit=40, + include_explanation=True, + ) + + assert store.seen_kwargs["retrieval_mode"] == "hybrid" + assert store.seen_kwargs["fusion"] == "weighted" + assert store.seen_kwargs["vector_weight"] == 0.25 + assert store.seen_kwargs["fts_weight"] == 0.75 + assert store.seen_kwargs["rrf_k"] == 24 + assert store.seen_kwargs["candidate_limit"] == 40 + assert store.seen_kwargs["include_explanation"] is True + assert store.seen_kwargs["filters"] == {"scope": "personal"} + + def test_adapter_preserves_flat_vector_shape_for_pgvector_like_store(self): + """Non-SQLite stores should still receive a flat query vector.""" + store = FlatVectorStore() + adapter = StorageAdapter(store) + query_embedding = [0.1, 0.2, 0.3] + + adapter.search_memories( + query_embedding=query_embedding, + query="flat vector", + limit=5, + ) + + assert store.seen_vectors == query_embedding + + def test_adapter_returns_empty_for_vector_only_store_without_embedding(self): + """Vector-only stores should not receive vectors=None after embed failure.""" + store = FlatVectorStore() + adapter = StorageAdapter(store) + + results = adapter.search_memories( + query_embedding=None, + query="flat vector", + limit=5, + ) + + assert results == [] + assert store.called is False + + @pytest.mark.parametrize( + "kwargs", + [ + {"retrieval_mode": "semantic"}, + {"fusion": "sum"}, + {"vector_weight": 1.1}, + {"fts_weight": -0.1}, + {"rrf_k": 0}, + {"candidate_limit": 0}, + ], + ) + def test_search_rejects_invalid_direct_retrieval_parameters(self, store, kwargs): + with pytest.raises(ValueError): + store.search(query="python", vectors=None, **kwargs) + + def test_adapter_rejects_invalid_direct_retrieval_parameters(self, store): + adapter = StorageAdapter(store) + + with pytest.raises(ValueError): + adapter.search_memories( + query_embedding=None, + query="python", + rrf_k=0, + ) + # --------------------------------------------------------------------------- # 7. Update syncs to FTS