Completes OPEN-5839 Don't require OPENLAYER_PROJECT_NAME if inference pipeline id is present and Completes OPEN-5840 Make sure an inference pipeline is created if name is specified and it doesn’t exist

Author: gustavocidornelas

openlayer/services/data_streamer.py

@@ -96,37 +96,36 @@ def _validate_attributes(self) -> None:
             if not self.openlayer_api_key:
                 raise ValueError(
                     "An Openlayer API key is required for publishing."
-                    " Please provide `openlayer_api_key` or set the"
-                    " OPENLAYER_API_KEY environment variable."
+                    " Please set it as environment variable named OPENLAYER_API_KEY."
                 )
 
-            if not self.openlayer_project_name:
+            if (
+                not self.openlayer_project_name
+                and not self.openlayer_inference_pipeline_name
+                and not self.openlayer_inference_pipeline_id
+            ):
                 raise ValueError(
-                    "You must specify the name of the project on Openlayer"
-                    " that you want to publish to. Please provide"
-                    " `openlayer_project_name` or set the OPENLAYER_PROJECT_NAME"
-                    " environment variable."
+                    "You must provide more information about the project and"
+                    " inference pipeline on Openlayer to publish data."
+                    " Please provide either: "
+                    " - the project name and inference pipeline name, or"
+                    " - the inference pipeline id."
+                    " You can set them as environment variables named"
+                    " OPENLAYER_PROJECT_NAME, OPENLAYER_INFERENCE_PIPELINE_NAME, "
+                    "and OPENLAYER_INFERENCE_PIPELINE_ID."
                 )
 
-        if (
-            not self.openlayer_inference_pipeline_id
-            and not self.openlayer_inference_pipeline_name
-        ):
-            raise ValueError(
-                "Either an inference pipeline id or name is required."
-                " Please provide `openlayer_inference_pipeline_id` or"
-                " `openlayer_inference_pipeline_name`, "
-                "or set the OPENLAYER_INFERENCE_PIPELINE_ID or"
-                " OPENLAYER_INFERENCE_PIPELINE_NAME environment variables."
-            )
-        logger.info(
-            "Data will be streamed to Openlayer project %s and inference pipeline %s.",
-            self.openlayer_project_name,
-            (
-                self.openlayer_inference_pipeline_id
-                or self.openlayer_inference_pipeline_name
-            ),
-        )
+            if (
+                self.openlayer_inference_pipeline_name
+                and not self.openlayer_project_name
+                and not self.openlayer_inference_pipeline_id
+            ):
+                raise ValueError(
+                    "You must provide the Openlayer project name where the inference"
+                    " pipeline is located."
+                    " Please set it as the environment variable"
+                    " OPENLAYER_PROJECT_NAME."
+                )
 
     def stream_data(self, data: Dict[str, any], config: Dict[str, any]) -> None:
         """Stream data to the Openlayer platform.
@@ -157,32 +156,35 @@ def _load_inference_pipeline(self) -> None:
         If no platform/project information is provided, it is set to None.
         """
         inference_pipeline = None
-        if self.openlayer_api_key:
-            client = openlayer.OpenlayerClient(
-                api_key=self.openlayer_api_key, verbose=False
+        client = openlayer.OpenlayerClient(
+            api_key=self.openlayer_api_key, verbose=False
+        )
+
+        # Prioritize the inference pipeline id over the name
+        if self.openlayer_inference_pipeline_id:
+            inference_pipeline = inference_pipelines.InferencePipeline(
+                client=client,
+                upload=None,
+                json={"id": self.openlayer_inference_pipeline_id, "projectId": None},
+                task_type=tasks.TaskType.LLM,
             )
-            if self.openlayer_inference_pipeline_id:
-                # Load inference pipeline directly from the id
-                inference_pipeline = inference_pipelines.InferencePipeline(
-                    client=client,
-                    upload=None,
-                    json={
-                        "id": self.openlayer_inference_pipeline_id,
-                        "projectId": None,
-                    },
-                    task_type=tasks.TaskType.LLM,
+        elif self.openlayer_inference_pipeline_name:
+            with utils.HidePrints():
+                project = client.create_project(
+                    name=self.openlayer_project_name, task_type=tasks.TaskType.LLM
                 )
-            else:
-                if self.openlayer_project_name:
-                    with utils.HidePrints():
-                        project = client.create_project(
-                            name=self.openlayer_project_name,
-                            task_type=tasks.TaskType.LLM,
-                        )
-                        inference_pipeline = project.create_inference_pipeline(
-                            name=self.openlayer_inference_pipeline_name
-                        )
-
+                inference_pipeline = project.create_inference_pipeline(
+                    name=self.openlayer_inference_pipeline_name
+                )
+        if inference_pipeline:
+            logger.info(
+                "Going to try to stream data to the inference pipeline with id %s.",
+                inference_pipeline.id,
+            )
+        else:
+            logger.warn(
+                "No inference pipeline found. Data will not be streamed to Openlayer."
+            )
         self.inference_pipeline = inference_pipeline
 
     def publish_batch_data(self, df: pd.DataFrame, config: Dict[str, any]) -> None: