Repository: microsoft/promptflow
File: src/promptflow-evals/promptflow/evals/evaluate/_batch_run_client/proxy_client.py (Python)

Code before:
from promptflow.client import PFClient
from promptflow.tracing import ThreadPoolExecutorWithContext as ThreadPoolExecutor

from ..._constants import PF_BATCH_TIMEOUT_DEFAULT, PF_BATCH_TIMEOUT_SEC

LOGGER = logging.getLogger(__name__)


class ProxyRun:
    def __init__(self, run, **kwargs):
        self.run = run


class ProxyClient:
    def __init__(self, pf_client: PFClient):
        self._pf_client = pf_client
        self._thread_pool = ThreadPoolExecutor(thread_name_prefix="evaluators_thread")

    def run(self, flow, data, column_mapping=None, **kwargs):
        eval_future = self._thread_pool.submit(
            self._pf_client.run, flow, data=data, column_mapping=column_mapping, **kwargs
        )
        return ProxyRun(run=eval_future)

    def get_details(self, proxy_run, all_results=False):
        batch_run_timeout = get_int_env_var(PF_BATCH_TIMEOUT_SEC, PF_BATCH_TIMEOUT_DEFAULT)
        run = proxy_run.run.result(timeout=batch_run_timeout)
        result_df = self._pf_client.get_details(run, all_results=all_results)
        result_df.replace("(Failed)", np.nan, inplace=True)
        return result_df

    def get_metrics(self, proxy_run):
        batch_run_timeout = get_int_env_var(PF_BATCH_TIMEOUT_SEC, PF_BATCH_TIMEOUT_DEFAULT)
        run = proxy_run.run.result(timeout=batch_run_timeout)
        return self._pf_client.get_metrics(run)


Review comment (performance):
Unlike CodeClient, ProxyClient relies on pfclient.run. The underlying Batch Engine honors the PF_BATCH_TIMEOUT_SEC environment variable and handles the timeout already. Therefore, it's not necessary to pass the timeout here. Once the Batch Engine times out, proxy_run.run.result() will return immediately.

Instead of making changes in this class, I suggest the following modifications in batch_run_context.py:

In "\_\_enter\_\_", if PF_BATCH_TIMEOUT_SEC is not set, assign it a default value.
In "\_\_exit\_\_", ~~revert the PF_BATCH_TIMEOUT_SEC change~~. No change is needed in method.

Code after:
from promptflow.client import PFClient
from promptflow.tracing import ThreadPoolExecutorWithContext as ThreadPoolExecutor

LOGGER = logging.getLogger(__name__)


class ProxyRun:
    def __init__(self, run, **kwargs):
        self.run = run


class ProxyClient:
    def __init__(self, pf_client: PFClient):
        self._pf_client = pf_client
        self._thread_pool = ThreadPoolExecutor(thread_name_prefix="evaluators_thread")

    def run(self, flow, data, column_mapping=None, **kwargs):
        flow_to_run = flow
        if hasattr(flow, "_to_async"):
            flow_to_run = flow._to_async()

        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,
            batch_use_async=batch_use_async,
            **kwargs
        )
        return ProxyRun(run=eval_future)

    def get_details(self, proxy_run, all_results=False):
        run = proxy_run.run.result()
        result_df = self._pf_client.get_details(run, all_results=all_results)
        result_df.replace("(Failed)", np.nan, inplace=True)
        return result_df

    def get_metrics(self, proxy_run):
        run = proxy_run.run.result()
        return self._pf_client.get_metrics(run)

    @staticmethod
    def _should_batch_use_async(flow):
        if hasattr(flow, "__call__") and inspect.iscoroutinefunction(flow.__call__):
            return True
        elif inspect.iscoroutinefunction(flow):
            return True
        else:
            return False