Repository: microsoft/promptflow
File: src/promptflow-devkit/promptflow/batch/_batch_engine.py (Python)

Code before:
        elif self._executor_proxy.allow_aggregation:
            # execute aggregation nodes
            aggr_exec_result = await self._exec_aggregation(batch_inputs, line_results, run_id)
            # use the execution result to update aggr_result to make sure we can get the aggr_result in _exec_in_task
            self._update_aggr_result(aggr_result, aggr_exec_result)
        # summary some infos from line results and aggr results to batch result
        return BatchResult.create(self._start_time, datetime.utcnow(), line_results, aggr_result, exception=ex)

    async def _exec_batch(
        self,
        line_results: List[LineResult],
        batch_inputs: List[Mapping[str, Any]],
        run_id: Optional[str] = None,
    ) -> List[LineResult]:
        # line_results as input parameter, so that the completed line results can be summarized
        # when batch run is canceled.
        worker_count = self._worker_count or DEFAULT_CONCURRENCY
        semaphore = asyncio.Semaphore(worker_count)

        batch_inputs_dict = {}
        if any(LINE_NUMBER_KEY not in input for input in batch_inputs):
            bulk_logger.warning("There are lines without line number in the batch inputs, use index as line number.")
            batch_inputs_dict = {i: input for i, input in enumerate(batch_inputs)}
        else:
            batch_inputs_dict = {input[LINE_NUMBER_KEY]: input for input in batch_inputs}

        pending = [
            asyncio.create_task(self._exec_line_under_semaphore(semaphore, line_inputs, i, run_id))
            for i, line_inputs in batch_inputs_dict.items()
        ]

        total_lines = len(batch_inputs)
        completed_line = 0
        last_log_count = 0
        while completed_line < total_lines:
            # wait for any task to complete
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            completed_line_results = [task.result() for task in done]
            # persist node run infos and flow run info in line result to storage
            self._persist_run_info(completed_line_results)
            line_results.extend(completed_line_results)
            # update the progress log
            completed_line += len(completed_line_results)
            last_log_count = log_progress(
                run_start_time=self._start_time,
                total_count=total_lines,
                current_count=completed_line,
                last_log_count=last_log_count,
            )

    async def _exec_line_under_semaphore(

Review comment (suggestion):
Doesn’t local python and single container need this logic? Should it all be consolidated in the code shown in the figure below?
![image](https://github.com/microsoft/promptflow/assets/111329184/0cbf0b89-6690-40cd-833b-79734b5f3e19)

Code after:
        elif self._executor_proxy.allow_aggregation:
            # execute aggregation nodes
            aggr_exec_result = await self._exec_aggregation(batch_inputs, line_results, run_id)
            # use the execution result to update aggr_result to make sure we can get the aggr_result in _exec_in_task
            self._update_aggr_result(aggr_result, aggr_exec_result)
        # summary some infos from line results and aggr results to batch result
        return BatchResult.create(self._start_time, datetime.utcnow(), line_results, aggr_result, exception=ex)

    async def _exec_batch(
        self,
        line_results: List[LineResult],
        batch_inputs: List[Mapping[str, Any]],
        run_id: Optional[str] = None,
    ) -> List[LineResult]:
        # line_results as input parameter, so that the completed line results can be summarized
        # when batch run is canceled.
        worker_count = self._worker_count or DEFAULT_CONCURRENCY
        semaphore = asyncio.Semaphore(worker_count)

        pending = [
            asyncio.create_task(
                self._exec_line_under_semaphore(semaphore, line_input, line_input[LINE_NUMBER_KEY], run_id)
            )
            for line_input in batch_inputs
        ]

        total_lines = len(batch_inputs)
        completed_line = 0
        last_log_count = 0
        while completed_line < total_lines:
            # wait for any task to complete
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            completed_line_results = [task.result() for task in done]
            # persist node run infos and flow run info in line result to storage
            self._persist_run_info(completed_line_results)
            line_results.extend(completed_line_results)
            # update the progress log
            completed_line += len(completed_line_results)
            last_log_count = log_progress(
                run_start_time=self._start_time,
                total_count=total_lines,
                current_count=completed_line,
                last_log_count=last_log_count,
            )

    async def _exec_line_under_semaphore(
        self,
        semaphore,
        inputs: Mapping[str, Any],
        index: Optional[int] = None,
        run_id: Optional[str] = None,