def deploy_elser()

in example-apps/search-tutorial/v3/search-tutorial/search.py [0:0]

• 28 lines of code
• 3 McCabe index (conditional complexity)

    def deploy_elser(self):
        # download ELSER v2
        self.es.ml.put_trained_model(
            model_id=".elser_model_2", input={"field_names": ["text_field"]}
        )

        # wait until ready
        while True:
            status = self.es.ml.get_trained_models(
                model_id=".elser_model_2", include="definition_status"
            )
            if status["trained_model_configs"][0]["fully_defined"]:
                # model is ready
                break
            time.sleep(1)

        # deploy the model
        self.es.ml.start_trained_model_deployment(model_id=".elser_model_2")

        # define a pipeline
        self.es.ingest.put_pipeline(
            id="elser-ingest-pipeline",
            processors=[
                {
                    "inference": {
                        "model_id": ".elser_model_2",
                        "input_output": [
                            {
                                "input_field": "summary",
                                "output_field": "elser_embedding",
                            }
                        ],
                    }
                }
            ],
        )