分析evalscope中有关perf的代码，梳理清楚 TTFT/ TPOT 等性能指标的定义和计算标准
TTFT #
定义：TTFT(time to first token)，从用户输入提示词（Prompt）到模型生成第一个输出 Token 的时间，仅统计流式响应。
代码实现：通过异步读取流式 HTTP 响应，在收到首个有效响应 chunk 时用客户端计时器计算 当前时间 - 请求起始时间。
# 单请求TTFT的定义:
# `evalscope/evalscope/perf/utils/benchmark_util.py` 中的 `first_chunk_latency`
# 平均TTFT= 所有**成功**请求的 `first_chunk_latency` 之和 / 成功请求数量：
self.total_first_chunk_latency += data.first_chunk_latency
avg_first_chunk_latency = _safe_div(self.total_first_chunk_latency, n)
只要首个 SSE JSON 中存在非空
choices，即使内容只是如下也会被记录为TTFT：{ "choices": [ { "delta": { "role": "assistant" } } ] }相关代码：
if choices := data.get('choices'): if data.get('object') == 'text_completion': content = choices[0].get('text') or '' else: delta = choices[0].get('delta', {}) content = (delta.get('content') or '') + (delta.get('reasoning_content') or '') # First token if ttft == 0.0: ttft = timestamp - st output.first_chunk_latency = ttft如何区分每个请求的TTFT的
class DefaultApiPlugin(ApiPluginBase): """Default implementation of API plugin with common HTTP handling methods.""" def __init__(self, param: Arguments): super().__init__(param) async def process_request( self, client_session: aiohttp.ClientSession, url: str, headers: Dict, body: Dict ) -> BenchmarkData: """Process the HTTP request and handle the response. Args: client_session: The aiohttp client session url: The request URL headers: The request headers body: The request body Returns: BenchmarkData: Aggregated benchmarking data for the request/response. """ headers = {'Content-Type': 'application/json', **headers} data = json.dumps(body, ensure_ascii=False) # serialize to JSON output = BenchmarkData() # 当前请求独有 ttft = 0.0 # 当前请求独有 generated_text = '' st = time.perf_counter() # 当前请求的开始时间 output.start_time = st output.request = data most_recent_timestamp = st try: async with client_session.post(url=url, data=data, headers=headers) as response: content_type = response.headers.get('Content-Type', '') if response.status == 200: # Handle streaming responses (SSE) if 'text/event-stream' in content_type: handler = StreamedResponseHandler() # 当前响应独有 async for chunk_bytes in response.content.iter_any(): if not chunk_bytes: continue messages = handler.add_chunk(chunk_bytes) for message in messages: # NOTE: SSE comments (often used as pings) start with # a colon. These are not JSON data payload and should # be skipped. if message.startswith(':'): continue chunk = message.removeprefix('data:').strip() if chunk != '[DONE]': timestamp = time.perf_counter() data = json.loads(chunk) if choices := data.get('choices'): if data.get('object') == 'text_completion': content = choices[0].get('text') or '' else: delta = choices[0].get('delta', {}) content = (delta.get('content') or '') + (delta.get('reasoning_content') or '') # First token if ttft == 0.0: ttft = timestamp - st output.first_chunk_latency = ttft即使并发度是 100，逻辑上也是：
共享一个 ClientSession ├── 请求 A → response A → TTFT A ├── 请求 B → response B → TTFT B ├── 请求 C → response C → TTFT C └── ...不会出现请求 A 的首个 SSE chunk 被记到请求 B 上。
aiohttp会把每条 HTTP 响应流交给各自的ClientResponse对象。共享 ClientSession 不代表共享响应流，共享ClientSession负责TCP连接池等。EvalScope 的 TTFT 计算通常并不依赖模型返回的：{"id": "chatcmpl-xxx"}，也不靠conversation_id或session_id来匹配响应。关联关系是在客户端调用栈中天然建立的。
TPOT #
定义：TPOT(time per output token)，解码（Decoder）阶段生成每个输出 Token 的平均时间（不包括首个 Token），每个输出 Token 的平均耗时，仅统计流式响应。
代码实现：TPOT = E2E Latency − TTFT / CompletionTokens − 1，分母减 1，是因为首 Token 的耗时已经计入 TTFT，不再算作后续解码 Token。是一个平均值。
# 单请求TPOP的定义：`evalscope/evalscope/perf/utils/benchmark_util.py` 中的 `time_per_output_token`
def finalize(self, api_plugin) -> None:
"""Parse token counts and compute all derived timing metrics.
Must be called after the response is fully received. Idempotent:
token counts already present will not be re-parsed.
"""
if self.prompt_tokens is None or self.completion_tokens is None:
self.prompt_tokens, self.completion_tokens = api_plugin.parse_responses(
self.response_messages, request=self.request
)
# tpot = (latency - ttft) / (output_len - 1)
if self.completion_tokens and self.completion_tokens > 1:
self.time_per_output_token = ((self.query_latency - self.first_chunk_latency) /
(self.completion_tokens - 1))
output.completed_time = time.perf_counter()
output.query_latency = output.completed_time - output.start_time
# 然后可以进一步算个平均
EvalScope 记录 TTFT 的条件并不是“收到第一个真实Token”，而是第一个chunk，如果空 chunk 与首 Token 之间有明显间隔，EvalScope 的 TPOT 会被轻微高估。如果第一个chunk包含了多个token，EvalScope 的 TPOT 会偏小。
completion_tokens有两种来源，优先使用服务端返回的usage.completion_tokens；服务端不返回时，才用本地 tokenizer 对完整输出文本重新分词统计。
OTPS #
定义：OTPS(Output Tokens/s)，每秒输出 Token 数（Output Tokens/s），只计算生成内容。
代码实现：OTPS = 成功请求的 completion_tokens 总和 / (最晚请求开始时间 - 最早请求完成时间)
def to_result(self) -> 'BenchmarkMetrics':
t = self.wall_time
def _safe_div(numerator, denominator, default=-1):
return numerator / denominator if denominator else default
try:
...
avg_output_token_throughput = _safe_div(self.total_completion_tokens, t)
OTPS / Output Token Throughput 计算是全压测窗口的聚合吞吐，不是逐请求 output_tokens / latency 后再求平均
wall_time，表示 现实世界中钟表经过的时间（墙上时钟实际走过了多久），英文叫 wall-clock time，简称wall_time。
wall_time要取整个压测中所有请求共同覆盖的完整时间区间，min(start_time)是第一个正式请求开始的时刻，max(completed_time)是最后一个请求完成的时刻。单请求 OTPS ≈ 1 / TPOT
TPS #
定义：TPM(tokens per second)，模型在一秒内能够处理的 Token 总数。服务整体吞吐量指标，会随并发数增加而变化。
代码实现：分为input/output/total，代码中对应的是 Token Throughput，然后除以时间单位，得到TPS，单位为 tok/s。
# 单请求`token throughput`的定义`total_prompt_tokens `/`total_completion_tokens`
def update(self, data: BenchmarkData, api_plugin) -> None:
"""Ingest one completed request and update all running totals."""
if data.is_warmup:
return # Warmup requests are excluded from all metrics
self.n_total += 1
if data.success:
self.n_success += 1
data.finalize(api_plugin)
self.total_latency += data.query_latency
self.total_first_chunk_latency += data.first_chunk_latency
self.total_prompt_tokens += data.prompt_tokens
self.total_completion_tokens += data.completion_tokens
def _update_wall_time(self, data: BenchmarkData) -> None:
"""Expand the wall-clock window to cover *data*'s lifecycle."""
if self._wall_start is None:
self._wall_start = data.start_time
else:
self._wall_start = min(self._wall_start, data.start_time)
if self._wall_end is None:
self._wall_end = data.completed_time
else:
self._wall_end = max(self._wall_end, data.completed_time)
@property
def wall_time(self) -> float:
if self._wall_start is not None and self._wall_end is not None:
return max(self._wall_end - self._wall_start, 0.0) # 这里的 max 是防御性保护，保证 wall_time 不会是负数
return 1.0
avg_input_token_throughput = _safe_div(self.total_prompt_tokens, self.total_first_chunk_latency)
avg_output_token_throughput = _safe_div(self.total_completion_tokens, t)
avg_total_token_throughput = _safe_div(self.total_prompt_tokens + self.total_completion_tokens, t)
Warmup 请求就是正式统计前，先发给服务端的一小批预热请求，把冷启动性能和稳定运行性能分开