fix: Remove sem/gather and improve handling of alpha default values

Author: maxdswain

integrations/weaviate/src/haystack_integrations/components/retrievers/weaviate/hybrid_retriever.py

@@ -9,6 +9,7 @@
 from haystack.document_stores.types.filter_policy import apply_filter_policy
 
 from haystack_integrations.document_stores.weaviate import WeaviateDocumentStore
+from haystack_integrations.document_stores.weaviate.document_store import DEFAULT_ALPHA
 
 
 @component
@@ -23,7 +24,7 @@ def __init__(
         document_store: WeaviateDocumentStore,
         filters: dict[str, Any] | None = None,
         top_k: int = 10,
-        alpha: float | None = None,
+        alpha: float = DEFAULT_ALPHA,
         max_vector_distance: float | None = None,
         filter_policy: str | FilterPolicy = FilterPolicy.REPLACE,
     ):
@@ -46,7 +47,7 @@ def __init__(
             - `alpha = 1.0`: only vector similarity scoring is used.
             - Values in between blend the two; higher values favor the vector score, lower values favor BM25.
 
-            If `None`, the Weaviate server default is used.
+            By default, 0.7 is used which is the Weaviate server default.
 
             See the official Weaviate docs on Hybrid Search parameters for more details:
             - [Hybrid search parameters](https://weaviate.io/developers/weaviate/search/hybrid#parameters)
@@ -66,7 +67,7 @@ def __init__(
             Policy to determine how filters are applied.
         """
 
-        if alpha is not None and not 0.0 <= alpha <= 1.0:
+        if not 0.0 <= alpha <= 1.0:
             msg = f"alpha ({alpha}) must be in the range [0.0, 1.0]"
             raise ValueError(msg)
 

integrations/weaviate/src/haystack_integrations/document_stores/weaviate/document_store.py

@@ -5,17 +5,14 @@
 import base64
 import datetime
 import json
-from asyncio import Semaphore, gather
 from dataclasses import asdict
 from typing import Any, NoReturn
-from uuid import UUID
 
 from haystack import logging
 from haystack.core.serialization import default_from_dict, default_to_dict
 from haystack.dataclasses.document import Document
 from haystack.document_stores.errors import DocumentStoreError, DuplicateDocumentError
 from haystack.document_stores.types.policy import DuplicatePolicy
-from more_itertools import batched
 
 import weaviate
 from weaviate.collections.classes.aggregate import (
@@ -69,6 +66,9 @@
 # See WeaviateDocumentStore._query_with_filters() for more information.
 DEFAULT_QUERY_LIMIT = 9999
 
+# See weaviate.collections.queries.hybrid.query.sync.pyi for the default value of alpha
+DEFAULT_ALPHA = 0.7
+
 
 class WeaviateDocumentStore:
     """
@@ -112,7 +112,6 @@ def __init__(
         additional_config: AdditionalConfig | None = None,
         grpc_port: int = 50051,
         grpc_secure: bool = False,
-        concurrency_limit: int = 5,
     ) -> None:
         """
         Create a new instance of WeaviateDocumentStore and connects to the Weaviate instance.
@@ -156,8 +155,6 @@ def __init__(
             The port to use for the gRPC connection.
         :param grpc_secure:
             Whether to use a secure channel for the underlying gRPC API.
-        :param concurrency_limit:
-            Number of parallel requests to make. Default is 5.
         """
         self._url = url
         self._auth_client_secret = auth_client_secret
@@ -166,7 +163,6 @@ def __init__(
         self._additional_config = additional_config
         self._grpc_port = grpc_port
         self._grpc_secure = grpc_secure
-        self.concurrency_limit = concurrency_limit
         self._client: weaviate.WeaviateClient | None = None
         self._async_client: weaviate.WeaviateAsyncClient | None = None
         self._collection: weaviate.Collection | None = None
@@ -1225,7 +1221,6 @@ async def delete_all_documents_async(self, *, recreate_index: bool = False, batc
             Reference: https://docs.weaviate.io/weaviate/manage-objects/delete#delete-all-objects
         """
         client = await self.async_client
-        sem = Semaphore(max(1, self.concurrency_limit))
 
         if recreate_index:
             # get current up-to-date config from server, so we can recreate the collection faithfully
@@ -1245,18 +1240,23 @@ async def delete_all_documents_async(self, *, recreate_index: bool = False, batc
         collection = await self.async_collection
         async for obj in collection.iterator(return_properties=[], include_vector=False):
             uuids.append(obj.uuid)
-
-        async def _runner(uuids: list[UUID]) -> None:
-            async with sem:
+            if len(uuids) >= batch_size:
                 res = await collection.data.delete_many(where=weaviate.classes.query.Filter.by_id().contains_any(uuids))
+                if res.successful < len(uuids):
+                    logger.warning(
+                        "Not all documents in the batch have been deleted. "
+                        "Make sure to specify a deletion `batch_size` which is less than `QUERY_MAXIMUM_RESULTS`.",
+                    )
+                uuids.clear()
+
+        if uuids:
+            res = await collection.data.delete_many(where=weaviate.classes.query.Filter.by_id().contains_any(uuids))
             if res.successful < len(uuids):
                 logger.warning(
-                    "Not all documents in the batch have been deleted. "
+                    "Not all documents have been deleted. "
                     "Make sure to specify a deletion `batch_size` which is less than `QUERY_MAXIMUM_RESULTS`.",
                 )
 
-        await gather(*[_runner(list(batch)) for batch in batched(uuids, batch_size)])
-
     def delete_by_filter(self, filters: dict[str, Any]) -> int:
         """
         Deletes all documents that match the provided filters.
@@ -1572,12 +1572,10 @@ def _hybrid_retrieval(
         query_embedding: list[float],
         filters: dict[str, Any] | None = None,
         top_k: int | None = None,
-        alpha: float | None = None,
+        alpha: float = DEFAULT_ALPHA,
         max_vector_distance: float | None = None,
     ) -> list[Document]:
         properties = [p.name for p in self.collection.config.get().properties]
-        if alpha is None:
-            alpha = 0.7
         result = self.collection.query.hybrid(
             query=query,
             vector=query_embedding,
@@ -1599,14 +1597,12 @@ async def _hybrid_retrieval_async(
         query_embedding: list[float],
         filters: dict[str, Any] | None = None,
         top_k: int | None = None,
-        alpha: float | None = None,
+        alpha: float = DEFAULT_ALPHA,
         max_vector_distance: float | None = None,
     ) -> list[Document]:
         collection = await self.async_collection
         config = await collection.config.get()
         properties = [p.name for p in config.properties]
-        if alpha is None:
-            alpha = 0.7
         result = await collection.query.hybrid(
             query=query,
             vector=query_embedding,

integrations/weaviate/tests/test_document_store_async.py

@@ -588,4 +588,4 @@ async def test_delete_all_documents_excessive_batch_size_async(
         with caplog.at_level(logging.WARNING):
             await document_store.delete_all_documents_async(batch_size=20000)
         assert await document_store.count_documents_async() == 5
-        assert "Not all documents in the batch have been deleted." in caplog.text
+        assert "Not all documents have been deleted." in caplog.text

integrations/weaviate/tests/test_hybrid_retriever.py

@@ -9,6 +9,7 @@
 
 from haystack_integrations.components.retrievers.weaviate import WeaviateHybridRetriever
 from haystack_integrations.document_stores.weaviate import WeaviateDocumentStore
+from haystack_integrations.document_stores.weaviate.document_store import DEFAULT_ALPHA
 
 
 def test_init_default():
@@ -17,7 +18,7 @@ def test_init_default():
     assert retriever._document_store == mock_document_store
     assert retriever._filters == {}
     assert retriever._top_k == 10
-    assert retriever._alpha is None
+    assert retriever._alpha == DEFAULT_ALPHA
     assert retriever._max_vector_distance is None
     assert retriever._filter_policy == FilterPolicy.REPLACE
 
@@ -56,7 +57,7 @@ def test_to_dict(_mock_weaviate):
         "init_parameters": {
             "filters": {},
             "top_k": 10,
-            "alpha": None,
+            "alpha": DEFAULT_ALPHA,
             "max_vector_distance": None,
             "filter_policy": "replace",
             "document_store": {
@@ -112,7 +113,7 @@ def test_from_dict(_mock_weaviate):
             "init_parameters": {
                 "filters": {},
                 "top_k": 10,
-                "alpha": None,
+                "alpha": DEFAULT_ALPHA,
                 "max_vector_distance": None,
                 "filter_policy": "replace",
                 "document_store": {
@@ -142,7 +143,7 @@ def test_from_dict(_mock_weaviate):
     assert retriever._document_store
     assert retriever._filters == {}
     assert retriever._top_k == 10
-    assert retriever._alpha is None
+    assert retriever._alpha == DEFAULT_ALPHA
     assert retriever._max_vector_distance is None
 
 
@@ -200,7 +201,12 @@ def test_run_basic():
     assert "documents" in result
     assert len(result["documents"]) == 1
     mock_document_store._hybrid_retrieval.assert_called_once_with(
-        query="test query", query_embedding=[0.1, 0.2, 0.3], filters={}, top_k=10, alpha=None, max_vector_distance=None
+        query="test query",
+        query_embedding=[0.1, 0.2, 0.3],
+        filters={},
+        top_k=10,
+        alpha=DEFAULT_ALPHA,
+        max_vector_distance=None,
     )
 
 
@@ -217,7 +223,7 @@ def test_run_with_runtime_filters():
         query_embedding=[0.1, 0.2, 0.3],
         filters={"runtime": "filter"},
         top_k=10,
-        alpha=None,
+        alpha=DEFAULT_ALPHA,
         max_vector_distance=None,
     )
 
@@ -259,7 +265,7 @@ def test_run_empty_query():
     assert "documents" in result
     assert len(result["documents"]) == 0
     mock_document_store._hybrid_retrieval.assert_called_once_with(
-        query="", query_embedding=[0.1, 0.2, 0.3], filters={}, top_k=10, alpha=None, max_vector_distance=None
+        query="", query_embedding=[0.1, 0.2, 0.3], filters={}, top_k=10, alpha=DEFAULT_ALPHA, max_vector_distance=None
     )
 
 
@@ -288,7 +294,7 @@ def test_from_dict_no_filter_policy(_mock_weaviate):
             "init_parameters": {
                 "filters": {},
                 "top_k": 10,
-                "alpha": None,
+                "alpha": DEFAULT_ALPHA,
                 "max_vector_distance": None,
                 # filter_policy intentionally omitted
                 "document_store": {
@@ -318,7 +324,7 @@ def test_from_dict_no_filter_policy(_mock_weaviate):
     assert retriever._document_store
     assert retriever._filters == {}
     assert retriever._top_k == 10
-    assert retriever._alpha is None
+    assert retriever._alpha == DEFAULT_ALPHA
     assert retriever._max_vector_distance is None
     assert retriever._filter_policy == FilterPolicy.REPLACE
 
@@ -381,7 +387,7 @@ def test_run_with_max_vector_distance_zero_runtime():
         query_embedding=[0.1, 0.2],
         filters={},
         top_k=10,
-        alpha=None,
+        alpha=DEFAULT_ALPHA,
         max_vector_distance=0.0,
     )
 
@@ -402,7 +408,7 @@ def test_run_with_max_vector_distance_zero_init_and_none_runtime():
         query_embedding=[0.1, 0.2],
         filters={},
         top_k=10,
-        alpha=None,
+        alpha=DEFAULT_ALPHA,
         max_vector_distance=0.0,
     )