add elapsed time metrics in api serve framework

Author: SdeeRK

llm/benchmark/rl/api_serve.py

@@ -142,20 +142,6 @@ class ResponsePayload:
 
 
 class StatisticsManager:
-    def __init__(self, responses_num: int):
-        self.responses_num = responses_num
-        self.batch_index = 0
-
-    def res_stats(self, response: List[ResponsePayload]):
-        batch_group_pd = pd.DataFrame(response)
-        res_batch_pd = batch_group_pd[["idx", "question", "responses"]]
-        responses_batch_pd = pd.DataFrame(
-            res_batch_pd["responses"].to_list(), columns=[f"response_{i+1}" for i in range(self.responses_num)]
-        )
-        res_batch_pd = pd.concat([res_batch_pd[["idx", "question"]], responses_batch_pd], axis=1)
-
-        res_batch_pd.to_json(self.res_path, orient="records", lines=True, force_ascii=False, mode="a")
-
     def dispersed_stats(self, responses: List[ResponsePayload], batch_elapsed_time: float):
         batch_group_pd = pd.DataFrame(responses)
 
@@ -168,6 +154,10 @@ def dispersed_stats(self, responses: List[ResponsePayload], batch_elapsed_time:
             "completion_time": batch_elapsed_time,
             "throughput_tokens_per_sec": batch_group_pd["token_lengths"].apply((lambda x: sum(x))).sum()
             / batch_elapsed_time,
+            "elapsed_times": batch_group_pd["elapsed_times"].to_list(),
+            "min_time": batch_group_pd["elapsed_times"].apply(lambda x: min(x)).tolist(),
+            "max_time": batch_group_pd["elapsed_times"].apply(lambda x: max(x)).tolist(),
+            "avg_time": batch_group_pd["elapsed_times"].apply(lambda x: sum(x) / len(x)).tolist(),
         }
 
         return dispersed_stats_dict
@@ -186,6 +176,8 @@ def global_stats(self, responses: List[ResponsePayload], batch_elapsed_time: flo
         global_stats_dict["avg_response_tokens"] = total_response_tokens / len(responses)
         global_stats_dict["total_response_tokens"] = total_response_tokens
         global_stats_dict["group_max_response_tokens"] = dispersed_stats_dict["max_length"]
+        global_stats_dict["min_time"] = min(dispersed_stats_dict["min_time"])
+        global_stats_dict["avg_time"] = sum(dispersed_stats_dict["avg_time"]) / len(responses)
         global_stats_dict["completion_time"] = dispersed_stats_dict["completion_time"]
         global_stats_dict["throughput_tokens_per_sec"] = dispersed_stats_dict["throughput_tokens_per_sec"]
 
@@ -211,7 +203,7 @@ def __init__(self, args, max_concurrency: int = 1000):
         self.rollout_details_path = self.output_dir / "rollout_details.jsonl"
         self.status_file_path = self.output_dir / "status.txt"
 
-        self.stats_manager = StatisticsManager(args.rollout_output_num)
+        self.stats_manager = StatisticsManager()
 
         self._load_status()
 
@@ -282,8 +274,8 @@ async def call(self, request: RequestPayload) -> Tuple[str, float]:
                 text = "".join(chunks)
                 end_time = time.perf_counter()
                 elapsed_time = end_time - start_time
-                logger.debug("Streaming response took %.4f seconds", elapsed_time)
-                return text, elapsed_time
+                logger.debug("Streaming response took %.2f seconds", elapsed_time)
+                return text, round(elapsed_time, 2)
 
         except Exception as e:
             logger.error("Error while streaming: %s", e)
@@ -333,6 +325,8 @@ def execute(self):
                         "avg_response_tokens",
                         "total_response_tokens",
                         "group_max_response_tokens",
+                        "min_time",
+                        "avg_time",
                         "completion_time",
                         "throughput_tokens_per_sec",
                     ]
@@ -346,6 +340,10 @@ def execute(self):
                         "avg_length",
                         "completion_time",
                         "throughput_tokens_per_sec",
+                        "elapsed_times",
+                        "min_time",
+                        "max_time",
+                        "avg_time",
                     ]
                 )
 
@@ -374,6 +372,8 @@ def execute(self):
                         round(global_stats_dict["avg_response_tokens"], 2),
                         global_stats_dict["total_response_tokens"],
                         global_stats_dict["group_max_response_tokens"],
+                        global_stats_dict["min_time"],
+                        global_stats_dict["avg_time"],
                         round(global_stats_dict["completion_time"], 2),
                         round(global_stats_dict["throughput_tokens_per_sec"], 2),
                     ]
@@ -388,6 +388,10 @@ def execute(self):
                         dispersed_stats_dict["avg_length"],
                         round(dispersed_stats_dict["completion_time"], 2),
                         round(dispersed_stats_dict["throughput_tokens_per_sec"], 2),
+                        dispersed_stats_dict["elapsed_times"],
+                        dispersed_stats_dict["min_time"],
+                        dispersed_stats_dict["max_time"],
+                        dispersed_stats_dict["avg_time"],
                     ]
                 )
 

llm/benchmark/rl/paddle_infer.py

@@ -20,6 +20,7 @@
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import List
+from utils import RangeSet
 
 import paddle
 import pandas as pd
@@ -36,93 +37,6 @@
 from paddlenlp.transformers import AutoModelForCausalLM, AutoTokenizer
 from paddlenlp.utils.log import logger
 
-
-@dataclass
-class RangeSet:
-    """Manage processed line ranges with efficient storage and querying"""
-
-    ranges: List[tuple]
-
-    def add(self, number: int):
-        """Add a number to the range set and merge adjacent ranges"""
-        new_ranges = []
-        added = False
-        for start, end in sorted(self.ranges):
-            if number < start - 1:
-                if not added:
-                    new_ranges.append((number, number))
-                    added = True
-                new_ranges.append((start, end))
-            elif number == start - 1:
-                new_ranges.append((number, end))
-                added = True
-            elif number <= end:
-                new_ranges.append((start, end))
-                added = True
-            else:
-                new_ranges.append((start, end))
-        if not added:
-            new_ranges.append((number, number))
-        self.ranges = self.merge_ranges(new_ranges)
-
-    @staticmethod
-    def merge_ranges(ranges: List[tuple]) -> List[tuple]:
-        """Merge overlapping or adjacent ranges"""
-        if not ranges:
-            return []
-        sorted_ranges = sorted(ranges)
-        merged = [sorted_ranges[0]]
-        for current in sorted_ranges[1:]:
-            last = merged[-1]
-            if current[0] <= last[1] + 1:
-                merged[-1] = (last[0], max(last[1], current[1]))
-            else:
-                merged.append(current)
-        return merged
-
-    def contains(self, number: int) -> bool:
-        """Check if a number exists in any range"""
-        for start, end in self.ranges:
-            if start <= number <= end:
-                return True
-        return False
-
-    def to_file_format(self) -> str:
-        """Serialize ranges to compact string format"""
-        return ",".join(f"{start}-{end}" if start != end else str(start) for start, end in self.ranges)
-
-    @classmethod
-    def from_file(cls, content: str) -> "RangeSet":
-        """Deserialize from string format"""
-        if not content:
-            return cls(ranges=[])
-        ranges = []
-        for part in content.split(","):
-            if "-" in part:
-                start, end = map(int, part.split("-"))
-                ranges.append((start, end))
-            else:
-                num = int(part)
-                ranges.append((num, num))
-        return cls(ranges=ranges)
-
-    @property
-    def processed_count(self) -> int:
-        """Total number of processed items"""
-        return sum(end - start + 1 for start, end in self.ranges)
-
-
-@contextmanager
-def switch_level_context(level="ERROR"):
-    original_level = logger.logLevel
-    logger.set_level(level)
-
-    try:
-        yield
-    finally:
-        logger.set_level(original_level)
-
-
 def chunk(all_input_ids, size):
     if size <= 0:
         raise ValueError("Size must be greater than 0")

llm/benchmark/rl/torch_infer.py

@@ -18,13 +18,12 @@
 import math
 import time
 from contextlib import contextmanager
-from dataclasses import dataclass
 from pathlib import Path
-from typing import List
 
 import pandas as pd
 import tqdm
 from transformers import AutoTokenizer
+from utils import RangeSet
 from vllm import LLM, SamplingParams
 
 from paddlenlp.utils.log import logger
@@ -70,81 +69,6 @@ def switch_level_context(level="ERROR"):
         logger.set_level(original_level)
 
 
-@dataclass
-class RangeSet:
-    """Manage processed line ranges with efficient storage and querying"""
-
-    ranges: List[tuple]
-
-    def add(self, number: int):
-        """Add a number to the range set and merge adjacent ranges"""
-        new_ranges = []
-        added = False
-        for start, end in sorted(self.ranges):
-            if number < start - 1:
-                if not added:
-                    new_ranges.append((number, number))
-                    added = True
-                new_ranges.append((start, end))
-            elif number == start - 1:
-                new_ranges.append((number, end))
-                added = True
-            elif number <= end:
-                new_ranges.append((start, end))
-                added = True
-            else:
-                new_ranges.append((start, end))
-        if not added:
-            new_ranges.append((number, number))
-        self.ranges = self.merge_ranges(new_ranges)
-
-    @staticmethod
-    def merge_ranges(ranges: List[tuple]) -> List[tuple]:
-        """Merge overlapping or adjacent ranges"""
-        if not ranges:
-            return []
-        sorted_ranges = sorted(ranges)
-        merged = [sorted_ranges[0]]
-        for current in sorted_ranges[1:]:
-            last = merged[-1]
-            if current[0] <= last[1] + 1:
-                merged[-1] = (last[0], max(last[1], current[1]))
-            else:
-                merged.append(current)
-        return merged
-
-    def contains(self, number: int) -> bool:
-        """Check if a number exists in any range"""
-        for start, end in self.ranges:
-            if start <= number <= end:
-                return True
-        return False
-
-    def to_file_format(self) -> str:
-        """Serialize ranges to compact string format"""
-        return ",".join(f"{start}-{end}" if start != end else str(start) for start, end in self.ranges)
-
-    @classmethod
-    def from_file(cls, content: str) -> "RangeSet":
-        """Deserialize from string format"""
-        if not content:
-            return cls(ranges=[])
-        ranges = []
-        for part in content.split(","):
-            if "-" in part:
-                start, end = map(int, part.split("-"))
-                ranges.append((start, end))
-            else:
-                num = int(part)
-                ranges.append((num, num))
-        return cls(ranges=ranges)
-
-    @property
-    def processed_count(self) -> int:
-        """Total number of processed items"""
-        return sum(end - start + 1 for start, end in self.ranges)
-
-
 class DumpyInferenceTask:
     def __init__(self, args):
         self.args = args

llm/benchmark/rl/utils.py

@@ -0,0 +1,91 @@
+# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved.
+#
+# 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 dataclasses import dataclass
+from typing import List
+
+
+@dataclass
+class RangeSet:
+    """Manage processed line ranges with efficient storage and querying"""
+
+    ranges: List[tuple]
+
+    def add(self, number: int):
+        """Add a number to the range set and merge adjacent ranges"""
+        new_ranges = []
+        added = False
+        for start, end in sorted(self.ranges):
+            if number < start - 1:
+                if not added:
+                    new_ranges.append((number, number))
+                    added = True
+                new_ranges.append((start, end))
+            elif number == start - 1:
+                new_ranges.append((number, end))
+                added = True
+            elif number <= end:
+                new_ranges.append((start, end))
+                added = True
+            else:
+                new_ranges.append((start, end))
+        if not added:
+            new_ranges.append((number, number))
+        self.ranges = self.merge_ranges(new_ranges)
+
+    @staticmethod
+    def merge_ranges(ranges: List[tuple]) -> List[tuple]:
+        """Merge overlapping or adjacent ranges"""
+        if not ranges:
+            return []
+        sorted_ranges = sorted(ranges)
+        merged = [sorted_ranges[0]]
+        for current in sorted_ranges[1:]:
+            last = merged[-1]
+            if current[0] <= last[1] + 1:
+                merged[-1] = (last[0], max(last[1], current[1]))
+            else:
+                merged.append(current)
+        return merged
+
+    def contains(self, number: int) -> bool:
+        """Check if a number exists in any range"""
+        for start, end in self.ranges:
+            if start <= number <= end:
+                return True
+        return False
+
+    def to_file_format(self) -> str:
+        """Serialize ranges to compact string format"""
+        return ",".join(f"{start}-{end}" if start != end else str(start) for start, end in self.ranges)
+
+    @classmethod
+    def from_file(cls, content: str) -> "RangeSet":
+        """Deserialize from string format"""
+        if not content:
+            return cls(ranges=[])
+        ranges = []
+        for part in content.split(","):
+            if "-" in part:
+                start, end = map(int, part.split("-"))
+                ranges.append((start, end))
+            else:
+                num = int(part)
+                ranges.append((num, num))
+        return cls(ranges=ranges)
+
+    @property
+    def processed_count(self) -> int:
+        """Total number of processed items"""
+        return sum(end - start + 1 for start, end in self.ranges)