Add Evaluation Service 1DP Client (#40683)

Author: singankit

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_common/evaluation_onedp_client.py

@@ -0,0 +1,115 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+import logging
+from typing import Union, Any
+from azure.core.credentials import AzureKeyCredential, TokenCredential
+from azure.ai.evaluation._common.onedp import AIProjectClient as RestEvaluationServiceClient
+from azure.ai.evaluation._common.onedp.models import (PendingUploadRequest, PendingUploadType, EvaluationResult,
+                                                      ResultType, AssetCredentialRequest, EvaluationUpload, InputDataset)
+from azure.storage.blob import ContainerClient
+from .utils import upload
+
+LOGGER = logging.getLogger(__name__)
+
+class EvaluationServiceOneDPClient:
+
+    def __init__(self, endpoint: str, credential: Union[AzureKeyCredential, "TokenCredential"], **kwargs: Any) -> None:
+        self.rest_client = RestEvaluationServiceClient(
+            endpoint=endpoint,
+            credential=credential,
+            **kwargs,
+        )
+
+    def create_evaluation_result(self, *, name: str, path: str, version=1, **kwargs) -> None:
+        """Create and upload evaluation results to Azure evaluation service.
+
+        This method uploads evaluation results from a local path to Azure Blob Storage
+        and registers them with the evaluation service. The process involves:
+        1. Starting a pending upload with the evaluation service
+        2. Getting a SAS token for the blob container
+        3. Uploading the local evaluation results to the blob container
+        4. Creating a version record for the evaluation results
+
+        :param name: The name to identify the evaluation results
+        :type name: str
+        :param path: The local path to the evaluation results file or directory
+        :type path: str
+        :param version: The version number for the evaluation results, defaults to 1
+        :type version: int, optional
+        :param kwargs: Additional keyword arguments to pass to the underlying API calls
+        :return: The response from creating the evaluation result version
+        :rtype: EvaluationResult
+        :raises: Various exceptions from the underlying API calls or upload process
+        """
+
+        LOGGER.debug(f"Creating evaluation result for {name} with version {version} from path {path}")
+        start_pending_upload_response = self.rest_client.evaluation_results.start_pending_upload(
+            name=name,
+            version=version,
+            body=PendingUploadRequest(pending_upload_type=PendingUploadType.TEMPORARY_BLOB_REFERENCE),
+            **kwargs
+        )
+
+        LOGGER.debug(f"Uploading {path} to {start_pending_upload_response.blob_reference_for_consumption.blob_uri}")
+        with ContainerClient.from_container_url(
+            start_pending_upload_response.blob_reference_for_consumption.credential.sas_uri) as container_client:
+            upload(path=path, container_client=container_client, logger=LOGGER)
+
+        LOGGER.debug(f"Creating evaluation result version for {name} with version {version}")
+        create_version_response = self.rest_client.evaluation_results.create_version(
+            body=EvaluationResult(
+                blob_uri=start_pending_upload_response.blob_reference_for_consumption.blob_uri,
+                result_type=ResultType.EVALUATION,
+                name=name,
+                version=version
+            ),
+            name=name,
+            version=version,
+            **kwargs
+        )
+
+        return create_version_response
+
+    def start_evaluation_run(self, *, evaluation: EvaluationUpload, **kwargs) -> EvaluationUpload:
+        """Start a new evaluation run in the Azure evaluation service.
+
+        This method creates a new evaluation run with the provided configuration details.
+
+        :param evaluation: The evaluation configuration to upload
+        :type evaluation: EvaluationUpload
+        :param kwargs: Additional keyword arguments to pass to the underlying API calls
+        :return: The created evaluation run object
+        :rtype: EvaluationUpload
+        :raises: Various exceptions from the underlying API calls
+        """
+        upload_run_response = self.rest_client.evaluations.upload_run(
+            evaluation=evaluation,
+            **kwargs
+        )
+
+        return upload_run_response
+
+    def update_evaluation_run(self, *, name: str, evaluation: EvaluationUpload, **kwargs) -> EvaluationUpload:
+        """Update an existing evaluation run in the Azure evaluation service.
+
+        This method updates an evaluation run with new information such as status changes,
+        result references, or other metadata.
+
+        :param name: The identifier of the evaluation run to update
+        :type name: str
+        :param evaluation: The updated evaluation configuration
+        :type evaluation: EvaluationUpload
+        :param kwargs: Additional keyword arguments to pass to the underlying API calls
+        :return: The updated evaluation run object
+        :rtype: EvaluationUpload
+        :raises: Various exceptions from the underlying API calls
+        """
+        update_run_response = self.rest_client.evaluations.upload_update_run(
+            name=name,
+            evaluation=evaluation,
+            **kwargs
+        )
+
+        return update_run_response

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_common/utils.py

@@ -1,13 +1,15 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
-
+import os
+import posixpath
 import re
 import math
 import threading
 from typing import Any, List, Literal, Mapping, Type, TypeVar, Tuple, Union, cast, get_args, get_origin
 
 import nltk
+from azure.storage.blob import ContainerClient
 from typing_extensions import NotRequired, Required, TypeGuard
 from azure.ai.evaluation._legacy._adapters._errors import MissingRequiredPackage
 from azure.ai.evaluation._constants import AZURE_OPENAI_TYPE, OPENAI_TYPE
@@ -463,3 +465,64 @@ def raise_exception(msg, target):
             "User and assistant role expected as the only role in each message.",
             ErrorTarget.CONTENT_SAFETY_CHAT_EVALUATOR,
         )
+
+def upload(path: str, container_client: ContainerClient, logger=None):
+    """Upload files or directories to Azure Blob Storage using a container client.
+
+    This function uploads a file or all files in a directory (recursively) to Azure Blob Storage.
+    When uploading a directory, the relative path structure is preserved in the blob container.
+
+    :param path: The local path to a file or directory to upload
+    :type path: str
+    :param container_client: The Azure Blob Container client to use for uploading
+    :type container_client: azure.storage.blob.ContainerClient
+    :param logger: Optional logger for debug output, defaults to None
+    :type logger: logging.Logger, optional
+    :raises EvaluationException: If the path doesn't exist or errors occur during upload
+    """
+
+    if not os.path.isdir(path) and not os.path.isfile(path):
+        raise EvaluationException(
+            message=f"Path '{path}' is not a directory or a file",
+            internal_message=f"Path '{path}' is not a directory or a file",
+            target=ErrorTarget.RAI_CLIENT,
+            category=ErrorCategory.INVALID_VALUE,
+            blame=ErrorBlame.SYSTEM_ERROR,
+        )
+
+    remote_paths = []
+    local_paths = []
+
+    if os.path.isdir(path):
+        for (root, _, filenames) in os.walk(path):
+            upload_path = ""
+            if root != path:
+                rel_path = os.path.relpath(root, path)
+                upload_path = posixpath.join(rel_path)
+            for f in filenames:
+                remote_file_path = posixpath.join(upload_path, f)
+                remote_paths.append(remote_file_path)
+                local_file_path = os.path.join(root, f)
+                local_paths.append(local_file_path)
+
+    if os.path.isfile(path):
+        remote_paths = [os.path.basename(path)]
+        local_paths = [path]
+
+    try:
+        # Open the file in binary read mode
+        for local, remote in zip(local_paths, remote_paths):
+            with open(local, "rb") as data:
+                # Upload the file to Azure Blob Storage
+                container_client.upload_blob(data=data, name=remote)
+            if logger:
+                logger.debug(f"File '{local}' uploaded successfully")
+
+    except Exception as e:
+        raise EvaluationException(
+            message=f"Error uploading file: {e}",
+            internal_message=f"Error uploading file: {e}",
+            target=ErrorTarget.RAI_CLIENT,
+            category=ErrorCategory.UPLOAD_ERROR,
+            blame=ErrorBlame.SYSTEM_ERROR,
+        )

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_exceptions.py

@@ -38,6 +38,7 @@ class ErrorCategory(Enum):
     FAILED_REMOTE_TRACKING = "FAILED REMOTE TRACKING"
     PROJECT_ACCESS_ERROR = "PROJECT ACCESS ERROR"
     UNKNOWN = "UNKNOWN"
+    UPLOAD_ERROR = "UPLOAD ERROR"
 
 
 class ErrorBlame(Enum):