Repository: ray-project/ray
File: python/ray/llm/_internal/batch/stages/vllm_engine_stage.py (Python)

Code before:
            logger.error(
                f"[vLLM] Fatal engine error, exiting actor to trigger restart: {e}"
            )
            os._exit(1)
        except Exception as e:
            if not self.should_continue_on_error:
                raise
            error_msg = f"{type(e).__name__}: {str(e)}"
            logger.warning(
                "[vLLM] Inference failed for row %d in batch %s: %s",
                idx_in_batch,
                batch_uuid.hex,
                error_msg,
            )
            # Include snippet of failed prompt for debuggability
            prompt = truncate_str(row.get("prompt", ""), _MAX_PROMPT_LENGTH_IN_ERROR)

            # Construct default vLLMOutputData for schema consistency with success rows
            default_output = vLLMOutputData(
                prompt=prompt,
                prompt_token_ids=None,
                num_input_tokens=0,
            )
            return {
                **default_output.model_dump(),
                "request_id": "",
                self.IDX_IN_BATCH_COLUMN: idx_in_batch,
                "batch_uuid": batch_uuid.hex,
                "time_taken_llm": -1,
                "params": "",
                "__inference_error__": error_msg,
            }

    async def udf(self, batch: List[Dict[str, Any]]) -> AsyncIterator[Dict[str, Any]]:
        """Run the vLLM engine.

        Args:
            batch: A list of rows to run the vLLM engine on.

        Returns:
            The response of the vLLM engine.
        """
        batch_uuid = uuid.uuid4()
        batch_start_time = time.perf_counter()

        tasks = [
            asyncio.create_task(self._generate_with_error_handling(row, batch_uuid))
            for row in batch
        ]

        for resp in asyncio.as_completed(tasks):

Review comment (suggestion):
```suggestion
                "request_id": -1,
```

Code after:
            logger.error(
                f"[vLLM] Fatal engine error, exiting actor to trigger restart: {e}"
            )
            os._exit(1)
        except Exception as e:
            if not self.should_continue_on_error:
                raise
            error_msg = f"{type(e).__name__}: {str(e)}"
            logger.warning(
                "[vLLM] Inference failed for row %d in batch %s: %s",
                idx_in_batch,
                batch_uuid.hex,
                error_msg,
            )
            # Include snippet of failed prompt for debuggability
            prompt = truncate_str(row.get("prompt", ""), _MAX_PROMPT_LENGTH_IN_ERROR)

            # Construct default vLLMOutputData for schema consistency with success rows
            default_output = vLLMOutputData(
                prompt=prompt,
                prompt_token_ids=None,
                num_input_tokens=0,
            )
            return {
                **default_output.model_dump(),
                "request_id": -1,
                self.IDX_IN_BATCH_COLUMN: idx_in_batch,
                "batch_uuid": batch_uuid.hex,
                "time_taken_llm": -1,
                "params": "",
                "__inference_error__": error_msg,
            }

    async def udf(self, batch: List[Dict[str, Any]]) -> AsyncIterator[Dict[str, Any]]:
        """Run the vLLM engine.

        Args:
            batch: A list of rows to run the vLLM engine on.

        Returns:
            The response of the vLLM engine.
        """
        batch_uuid = uuid.uuid4()
        batch_start_time = time.perf_counter()

        tasks = [
            asyncio.create_task(self._generate_with_error_handling(row, batch_uuid))
            for row in batch
        ]

        for resp in asyncio.as_completed(tasks):