sgl-project--sglang/python/sglang/multimodal_gen/runtime/utils/request_logger.py

# Copyright 2026 SGLang Team
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================


from typing import Any, Optional

from sglang.srt.environ import envs
from sglang.srt.utils.log_utils import create_log_targets, log_json
from sglang.srt.utils.request_logger import (
    _dataclass_to_string_truncated,
    _transform_data_for_logging,
)

# Core generation knobs logged per record. Prompt text is logged separately,
# gated by the level, so it is excluded here.
_SAMPLING_CONFIG_FIELDS = (
    "data_type",
    "seed",
    "num_inference_steps",
    "guidance_scale",
    "true_cfg_scale",
    "width",
    "height",
    "num_frames",
    "fps",
    "num_outputs_per_prompt",
)

# Level 2 truncates prompt text; level 3 keeps it whole. Lower levels log no
# prompt at all.
_TRUNCATE_LENGTH = 2048
_UNLIMITED = 1 << 30


class DiffusionRequestLogger:
    def __init__(
        self,
        log_requests: bool,
        log_requests_level: int,
        log_requests_format: str,
        log_requests_target: Optional[list],
    ):
        self.log_requests = log_requests
        self.log_requests_level = log_requests_level
        self.log_requests_format = log_requests_format
        self.log_requests_target = log_requests_target
        self.targets = create_log_targets(
            targets=log_requests_target, name_prefix=__name__
        )
        self.log_exceeded_ms = envs.SGLANG_LOG_REQUEST_EXCEEDED_MS.get()
        self._max_length = (
            _TRUNCATE_LENGTH if self.log_requests_level == 2 else _UNLIMITED
        )

    @classmethod
    def from_server_args(cls, server_args: Any) -> "DiffusionRequestLogger":
        """Build a logger from server args."""
        return cls(
            log_requests=server_args.log_requests,
            log_requests_level=server_args.log_requests_level,
            log_requests_format=server_args.log_requests_format,
            log_requests_target=server_args.log_requests_target,
        )

    @staticmethod
    def _request_id(req: Any) -> Optional[str]:
        """The request's id, or ``None`` if absent."""
        return getattr(req, "request_id", None)

    def _config_view(self, req: Any, *, drop_seed: bool = False) -> dict:
        """Sampling config + (level >= 2) prompt, gated by the log level.
        Returns ``{}`` below level 1."""
        sp = getattr(req, "sampling_params", None)
        if sp is None or self.log_requests_level < 1:
            return {}
        cfg = {name: getattr(sp, name, None) for name in _SAMPLING_CONFIG_FIELDS}
        if drop_seed:
            cfg.pop("seed", None)
        view: dict = {"sampling_params": cfg}
        if self.log_requests_level >= 2:
            view["prompt"] = getattr(sp, "prompt", None)
            view["negative_prompt"] = getattr(sp, "negative_prompt", None)
        return view

    def _result_view(self, result: Any) -> dict:
        """Result-side fields for a finished record: latency and error."""
        e2e_latency = 0.0
        metrics = getattr(result, "metrics", None) if result is not None else None
        if metrics is not None:
            e2e_latency = getattr(metrics, "total_duration_s", 0.0) or 0.0
        return {
            "meta_info": {"e2e_latency": e2e_latency},
            "error": getattr(result, "error", None) if result is not None else None,
        }

    def _emit(self, msg: str) -> None:
        for target in self.targets:
            target.info(msg)

    def _per_request_view(self, req: Any) -> dict:
        """Per-output identity within a batch: ``request_id`` plus ``seed`` at
        level >= 1."""
        sp = getattr(req, "sampling_params", None)
        view: dict = {"request_id": self._request_id(req)}
        if self.log_requests_level >= 1:
            view["seed"] = getattr(sp, "seed", None) if sp is not None else None
        return view

    def _batch_record(self, reqs: list) -> tuple:
        """Build the ``rid`` / ``obj`` for one forward call"""
        rids = [self._request_id(req) or "unknown" for req in reqs]
        if len(reqs) == 1:
            # Single request: scalar rid + flat dict obj (id + config).
            req = reqs[0]
            obj = {"request_id": self._request_id(req), **self._config_view(req)}
            return rids[0], obj

        shared_views = [self._config_view(req, drop_seed=True) for req in reqs]
        if all(view == shared_views[0] for view in shared_views):
            obj = {
                **shared_views[0],
                "outputs": [self._per_request_view(req) for req in reqs],
            }
        else:
            # Configs genuinely differ: list each request's full payload verbatim.
            obj = [
                {"request_id": self._request_id(req), **self._config_view(req)}
                for req in reqs
            ]
        return rids, obj

    def _loggable(self, req: Any) -> bool:
        """Whether ``req`` should be recorded: logging is on, it's a real
        generation request (control messages -- LoRA / weight / stats / shutdown
        -- have no ``sampling_params`` and are skipped), and it's not a warmup."""
        return (
            self.log_requests
            and getattr(req, "sampling_params", None) is not None
            and not getattr(req, "is_warmup", False)
        )

    def _logged_reqs(self, batch: Any) -> list:
        """Normalize ``batch`` to a list and drop control / warmup requests."""
        reqs = batch if isinstance(batch, (list, tuple)) else [batch]
        return [r for r in reqs if self._loggable(r)]

    def log_received_request(self, batch: Any) -> None:
        reqs = self._logged_reqs(batch)
        if not reqs:
            return

        rid, obj = self._batch_record(reqs)
        max_length = self._max_length

        if self.log_requests_format == "json":
            log_json(
                self.targets,
                "request.received",
                {"rid": rid, "obj": _transform_data_for_logging(obj, max_length)},
            )
        else:
            self._emit(
                f"Receive: obj={_dataclass_to_string_truncated(obj, max_length)}"
            )

    def log_finished_request(self, batch: Any, result: Any) -> None:
        reqs = self._logged_reqs(batch)
        if not reqs:
            return

        out = self._result_view(result)
        e2e_latency_ms = out["meta_info"]["e2e_latency"] * 1000
        if self.log_exceeded_ms > 0 and e2e_latency_ms < self.log_exceeded_ms:
            return

        rid, obj = self._batch_record(reqs)
        max_length = self._max_length

        if self.log_requests_format == "json":
            log_json(
                self.targets,
                "request.finished",
                {
                    "rid": rid,
                    "obj": _transform_data_for_logging(obj, max_length),
                    "out": _transform_data_for_logging(out, max_length),
                },
            )
        else:
            self._emit(
                f"Finish: obj={_dataclass_to_string_truncated(obj, max_length)}"
                f", out={_dataclass_to_string_truncated(out, max_length)}"
            )