)
        self.index = self.index.update_embeddings(
            contents_delta_uri=f"gs://{self.gcs_bucket_name}/{filename_prefix}/"
        )
        logger.debug("Updated index with new configuration.")
        return ids
    def _upload_to_gcs(self, data: str, gcs_location: str) -> None:
        """Uploads data to gcs_location.
        Args:
            data: The data that will be stored.
            gcs_location: The location where the data will be stored.
        """
        bucket = self.gcs_client.get_bucket(self.gcs_bucket_name)
        blob = bucket.blob(gcs_location)
        blob.upload_from_string(data)
[docs]    def similarity_search(
        self, query: str, k: int = 4, **kwargs: Any
    ) -> List[Document]:
        """Return docs most similar to query.
        Args:
            query: The string that will be used to search for similar documents.
            k: The amount of neighbors that will be retrieved.
        Returns:
            A list of k matching documents.
        """
        logger.debug(f"Embedding query {query}.")
        embedding_query = self.embedding.embed_documents([query])
        # If the endpoint is public we use the find_neighbors function.
        if self.endpoint._public_match_client:
            response = self.endpoint.find_neighbors(
                deployed_index_id=self._get_index_id(),
                queries=embedding_query,
                num_neighbors=k,
            )
        else:
            response = self.endpoint.match(
                deployed_index_id=self._get_index_id(),
                queries=embedding_query,
                num_neighbors=k,
            )
        if len(response) == 0:
            return []