0819 fix itl

ais_bench/benchmark/calculators/stable_perf_metric_calculator.py

@@ -10,18 +10,17 @@
 from ais_bench.benchmark.utils.logging.error_codes import CALC_CODES
 from ais_bench.benchmark.utils.logging.exceptions import AISBenchDataContentError
 
-WAVE_OFFSET = 0.02
-INTERVAL_OFFSET = 0.001
 
 @PERF_METRIC_CALCULATORS.register_module()
 class StablePerfMetricCalculator(BasePerfMetricCalculator):
     """
     Performance metric calculator for stable stage analysis.
 
-    This calculator focuses on analyzing the stable phase of benchmark execution,
-    where the system operates at maximum concurrency with minimal fluctuations.
-    It identifies and analyzes the stable period to provide more accurate
-    performance metrics.
+    This calculator identifies the stable phase as the interval from the first
+    time concurrency reaches max_concurrency to the last time it is at that level.
+    All requests starting within this interval are included regardless of
+    concurrency fluctuations, providing a robust steady-state performance measurement
+    without requiring manual parameter tuning.
 
     Args:
         stats_list (list, optional): List of statistics to calculate
@@ -63,18 +62,24 @@ def _get_requests_id(self, perf_details: dict) -> list:
         """
         Identify requests that belong to the stable stage.
 
+        The stable stage is defined as the interval from the first moment
+        concurrency reaches max_concurrency to the last moment concurrency
+        is at max_concurrency. All requests with start_time within this
+        closed interval are included, regardless of concurrency fluctuations.
+
         Args:
             perf_details (dict): Performance details dictionary
 
         Returns:
             list: List of request IDs in the stable stage
 
         Raises:
-            RuntimeError: If no stable stage can be identified
+            AISBenchDataContentError: If no stable stage can be identified
         """
         # Calculate the minimum start time as the baseline
         min_start_time = min(perf_details["start_time"])
-        time_point_concurrency = [0] * 2 * len(perf_details["id"])
+
+        # Build sorted list of start/end events for all requests
         request_time_sections = []
         for id in range(len(perf_details["id"])):
             request_time_sections.append(
@@ -92,84 +97,52 @@ def _get_requests_id(self, perf_details: dict) -> list:
                 }
             )
         sorted_time_sections = sorted(request_time_sections, key=lambda x: x["time"])
-        id_lists = []
+
         self.logger.info("Starting stable stage calculation...")
-        requested = 0
+        first_max_time = None
+        last_max_time = None
+        concurrency = 0
+
         progress_bar = tqdm(
             total=len(sorted_time_sections),
             desc="Calculating stable stage",
             unit=" req",
         )
-        for i, section in enumerate(sorted_time_sections):
+
+        for section in sorted_time_sections:
             if section["attr"] == "start":
-                time_point_concurrency[i] = time_point_concurrency[i - 1] + 1
-                requested += 1
+                concurrency += 1
             else:
-                time_point_concurrency[i] = time_point_concurrency[i - 1] - 1
-            if (
-                section["attr"] == "start"
-                and time_point_concurrency[i] == self.max_concurrency
-            ):
-                id_lists.append(section["id"])
-                if len(id_lists) == 2:
-                    self.stage_section[0] = section["time"]  # total start time
-            elif (
-                section["attr"] == "start"
-                and time_point_concurrency[i]
-                >= int(self.max_concurrency * (1 - WAVE_OFFSET))
-                and len(id_lists) > 2
-            ):
-                id_lists.append(section["id"])
-            elif requested == len(perf_details["id"]) and section["attr"] == "end":
-                self.stage_section[1] = section["time"]
-                progress_bar.update(len(sorted_time_sections) - progress_bar.n)
-                break
-            elif (
-                len(id_lists) > 1
-                and section["attr"] == "end"
-                and time_point_concurrency[i]
-                < int(self.max_concurrency * (1 - WAVE_OFFSET))
-            ):
-                # Check if there's a start event within INTERVAL_OFFSET that recovers concurrency
-                # Skip consecutive end events and look ahead for start events
-                should_exit_stable = True
-                current_time = section["time"]
-                current_concurrency = time_point_concurrency[i]
-
-                # Look ahead to find the next start event within INTERVAL_OFFSET
-                for j in range(i + 1, len(sorted_time_sections)):
-                    next_section = sorted_time_sections[j]
-                    time_interval = next_section["time"] - current_time
-
-                    # If time interval exceeds INTERVAL_OFFSET, exit stable stage
-                    if time_interval > INTERVAL_OFFSET:
-                        break
-
-                    # Update concurrency based on the event type
-                    if next_section["attr"] == "end":
-                        current_concurrency -= 1
-                    else:  # start event
-                        current_concurrency += 1
-                        # If concurrency recovers to threshold, don't exit stable stage
-                        if current_concurrency >= int(
-                            self.max_concurrency * (1 - WAVE_OFFSET)
-                        ):
-                            should_exit_stable = False
-                            break
-
-                if should_exit_stable:
-                    self.stage_section[1] = section["time"]
-                    progress_bar.update(len(sorted_time_sections) - progress_bar.n)
-                    break
+                concurrency -= 1
+
+            if concurrency == self.max_concurrency:
+                if first_max_time is None:
+                    first_max_time = section["time"]
+                last_max_time = section["time"]
+
             progress_bar.update(1)
         progress_bar.close()
-        if len(id_lists) > 0:
-            id_lists.pop(0)  # ignore first request that reached max concurrency
+
+        if first_max_time is None:
+            raise AISBenchDataContentError(
+                CALC_CODES.CAN_NOT_FIND_STABLE_STAGE,
+                "Can not find a stable stage from performance results! Please check the conccurency plot.",
+            )
+
+        self.stage_section = [first_max_time, last_max_time]
+
+        # Collect all requests whose start_time falls within the stable interval
+        id_lists = [
+            id for id in range(len(perf_details["id"]))
+            if first_max_time <= perf_details["start_time"][id] <= last_max_time
+        ]
+
         if len(id_lists) == 0:
             raise AISBenchDataContentError(
                 CALC_CODES.CAN_NOT_FIND_STABLE_STAGE,
                 "Can not find a stable stage from performance results! Please check the conccurency plot.",
             )
+
         # Convert to relative time based on minimum start time
         relative_start_time = self.stage_section[0] - min_start_time
         relative_end_time = self.stage_section[1] - min_start_time

ais_bench/benchmark/models/api_models/base_api.py

@@ -226,6 +226,10 @@ async def parse_stream_response(self, data, output):
             f"{self.__class__.__name__} should be implemented if stream is True",
         )
 
+    def should_record_stream_time_point(self, data: dict) -> bool:
+        """Return whether a parsed stream response represents an output event."""
+        return True
+
     async def generate(
         self,
         input_data: PromptType,
@@ -295,7 +299,6 @@ async def stream_infer(self, request_body: dict, output: Output):
                     chunk = chunk.removeprefix("data:").strip()
                     if chunk == "[DONE]":
                         break
-                    await output.record_time_point()
                     try:
                         data = json.loads(chunk)
                     except json.JSONDecodeError as e:
@@ -305,6 +308,8 @@ async def stream_infer(self, request_body: dict, output: Output):
                             MODEL_CODES.PARSE_TEXT_RSP_INVALID_FORMAT,
                             f"Unexpected response format. Please check 'error_info' in ***_failed.jsonl for more information.",
                         )
+                    if self.should_record_stream_time_point(data):
+                        await output.record_time_point()
                     await self.parse_stream_response(data, output)
                 output.success = True
             else:

ais_bench/benchmark/models/api_models/vllm_custom_api.py

@@ -148,6 +148,9 @@ async def parse_text_response(self, api_response: dict, output: Output):
         await self._parse_usage(api_response, output)
         self.logger.debug(f"Output content: {output.content}")
 
+    def should_record_stream_time_point(self, data: dict) -> bool:
+        return bool(data.get("choices"))
+
     async def parse_stream_response(self, api_response: dict, output: Output):
         generated_text = ""
         if len(api_response.get("choices", [])) > 0:

ais_bench/benchmark/models/api_models/vllm_custom_api_chat.py

@@ -186,6 +186,9 @@ async def _parse_usage(self, json_content: dict, output: Output):
             output.input_tokens = json_content["usage"].get("prompt_tokens", 0)
             output.output_tokens = json_content["usage"].get("completion_tokens", 0)
 
+    def should_record_stream_time_point(self, data: dict) -> bool:
+        return bool(data.get("choices"))
+
     async def parse_stream_response(self, json_content, output):
         for item in json_content.get("choices", []):
             if item["delta"].get("content"):

docs/superpowers/specs/2026-08-19-vllm-itl-alignment-design.md

@@ -0,0 +1,76 @@
+# aisbench ITL 计时与 vLLM Completions 对齐设计
+
+## 目标
+
+使 aisbench 的 VLLM 流式计时语义与 `vllm bench serve --backend vllm` 保持一致：仅将 `choices` 非空的响应视为输出计时事件；usage-only 响应和 `[DONE]` 消息均不影响 ITL、TPOT 或 E2EL。
+
+## 范围
+
+- 对 `VLLMCustomAPI` 和 `VLLMCustomAPIChat` 应用 token 事件过滤。
+- 保持 TGI、Triton、MindIE 和 VITA 适配器现有的流式计时行为不变。
+- 继续解析 usage-only 响应，确保 prompt token 和 completion token 数量仍能正确获取。
+- 删除 `base_api.py` 中当前未提交的 `[ITL-DIAG]` 临时诊断代码。
+- 修改生产代码前先添加回归测试。
+- 本次变更不修改 ITL 聚合方式、百分位计算、CSV 列或报告中 `N` 字段的含义。
+
+## 设计
+
+### 协议感知的计时判断方法
+
+在 `BaseAPIModel` 中增加一个可覆盖的方法：
+
+```python
+def should_record_stream_time_point(self, data: dict) -> bool:
+    return True
+```
+
+基类默认返回 `True`，从而保持所有非 VLLM 适配器的现有行为。两个 VLLM 适配器均覆盖该方法：
+
+```python
+def should_record_stream_time_point(self, data: dict) -> bool:
+    return bool(data.get("choices"))
+```
+
+### 流式数据处理流程
+
+对于每一条非空、非注释且不等于 `[DONE]` 的流式消息：
+
+1. 解码并解析 JSON 数据。
+2. 调用 `should_record_stream_time_point(data)`。
+3. 仅当该方法返回 `True` 时记录时间点。
+4. 始终调用 `parse_stream_response(data, output)`，确保 usage-only 响应仍可写入 token 数量。
+
+请求开始时间点的记录方式保持不变。如果响应中包含 1024 个携带 `choices` 的输出 chunk，最终计时数组将包含一个请求开始时间点和 1024 个输出事件，因此生成 1023 个 ITL。
+
+## 兼容性
+
+- 该设计与 vLLM v0.26 Completions 客户端一致：仅在 `choices` 非空分支内记录时间点。
+- Chat 适配器采用相同过滤规则，使当前配置的 aisbench Chat 接口也可以按照同一套计时语义进行比较。
+- 其他适配器继承基类默认判断，继续为每个已解析的 JSON chunk 记录时间点。
+- JSON 解析失败时继续沿用现有错误处理路径，并且不记录时间点。
+
+## 测试设计
+
+新增针对性的异步测试，模拟包含以下内容的响应流：两个携带 `choices` 的 chunk、一个 usage-only chunk，以及 `[DONE]`。
+
+必须验证：
+
+- VLLM 适配器共记录三个时间点：一个请求开始时间点和两个 choice 事件时间点。
+- usage-only chunk 不增加时间点。
+- prompt token 和 completion token 数量可从 usage 响应中正确获取。
+- 最终只产生一个 ITL。
+- 选取一个有代表性的非 VLLM 适配器，验证其仍为每个已解析的数据 chunk 记录时间点。
+
+在修改生产代码之前，测试必须在当前实现上因 usage-only chunk 被错误计时而失败。完成最小实现后，测试必须通过。
+
+## 错误处理
+
+本次变更不引入新的错误类型。现有 HTTP 错误、JSON 解析错误、重试和空响应处理逻辑均保持不变。
+
+## 验收标准
+
+- VLLM usage-only 响应不再产生亚毫秒级的尾部 ITL。
+- VLLM usage token 统计保持正确。
+- 非 VLLM 流式计时行为不发生变化。
+- 针对性测试及相关现有 API 模型测试全部通过。
+- 删除临时诊断代码后，`base_api.py` 能够通过 Python 语法解析。

tests/UT/calculators/test_stable_perf_metric_calculator.py

@@ -300,22 +300,23 @@ def test_convert_result_integration(self):
 
     def test_edge_case_empty_stable_stage(self):
         # 测试边缘情况 - 稳定阶段只有少量请求
+        # 新算法：单个请求达到 max_concurrency=1 时也是有效的稳定阶段
         calculator = StablePerfMetricCalculator()
         calculator.max_concurrency = 1
         calculator.logger = self.mock_logger
         calculator.stage_section = [0, 0]  # 初始化必要的属性
-        
+
         # 准备测试数据 - 只有一个请求
         perf_details = {
             "id": [0],
             "start_time": [1.0],
             "end_time": [2.0],
             "success": [True]
         }
-        
-        # 验证抛出异常
-        with self.assertRaises(AISBenchDataContentError):
-            calculator._get_requests_id(perf_details)
+
+        result = calculator._get_requests_id(perf_details)
+        self.assertEqual(result, [0])
+        self.assertEqual(calculator.stage_section, [1.0, 1.0])
 
 
 if __name__ == "__main__":

tests/UT/models/api_models/test_base_api.py

@@ -284,6 +284,35 @@ async def run():
 
         asyncio.run(run())
 
+    def test_stream_infer_default_records_each_json_chunk(self):
+        async def mock_content():
+            yield b'data: {"chunk": "hello"}\n\n'
+            yield b'data: {"chunk": " world"}\n\n'
+            yield b"data: [DONE]\n\n"
+
+        class MockPostContext:
+            async def __aenter__(self):
+                response = mock.MagicMock()
+                response.status = 200
+                response.content = mock_content()
+                return response
+
+            async def __aexit__(self, exc_type, exc_value, traceback):
+                return False
+
+        model = self.model_class(**self.default_kwargs)
+        model.stream = True
+        model.session = mock.MagicMock()
+        model.session.post.return_value = MockPostContext()
+        output = Output(perf_mode=True)
+        output.text = ""
+
+        asyncio.run(model.stream_infer({"prompt": "test"}, output))
+
+        self.assertTrue(output.success)
+        self.assertEqual(output.text, "hello world")
+        self.assertEqual(len(output.time_points), 3)
+
     @mock.patch("aiohttp.ClientSession.post")
     async def test_stream_infer_json_decode_error(self, mock_post):
         async def mock_content():