Skip to content

CouchbaseSearchDocumentStore¤

CouchbaseSearchDocumentStore is a DocumentStore implementation that uses Couchbase capella service that is easy to deploy, operate, and scale.

The document store supports both scope-level and global-level vector search indexes:

  • Scope-level indexes (default): The vector search index is created at the scope level and only searches documents within that scope
  • Global-level indexes: The vector search index is created at the bucket level and can search across all scopes and collections in the bucket

The index level is specified using the is_global_level_index parameter during initialization.

Source code in src/couchbase_haystack/document_stores/document_store.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
class CouchbaseSearchDocumentStore:
    """CouchbaseSearchDocumentStore is a DocumentStore implementation that uses
    [Couchbase capella](https://cloud.couchbase.com) service that is easy to deploy, operate, and scale.

    The document store supports both scope-level and global-level vector search indexes:

    - Scope-level indexes (default): The vector search index is created at the scope level and only searches
      documents within that scope
    - Global-level indexes: The vector search index is created at the bucket level and can search across all
      scopes and collections in the bucket

    The index level is specified using the `is_global_level_index` parameter during initialization.
    """

    def __init__(
        self,
        *,
        cluster_connection_string: Secret = Secret.from_env_var("CB_CONNECTION_STRING"),
        authenticator: Union[CouchbasePasswordAuthenticator, CouchbaseCertificateAuthenticator],
        cluster_options: CouchbaseClusterOptions = CouchbaseClusterOptions(),
        bucket: str,
        scope: str,
        collection: str,
        vector_search_index: str,
        is_global_level_index: bool = False,
        **kwargs: Dict[str, Any],
    ):
        """Creates a new CouchbaseSearchDocumentStore instance.

        Parameters:
            cluster_connection_string:
                Connection string for the Couchbase cluster

            authenticator:
                Authentication method (password or certificate based)

            cluster_options:
                Options for configuring the cluster connection

            bucket:
                Name of the Couchbase bucket to use

            scope:
                Name of the scope within the bucket

            collection:
                Name of the collection within the scope

            vector_search_index:
                Name of the vector search index to use

            is_global_level_index:
                If True, uses a global (bucket-level) vector search index that can search across all
                scopes and collections. If False (default), uses a scope-level index that only searches
                within the specified scope.

            kwargs:
                Additional keyword arguments passed to the Cluster constructor

        Raises:
            ValueError:
                If the collection name contains invalid characters.
        """
        if collection and not bool(re.match(r"^[a-zA-Z0-9\-_]+$", collection)):
            msg = f'Invalid collection name: "{collection}". It can only contain letters, numbers, -, or _.'
            raise ValueError(msg)

        self.cluster_connection_string = cluster_connection_string
        self.authenticator = authenticator
        self.cluster_options = cluster_options
        self.bucket = bucket
        self.scope_name = scope
        self.collection_name = collection
        self.vector_search_index = vector_search_index
        self.is_global_level_index = is_global_level_index
        self._connection: Optional[Cluster] = None
        self._scope: Optional[Scope] = None
        self._collection: Optional[Collection] = None
        self._kwargs = kwargs

    @property
    def connection(self) -> Cluster:
        if self._connection is None:
            cluster_options = self.cluster_options.get_cluster_options(self.authenticator.get_cb_auth())
            if self.cluster_options.get("profile") is not None:
                cluster_options.apply_profile(self.cluster_options["profile"])
            self._connection = Cluster(
                self.cluster_connection_string.resolve_value(),
                cluster_options,
                **self._kwargs,
            )
            self._connection.wait_until_ready(timeout=timedelta(seconds=60))
        return self._connection

    @property
    def scope(self) -> Scope:
        if self._scope is None:
            bucket = self.connection.bucket(self.bucket)
            scopes_specs = bucket.collections().get_all_scopes()
            scope_found = False
            collection_found = False
            for scope_spec in scopes_specs:
                if scope_spec.name == self.scope_name:
                    scope_found = True
                    for col_spec in scope_spec.collections:
                        if col_spec.name == self.collection_name:
                            collection_found = True
            if not scope_found:
                msg = f"Scope '{self.scope_name}' does not exist in bucket '{self.bucket}'."
                raise ValueError(msg)
            if not collection_found:
                msg = f"Collection '{self.collection_name}' does not exist in scope '{self.scope_name}'."
                raise ValueError(msg)
            self._scope = self.connection.bucket(self.bucket).scope(self.scope_name)
        return self._scope

    @property
    def collection(self) -> Collection:
        if self._collection is None:
            self._collection = self.scope.collection(self.collection_name)
        return self._collection

    def to_dict(self) -> Dict[str, Any]:
        """Serializes the component to a dictionary.

        Returns:
            Dictionary with serialized data.
        """
        return default_to_dict(
            self,
            cluster_connection_string=self.cluster_connection_string.to_dict(),
            authenticator=self.authenticator.to_dict(),
            cluster_options=self.cluster_options.to_dict(),
            bucket=self.bucket,
            scope=self.scope_name,
            collection=self.collection_name,
            vector_search_index=self.vector_search_index,
            is_global_level_index=self.is_global_level_index,
            **self._kwargs,
        )

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "CouchbaseSearchDocumentStore":
        """Deserializes the component from a dictionary.

        Args:
            data:
                Dictionary to deserialize from.

        Returns:
            Deserialized component.
        """
        if data["init_parameters"]["authenticator"]["type"] == generate_qualified_class_name(CouchbasePasswordAuthenticator):
            data["init_parameters"]["authenticator"] = CouchbasePasswordAuthenticator.from_dict(
                data["init_parameters"]["authenticator"]
            )
        else:
            data["init_parameters"]["authenticator"] = CouchbaseCertificateAuthenticator.from_dict(
                data["init_parameters"]["authenticator"]
            )
        data["init_parameters"]["cluster_options"] = CouchbaseClusterOptions.from_dict(data["init_parameters"]["cluster_options"])
        deserialize_secrets_inplace(data["init_parameters"], keys=["cluster_connection_string"])
        return default_from_dict(cls, data)

    def _get_search_interface(self):
        """Returns the appropriate search interface based on the index level configuration.

        Returns:
            Either scope.search_indexes() for scope-level or connection.search_indexes() for global-level
        """
        if not self.is_global_level_index:
            return self.scope.search_indexes()
        return self.connection.search_indexes()

    def count_documents(self) -> int:
        """Returns how many documents are present in the document store.

        Returns:
            The number of documents in the document store.
        """
        search_interface = self._get_search_interface()
        return search_interface.get_indexed_documents_count(self.vector_search_index)

    def filter_documents(self, filters: Optional[Dict[str, Any]] = None) -> List[Document]:
        """Returns the documents that match the filters provided.

        For a detailed specification of the filters,
        refer to the Haystack [documentation](https://docs.haystack.deepset.ai/v2.0/docs/metadata-filtering).

        Args:
            filters: The filters to apply. It returns only the documents that match the filters.

        Returns:
            A list of Documents that match the given filters.
        """
        search_filters: SearchQuery
        if filters:
            search_filters = _normalize_filters(filters)
        else:
            search_filters = search.MatchAllQuery()
        logger.debug(search_filters.encodable)
        request = search.SearchRequest(search_filters)
        options = SearchOptions(fields=["*"], limit=10000)

        if not self.is_global_level_index:
            response = self.scope.search(self.vector_search_index, request, options)
        else:
            response = self.connection.search(self.vector_search_index, request, options)

        return self.__get_doc_from_kv(response)

    def write_documents(self, documents: List[Document], policy: DuplicatePolicy = DuplicatePolicy.NONE) -> int:
        """Writes documents into the couchbase collection.

        Args:
            documents: A list of Documents to write to the document store.
            policy: The duplicate policy to use when writing documents.

        Raises:
            DuplicateDocumentError: If a document with the same ID already exists in the document store
                and the policy is set to DuplicatePolicy.FAIL (or not specified).
            ValueError: If the documents are not of type Document.

        Returns:
            The number of documents written to the document store.
        """
        if len(documents) > 0:
            if not isinstance(documents[0], Document):
                msg = "param 'documents' must contain a list of objects of type Document"
                raise ValueError(msg)

        if policy == DuplicatePolicy.NONE:
            policy = DuplicatePolicy.FAIL

        cb_documents = []
        for doc in documents:
            doc_dict = doc.to_dict(flatten=False)
            doc_dict = {k: v for k, v in doc_dict.items() if v is not None}
            if "sparse_embedding" in doc_dict:
                sparse_embedding = doc_dict.pop("sparse_embedding", None)
                if sparse_embedding:
                    logger.warning(
                        "Document %s has the `sparse_embedding` field set,"
                        "but storing sparse embeddings in Couchbase is not currently supported."
                        "The `sparse_embedding` field will be ignored.",
                        doc.id,
                    )
            cb_documents.append(doc_dict)
        written_docs = len(documents)

        operations = {doc["id"]: doc for doc in cb_documents}
        try:
            result: MultiMutationResult
            if policy == DuplicatePolicy.FAIL:
                result = self.collection.insert_multi(operations)
            else:
                result = self.collection.upsert_multi(operations)
        except Exception as e:
            logger.error("write error {e}")
            msg = f"Failed to write documents to Couchbase. Error: {e}"
            raise DocumentStoreError(msg) from e
        if not result.all_ok and result.exceptions:
            duplicate_ids = []
            other_errors = []
            for id, ex in result.exceptions.items():
                if isinstance(ex, DocumentExistsException):
                    duplicate_ids.append(id)
                else:
                    other_errors.append({"id": id, "exception": ex})
            if len(duplicate_ids) > 0:
                msg = f"IDs '{', '.join(duplicate_ids)}' already exist in the document store."
                raise DuplicateDocumentError(msg)
            if len(other_errors) > 0:
                msg = f"Failed to write documents to couchbase. Errors:\n{other_errors}"
                raise DocumentStoreError(msg)
        logger.debug("date written")
        return written_docs

    def delete_documents(self, document_ids: List[str]) -> None:
        """Deletes all documents with a matching document_ids from the document store.

        Args:
            document_ids: the document ids to delete
        """
        if not document_ids:
            return
        self.collection.remove_multi(keys=document_ids)

    def _embedding_retrieval(
        self,
        query_embedding: List[float],
        top_k: int = 10,
        search_query: SearchQuery = None,
        limit: Optional[int] = None,
    ) -> List[Document]:
        """Find the documents that are most similar to the provided `query_embedding` by using a vector similarity metric.

        Args:
            query_embedding: Embedding of the query
            top_k: How many documents to be returned by the vector query
            search_query: Search filters param which is parsed to the Couchbase search query. The vector query and
                search query are ORed operation.
            limit: Maximum number of Documents to return. Defaults to top_k if not specified.

        Returns:
            A list of Documents that are most similar to the given `query_embedding`

        Raises:
            ValueError: If `query_embedding` is empty
            DocumentStoreError: If the retrieval of documents from Couchbase fails
        """
        if not query_embedding:
            msg = "Query embedding must not be empty"
            raise ValueError(msg)

        vector_search = VectorSearch.from_vector_query(
            VectorQuery(field_name="embedding", vector=query_embedding, num_candidates=top_k)
        )
        request = search.SearchRequest.create(vector_search)
        if search_query:
            request.with_search_query(search_query)

        if limit is None:
            limit = top_k
        options = SearchOptions(fields=["*"], limit=limit)

        if not self.is_global_level_index:
            response = self.scope.search(self.vector_search_index, request, options)
        else:
            response = self.connection.search(self.vector_search_index, request, options)

        return self.__get_doc_from_kv(response)

    def __get_doc_from_kv(self, response: SearchResult) -> List[Document]:
        """Gets documents from Couchbase KV store based on search results.

        Args:
            response: Search results from Couchbase

        Returns:
            List of Document objects

        Raises:
            DocumentStoreError: If retrieving documents from Couchbase fails
        """
        documents: List[Document] = []
        ids: List[str] = []
        scores: List[float] = []
        for doc in response.rows():
            ids.append(doc.id)
            scores.append(doc.score)
        kv_response = self.collection.get_multi(keys=ids)
        if not kv_response.all_ok and kv_response.exceptions:
            errors = []
            for id, ex in kv_response.exceptions.items():
                errors.append({"id": id, "exception": ex})
            if len(errors) > 0:
                msg = f"Failed to write documents to couchbase. Errors:\n{errors}"
                raise DocumentStoreError(msg)
        for i, id in enumerate(ids):
            get_result = kv_response.results.get(id)
            if get_result is not None and get_result.success:
                value = get_result.value
                value["id"] = id
                value["score"] = scores[i]
            documents.append(Document.from_dict(value))
        return documents

__init__ ¤

__init__(
    *,
    cluster_connection_string: Secret = Secret.from_env_var(
        "CB_CONNECTION_STRING"
    ),
    authenticator: Union[
        CouchbasePasswordAuthenticator, CouchbaseCertificateAuthenticator
    ],
    cluster_options: CouchbaseClusterOptions = CouchbaseClusterOptions(),
    bucket: str,
    scope: str,
    collection: str,
    vector_search_index: str,
    is_global_level_index: bool = False,
    **kwargs: Dict[str, Any]
)

Parameters:

  • cluster_connection_string (Secret, default: from_env_var('CB_CONNECTION_STRING') ) –

    Connection string for the Couchbase cluster

  • authenticator (Union[CouchbasePasswordAuthenticator, CouchbaseCertificateAuthenticator]) –

    Authentication method (password or certificate based)

  • cluster_options (CouchbaseClusterOptions, default: CouchbaseClusterOptions() ) –

    Options for configuring the cluster connection

  • bucket (str) –

    Name of the Couchbase bucket to use

  • scope (str) –

    Name of the scope within the bucket

  • collection (str) –

    Name of the collection within the scope

  • vector_search_index (str) –

    Name of the vector search index to use

  • is_global_level_index (bool, default: False ) –

    If True, uses a global (bucket-level) vector search index that can search across all scopes and collections. If False (default), uses a scope-level index that only searches within the specified scope.

  • kwargs (Dict[str, Any], default: {} ) –

    Additional keyword arguments passed to the Cluster constructor

Raises:

  • ValueError –

    If the collection name contains invalid characters.

Source code in src/couchbase_haystack/document_stores/document_store.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def __init__(
    self,
    *,
    cluster_connection_string: Secret = Secret.from_env_var("CB_CONNECTION_STRING"),
    authenticator: Union[CouchbasePasswordAuthenticator, CouchbaseCertificateAuthenticator],
    cluster_options: CouchbaseClusterOptions = CouchbaseClusterOptions(),
    bucket: str,
    scope: str,
    collection: str,
    vector_search_index: str,
    is_global_level_index: bool = False,
    **kwargs: Dict[str, Any],
):
    """Creates a new CouchbaseSearchDocumentStore instance.

    Parameters:
        cluster_connection_string:
            Connection string for the Couchbase cluster

        authenticator:
            Authentication method (password or certificate based)

        cluster_options:
            Options for configuring the cluster connection

        bucket:
            Name of the Couchbase bucket to use

        scope:
            Name of the scope within the bucket

        collection:
            Name of the collection within the scope

        vector_search_index:
            Name of the vector search index to use

        is_global_level_index:
            If True, uses a global (bucket-level) vector search index that can search across all
            scopes and collections. If False (default), uses a scope-level index that only searches
            within the specified scope.

        kwargs:
            Additional keyword arguments passed to the Cluster constructor

    Raises:
        ValueError:
            If the collection name contains invalid characters.
    """
    if collection and not bool(re.match(r"^[a-zA-Z0-9\-_]+$", collection)):
        msg = f'Invalid collection name: "{collection}". It can only contain letters, numbers, -, or _.'
        raise ValueError(msg)

    self.cluster_connection_string = cluster_connection_string
    self.authenticator = authenticator
    self.cluster_options = cluster_options
    self.bucket = bucket
    self.scope_name = scope
    self.collection_name = collection
    self.vector_search_index = vector_search_index
    self.is_global_level_index = is_global_level_index
    self._connection: Optional[Cluster] = None
    self._scope: Optional[Scope] = None
    self._collection: Optional[Collection] = None
    self._kwargs = kwargs

to_dict ¤

to_dict() -> Dict[str, Any]

Serializes the component to a dictionary.

Returns:

  • Dict[str, Any] –

    Dictionary with serialized data.

Source code in src/couchbase_haystack/document_stores/document_store.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def to_dict(self) -> Dict[str, Any]:
    """Serializes the component to a dictionary.

    Returns:
        Dictionary with serialized data.
    """
    return default_to_dict(
        self,
        cluster_connection_string=self.cluster_connection_string.to_dict(),
        authenticator=self.authenticator.to_dict(),
        cluster_options=self.cluster_options.to_dict(),
        bucket=self.bucket,
        scope=self.scope_name,
        collection=self.collection_name,
        vector_search_index=self.vector_search_index,
        is_global_level_index=self.is_global_level_index,
        **self._kwargs,
    )

from_dict classmethod ¤

from_dict(data: Dict[str, Any]) -> CouchbaseSearchDocumentStore

Deserializes the component from a dictionary.

Parameters:

  • data (Dict[str, Any]) –

    Dictionary to deserialize from.

Returns:

Source code in src/couchbase_haystack/document_stores/document_store.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "CouchbaseSearchDocumentStore":
    """Deserializes the component from a dictionary.

    Args:
        data:
            Dictionary to deserialize from.

    Returns:
        Deserialized component.
    """
    if data["init_parameters"]["authenticator"]["type"] == generate_qualified_class_name(CouchbasePasswordAuthenticator):
        data["init_parameters"]["authenticator"] = CouchbasePasswordAuthenticator.from_dict(
            data["init_parameters"]["authenticator"]
        )
    else:
        data["init_parameters"]["authenticator"] = CouchbaseCertificateAuthenticator.from_dict(
            data["init_parameters"]["authenticator"]
        )
    data["init_parameters"]["cluster_options"] = CouchbaseClusterOptions.from_dict(data["init_parameters"]["cluster_options"])
    deserialize_secrets_inplace(data["init_parameters"], keys=["cluster_connection_string"])
    return default_from_dict(cls, data)

_get_search_interface ¤

_get_search_interface()

Returns the appropriate search interface based on the index level configuration.

Returns:

  • –

    Either scope.search_indexes() for scope-level or connection.search_indexes() for global-level

Source code in src/couchbase_haystack/document_stores/document_store.py 
198
199
200
201
202
203
204
205
206
def _get_search_interface(self):
    """Returns the appropriate search interface based on the index level configuration.

    Returns:
        Either scope.search_indexes() for scope-level or connection.search_indexes() for global-level
    """
    if not self.is_global_level_index:
        return self.scope.search_indexes()
    return self.connection.search_indexes()

count_documents ¤

count_documents() -> int

Returns how many documents are present in the document store.

Returns:

  • int –

    The number of documents in the document store.

Source code in src/couchbase_haystack/document_stores/document_store.py 
208
209
210
211
212
213
214
215
def count_documents(self) -> int:
    """Returns how many documents are present in the document store.

    Returns:
        The number of documents in the document store.
    """
    search_interface = self._get_search_interface()
    return search_interface.get_indexed_documents_count(self.vector_search_index)

filter_documents ¤

filter_documents(filters: Optional[Dict[str, Any]] = None) -> List[Document]

Returns the documents that match the filters provided.

For a detailed specification of the filters, refer to the Haystack documentation.

Parameters:

  • filters (Optional[Dict[str, Any]], default: None ) –

    The filters to apply. It returns only the documents that match the filters.

Returns:

  • List[Document] –

    A list of Documents that match the given filters.

Source code in src/couchbase_haystack/document_stores/document_store.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def filter_documents(self, filters: Optional[Dict[str, Any]] = None) -> List[Document]:
    """Returns the documents that match the filters provided.

    For a detailed specification of the filters,
    refer to the Haystack [documentation](https://docs.haystack.deepset.ai/v2.0/docs/metadata-filtering).

    Args:
        filters: The filters to apply. It returns only the documents that match the filters.

    Returns:
        A list of Documents that match the given filters.
    """
    search_filters: SearchQuery
    if filters:
        search_filters = _normalize_filters(filters)
    else:
        search_filters = search.MatchAllQuery()
    logger.debug(search_filters.encodable)
    request = search.SearchRequest(search_filters)
    options = SearchOptions(fields=["*"], limit=10000)

    if not self.is_global_level_index:
        response = self.scope.search(self.vector_search_index, request, options)
    else:
        response = self.connection.search(self.vector_search_index, request, options)

    return self.__get_doc_from_kv(response)

write_documents ¤

write_documents(
    documents: List[Document], policy: DuplicatePolicy = DuplicatePolicy.NONE
) -> int

Writes documents into the couchbase collection.

Parameters:

  • documents (List[Document]) –

    A list of Documents to write to the document store.

  • policy (DuplicatePolicy, default: NONE ) –

    The duplicate policy to use when writing documents.

Raises:

  • DuplicateDocumentError –

    If a document with the same ID already exists in the document store and the policy is set to DuplicatePolicy.FAIL (or not specified).

  • ValueError –

    If the documents are not of type Document.

Returns:

  • int –

    The number of documents written to the document store.

Source code in src/couchbase_haystack/document_stores/document_store.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def write_documents(self, documents: List[Document], policy: DuplicatePolicy = DuplicatePolicy.NONE) -> int:
    """Writes documents into the couchbase collection.

    Args:
        documents: A list of Documents to write to the document store.
        policy: The duplicate policy to use when writing documents.

    Raises:
        DuplicateDocumentError: If a document with the same ID already exists in the document store
            and the policy is set to DuplicatePolicy.FAIL (or not specified).
        ValueError: If the documents are not of type Document.

    Returns:
        The number of documents written to the document store.
    """
    if len(documents) > 0:
        if not isinstance(documents[0], Document):
            msg = "param 'documents' must contain a list of objects of type Document"
            raise ValueError(msg)

    if policy == DuplicatePolicy.NONE:
        policy = DuplicatePolicy.FAIL

    cb_documents = []
    for doc in documents:
        doc_dict = doc.to_dict(flatten=False)
        doc_dict = {k: v for k, v in doc_dict.items() if v is not None}
        if "sparse_embedding" in doc_dict:
            sparse_embedding = doc_dict.pop("sparse_embedding", None)
            if sparse_embedding:
                logger.warning(
                    "Document %s has the `sparse_embedding` field set,"
                    "but storing sparse embeddings in Couchbase is not currently supported."
                    "The `sparse_embedding` field will be ignored.",
                    doc.id,
                )
        cb_documents.append(doc_dict)
    written_docs = len(documents)

    operations = {doc["id"]: doc for doc in cb_documents}
    try:
        result: MultiMutationResult
        if policy == DuplicatePolicy.FAIL:
            result = self.collection.insert_multi(operations)
        else:
            result = self.collection.upsert_multi(operations)
    except Exception as e:
        logger.error("write error {e}")
        msg = f"Failed to write documents to Couchbase. Error: {e}"
        raise DocumentStoreError(msg) from e
    if not result.all_ok and result.exceptions:
        duplicate_ids = []
        other_errors = []
        for id, ex in result.exceptions.items():
            if isinstance(ex, DocumentExistsException):
                duplicate_ids.append(id)
            else:
                other_errors.append({"id": id, "exception": ex})
        if len(duplicate_ids) > 0:
            msg = f"IDs '{', '.join(duplicate_ids)}' already exist in the document store."
            raise DuplicateDocumentError(msg)
        if len(other_errors) > 0:
            msg = f"Failed to write documents to couchbase. Errors:\n{other_errors}"
            raise DocumentStoreError(msg)
    logger.debug("date written")
    return written_docs

delete_documents ¤

delete_documents(document_ids: List[str]) -> None

Deletes all documents with a matching document_ids from the document store.

Parameters:

  • document_ids (List[str]) –

    the document ids to delete

Source code in src/couchbase_haystack/document_stores/document_store.py 
312
313
314
315
316
317
318
319
320
def delete_documents(self, document_ids: List[str]) -> None:
    """Deletes all documents with a matching document_ids from the document store.

    Args:
        document_ids: the document ids to delete
    """
    if not document_ids:
        return
    self.collection.remove_multi(keys=document_ids)

_embedding_retrieval ¤

_embedding_retrieval(
    query_embedding: List[float],
    top_k: int = 10,
    search_query: SearchQuery = None,
    limit: Optional[int] = None,
) -> List[Document]

Find the documents that are most similar to the provided query_embedding by using a vector similarity metric.

Parameters:

  • query_embedding (List[float]) –

    Embedding of the query

  • top_k (int, default: 10 ) –

    How many documents to be returned by the vector query

  • search_query (SearchQuery, default: None ) –

    Search filters param which is parsed to the Couchbase search query. The vector query and search query are ORed operation.

  • limit (Optional[int], default: None ) –

    Maximum number of Documents to return. Defaults to top_k if not specified.

Returns:

  • List[Document] –

    A list of Documents that are most similar to the given query_embedding

Raises:

  • ValueError –

    If query_embedding is empty

  • DocumentStoreError –

    If the retrieval of documents from Couchbase fails

Source code in src/couchbase_haystack/document_stores/document_store.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
def _embedding_retrieval(
    self,
    query_embedding: List[float],
    top_k: int = 10,
    search_query: SearchQuery = None,
    limit: Optional[int] = None,
) -> List[Document]:
    """Find the documents that are most similar to the provided `query_embedding` by using a vector similarity metric.

    Args:
        query_embedding: Embedding of the query
        top_k: How many documents to be returned by the vector query
        search_query: Search filters param which is parsed to the Couchbase search query. The vector query and
            search query are ORed operation.
        limit: Maximum number of Documents to return. Defaults to top_k if not specified.

    Returns:
        A list of Documents that are most similar to the given `query_embedding`

    Raises:
        ValueError: If `query_embedding` is empty
        DocumentStoreError: If the retrieval of documents from Couchbase fails
    """
    if not query_embedding:
        msg = "Query embedding must not be empty"
        raise ValueError(msg)

    vector_search = VectorSearch.from_vector_query(
        VectorQuery(field_name="embedding", vector=query_embedding, num_candidates=top_k)
    )
    request = search.SearchRequest.create(vector_search)
    if search_query:
        request.with_search_query(search_query)

    if limit is None:
        limit = top_k
    options = SearchOptions(fields=["*"], limit=limit)

    if not self.is_global_level_index:
        response = self.scope.search(self.vector_search_index, request, options)
    else:
        response = self.connection.search(self.vector_search_index, request, options)

    return self.__get_doc_from_kv(response)

__get_doc_from_kv ¤

__get_doc_from_kv(response: SearchResult) -> List[Document]

Gets documents from Couchbase KV store based on search results.

Parameters:

Returns:

  • List[Document] –

    List of Document objects

Raises:

  • DocumentStoreError –

    If retrieving documents from Couchbase fails

Source code in src/couchbase_haystack/document_stores/document_store.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def __get_doc_from_kv(self, response: SearchResult) -> List[Document]:
    """Gets documents from Couchbase KV store based on search results.

    Args:
        response: Search results from Couchbase

    Returns:
        List of Document objects

    Raises:
        DocumentStoreError: If retrieving documents from Couchbase fails
    """
    documents: List[Document] = []
    ids: List[str] = []
    scores: List[float] = []
    for doc in response.rows():
        ids.append(doc.id)
        scores.append(doc.score)
    kv_response = self.collection.get_multi(keys=ids)
    if not kv_response.all_ok and kv_response.exceptions:
        errors = []
        for id, ex in kv_response.exceptions.items():
            errors.append({"id": id, "exception": ex})
        if len(errors) > 0:
            msg = f"Failed to write documents to couchbase. Errors:\n{errors}"
            raise DocumentStoreError(msg)
    for i, id in enumerate(ids):
        get_result = kv_response.results.get(id)
        if get_result is not None and get_result.success:
            value = get_result.value
            value["id"] = id
            value["score"] = scores[i]
        documents.append(Document.from_dict(value))
    return documents