) -> None:
        self.embedding = embedding
        self.index_name = index_name
        self.query_field = kwargs.get("query_field", "text")
        self.vector_query_field = kwargs.get("vector_query_field", "vector")
        self.space_type = kwargs.get("space_type", "cosine")
        self.index_type = kwargs.get("index_type", "linear")
        self.index_params = kwargs.get("index_params") or {}
        if bes_url is not None:
            self.client = BESVectorStore.bes_client(
                bes_url=bes_url, username=user, password=password
            )
        else:
            raise ValueError("""Please specified a bes connection url.""")
    @property
    def embeddings(self) -> Optional[Embeddings]:
        return self.embedding
[docs]    @staticmethod
    def bes_client(
        *,
        bes_url: Optional[str] = None,
        username: Optional[str] = None,
        password: Optional[str] = None,
    ) -> "Elasticsearch":
        try:
            import elasticsearch
        except ImportError:
            raise ImportError(
                "Could not import elasticsearch python package. "
                "Please install it with `pip install elasticsearch`."
            )
        connection_params: Dict[str, Any] = {}
        connection_params["hosts"] = [bes_url]
        if username and password:
            connection_params["basic_auth"] = (username, password)
        es_client = elasticsearch.Elasticsearch(**connection_params)
        try:
            es_client.info()
        except Exception as e:
            logger.error(f"Error connecting to Elasticsearch: {e}")
            raise e
        return es_client
    def _create_index_if_not_exists(self, dims_length: Optional[int] = None) -> None: