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

Code before:
        "If None, the request will be sent sequentially.",
    )
    max_retries: int = Field(
        default=0,
        description="The maximum number of retries per request in the event of failures.",
    )
    base_retry_wait_time_in_s: float = Field(
        default=1,
        description="The base wait time for a retry during exponential backoff.",
    )
    # Since `session_factory` is a callable, we use type Any to avoid pydantic serialization issues
    session_factory: Optional[Any] = Field(
        default=None,
        description="Optional session factory to be used for initializing a client session. Type: Callable[[], ClientSession]",
        # exclude from JSON serialization since `session_factory` is a callable
        exclude=True,
    )
    num_cpus: Optional[float] = Field(
        default=None,
        description="Number of CPUs per HttpRequestUDF worker. Defaults to 1 if None. "
        "For I/O-bound workloads, use fractional values (e.g., 0.1).",
    )
    memory: Optional[float] = Field(
        default=None,
        description="Heap memory in bytes to reserve for each HttpRequestUDF worker.",
    )


def build_http_request_processor(
    config: HttpRequestProcessorConfig,
    preprocess: Optional[UserDefinedFunction] = None,
    postprocess: Optional[UserDefinedFunction] = None,
    preprocess_map_kwargs: Optional[Dict[str, Any]] = None,
    postprocess_map_kwargs: Optional[Dict[str, Any]] = None,
) -> Processor:
    """Construct a Processor and configure stages.

    Args:
        config: The configuration for the processor.
        preprocess: An optional lambda function that takes a row (dict) as input
            and returns a preprocessed row (dict). The output row must contain the
            required fields for the following processing stages.
        postprocess: An optional lambda function that takes a row (dict) as input
            and returns a postprocessed row (dict).
        preprocess_map_kwargs: Optional kwargs to pass to Dataset.map() for the
            preprocess stage (e.g., num_cpus, memory, concurrency).
        postprocess_map_kwargs: Optional kwargs to pass to Dataset.map() for the
            postprocess stage (e.g., num_cpus, memory, concurrency).

    Returns:
        The constructed processor.

Review comment (performance):
The direction makes sense, but let's do this at the stage layer (not the processor layer) to be consistent with how other stages configure `num_cpus` and `memory` although there is only 1 stage in the HttpRequestProcessor. We can take advantage of https://github.com/ray-project/ray/blob/d406677817f270675b02e0499018bd87e12510b4/python/ray/llm/_internal/batch/processor/utils.py#L44 and will not need to invoke `extract_resource_kwargs` directly. You can follow this [example](https://github.com/ray-project/ray/blob/master/python/ray/llm/_internal/batch/stages/configs.py#L50).

Code after:
    qps: Optional[int] = Field(
        default=None,
        description="The maximum number of requests per second to avoid rate limit. "
        "If None, the request will be sent sequentially.",
    )
    max_retries: int = Field(
        default=0,
        description="The maximum number of retries per request in the event of failures.",
    )
    base_retry_wait_time_in_s: float = Field(
        default=1,
        description="The base wait time for a retry during exponential backoff.",
    )
    # Since `session_factory` is a callable, we use type Any to avoid pydantic serialization issues
    session_factory: Optional[Any] = Field(
        default=None,
        description="Optional session factory to be used for initializing a client session. Type: Callable[[], ClientSession]",
        # exclude from JSON serialization since `session_factory` is a callable
        exclude=True,
    )
    http_request_stage: Any = Field(
        default=True,
        description="Http request stage config (bool | dict | HttpRequestStageConfig).",
    )


def build_http_request_processor(
    config: HttpRequestProcessorConfig,
    preprocess: Optional[UserDefinedFunction] = None,
    postprocess: Optional[UserDefinedFunction] = None,
    preprocess_map_kwargs: Optional[Dict[str, Any]] = None,
    postprocess_map_kwargs: Optional[Dict[str, Any]] = None,
) -> Processor:
    """Construct a Processor and configure stages.

    Args:
        config: The configuration for the processor.
        preprocess: An optional lambda function that takes a row (dict) as input
            and returns a preprocessed row (dict). The output row must contain the
            required fields for the following processing stages.
        postprocess: An optional lambda function that takes a row (dict) as input
            and returns a postprocessed row (dict).
        preprocess_map_kwargs: Optional kwargs to pass to Dataset.map() for the
            preprocess stage (e.g., num_cpus, memory, concurrency).
        postprocess_map_kwargs: Optional kwargs to pass to Dataset.map() for the
            postprocess stage (e.g., num_cpus, memory, concurrency).

    Returns:
        The constructed processor.
    """