vllm.entrypoints.launchers.run_batch

Classes:

• BatchFrontendArgs –

  Arguments for the batch runner frontend.

• BatchRequestInput –

  The per-line object of the batch input file.

• BatchRequestOutput –

  The per-line object of the batch output and error files

• BatchTranscriptionRequest –

  Batch transcription request that uses file_url instead of file.

• BatchTranslationRequest –

  Batch translation request that uses file_url instead of file.

• EndpointConfig –

  How a batch request URL is matched to the handler that serves it.

Functions:

• batch_output_writer –

  Yield a file that receives responses in input order.

• build_endpoint_registry –

  Build the endpoint registry with all serving objects and handler configurations.

• dispatch_batch –

  Run every request with at most max_inflight alive at once.

• download_bytes_from_url –

  Download data from a URL or decode from a data URL.

• handle_endpoint_request –

  Generic handler for endpoint requests.

• ignore_admission_limits –

  Clear the engine limits that reject requests with a 503.

• is_same_local_file –

  Whether both paths name the same regular file; only those can be truncated.

• make_transcription_wrapper –

  Factory function to create a wrapper for transcription/translation handlers.

• open_batch_input –

  Yield the batch as a seekable text file, so it can be read twice.

• run_one_request –

  Route a single line of the batch to its endpoint handler.

• staging_file –

  In-memory buffer, or a temporary file when a directory is given.

• upload_data –

  Upload a local file to a URL.

• url_matches –

  Unversioned endpoints (/score, /rerank) match by suffix; others exactly.

• validate_batch –

  Parse every request and return the count; requests are re-parsed when run.

BatchFrontendArgs

Bases: BaseFrontendArgs

Arguments for the batch runner frontend.

Attributes:

• enable_metrics (bool) –

  Enable Prometheus metrics

• host (str | None) –

  Host name for the Prometheus metrics server

• input_file (str | None) –

  The path or url to a single input file. Currently supports local file

• max_inflight (int | None) –

  Maximum number of requests queued at the engine at once. Responses are

• output_file (str | None) –

  The path or url to a single output file. Currently supports

• output_tmp_dir (str | None) –

  Directory for staging a URL input, a piped input, or the output before

• port (int) –

  Port number for the Prometheus metrics server

• url (str) –

  [DEPRECATED] Host name for the Prometheus metrics server

Source code in vllm/entrypoints/launchers/run_batch.py

@config
class BatchFrontendArgs(BaseFrontendArgs):
    """Arguments for the batch runner frontend."""

    input_file: str | None = None
    """The path or url to a single input file. Currently supports local file
    paths, or the http protocol (http or https). If a URL is specified,
    the file should be available via HTTP GET."""
    output_file: str | None = None
    """The path or url to a single output file. Currently supports
    local file paths, or web (http or https) urls. If a URL is specified,
    the file should be available via HTTP PUT."""
    output_tmp_dir: str | None = None
    """Directory for staging a URL input, a piped input, or the output before
    it is uploaded to a URL. When unset, piped input and output are held in
    memory and URL input is downloaded to the system temporary directory."""
    max_inflight: int | None = None
    """Maximum number of requests queued at the engine at once. Responses are
    written in input order, so up to 16 times this many finished responses may
    be held behind a slow request. Defaults to twice max_num_seqs across all
    data-parallel engines, at least 1024. --max-num-queued-reqs and
    --max-num-queued-tokens are ignored."""
    enable_metrics: bool = False
    """Enable Prometheus metrics"""
    host: str | None = None
    """Host name for the Prometheus metrics server
    (only needed if enable-metrics is set)."""
    port: int = 8000
    """Port number for the Prometheus metrics server
    (only needed if enable-metrics is set)."""
    url: str = "0.0.0.0"
    """[DEPRECATED] Host name for the Prometheus metrics server
    (only needed if enable-metrics is set). Use --host instead."""

    @classmethod
    def _customize_cli_kwargs(
        cls,
        frontend_kwargs: dict[str, Any],
    ) -> dict[str, Any]:
        frontend_kwargs = super()._customize_cli_kwargs(frontend_kwargs)

        frontend_kwargs["input_file"]["flags"] = ["-i"]
        frontend_kwargs["input_file"]["required"] = True
        frontend_kwargs["output_file"]["flags"] = ["-o"]
        frontend_kwargs["output_file"]["required"] = True

        frontend_kwargs["enable_metrics"]["action"] = "store_true"

        frontend_kwargs["url"]["deprecated"] = True
        return frontend_kwargs

enable_metrics = False class-attribute instance-attribute

Enable Prometheus metrics

host = None class-attribute instance-attribute

Host name for the Prometheus metrics server (only needed if enable-metrics is set).

input_file = None class-attribute instance-attribute

The path or url to a single input file. Currently supports local file paths, or the http protocol (http or https). If a URL is specified, the file should be available via HTTP GET.

max_inflight = None class-attribute instance-attribute

Maximum number of requests queued at the engine at once. Responses are written in input order, so up to 16 times this many finished responses may be held behind a slow request. Defaults to twice max_num_seqs across all data-parallel engines, at least 1024. --max-num-queued-reqs and --max-num-queued-tokens are ignored.

output_file = None class-attribute instance-attribute

The path or url to a single output file. Currently supports local file paths, or web (http or https) urls. If a URL is specified, the file should be available via HTTP PUT.

output_tmp_dir = None class-attribute instance-attribute

Directory for staging a URL input, a piped input, or the output before it is uploaded to a URL. When unset, piped input and output are held in memory and URL input is downloaded to the system temporary directory.

port = 8000 class-attribute instance-attribute

Port number for the Prometheus metrics server (only needed if enable-metrics is set).

url = '0.0.0.0' class-attribute instance-attribute

[DEPRECATED] Host name for the Prometheus metrics server (only needed if enable-metrics is set). Use --host instead.

BatchRequestInput

Bases: OpenAIBaseModel

The per-line object of the batch input file.

url selects the request type that body is parsed as.

Source code in vllm/entrypoints/launchers/run_batch.py

class BatchRequestInput(OpenAIBaseModel):
    """The per-line object of the batch input file.

    ``url`` selects the request type that ``body`` is parsed as.
    """

    # A developer-provided per-request id that will be used to match outputs to
    # inputs. Must be unique for each request in a batch.
    custom_id: str

    # The HTTP method to be used for the request. Currently only POST is
    # supported.
    method: str

    # The OpenAI API relative URL to be used for the request. Currently
    # /v1/chat/completions is supported.
    url: str

    # The parameters of the request.
    body: BatchRequestInputBody

    @field_validator("body", mode="plain")
    @classmethod
    def check_type_for_url(cls, value: Any, info: ValidationInfo):
        # Use url to disambiguate models
        url: str = info.data["url"]
        if url == "/v1/chat/completions":
            return ChatCompletionRequest.model_validate(value)
        if url == "/v1/embeddings":
            return TypeAdapter(EmbeddingRequest).validate_python(value)
        if url.endswith("/score"):
            return TypeAdapter(ScoreRequest).validate_python(value)
        if url.endswith("/rerank"):
            return RerankRequest.model_validate(value)
        if url == "/v1/audio/transcriptions":
            return BatchTranscriptionRequest.model_validate(value)
        if url == "/v1/audio/translations":
            return BatchTranslationRequest.model_validate(value)
        return TypeAdapter(BatchRequestInputBody).validate_python(value)

BatchRequestOutput

Bases: OpenAIBaseModel

The per-line object of the batch output and error files

Source code in vllm/entrypoints/launchers/run_batch.py

class BatchRequestOutput(OpenAIBaseModel):
    """The per-line object of the batch output and error files"""

    id: str

    # A developer-provided per-request id that will be used to match outputs to
    # inputs.
    custom_id: str

    response: BatchResponseData | None

    # For requests that failed with a non-HTTP error, this will contain more
    # information on the cause of the failure.
    error: Any | None

BatchTranscriptionRequest

Bases: TranscriptionRequest

Batch transcription request that uses file_url instead of file.

This class extends TranscriptionRequest but replaces the file field with file_url to support batch processing from audio files written in JSON format.

Methods:

• validate_no_file –

  Ensure file field is not provided in batch requests.

Source code in vllm/entrypoints/launchers/run_batch.py

class BatchTranscriptionRequest(TranscriptionRequest):
    """Batch transcription request that uses file_url instead of file.

    This class extends TranscriptionRequest but replaces the file field
    with file_url to support batch processing from audio files written in JSON format.
    """

    file_url: str = Field(
        ...,
        description=(
            "Either a URL of the audio or a data URL with base64 encoded audio data. "
        ),
    )

    # Override file to be optional and unused for batch processing
    file: UploadFile | None = Field(default=None, exclude=True)  # type: ignore[assignment]

    @model_validator(mode="before")
    @classmethod
    def validate_no_file(cls, data: Any):
        """Ensure file field is not provided in batch requests."""
        if isinstance(data, dict) and "file" in data:
            raise VLLMValidationError(
                "The 'file' field is not supported in batch requests. "
                "Use 'file_url' instead.",
                parameter="file",
            )
        return data

validate_no_file(data) classmethod

Ensure file field is not provided in batch requests.

Source code in vllm/entrypoints/launchers/run_batch.py

@model_validator(mode="before")
@classmethod
def validate_no_file(cls, data: Any):
    """Ensure file field is not provided in batch requests."""
    if isinstance(data, dict) and "file" in data:
        raise VLLMValidationError(
            "The 'file' field is not supported in batch requests. "
            "Use 'file_url' instead.",
            parameter="file",
        )
    return data

BatchTranslationRequest

Bases: TranslationRequest

Batch translation request that uses file_url instead of file.

This class extends TranslationRequest but replaces the file field with file_url to support batch processing from audio files written in JSON format.

Methods:

• validate_no_file –

  Ensure file field is not provided in batch requests.

Source code in vllm/entrypoints/launchers/run_batch.py

class BatchTranslationRequest(TranslationRequest):
    """Batch translation request that uses file_url instead of file.

    This class extends TranslationRequest but replaces the file field
    with file_url to support batch processing from audio files written in JSON format.
    """

    file_url: str = Field(
        ...,
        description=(
            "Either a URL of the audio or a data URL with base64 encoded audio data. "
        ),
    )

    # Override file to be optional and unused for batch processing
    file: UploadFile | None = Field(default=None, exclude=True)  # type: ignore[assignment]

    @model_validator(mode="before")
    @classmethod
    def validate_no_file(cls, data: Any):
        """Ensure file field is not provided in batch requests."""
        if isinstance(data, dict) and "file" in data:
            raise VLLMValidationError(
                "The 'file' field is not supported in batch requests. "
                "Use 'file_url' instead.",
                parameter="file",
            )
        return data

validate_no_file(data) classmethod

Ensure file field is not provided in batch requests.

Source code in vllm/entrypoints/launchers/run_batch.py

@model_validator(mode="before")
@classmethod
def validate_no_file(cls, data: Any):
    """Ensure file field is not provided in batch requests."""
    if isinstance(data, dict) and "file" in data:
        raise VLLMValidationError(
            "The 'file' field is not supported in batch requests. "
            "Use 'file_url' instead.",
            parameter="file",
        )
    return data

EndpointConfig

Bases: TypedDict

How a batch request URL is matched to the handler that serves it.

Source code in vllm/entrypoints/launchers/run_batch.py

class EndpointConfig(TypedDict):
    """How a batch request URL is matched to the handler that serves it."""

    url: str
    handler_getter: Callable[[], Callable | None]
    wrapper_fn: WrapperFn | None

batch_output_writer(path_or_url, output_tmp_dir) async

Yield a file that receives responses in input order.

A local file holds the responses for a prefix of the input if the run fails. A URL destination is staged and uploaded only once the batch completes.

Source code in vllm/entrypoints/launchers/run_batch.py

@contextlib.asynccontextmanager
async def batch_output_writer(
    path_or_url: str, output_tmp_dir: str | None
) -> AsyncIterator[IO[str]]:
    """Yield a file that receives responses in input order.

    A local file holds the responses for a prefix of the input if the run
    fails. A URL destination is staged and uploaded only once the batch
    completes.
    """
    if not is_url(path_or_url):
        logger.info("Writing outputs to local file %s", path_or_url)
        with open(path_or_url, "w", encoding="utf-8") as f:
            yield f
        return

    with staging_file(output_tmp_dir, "tmp_batch_output_") as f:
        yield f
        f.flush()
        logger.info("Uploading outputs to %s", path_or_url)
        if isinstance(f, StringIO):
            data = f.getvalue().strip().encode("utf-8")
            await upload_data(path_or_url, data, from_file=False)
        else:
            await upload_data(path_or_url, f.name, from_file=True)

build_endpoint_registry(engine_client, args) async

Build the endpoint registry with all serving objects and handler configurations.

Parameters:

• engine_client

  (EngineClient) –

  The engine client

• args

  (Namespace) –

  Command line arguments

Returns:

• dict[str, EndpointConfig] –

  Dictionary mapping endpoint keys to their configurations

Source code in vllm/entrypoints/launchers/run_batch.py

async def build_endpoint_registry(
    engine_client: EngineClient,
    args: Namespace,
) -> dict[str, EndpointConfig]:
    """Build the endpoint registry with all serving objects and handler configurations.

    Args:
        engine_client: The engine client
        args: Command line arguments

    Returns:
        Dictionary mapping endpoint keys to their configurations

    """
    supported_tasks = await engine_client.get_supported_tasks()
    logger.info("Supported tasks: %s", supported_tasks)

    # Create a state object to hold serving objects
    state = State()

    # Initialize all serving objects using init_app_state
    # This provides full functionality including chat template processing,
    # LoRA support, tool servers, etc.
    await init_app_state(engine_client, state, args, supported_tasks)

    # Get serving objects from state (defaulting to None if not set)
    openai_serving_chat = getattr(state, "openai_serving_chat", None)
    openai_serving_transcription = getattr(state, "openai_serving_transcription", None)
    openai_serving_translation = getattr(state, "openai_serving_translation", None)
    serving_embedding = getattr(state, "serving_embedding", None)
    serving_scores = getattr(state, "serving_scores", None)

    allowed_media_domains = getattr(args, "allowed_media_domains", None)

    # Registry of endpoint configurations
    endpoint_registry: dict[str, EndpointConfig] = {
        "completions": {
            "url": "/v1/chat/completions",
            "handler_getter": lambda: (
                openai_serving_chat.create_chat_completion
                if openai_serving_chat is not None
                else None
            ),
            "wrapper_fn": None,
        },
        "embeddings": {
            "url": "/v1/embeddings",
            "handler_getter": lambda: (
                serving_embedding if serving_embedding is not None else None
            ),
            "wrapper_fn": None,
        },
        "score": {
            "url": "/score",
            "handler_getter": lambda: (
                serving_scores if serving_scores is not None else None
            ),
            "wrapper_fn": None,
        },
        "rerank": {
            "url": "/rerank",
            "handler_getter": lambda: (
                serving_scores if serving_scores is not None else None
            ),
            "wrapper_fn": None,
        },
        "transcriptions": {
            "url": "/v1/audio/transcriptions",
            "handler_getter": lambda: (
                openai_serving_transcription.create_transcription
                if openai_serving_transcription is not None
                else None
            ),
            "wrapper_fn": make_transcription_wrapper(
                is_translation=False,
                allowed_media_domains=allowed_media_domains,
            ),
        },
        "translations": {
            "url": "/v1/audio/translations",
            "handler_getter": lambda: (
                openai_serving_translation.create_translation
                if openai_serving_translation is not None
                else None
            ),
            "wrapper_fn": make_transcription_wrapper(
                is_translation=True,
                allowed_media_domains=allowed_media_domains,
            ),
        },
    }

    return endpoint_registry

dispatch_batch(input_file, output_file, endpoint_registry, tracker, max_inflight, window=None, engine_client=None) async

Run every request with at most max_inflight alive at once.

Responses are written in input order, with at most window requests, max_inflight by default, dispatched but not yet written. The first request to raise, rather than return an error response, cancels the rest, as does an error response once engine_client is dead.

Source code in vllm/entrypoints/launchers/run_batch.py

async def dispatch_batch(
    input_file: IO[str],
    output_file: IO[str],
    endpoint_registry: dict[str, EndpointConfig],
    tracker: BatchProgressTracker,
    max_inflight: int,
    window: int | None = None,
    engine_client: EngineClient | None = None,
) -> None:
    """Run every request with at most `max_inflight` alive at once.

    Responses are written in input order, with at most `window` requests,
    `max_inflight` by default, dispatched but not yet written. The first
    request to raise, rather than return an error response, cancels the rest,
    as does an error response once `engine_client` is dead.
    """
    window = max(window or max_inflight, max_inflight)
    running = asyncio.Semaphore(max_inflight)
    unwritten = asyncio.Semaphore(window)
    inflight: set[asyncio.Task[None]] = set()
    finished: dict[int, str] = {}
    next_index = 0
    failure: BaseException | None = None
    stopping = False

    async def run_and_write(index: int, request_json: str) -> None:
        nonlocal failure, next_index
        try:
            try:
                response = await run_one_request(request_json, endpoint_registry)
            finally:
                running.release()
            # Handlers answer a cancellation with an error response; drop it.
            if stopping:
                return
            # A dead engine answers every request with an error; keep a prefix.
            if (
                response.error is not None
                and engine_client is not None
                and engine_client.errored
            ):
                raise engine_client.dead_error
            finished[index] = response.model_dump_json()
            while next_index in finished:
                print(finished.pop(next_index), file=output_file)
                next_index += 1
                tracker.completed()
                unwritten.release()
            output_file.flush()
        except BaseException as exc:
            if failure is None:
                failure = exc
            # The dispatch loop may be waiting for a slot this request holds.
            unwritten.release()
            if isinstance(exc, asyncio.CancelledError):
                raise

    try:
        index = 0
        for request_json in input_file:
            if not request_json.strip():
                continue
            await unwritten.acquire()
            await running.acquire()
            if failure is not None:
                break
            task = asyncio.create_task(run_and_write(index, request_json))
            index += 1
            inflight.add(task)
            task.add_done_callback(inflight.discard)
        else:
            # Reacquiring every slot waits for the last response to be written.
            for _ in range(window):
                await unwritten.acquire()
                if failure is not None:
                    break

        if isinstance(failure, asyncio.CancelledError):
            raise RuntimeError(
                "A batch request was cancelled before it completed"
            ) from failure
        if failure is not None:
            raise failure
    finally:
        stopping = True
        for task in inflight:
            task.cancel()
        if inflight:
            await asyncio.gather(*inflight, return_exceptions=True)

download_bytes_from_url(url, allowed_media_domains=None) async

Download data from a URL or decode from a data URL.

Parameters:

• url

  (str) –

  Either an HTTP/HTTPS URL or a data URL (data:...;base64,...)

• allowed_media_domains

  (list[str] | None, default: None ) –

  If set, only HTTP/HTTPS URLs whose hostname is in this list are permitted. data: URLs are not subject to this restriction.

Returns:

• bytes –

  Data as bytes

Source code in vllm/entrypoints/launchers/run_batch.py

async def download_bytes_from_url(
    url: str,
    allowed_media_domains: list[str] | None = None,
) -> bytes:
    """Download data from a URL or decode from a data URL.

    Args:
        url: Either an HTTP/HTTPS URL or a data URL (data:...;base64,...)
        allowed_media_domains: If set, only HTTP/HTTPS URLs whose hostname
            is in this list are permitted. data: URLs are not subject to
            this restriction.

    Returns:
        Data as bytes

    """
    parsed = urlparse(url)

    # Handle data URLs (base64 encoded) - not subject to domain restrictions
    if parsed.scheme == "data":
        # Format: data:...;base64,<base64_data>
        if "," in url:
            header, data = url.split(",", 1)
            if "base64" in header:
                decoded_size_bytes = len(data.rstrip("=")) * 3 // 4
                if decoded_size_bytes > _get_audio_download_limit_bytes():
                    _raise_audio_filesize_limit(decoded_size_bytes)
                return base64.b64decode(data)
            else:
                raise VLLMValidationError(
                    f"Unsupported data URL encoding: {header}",
                    parameter="url",
                )
        else:
            raise VLLMValidationError(
                f"Invalid data URL format: {url}",
                parameter="url",
            )

    # Handle HTTP/HTTPS URLs
    elif parsed.scheme in ("http", "https"):
        if allowed_media_domains is not None:
            url_spec = parse_url(url)
            if url_spec.hostname not in allowed_media_domains:
                raise VLLMValidationError(
                    f"The URL must be from one of the allowed domains: "
                    f"{allowed_media_domains}. Input URL domain: "
                    f"{url_spec.hostname}",
                    parameter="url",
                    value=url_spec.hostname,
                )
            # Use the normalized URL to prevent parsing discrepancies
            # between urllib3 and aiohttp (e.g. backslash-@ attacks).
            url = url_spec.url

        try:
            return await global_http_connection.async_get_bytes(
                url,
                allow_redirects=envs.VLLM_MEDIA_URL_ALLOW_REDIRECTS,
                max_bytes=_get_audio_download_limit_bytes(),
            )
        except HTTPResponseSizeExceededError as exc:
            _raise_audio_filesize_limit(exc.received_bytes)

    else:
        raise VLLMValidationError(
            f"Unsupported URL scheme: {parsed.scheme}. "
            "Supported schemes: http, https, data",
            parameter="url",
            value=parsed.scheme,
        )

handle_endpoint_request(request, config)

Generic handler for endpoint requests.

Parameters:

• request

  (BatchRequestInput) –

  The batch request input

• config

  (EndpointConfig) –

  The endpoint's URL matching rule and handler

Returns:

• Awaitable[BatchRequestOutput] | None –

  Awaitable[BatchRequestOutput] if the request was handled,

• Awaitable[BatchRequestOutput] | None –

  None if URL didn't match

Source code in vllm/entrypoints/launchers/run_batch.py

def handle_endpoint_request(
    request: BatchRequestInput,
    config: EndpointConfig,
) -> Awaitable[BatchRequestOutput] | None:
    """Generic handler for endpoint requests.

    Args:
        request: The batch request input
        config: The endpoint's URL matching rule and handler

    Returns:
        Awaitable[BatchRequestOutput] if the request was handled,
        None if URL didn't match

    """
    if not url_matches(config["url"], request.url):
        return None

    handler_fn = config["handler_getter"]()
    if handler_fn is None:
        error_msg = f"Model does not support endpoint: {request.url}"
        return make_async_error_request_output(request, error_msg=error_msg)

    # Apply wrapper if provided (e.g., for transcriptions/translations)
    wrapper_fn = config["wrapper_fn"]
    if wrapper_fn is not None:
        handler_fn = wrapper_fn(handler_fn)

    return run_request(handler_fn, request)

ignore_admission_limits(args)

Clear the engine limits that reject requests with a 503.

They let an online client retry on another instance or meet a TTFT target; run-batch could only write each rejected request out as an error.

Source code in vllm/entrypoints/launchers/run_batch.py

def ignore_admission_limits(args: Namespace) -> None:
    """Clear the engine limits that reject requests with a 503.

    They let an online client retry on another instance or meet a TTFT target;
    run-batch could only write each rejected request out as an error.
    """
    flags = [
        flag
        for flag, value in (
            ("--max-num-queued-reqs", args.max_num_queued_reqs),
            ("--max-num-queued-tokens", args.max_num_queued_tokens),
        )
        if value is not None
    ]
    if flags:
        logger.warning(
            "Ignoring %s: run-batch does not reject requests. "
            "Use --max-inflight to bound how many run at once.",
            " and ".join(flags),
        )
    args.max_num_queued_reqs = args.max_num_queued_tokens = None

is_same_local_file(input_file, output_file)

Whether both paths name the same regular file; only those can be truncated.

Source code in vllm/entrypoints/launchers/run_batch.py

def is_same_local_file(input_file: str, output_file: str) -> bool:
    """Whether both paths name the same regular file; only those can be truncated."""
    if is_url(input_file) or is_url(output_file):
        return False
    try:
        input_stat = os.stat(input_file)
        output_stat = os.stat(output_file)
    except OSError:
        return False
    if not (stat.S_ISREG(input_stat.st_mode) and stat.S_ISREG(output_stat.st_mode)):
        return False
    return os.path.samestat(input_stat, output_stat)

make_transcription_wrapper(is_translation, allowed_media_domains=None)

Factory function to create a wrapper for transcription/translation handlers. The wrapper converts BatchTranscriptionRequest or BatchTranslationRequest to TranscriptionRequest or TranslationRequest and calls the appropriate handler.

Parameters:

• is_translation

  (bool) –

  If True, process as translation; otherwise process as transcription

• allowed_media_domains

  (list[str] | None, default: None ) –

  If set, only URLs from these domains are permitted for HTTP/HTTPS fetches.

Returns:

• WrapperFn –

  A function that takes a handler and returns a wrapped handler

Source code in vllm/entrypoints/launchers/run_batch.py

def make_transcription_wrapper(
    is_translation: bool,
    allowed_media_domains: list[str] | None = None,
) -> WrapperFn:
    """Factory function to create a wrapper for transcription/translation handlers.
    The wrapper converts BatchTranscriptionRequest or BatchTranslationRequest
    to TranscriptionRequest or TranslationRequest and calls the appropriate handler.

    Args:
        is_translation: If True, process as translation; otherwise process
            as transcription
        allowed_media_domains: If set, only URLs from these domains are
            permitted for HTTP/HTTPS fetches.

    Returns:
        A function that takes a handler and returns a wrapped handler

    """

    def wrapper(handler_fn: Callable):
        async def transcription_wrapper(
            batch_request_body: (BatchTranscriptionRequest | BatchTranslationRequest),
        ) -> (
            TranscriptionResponseVariant
            | TranslationResponse
            | TranslationResponseVerbose
            | ErrorResponse
        ):
            try:
                # Download data from URL
                audio_data = await download_bytes_from_url(
                    batch_request_body.file_url,
                    allowed_media_domains=allowed_media_domains,
                )

                # Create a mock file from the downloaded audio data
                mock_file = UploadFile(
                    file=BytesIO(audio_data),
                    filename="audio.bin",
                )

                # Convert batch request to regular request
                # by copying all fields except file_url and setting file to mock_file
                request_dict = batch_request_body.model_dump(exclude={"file_url"})
                request_dict["file"] = mock_file

                if is_translation:
                    # Create TranslationRequest from BatchTranslationRequest
                    translation_request = TranslationRequest.model_validate(
                        request_dict
                    )
                    return await handler_fn(audio_data, translation_request)
                else:
                    # Create TranscriptionRequest from BatchTranscriptionRequest
                    transcription_request = TranscriptionRequest.model_validate(
                        request_dict
                    )
                    return await handler_fn(audio_data, transcription_request)
            except Exception as e:
                operation = "translation" if is_translation else "transcription"
                return ErrorResponse(
                    error=ErrorInfo(
                        message=f"Failed to process {operation}: {str(e)}",
                        type="BadRequestError",
                        code=HTTPStatus.BAD_REQUEST.value,
                    )
                )

        return transcription_wrapper

    return wrapper

open_batch_input(path_or_url, tmp_dir) async

Yield the batch as a seekable text file, so it can be read twice.

A non-seekable source is staged via staging_file. A URL is downloaded to a file, since the in-memory fetch is capped at the media download limit.

Source code in vllm/entrypoints/launchers/run_batch.py

@contextlib.asynccontextmanager
async def open_batch_input(
    path_or_url: str, tmp_dir: str | None
) -> AsyncIterator[IO[str]]:
    """Yield the batch as a seekable text file, so it can be read twice.

    A non-seekable source is staged via `staging_file`. A URL is downloaded to
    a file, since the in-memory fetch is capped at the media download limit.
    """
    with contextlib.ExitStack() as stack:
        if is_url(path_or_url):
            download_dir = stack.enter_context(
                tempfile.TemporaryDirectory(dir=tmp_dir, prefix="tmp_batch_input_")
            )
            download_path = Path(download_dir, "input.jsonl")
            logger.info("Downloading %s to %s", path_or_url, download_path)
            await global_http_connection.async_download_file(
                path_or_url,
                download_path,
                timeout=_INPUT_DOWNLOAD_TIMEOUT_S,
                chunk_size=_STAGING_CHUNK_SIZE,
            )
            yield stack.enter_context(
                download_path.open(encoding="utf-8", newline="\n")
            )
            return

        # JSON allows a bare "\r" as whitespace, so split lines on "\n" only.
        source = stack.enter_context(open(path_or_url, encoding="utf-8", newline="\n"))
        if source.seekable():
            yield source
            return

        logger.info("Staging %s so it can be read twice", path_or_url)
        staged = stack.enter_context(staging_file(tmp_dir, "tmp_batch_input_"))
        shutil.copyfileobj(source, staged, _STAGING_CHUNK_SIZE)
        staged.seek(0)
        yield staged

run_one_request(request_json, endpoint_registry) async

Route a single line of the batch to its endpoint handler.

Source code in vllm/entrypoints/launchers/run_batch.py

async def run_one_request(
    request_json: str,
    endpoint_registry: dict[str, EndpointConfig],
) -> BatchRequestOutput:
    """Route a single line of the batch to its endpoint handler."""
    request = BatchRequestInput.model_validate_json(request_json)

    endpoint_key = request.url.split("/")[-1]

    result = None
    if endpoint_key in endpoint_registry:
        result = handle_endpoint_request(request, endpoint_registry[endpoint_key])

    if result is None:
        result = make_async_error_request_output(
            request,
            error_msg=f"URL {request.url} is not a supported endpoint. "
            "Supported endpoints: "
            + ", ".join(config["url"] for config in endpoint_registry.values())
            + ".",
        )

    return await result

staging_file(tmp_dir, prefix)

In-memory buffer, or a temporary file when a directory is given.

Source code in vllm/entrypoints/launchers/run_batch.py

def staging_file(tmp_dir: str | None, prefix: str) -> IO[str]:
    """In-memory buffer, or a temporary file when a directory is given."""
    if tmp_dir is None:
        return StringIO()
    return tempfile.NamedTemporaryFile(
        mode="w+",
        encoding="utf-8",
        newline="\n",
        dir=tmp_dir,
        prefix=prefix,
        suffix=".jsonl",
    )

upload_data(output_url, data_or_file, from_file) async

Upload a local file to a URL. output_url: The URL to upload the file to. data_or_file: Either the data to upload or the path to the file to upload. from_file: If True, data_or_file is the path to the file to upload.

Source code in vllm/entrypoints/launchers/run_batch.py

async def upload_data(
    output_url: str, data_or_file: str | bytes, from_file: bool
) -> None:
    """Upload a local file to a URL.
    output_url: The URL to upload the file to.
    data_or_file: Either the data to upload or the path to the file to upload.
    from_file: If True, data_or_file is the path to the file to upload.
    """
    # Timeout is a common issue when uploading large files.
    # We retry max_retries times before giving up.
    max_retries = 5
    # Number of seconds to wait before retrying.
    delay = 5

    for attempt in range(1, max_retries + 1):
        try:
            # We increase the timeout to 1000 seconds to allow
            # for large files (default is 300).
            async with aiohttp.ClientSession(
                timeout=aiohttp.ClientTimeout(total=1000)
            ) as session:
                if from_file:
                    with open(data_or_file, "rb") as file:
                        async with session.put(output_url, data=file) as response:
                            if not 200 <= response.status < 300:
                                error_text = await response.text()
                                raise Exception(
                                    f"Failed to upload file.\n"
                                    f"Status: {response.status}\n"
                                    f"Response: {error_text}"
                                )
                else:
                    async with session.put(output_url, data=data_or_file) as response:
                        if not 200 <= response.status < 300:
                            error_text = await response.text()
                            raise Exception(
                                f"Failed to upload data.\n"
                                f"Status: {response.status}\n"
                                f"Response: {error_text}"
                            )
            # Upload succeeded; do not retry.
            return
        except Exception as e:
            if attempt < max_retries:
                logger.error(
                    "Failed to upload data (attempt %d). Error message: %s.\nRetrying in %d seconds...",  # noqa: E501
                    attempt,
                    e,
                    delay,
                )
                await asyncio.sleep(delay)
            else:
                raise Exception(
                    f"Failed to upload data (attempt {attempt}). Error message: {str(e)}."  # noqa: E501
                ) from e

url_matches(endpoint_url, url)

Unversioned endpoints (/score, /rerank) match by suffix; others exactly.

Source code in vllm/entrypoints/launchers/run_batch.py

def url_matches(endpoint_url: str, url: str) -> bool:
    """Unversioned endpoints (/score, /rerank) match by suffix; others exactly."""
    if endpoint_url.startswith("/v1/"):
        return url == endpoint_url
    return url.endswith(endpoint_url)

validate_batch(input_file)

Parse every request and return the count; requests are re-parsed when run.

Source code in vllm/entrypoints/launchers/run_batch.py

def validate_batch(input_file: IO[str]) -> int:
    """Parse every request and return the count; requests are re-parsed when run."""
    num_requests = 0
    for line_number, request_json in enumerate(input_file, start=1):
        if not request_json.strip():
            continue
        try:
            BatchRequestInput.model_validate_json(request_json)
        except pydantic.ValidationError as e:
            raise ValueError(f"Invalid request on line {line_number}: {e}") from e
        num_requests += 1
    return num_requests