Skip to content

CouchbaseSearchEmbeddingRetriever¤

Retrieves documents from the CouchbaseSearchDocumentStore by embedding similarity.

The similarity is dependent on the vector_search_index used in the CouchbaseSearchDocumentStore and the chosen metric during the creation of the index (i.e. dot product, or l2 norm). See CouchbaseSearchDocumentStore for more information.

Usage example:

import numpy as np
from couchbase_haystack import CouchbaseSearchDocumentStore, CouchbaseSearchEmbeddingRetriever, CouchbasePasswordAuthenticator
from haystack.utils import Secret

store = CouchbaseSearchDocumentStore(
    cluster_connection_string=Secret.from_env_var("CB_CONNECTION_STRING"),
    authenticator=CouchbasePasswordAuthenticator(
        username=Secret.from_env_var("CB_USERNAME"),
        password=Secret.from_env_var("CB_PASSWORD")
    ),
    bucket="haystack_test_bucket",
    scope="scope_name",
    collection="collection_name",
    vector_search_index="vector_index"
)
retriever = CouchbaseSearchEmbeddingRetriever(document_store=store)

results = retriever.run(query_embedding=np.random.random(768).tolist())
print(results["documents"])

The example above retrieves the 10 most similar documents to a random query embedding from the CouchbaseSearchDocumentStore. Note that dimensions of the query_embedding must match the dimensions of the embeddings stored in the CouchbaseSearchDocumentStore.

Source code in src/couchbase_haystack/components/retrievers/embedding_retriever.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 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
@component
class CouchbaseSearchEmbeddingRetriever:
    """
    Retrieves documents from the CouchbaseSearchDocumentStore by embedding similarity.

    The similarity is dependent on the vector_search_index used in the CouchbaseSearchDocumentStore and the chosen metric
    during the creation of the index (i.e. dot product, or l2 norm). See CouchbaseSearchDocumentStore for more
    information.

    Usage example:

    ```python
    import numpy as np
    from couchbase_haystack import CouchbaseSearchDocumentStore, CouchbaseSearchEmbeddingRetriever, CouchbasePasswordAuthenticator
    from haystack.utils import Secret

    store = CouchbaseSearchDocumentStore(
        cluster_connection_string=Secret.from_env_var("CB_CONNECTION_STRING"),
        authenticator=CouchbasePasswordAuthenticator(
            username=Secret.from_env_var("CB_USERNAME"),
            password=Secret.from_env_var("CB_PASSWORD")
        ),
        bucket="haystack_test_bucket",
        scope="scope_name",
        collection="collection_name",
        vector_search_index="vector_index"
    )
    retriever = CouchbaseSearchEmbeddingRetriever(document_store=store)

    results = retriever.run(query_embedding=np.random.random(768).tolist())
    print(results["documents"])
    ```

    The example above retrieves the 10 most similar documents to a random query embedding from the
    CouchbaseSearchDocumentStore. Note that dimensions of the query_embedding must match the dimensions of the embeddings
    stored in the CouchbaseSearchDocumentStore.
    """

    def __init__(
        self,
        *,
        document_store: CouchbaseSearchDocumentStore,
        top_k: int = 10,
    ):
        """
        Create the CouchbaseSearchDocumentStore component.

        Note: Currently, the filter option is not supported with embedding queries.
        Instead, you can provide a couchbase search query while running the embedding query.
        The embedding query and search query are combined using an OR operation.

        Args:
            document_store: An instance of CouchbaseSearchDocumentStore.
            top_k: Maximum number of Documents to return.

        Raises:
            ValueError: If document_store is not an instance of CouchbaseSearchDocumentStore.
        """
        if not isinstance(document_store, CouchbaseSearchDocumentStore):
            msg = "document_store must be an instance of CouchbaseSearchDocumentStore"
            raise ValueError(msg)

        self.document_store = document_store
        self.top_k = top_k

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

        :returns:
            Dictionary with serialized data.
        """
        return default_to_dict(
            self,
            top_k=self.top_k,
            document_store=self.document_store.to_dict(),
        )

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

        :param data:
            Dictionary to deserialize from.
        :returns:
              Deserialized component.
        """
        data["init_parameters"]["document_store"] = CouchbaseSearchDocumentStore.from_dict(
            data["init_parameters"]["document_store"]
        )
        return default_from_dict(cls, data)

    @component.output_types(documents=List[Document])
    def run(
        self,
        query_embedding: List[float],
        top_k: Optional[int] = None,
        search_query: Optional[SearchQuery] = None,
        limit: Optional[int] = None,
    ) -> Dict[str, List[Document]]:
        """
        Retrieve documents from the CouchbaseSearchDocumentStore, based on the provided embedding similarity.

        Args:
            query_embedding: Embedding of the query.
            top_k: Maximum number of Documents to be returned from vector query.
                  Overrides the value specified at initialization.
            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 be return by the couchbase fts search request.
                  Default value is top_k.

        Returns:
            A dictionary with the following keys:
            - documents: List of Documents most similar to the given query_embedding
        """
        top_k = top_k or self.top_k

        docs = self.document_store._embedding_retrieval(
            query_embedding=query_embedding, top_k=top_k, search_query=search_query, limit=limit
        )
        return {"documents": docs}

__init__ ¤

__init__(*, document_store: CouchbaseSearchDocumentStore, top_k: int = 10)

Note: Currently, the filter option is not supported with embedding queries. Instead, you can provide a couchbase search query while running the embedding query. The embedding query and search query are combined using an OR operation.

Parameters:

  • document_store (CouchbaseSearchDocumentStore) –

    An instance of CouchbaseSearchDocumentStore.

  • top_k (int, default: 10 ) –

    Maximum number of Documents to return.

Raises:

  • ValueError –

    If document_store is not an instance of CouchbaseSearchDocumentStore.

Source code in src/couchbase_haystack/components/retrievers/embedding_retriever.py 
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
def __init__(
    self,
    *,
    document_store: CouchbaseSearchDocumentStore,
    top_k: int = 10,
):
    """
    Create the CouchbaseSearchDocumentStore component.

    Note: Currently, the filter option is not supported with embedding queries.
    Instead, you can provide a couchbase search query while running the embedding query.
    The embedding query and search query are combined using an OR operation.

    Args:
        document_store: An instance of CouchbaseSearchDocumentStore.
        top_k: Maximum number of Documents to return.

    Raises:
        ValueError: If document_store is not an instance of CouchbaseSearchDocumentStore.
    """
    if not isinstance(document_store, CouchbaseSearchDocumentStore):
        msg = "document_store must be an instance of CouchbaseSearchDocumentStore"
        raise ValueError(msg)

    self.document_store = document_store
    self.top_k = top_k

to_dict ¤

to_dict() -> Dict[str, Any]

Serializes the component to a dictionary.

:returns: Dictionary with serialized data.

Source code in src/couchbase_haystack/components/retrievers/embedding_retriever.py 
78
79
80
81
82
83
84
85
86
87
88
89
def to_dict(self) -> Dict[str, Any]:
    """
    Serializes the component to a dictionary.

    :returns:
        Dictionary with serialized data.
    """
    return default_to_dict(
        self,
        top_k=self.top_k,
        document_store=self.document_store.to_dict(),
    )

from_dict classmethod ¤

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

Deserializes the component from a dictionary.

:param data: Dictionary to deserialize from. :returns: Deserialized component.

Source code in src/couchbase_haystack/components/retrievers/embedding_retriever.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "CouchbaseSearchDocumentStore":
    """
    Deserializes the component from a dictionary.

    :param data:
        Dictionary to deserialize from.
    :returns:
          Deserialized component.
    """
    data["init_parameters"]["document_store"] = CouchbaseSearchDocumentStore.from_dict(
        data["init_parameters"]["document_store"]
    )
    return default_from_dict(cls, data)

run ¤

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

Retrieve documents from the CouchbaseSearchDocumentStore, based on the provided embedding similarity.

Parameters:

  • query_embedding (List[float]) –

    Embedding of the query.

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

    Maximum number of Documents to be returned from vector query. Overrides the value specified at initialization.

  • search_query (Optional[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 be return by the couchbase fts search request. Default value is top_k.

Returns:

  • Dict[str, List[Document]] –

    A dictionary with the following keys:

  • Dict[str, List[Document]] –
    • documents: List of Documents most similar to the given query_embedding
Source code in src/couchbase_haystack/components/retrievers/embedding_retriever.py 
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
@component.output_types(documents=List[Document])
def run(
    self,
    query_embedding: List[float],
    top_k: Optional[int] = None,
    search_query: Optional[SearchQuery] = None,
    limit: Optional[int] = None,
) -> Dict[str, List[Document]]:
    """
    Retrieve documents from the CouchbaseSearchDocumentStore, based on the provided embedding similarity.

    Args:
        query_embedding: Embedding of the query.
        top_k: Maximum number of Documents to be returned from vector query.
              Overrides the value specified at initialization.
        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 be return by the couchbase fts search request.
              Default value is top_k.

    Returns:
        A dictionary with the following keys:
        - documents: List of Documents most similar to the given query_embedding
    """
    top_k = top_k or self.top_k

    docs = self.document_store._embedding_retrieval(
        query_embedding=query_embedding, top_k=top_k, search_query=search_query, limit=limit
    )
    return {"documents": docs}