Enable use of the new leaner ported BatchEngine code path (Azure#40241)

- Added a new kwargs argument called `_use_run_submitter_client` that is set to true when calling evaluate() will enable the use of the new faster and leaner ported code path without the promptflow dependenecy. Please note that this is still in a "happy path" works state with some features missing (e.g. handling target function calls which will be added in a future PR). The goal here is to enable testing of this code path sooner to enable us to start finding bugs/issues
- Created a `BatchClient` protocol to standardize the existing `CodeClient`, `ProxyClient`, as well as the newly added `RunSubmitterClient`. This makes the evaluate logic simpler in the rest of the code
- Though still installed by default, made promptflow an optional dependency by:
  - Creating some adapters to handle the case where promptflow is not installed, and replace with either some stub code, or the closest ported version of the code as needed
  - Added some additional dependencies to setup.py that were implicitly brought in by the promptflow dependency that are still needed by the ported legacy code
  - Removed some unneeded code now that tracing support has been deprecated/disabled, as well as some unneeded test code

Author: ralph-msft

sdk/evaluation/azure-ai-evaluation/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "python",
   "TagPrefix": "python/evaluation/azure-ai-evaluation",
-  "Tag": "python/evaluation/azure-ai-evaluation_f423a24efa"
+  "Tag": "python/evaluation/azure-ai-evaluation_e33b6c53d7"
 }

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

@@ -15,7 +15,7 @@
 
 import jwt
 
-from promptflow.core._errors import MissingRequiredPackage
+from azure.ai.evaluation._legacy._adapters._errors import MissingRequiredPackage
 from azure.ai.evaluation._exceptions import ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException
 from azure.ai.evaluation._http_utils import AsyncHttpPipeline, get_async_http_client
 from azure.ai.evaluation._model_configurations import AzureAIProject

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

@@ -9,7 +9,7 @@
 
 import nltk
 from typing_extensions import NotRequired, Required, TypeGuard
-from promptflow.core._errors import MissingRequiredPackage
+from azure.ai.evaluation._legacy._adapters._errors import MissingRequiredPackage
 from azure.ai.evaluation._constants import AZURE_OPENAI_TYPE, OPENAI_TYPE
 from azure.ai.evaluation._exceptions import ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException
 from azure.ai.evaluation._model_configurations import (

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_converters/__init__.py

@@ -1,3 +1,3 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
-# ---------------------------------------------------------
+# ---------------------------------------------------------

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/code_client.py

@@ -6,17 +6,17 @@
 import logging
 import os
 from concurrent.futures import Future
-from pathlib import Path
-from typing import Any, Callable, Dict, Optional, Union, cast
+from typing import Any, Callable, Dict, Optional, Sequence, Union, cast
 
 import pandas as pd
-from promptflow.contracts.types import AttrDict
-from promptflow.tracing import ThreadPoolExecutorWithContext as ThreadPoolExecutor
+from azure.ai.evaluation._legacy._adapters.types import AttrDict
+from azure.ai.evaluation._legacy._adapters.tracing import ThreadPoolExecutorWithContext as ThreadPoolExecutor
 
 from azure.ai.evaluation._evaluate._utils import _apply_column_mapping, _has_aggregator, get_int_env_var, load_jsonl
 from azure.ai.evaluation._exceptions import ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException
 
 from ..._constants import PF_BATCH_TIMEOUT_SEC, PF_BATCH_TIMEOUT_SEC_DEFAULT
+from .batch_clients import BatchClientRun
 
 LOGGER = logging.getLogger(__name__)
 
@@ -84,7 +84,7 @@ def _calculate_metric(
             for param in inspect.signature(evaluator).parameters.values()
             if param.name not in ["args", "kwargs"]
         }
-        for value in input_df.to_dict("records"):
+        for value in cast(Sequence[Dict[str, Any]], input_df.to_dict("records")):
             # Filter out only the parameters that are present in the input data
             # if no parameters then pass data as is
             filtered_values = {k: v for k, v in value.items() if k in parameters} if len(parameters) > 0 else value
@@ -133,10 +133,10 @@ def _calculate_aggregations(evaluator: Callable, run: CodeRun) -> Any:
     def run(
         self,  # pylint: disable=unused-argument
         flow: Callable,
-        data: Union[os.PathLike, Path, pd.DataFrame],
-        evaluator_name: Optional[str] = None,
+        data: Union[str, os.PathLike, pd.DataFrame],
         column_mapping: Optional[Dict[str, str]] = None,
-        **kwargs,
+        evaluator_name: Optional[str] = None,
+        **kwargs: Any,
     ) -> CodeRun:
         input_df = data
         if not isinstance(input_df, pd.DataFrame):
@@ -157,7 +157,7 @@ def run(
             evaluator=flow,
             input_df=input_df,
             column_mapping=column_mapping,
-            evaluator_name=evaluator_name,
+            evaluator_name=evaluator_name or "",
         )
 
         return CodeRun(
@@ -169,11 +169,13 @@ def run(
             ),
         )
 
-    def get_details(self, run: CodeRun, all_results: bool = False) -> pd.DataFrame:
+    def get_details(self, client_run: BatchClientRun, all_results: bool = False) -> pd.DataFrame:
+        run = self._get_result(client_run)
         result_df = run.get_result_df(exclude_inputs=not all_results)
         return result_df
 
-    def get_metrics(self, run: CodeRun) -> Dict[str, Any]:
+    def get_metrics(self, client_run: BatchClientRun) -> Dict[str, Any]:
+        run = self._get_result(client_run)
         try:
             aggregated_metrics = run.get_aggregated_metrics()
             print("Aggregated metrics")
@@ -183,6 +185,10 @@ def get_metrics(self, run: CodeRun) -> Dict[str, Any]:
             return {}
         return aggregated_metrics
 
-    def get_run_summary(self, run: CodeRun) -> Any:  # pylint: disable=unused-argument
+    def get_run_summary(self, client_run: BatchClientRun) -> Any:  # pylint: disable=unused-argument
         # Not implemented
         return None
+
+    @staticmethod
+    def _get_result(run: BatchClientRun) -> CodeRun:
+        return cast(CodeRun, run)

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/eval_run_context.py

@@ -5,9 +5,9 @@
 import types
 from typing import Optional, Type, Union
 
-from promptflow._sdk._constants import PF_FLOW_ENTRY_IN_TMP, PF_FLOW_META_LOAD_IN_SUBPROCESS
-from promptflow._utils.user_agent_utils import ClientUserAgentUtil
-from promptflow.tracing._integrations._openai_injector import inject_openai_api, recover_openai_api
+from azure.ai.evaluation._legacy._adapters._constants import PF_FLOW_ENTRY_IN_TMP, PF_FLOW_META_LOAD_IN_SUBPROCESS
+from azure.ai.evaluation._legacy._adapters.utils import ClientUserAgentUtil
+from azure.ai.evaluation._legacy._adapters.tracing import inject_openai_api, recover_openai_api
 
 from azure.ai.evaluation._constants import (
     OTEL_EXPORTER_OTLP_TRACES_TIMEOUT,
@@ -19,6 +19,8 @@
 
 from ..._user_agent import USER_AGENT
 from .._utils import set_event_loop_policy
+from .batch_clients import BatchClient
+from ._run_submitter_client import RunSubmitterClient
 from .code_client import CodeClient
 from .proxy_client import ProxyClient
 
@@ -33,7 +35,7 @@ class EvalRunContext:
     ]
     """
 
-    def __init__(self, client: Union[CodeClient, ProxyClient]) -> None:
+    def __init__(self, client: BatchClient) -> None:
         self.client = client
         self._is_batch_timeout_set_by_system = False
         self._is_otel_timeout_set_by_system = False
@@ -64,6 +66,9 @@ def __enter__(self) -> None:
             # For addressing the issue of asyncio event loop closed on Windows
             set_event_loop_policy()
 
+        if isinstance(self.client, RunSubmitterClient):
+            set_event_loop_policy()
+
     def __exit__(
         self,
         exc_type: Optional[Type[BaseException]],

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/proxy_client.py

@@ -8,15 +8,21 @@
 import logging
 import math
 import os
+from datetime import datetime
 from collections import OrderedDict
 from concurrent.futures import Future
-from typing import Any, Callable, Dict, Optional, Union
+from typing import Any, Callable, Dict, Optional, Union, cast
 
+from azure.ai.evaluation._legacy._adapters.entities import Run
+from azure.ai.evaluation._legacy._adapters._configuration import Configuration
+from azure.ai.evaluation._legacy._adapters.client import PFClient
+from azure.ai.evaluation._legacy._adapters.tracing import ThreadPoolExecutorWithContext
 import pandas as pd
-from promptflow.client import PFClient
-from promptflow.entities import Run
-from promptflow.tracing import ThreadPoolExecutorWithContext as ThreadPoolExecutor
 
+from azure.ai.evaluation._evaluate._batch_run.batch_clients import BatchClientRun, HasAsyncCallable
+
+
+Configuration.get_instance().set_config("trace.destination", "none")
 LOGGER = logging.getLogger(__name__)
 
 
@@ -26,46 +32,56 @@ def __init__(self, run: Future, **kwargs) -> None:  # pylint: disable=unused-arg
 
 
 class ProxyClient:  # pylint: disable=client-accepts-api-version-keyword
-    def __init__(  # pylint: disable=missing-client-constructor-parameter-credential,missing-client-constructor-parameter-kwargs
-        self, pf_client: PFClient
+    def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
+        self,
+        **kwargs: Any,
     ) -> None:
-        self._pf_client = pf_client
-        self._thread_pool = ThreadPoolExecutor(thread_name_prefix="evaluators_thread")
+        self._pf_client = PFClient(**kwargs)
+        self._thread_pool = ThreadPoolExecutorWithContext(thread_name_prefix="evaluators_thread")
 
     def run(
         self,
-        flow: Union[str, os.PathLike, Callable],
-        data: Union[str, os.PathLike],
+        flow: Callable,
+        data: Union[str, os.PathLike, pd.DataFrame],
         column_mapping: Optional[Dict[str, str]] = None,
-        **kwargs
+        evaluator_name: Optional[str] = None,
+        **kwargs: Any,
     ) -> ProxyRun:
-        flow_to_run = flow
-        if os.getenv("AI_EVALS_BATCH_USE_ASYNC", "true").lower() == "true" and hasattr(flow, "_to_async"):
+        if isinstance(data, pd.DataFrame):
+            raise ValueError("Data cannot be a pandas DataFrame")
+
+        flow_to_run: Callable = flow
+        if os.getenv("AI_EVALS_BATCH_USE_ASYNC", "true").lower() == "true" and isinstance(flow, HasAsyncCallable):
             flow_to_run = flow._to_async()  # pylint: disable=protected-access
 
+        name: str = kwargs.pop("name", "")
+        if not name:
+            name = f"azure_ai_evaluation_evaluators_{evaluator_name}_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
+
         batch_use_async = self._should_batch_use_async(flow_to_run)
         eval_future = self._thread_pool.submit(
             self._pf_client.run,
             flow_to_run,
             data=data,
-            column_mapping=column_mapping,
+            column_mapping=column_mapping,  # type: ignore
             batch_use_async=batch_use_async,
-            **kwargs
+            name=name,
+            **kwargs,
         )
         return ProxyRun(run=eval_future)
 
-    def get_details(self, proxy_run: ProxyRun, all_results: bool = False) -> pd.DataFrame:
-        run: Run = proxy_run.run.result()
+    def get_details(self, client_run: BatchClientRun, all_results: bool = False) -> pd.DataFrame:
+        run: Run = self.get_result(client_run)
         result_df = self._pf_client.get_details(run, all_results=all_results)
         result_df.replace("(Failed)", math.nan, inplace=True)
         return result_df
 
-    def get_metrics(self, proxy_run: ProxyRun) -> Dict[str, Any]:
-        run: Run = proxy_run.run.result()
+    def get_metrics(self, client_run: BatchClientRun) -> Dict[str, Any]:
+        run: Run = self.get_result(client_run)
         return self._pf_client.get_metrics(run)
 
-    def get_run_summary(self, proxy_run: ProxyRun) -> Dict[str, Any]:
-        run = proxy_run.run.result()
+    def get_run_summary(self, client_run: BatchClientRun) -> Dict[str, Any]:
+        run: Run = self.get_result(client_run)
 
         # pylint: disable=protected-access
         completed_lines = run._properties.get("system_metrics", {}).get("__pf__.lines.completed", "NA")
@@ -81,13 +97,17 @@ def get_run_summary(self, proxy_run: ProxyRun) -> Dict[str, Any]:
         return OrderedDict(
             [
                 ("status", status),
-                ("duration", str(run._end_time - run._created_on)),
+                ("duration", str((run._end_time or run._created_on) - run._created_on)),
                 ("completed_lines", completed_lines),
                 ("failed_lines", failed_lines),
                 ("log_path", str(run._output_path)),
             ]
         )
 
+    @staticmethod
+    def get_result(run: BatchClientRun) -> Run:
+        return cast(ProxyRun, run).run.result()
+
     @staticmethod
     def _should_batch_use_async(flow):
         if os.getenv("AI_EVALS_BATCH_USE_ASYNC", "true").lower() == "true":

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/target_run_context.py

@@ -5,7 +5,7 @@
 import types
 from typing import Optional, Type
 
-from promptflow._sdk._constants import PF_FLOW_ENTRY_IN_TMP
+from azure.ai.evaluation._legacy._adapters._constants import PF_FLOW_ENTRY_IN_TMP
 from azure.ai.evaluation._constants import PF_DISABLE_TRACING
 
 

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_eval_run.py

@@ -13,7 +13,7 @@
 from typing import Any, Dict, List, Optional, Set, Type
 from urllib.parse import urlparse
 
-from promptflow._sdk.entities import Run
+from azure.ai.evaluation._legacy._adapters.entities import Run
 from typing_extensions import Self
 
 from azure.ai.evaluation._exceptions import ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException