HttpRequestProcessorConfig

class ray.data.llm.HttpRequestProcessorConfig(*, batch_size: int = 64, accelerator_type: str | None = None, concurrency: int | Tuple[int, int] = 1, url: str, headers: Dict[str, Any] | None = None, qps: int | None = None, max_retries: int = 0, base_retry_wait_time_in_s: float = 1, session_factory: Any | None = None, http_request_stage: Any = True)

The configuration for the HTTP request processor.

Parameters:

• batch_size (int) – The batch size to send to the HTTP request.

• url (str) – The URL to send the HTTP request to.

• headers (Dict[str, Any] | None) – The headers to send with the HTTP request.

• concurrency (int | Tuple[int, int]) – The number of concurrent requests to send. Default to 1. If concurrency is an int n, a fixed pool of n workers is used. If concurrency is a tuple (m, n), autoscaling strategy is used (1 <= m <= n).

Examples

import ray
from ray.data.llm import HttpRequestProcessorConfig, build_processor

config = HttpRequestProcessorConfig(
    url="https://api.openai.com/v1/chat/completions",
    headers={"Authorization": "Bearer sk-..."},
    concurrency=1,
)
processor = build_processor(
    config,
    preprocess=lambda row: dict(
        payload=dict(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are a calculator"},
                {"role": "user", "content": f"{row['id']} ** 3 = ?"},
            ],
            temperature=0.3,
            max_tokens=20,
        ),
    ),
    postprocess=lambda row: dict(
        resp=row["http_response"]["choices"][0]["message"]["content"],
    ),
)

ds = ray.data.range(10)
ds = processor(ds)
for row in ds.take_all():
    print(row)

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'protected_namespaces': (), 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].