Remove internal import from pipeline builder (Azure#28857)

Signed-off-by: Brynn Yin <[email protected]>

Author: brynn-code

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/_utils/_utils.py

@@ -4,6 +4,10 @@
 from typing import Dict, Optional, Union
 
 from azure.ai.ml import Input, Output
+from azure.ai.ml._internal._schema.input_output import SUPPORTED_INTERNAL_PARAM_TYPES
+from azure.ai.ml._utils.utils import get_all_enum_values_iter
+from azure.ai.ml.constants import AssetTypes
+from azure.ai.ml.constants._common import InputTypes
 from azure.ai.ml.constants._component import ComponentParameterTypes, IOConstants
 
 _INPUT_TYPE_ENUM = "enum"
@@ -127,3 +131,22 @@ def _from_base(cls, _output: Union[Output, Dict]) -> Optional["InternalOutput"]:
             _output.__class__ = InternalOutput
             return _output
         return InternalOutput(**_output)
+
+    def map_pipeline_output_type(self):
+        """Map output type to pipeline output type."""
+
+        def _map_primitive_type(_type):
+            """Convert double and float to number type."""
+            _type = _type.lower()
+            if _type in ["double", "float"]:
+                return InputTypes.NUMBER
+            return _type
+
+        if self.type in list(get_all_enum_values_iter(AssetTypes)):
+            return self.type
+        if self.type in SUPPORTED_INTERNAL_PARAM_TYPES:
+            return _map_primitive_type(self.type)
+        if self.type in ["AnyFile"]:
+            return AssetTypes.URI_FILE
+        # Handle AnyDirectory and the other types.
+        return AssetTypes.URI_FOLDER

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/entities/_input_outputs.py

@@ -11,7 +11,6 @@
 from inspect import Parameter, signature
 from typing import Callable, Union
 
-from azure.ai.ml._internal._utils._utils import _map_internal_output_type
 from azure.ai.ml._utils._func_utils import get_outputs_and_locals
 from azure.ai.ml._utils.utils import (
     is_valid_node_name,
@@ -259,6 +258,13 @@ def _build_pipeline_outputs(self, outputs: typing.Dict[str, NodeOutput]):
                     is_control=value.is_control,
                 )
 
+            # Hack: map internal output type to pipeline output type
+            def _map_internal_output_type(_meta):
+                """Map component output type to valid pipeline output type."""
+                if type(_meta).__name__ != "InternalOutput":
+                    return _meta.type
+                return _meta.map_pipeline_output_type()
+
             # Note: Here we set PipelineOutput as Pipeline's output definition as we need output binding.
             output_meta = Output(
                 type=_map_internal_output_type(meta),