model_id (str, optional): The ID of the model to use for transforming the
                texts into vectors.
            refresh_indices (bool, optional): Whether to refresh the Elasticsearch
                indices after adding the texts.
            **kwargs: Arbitrary keyword arguments.
        Returns:
            A list of IDs for the added texts.
        """
        # Check if the index exists.
        if not self.client.indices.exists(index=self.index_name):
            dims = kwargs.get("dims")
            if dims is None:
                raise ValueError("ElasticKnnSearch requires 'dims' parameter")
            similarity = kwargs.get("similarity")
            optional_args = {}
            if similarity is not None:
                optional_args["similarity"] = similarity
            mapping = self._default_knn_mapping(dims=dims, **optional_args)
            self.create_knn_index(mapping)
        embeddings = self.embedding.embed_documents(list(texts))
        # body = []
        body: List[Mapping[str, Any]] = []
        for text, vector in zip(texts, embeddings):
            body.extend(
                [
                    {"index": {"_index": self.index_name}},
                    {"text": text, "vector": vector},
                ]
            )
        responses = self.client.bulk(operations=body)
        ids = [
            item["index"]["_id"]
            for item in responses["items"]
            if item["index"]["result"] == "created"
        ]
        if refresh_indices:
            self.client.indices.refresh(index=self.index_name)
        return ids
[docs]    @classmethod
    def from_texts(
        cls,
        texts: List[str],
        embedding: Embeddings,
        metadatas: Optional[List[Dict[Any, Any]]] = None,