sgl-project--sglang/python/sglang/benchmark/datasets/autobench.py

import json
from argparse import Namespace
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple

import numpy as np
from transformers import PreTrainedTokenizerBase

from sglang.benchmark.datasets.common import BaseDataset, DatasetRow

AUTOBENCH_RESERVED_FIELDS = {
    "prompt",
    "messages",
    "prompt_origin",
    "output_len",
    "max_tokens",
    "max_completion_tokens",
    "completion_tokens",
    "prompt_len",
    "text_prompt_len",
    "vision_prompt_len",
    "image_data",
    "timestamp",
    "routing_key",
    "metadata",
    "extra_request_body",
    "param_send",
}


def _load_json_if_needed(value: Any) -> Any:
    if not isinstance(value, str):
        return value
    value = value.strip()
    if not value:
        return value
    if value[0] not in "[{":
        return value
    try:
        return json.loads(value)
    except json.JSONDecodeError:
        return value


def _normalize_messages(messages: Any) -> Optional[List[Dict[str, Any]]]:
    messages = _load_json_if_needed(messages)
    if not isinstance(messages, list) or not messages:
        return None
    if not all(isinstance(message, dict) for message in messages):
        return None

    normalized = []
    for message in messages:
        if "role" not in message:
            return None
        content = message.get("content")
        if content is None:
            return None
        normalized.append({"role": message["role"], "content": content})
    return normalized


def _normalize_legacy_system_content(
    system_prompt: Any, content_list: Any
) -> Optional[List[Dict[str, Any]]]:
    if not isinstance(content_list, list) or not content_list:
        return None

    messages: List[Dict[str, Any]] = []
    if system_prompt:
        messages.append({"role": "system", "content": str(system_prompt)})

    turns = [str(item) for item in content_list]
    # In the old auto_benchmark helpers, an even number of items usually means the
    # last assistant reply is present and should be removed before benchmarking.
    if len(turns) % 2 == 0:
        turns = turns[:-1]
    if not turns:
        return None

    for index, turn in enumerate(turns):
        role = "user" if index % 2 == 0 else "assistant"
        messages.append({"role": role, "content": turn})
    return messages


def _normalize_prompt(row: Dict[str, Any]) -> Tuple[Any, str]:
    prompt = row.get("prompt")
    messages = row.get("messages")
    prompt_origin = row.get("prompt_origin")

    if messages is not None:
        normalized = _normalize_messages(messages)
        if normalized is not None:
            return normalized, "messages"

    if prompt is not None:
        prompt = _load_json_if_needed(prompt)
        if isinstance(prompt, list) and prompt and isinstance(prompt[0], dict):
            normalized = _normalize_messages(prompt)
            if normalized is not None:
                return normalized, "messages"
        if (
            isinstance(prompt, list)
            and prompt
            and all(isinstance(item, str) for item in prompt)
        ):
            return prompt, "multi_turn"
        if (
            isinstance(prompt, list)
            and prompt
            and all(
                isinstance(item, list)
                and item
                and all(
                    isinstance(m, dict) and "role" in m and "content" in m for m in item
                )
                for item in prompt
            )
        ):
            # Multi-turn with N messages per round (e.g. tool observations).
            return prompt, "multi_turn"
        if (
            isinstance(prompt, list)
            and prompt
            and all(isinstance(item, int) for item in prompt)
        ):
            return prompt, "token_ids"
        if isinstance(prompt, str) and prompt:
            return prompt, "prompt"

    if prompt_origin is not None:
        normalized = _normalize_messages(prompt_origin)
        if normalized is not None:
            return normalized, "messages"

    if "system" in row and "content" in row:
        normalized = _normalize_legacy_system_content(
            row.get("system"), row.get("content")
        )
        if normalized is not None:
            return normalized, "messages"

    raise ValueError("Unsupported auto benchmark row: missing prompt/messages")


def _estimate_prompt_lens(
    prompt: Any,
    prompt_kind: str,
    tokenizer: PreTrainedTokenizerBase,
    row: Dict[str, Any],
) -> Tuple[int, int, int]:
    if row.get("prompt_len") is not None:
        prompt_len = int(row["prompt_len"])
        text_prompt_len = int(row.get("text_prompt_len", prompt_len))
        vision_prompt_len = int(row.get("vision_prompt_len", 0))
        return prompt_len, text_prompt_len, vision_prompt_len

    if prompt_kind == "messages":
        text_prompt_len = len(
            tokenizer.apply_chat_template(
                prompt, tokenize=True, add_generation_prompt=True
            )
        )
        vision_prompt_len = 0
        return text_prompt_len, text_prompt_len, vision_prompt_len

    if prompt_kind == "prompt":
        prompt_len = len(tokenizer.encode(prompt, add_special_tokens=False))
        return prompt_len, prompt_len, 0

    if prompt_kind == "token_ids":
        prompt_len = len(prompt)
        return prompt_len, prompt_len, 0

    # Multi-turn prompt lists are handled specially by the serving benchmark and do not
    # contribute reliable static prompt lengths.
    return 0, 0, 0


def _collect_extra_request_body(row: Dict[str, Any]) -> Dict[str, Any]:
    extra: Dict[str, Any] = {}

    param_send = row.get("param_send")
    if param_send is not None:
        parsed = _load_json_if_needed(param_send)
        if isinstance(parsed, dict):
            extra.update(parsed)

    for key, value in row.items():
        if key not in AUTOBENCH_RESERVED_FIELDS:
            extra[key] = value

    explicit_extra = row.get("extra_request_body")
    explicit_extra = _load_json_if_needed(explicit_extra)
    if isinstance(explicit_extra, dict):
        extra.update(explicit_extra)

    return extra


def serialize_dataset_row_to_autobench(
    row: DatasetRow, metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    record: Dict[str, Any] = {
        "prompt": row.prompt,
        "output_len": row.output_len,
    }
    if row.prompt_len:
        record["prompt_len"] = row.prompt_len
    if row.text_prompt_len not in (None, row.prompt_len):
        record["text_prompt_len"] = row.text_prompt_len
    if row.vision_prompt_len:
        record["vision_prompt_len"] = row.vision_prompt_len
    if row.image_data:
        record["image_data"] = row.image_data
    if row.timestamp is not None:
        record["timestamp"] = row.timestamp
    if row.routing_key is not None:
        record["routing_key"] = row.routing_key
    if row.extra_request_body:
        record["extra_request_body"] = row.extra_request_body
    if metadata:
        record["metadata"] = metadata
    return record


@dataclass
class AutoBenchmarkDataset(BaseDataset):
    dataset_path: str
    num_requests: int
    fixed_output_len: Optional[int]

    @classmethod
    def from_args(cls, args: Namespace) -> "AutoBenchmarkDataset":
        return cls(
            dataset_path=args.dataset_path,
            num_requests=args.num_prompts,
            fixed_output_len=args.sharegpt_output_len,
        )

    def load(
        self, tokenizer: PreTrainedTokenizerBase, model_id=None
    ) -> List[DatasetRow]:
        return sample_autobench_requests(
            dataset_path=self.dataset_path,
            num_requests=self.num_requests,
            tokenizer=tokenizer,
            fixed_output_len=self.fixed_output_len,
        )


def sample_autobench_requests(
    dataset_path: str,
    num_requests: int,
    tokenizer: PreTrainedTokenizerBase,
    fixed_output_len: Optional[int] = None,
) -> List[DatasetRow]:
    dataset: List[DatasetRow] = []

    with open(dataset_path, "r", encoding="utf-8") as f:
        for line in f:
            if num_requests > 0 and len(dataset) >= num_requests:
                break

            line = line.strip()
            if not line:
                continue

            row = json.loads(line)
            prompt, prompt_kind = _normalize_prompt(row)
            prompt_len, text_prompt_len, vision_prompt_len = _estimate_prompt_lens(
                prompt, prompt_kind, tokenizer, row
            )

            output_len = fixed_output_len or row.get("output_len")
            output_len = output_len or row.get("max_tokens")
            output_len = output_len or row.get("max_completion_tokens")
            output_len = output_len or row.get("completion_tokens")
            output_len = int(output_len or 256)

            dataset.append(
                DatasetRow(
                    prompt=prompt,
                    prompt_len=prompt_len,
                    output_len=output_len,
                    text_prompt_len=text_prompt_len,
                    vision_prompt_len=vision_prompt_len,
                    image_data=row.get("image_data"),
                    timestamp=row.get("timestamp"),
                    routing_key=row.get("routing_key"),
                    extra_request_body=_collect_extra_request_body(row),
                )
            )

    print(f"Loaded {len(dataset)} auto benchmark requests")
    print(f"#Input tokens: {np.sum([x.prompt_len for x in dataset])}")
    print(f"#Output tokens: {np.sum([x.output_len for x in dataset])}")
    return dataset