guidellm.schemas.base.request_stats

class GenerativeRequestStats(StandardBaseDict):
    """
    Request statistics for generative AI text generation workloads.

    Captures comprehensive performance metrics for individual generative requests,
    including token counts, timing measurements, and derived performance statistics.
    Provides computed properties for latency analysis, throughput calculations,
    and token generation metrics essential for benchmark evaluation.

    Example:
    ::
        stats = GenerativeRequestStats(
            request_id="req_123",
            info=request_info,
            input_metrics=input_usage,
            output_metrics=output_usage
        )
        throughput = stats.output_tokens_per_second
    """

    type_: Literal["generative_request_stats"] = "generative_request_stats"
    request_id: str = Field(description="Unique identifier for the request")
    response_id: str | None = Field(
        default=None, description="Unique identifier matching vLLM Response ID"
    )
    request_args: str | None = Field(
        default=None, description="Backend arguments used for this request"
    )
    output: str | None = Field(
        default=None, description="Generated text output from the request"
    )
    reasoning_output: str | None = Field(
        default=None,
        description="Reasoning/chain-of-thought text emitted before content",
    )
    tool_calls: list[ToolCall] | None = Field(
        default=None,
        description="Raw tool call payloads from the model response in OpenAI format",
    )
    info: RequestInfo = Field(description="Request metadata and timing information")
    input_metrics: UsageMetrics = Field(
        description="Token usage statistics for the input prompt"
    )
    response_metrics: dict[str, Any] | None = Field(
        default=None,
        description=(
            "Per-request metrics the backend reported alongside usage, passed through "
            "as received. vLLM populates this with its `metrics` object, which carries "
            "server-side timings and speculative-decoding acceptance."
        ),
    )
    output_metrics: UsageMetrics = Field(
        description="Token usage statistics for the generated output"
    )

    # Request stats
    @computed_field  # type: ignore[misc]
    @property
    def request_start_time(self) -> float | None:
        """
        :return: Timestamp when the request started, or None if unavailable
        """
        return (
            self.info.timings.request_start
            if self.info.timings.request_start is not None
            else self.info.timings.resolve_start
        )

    @computed_field  # type: ignore[misc]
    @property
    def request_end_time(self) -> float:
        """
        :return: Timestamp when the request ended, or None if unavailable
        """
        if self.info.timings.resolve_end is None:
            raise ValueError("resolve_end timings should be set but is None.")

        return (
            self.info.timings.request_end
            if self.info.timings.request_end is not None
            else self.info.timings.resolve_end
        )

    @computed_field  # type: ignore[misc]
    @property
    def request_latency(self) -> float | None:
        """
        End-to-end request processing latency in seconds.

        :return: Duration from request start to completion, or None if unavailable
        """
        start = self.info.timings.request_start
        end = self.info.timings.request_end
        if start is None or end is None:
            return None

        return end - start

    @computed_field  # type: ignore[misc]
    @property
    def request_dispatch_delay(self) -> float | None:
        """
        Delay between the scheduled arrival time and the actual dispatch in seconds.

        Non-zero when the scheduler could not issue the request at the time its
        strategy targeted, for example while a concurrency limit is saturated.
        Only describes arrival-schedule delay for strategies that define an
        arrival schedule (constant, poisson, and trace when the dataset supplies
        timestamps); the synchronous, concurrent, and throughput strategies
        target an ASAP start instead. See the metrics guide for full caveats.

        :return: Duration from targeted start to request start in seconds, or
            None if unavailable
        """
        targeted = self.info.timings.targeted_start
        start = self.info.timings.request_start
        if targeted is None or start is None:
            return None

        return start - targeted

    @computed_field  # type: ignore[misc]
    @property
    def request_scheduled_latency(self) -> float | None:
        """
        Request latency measured from the scheduled arrival time in seconds.

        Unlike :attr:`request_latency`, which starts when the request was
        dispatched, this includes any time the request waited for the scheduler
        to reach it. The two are equal when the scheduler keeps up with the
        configured rate and diverge once it falls behind. Carries the same
        strategy caveat as :attr:`request_dispatch_delay`.

        :return: Duration from targeted start to request completion in seconds,
            or None if unavailable
        """
        targeted = self.info.timings.targeted_start
        end = self.info.timings.request_end
        if targeted is None or end is None:
            return None

        return end - targeted

    # General token stats
    @computed_field  # type: ignore[misc]
    @property
    def prompt_tokens(self) -> int | None:
        """
        :return: Number of tokens in the input prompt, or None if unavailable
        """
        return self.input_metrics.total_tokens

    @computed_field  # type: ignore[misc]
    @property
    def cached_tokens(self) -> int | None:
        """
        :return: Number of input tokens served from the prefix cache, or None
        """
        return self.input_metrics.cached_tokens

    @computed_field  # type: ignore[misc]
    @property
    def output_tokens(self) -> int | None:
        """
        :return: Number of tokens in the generated output, or None if unavailable
        """
        # Fallback if we did not get usage metrics from the server
        # NOTE: This assumes each iteration is one token
        if self.output_metrics.total_tokens is None:
            return self.info.timings.token_iterations or None

        return self.output_metrics.total_tokens

    @computed_field  # type: ignore[misc]
    @property
    def total_tokens(self) -> int | None:
        """
        :return: Sum of prompt and output tokens, or None if both unavailable
        """
        input_tokens = self.prompt_tokens
        output_tokens = self.output_tokens

        if input_tokens is None and output_tokens is None:
            return None

        return (input_tokens or 0) + (output_tokens or 0)

    @computed_field  # type: ignore[misc]
    @property
    def time_to_first_token_ms(self) -> float | None:
        """
        :return: Time to first token generation in milliseconds, or None if unavailable
        """
        first_token = self.first_token_iteration
        start = self.info.timings.request_start
        if first_token is None or start is None:
            return None

        return 1000 * (first_token - start)

    @computed_field  # type: ignore[misc]
    @property
    def time_to_last_round_trip_ms(self) -> float | None:
        """
        Time from the last sent packet to the last received token in milliseconds.

        Only populated by the websocket backend, which records send timestamps;
        None for backends that do not record sends.

        :return: Last round-trip latency in milliseconds, or None if unavailable
        """
        last_received = self.info.timings.last_token_iteration
        last_sent = self.info.timings.last_request_sent
        if last_received is None or last_sent is None:
            return None

        return 1000 * (last_received - last_sent)

    @computed_field  # type: ignore[misc]
    @property
    def avg_round_trip_time_ms(self) -> float | None:
        """
        Approximate average round-trip time in milliseconds.

        Computed as the mean of received content-token timestamps minus the mean
        of sent-packet timestamps. This is an approximation that assumes sent
        packets and received tokens align uniformly in time. Only populated by
        the websocket backend; None otherwise.

        :return: Average round-trip time in milliseconds, or None if unavailable
        """
        timings = self.info.timings
        if timings.request_sent_count <= 0 or timings.token_received_count <= 0:
            return None

        mean_sent = timings.request_sent_sum / timings.request_sent_count
        mean_received = timings.token_received_sum / timings.token_received_count
        return 1000 * (mean_received - mean_sent)

    @computed_field  # type: ignore[misc]
    @property
    def time_per_output_token_ms(self) -> float | None:
        """
        Average time per output token in milliseconds including first token.

        :return: Average milliseconds per output token, or None if unavailable
        """
        if (
            (start := self.info.timings.request_start) is None
            or (
                (last_token := self.last_token_iteration or self.request_end_time)
                is None
            )
            or (output_tokens := self.output_tokens) is None
            or output_tokens == 0
        ):
            return None

        return 1000 * (last_token - start) / output_tokens

    @computed_field  # type: ignore[misc]
    @property
    def inter_token_latency_ms(self) -> float | None:
        """
        Average inter-token latency in milliseconds excluding first token.

        :return: Average milliseconds between token generations, or None if unavailable
        """
        first_token = self.first_token_iteration
        last_token = self.last_token_iteration
        output_tokens = self.output_tokens
        if (
            first_token is None
            or last_token is None
            or output_tokens is None
            or output_tokens <= 1
        ):
            return None

        return 1000 * (last_token - first_token) / (output_tokens - 1)

    @computed_field  # type: ignore[misc]
    @property
    def tokens_per_second(self) -> float | None:
        """
        :return: Total tokens per second throughput, or None if unavailable
        """
        if not (latency := self.request_latency) or self.total_tokens is None:
            return None

        return self.total_tokens / latency

    @computed_field  # type: ignore[misc]
    @property
    def output_tokens_per_second(self) -> float | None:
        """
        :return: Output token generation throughput, or None if unavailable
        """
        if not (latency := self.request_latency) or self.output_tokens is None:
            return None

        return self.output_tokens / latency

    @computed_field  # type: ignore[misc]
    @property
    def time_to_first_output_token_ms(self) -> float | None:
        """
        Time to first content (non-reasoning) token in milliseconds.

        When no reasoning tokens are emitted this equals TTFT.

        :return: Milliseconds from request start to first content token,
            or None if unavailable
        """
        first_output = self.first_output_token_iteration
        start = self.info.timings.request_start
        if first_output is None or start is None:
            return None
        return 1000 * (first_output - start)

    @computed_field  # type: ignore[misc]
    @property
    def iter_tokens_per_iteration(self) -> float | None:
        """
        :return: Average tokens per iteration excluding first token, or None if
            unavailable
        """
        if (
            self.output_tokens is None
            or self.output_tokens <= 1
            or self.token_iterations <= 1
        ):
            return None

        return (self.output_tokens - 1.0) / (
            self.token_iterations - 1.0
        )  # subtract 1 for first token from the prompt, assume first iter is 1 token

    @computed_field  # type: ignore[misc]
    @property
    def output_tokens_per_iteration(self) -> float | None:
        """
        :return: Average output tokens per iteration, or None if unavailable
        """
        if self.output_tokens is None or self.token_iterations < 1:
            return None

        return self.output_tokens / self.token_iterations

    @property
    def first_token_iteration(self) -> float | None:
        """
        :return: Timestamp of first token generation, or None if unavailable
        """
        return self.info.timings.first_token_iteration

    @property
    def first_output_token_iteration(self) -> float | None:
        """
        :return: Timestamp of first token generation, or None if unavailable
        """
        return self.info.timings.first_output_token_iteration

    @property
    def last_token_iteration(self) -> float | None:
        """
        :return: Timestamp of last token generation, or None if unavailable
        """
        return self.info.timings.last_token_iteration

    @property
    def token_iterations(self) -> int:
        """
        :return: Total number of token generation iterations
        """
        return self.info.timings.token_iterations

    @property
    def prompt_tokens_timing(self) -> tuple[float, float]:
        """
        :return: Tuple of (timestamp, token_count) for prompt processing
        :raises ValueError: If resolve_end timings are not set
        """
        return (
            (
                self.first_token_iteration
                if self.first_token_iteration is not None
                else self.request_end_time
            ),
            self.prompt_tokens or 0.0,
        )

    @property
    def output_tokens_timings(self) -> list[tuple[float, float]]:
        """
        :return: List of (timestamp, token_count) tuples for output token generations
        :raises ValueError: If resolve_end timings are not set
        """
        if (
            self.first_token_iteration is None
            or self.last_token_iteration is None
            or self.token_iterations <= 1
        ):
            # No iteration data, return single timing at end with all tokens
            return [
                (
                    (
                        self.last_token_iteration
                        if self.last_token_iteration is not None
                        else self.request_end_time
                    ),
                    self.output_tokens or 0.0,
                )
            ]

        # Return first token timing as 1 token plus per-iteration timings
        return [
            (self.first_token_iteration, 1.0 * bool(self.output_tokens))
        ] + self.iter_tokens_timings

    @property
    def iter_tokens_timings(self) -> list[tuple[float, float]]:
        """
        :return: List of (timestamp, token_count) tuples for iterations excluding
            first token
        """
        if (
            self.first_token_iteration is None
            or self.last_token_iteration is None
            or (tok_per_iter := self.iter_tokens_per_iteration) is None
            or self.token_iterations <= 1
        ):
            return []

        # evenly space the iterations since we don't have per-iteration timings
        # / we don't know the individual token counts per iteration
        iter_times = np.linspace(
            self.first_token_iteration,
            self.last_token_iteration,
            num=self.token_iterations,
        )[1:]  # skip first iteration

        return [(iter_time, tok_per_iter) for iter_time in iter_times]

    @property
    def total_tokens_timings(self) -> list[tuple[float, float]]:
        """
        :return: List of (timestamp, token_count) tuples for all token generations
        """
        prompt_timings = self.prompt_tokens_timing
        output_timings = self.output_tokens_timings

        return ([prompt_timings] if prompt_timings else []) + output_timings