Skip to content

[benchmark] update tools #8097

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
EmmonsCurse merged 1 commit into from
Jul 9, 2026
Merged

[benchmark] update tools #8097

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion benchmarks/backend_request_func.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@
class RequestFuncInput:
"""Input for requesting LLMs via API"""

no: int
no: str
prompt: str
history_QA: Optional[dict]
hyper_parameters: dict
Expand All @@ -61,6 +61,8 @@ class RequestFuncInput:
tokenizer_model: str = None
tokenizer_path: str = None
stream: bool = True
session_id: Optional[str] = None
turn_idx: Optional[int] = None


@dataclass
Expand Down
65 changes: 26 additions & 39 deletions benchmarks/backend_request_func_swe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import sys
import time
import traceback
import uuid
from dataclasses import dataclass, field
from typing import Optional

Expand Down Expand Up @@ -299,7 +300,7 @@ async def handle_non_stream_response(
# arrival_time:
output.arrival_time = []

has_text = output.generated_text.strip() or output.reasoning_content.strip()
has_text = bool(output.generated_text) or bool(output.reasoning_content)

has_tool = bool(output.tool_calls)

Expand Down Expand Up @@ -412,6 +413,7 @@ async def async_request_eb_openai_chat_completions(
"Content-Type": "application/json",
"Authorization": f"Bearer {os.environ.get('OPENAI_API_KEY')}",
}

if request_func_input.session_id is not None:
headers["X-SMG-Routing-Key"] = f"{request_func_input.session_id}"
if request_func_input.session_id is not None and request_func_input.turn_idx is not None:
Expand Down Expand Up @@ -619,7 +621,7 @@ async def async_request_eb_openai_chat_completions(
# 新增metrics统计，计算首token过滤空包
output.metrics = metrics_summary(metrics_list, token_timestamps[1:])

has_text = output.generated_text.strip() or output.reasoning_content.strip()
has_text = bool(output.generated_text) or bool(output.reasoning_content)
has_tool = getattr(output, "tool_calls", None)

# 如果前面已经有服务端错误，保留原错误
Expand Down Expand Up @@ -770,6 +772,7 @@ async def async_request_eb_openai_chat_completions_multi_turn(

# 只创建一次 session
session_start = time.perf_counter()
session_uuid = uuid.uuid4().hex
connector = aiohttp.TCPConnector(
limit=0,
limit_per_host=0,
Expand All @@ -788,7 +791,7 @@ async def async_request_eb_openai_chat_completions_multi_turn(
round_input = copy.deepcopy(request_func_input)
round_input.history_QA = history
round_input.no = f"{round_input.no}_{prompt_no}"
round_input.session_id = request_func_input.no
round_input.session_id = f"{session_uuid}:{request_func_input.no}"
round_input.turn_idx = prompt_no
if use_token_ids:
if len(input_ids_all) == 0:
Expand Down Expand Up @@ -841,16 +844,14 @@ async def async_request_eb_openai_chat_completions_multi_turn(
outputs.append(output)

if not output.success:
session_end = time.perf_counter()
metrics = SessionMetrics(
session_no=request_func_input.no,
session_e2e_time=session_end - session_start,
pure_llm_time=llm_time,
input_tokens=input_tokens,
output_tokens=output_tokens,
tool_calls=tool_call_count,
)
return outputs, metrics
if enable_tools:
# 有工具调用时，请求失败直接中断整个session
print(f"[SESSION STOP] {round_input.no} request failed with tools, stop session")
break
# SWE无工具调用时，跳过当前轮但继续session
print(f"[SESSION WARN] {round_input.no} request failed, continue session")
prompt_no += 1
continue

# llm_cost = s1 - s0
input_tokens += output.prompt_tokens
Expand All @@ -871,6 +872,7 @@ async def async_request_eb_openai_chat_completions_multi_turn(
max_prompt_len = json_data.get("max_prompt_len")
if not tool_url:
raise ValueError("tool_url is empty.")
session_stopped = False
for _ in range(max_loop):
t0 = time.perf_counter()
tool_result, is_tool_result, tool_name, tool_id = await simple_tool_call(
Expand All @@ -883,26 +885,14 @@ async def async_request_eb_openai_chat_completions_multi_turn(
# print(f"#### tool_result: {tool_result}")
# print(f"#### is_tool_result: {is_tool_result}")

# 工具调用失败
# 工具调用失败，中断整个session
if tool_name and not is_tool_result:
print(f"[SESSION FAIL] tool call failed: {tool_name}")
print(f"[SESSION STOP] tool call failed: {tool_name}, stop session")

output.success = False

session_end = time.perf_counter()
session_e2e_time = session_end - session_start
tool_call_count += 1

metrics = SessionMetrics(
session_no=request_func_input.no,
session_e2e_time=session_e2e_time,
pure_llm_time=llm_time,
input_tokens=input_tokens,
output_tokens=output_tokens,
tool_calls=tool_call_count,
)

return outputs, metrics
session_stopped = True
break

if not is_tool_result:
history.append(
Expand Down Expand Up @@ -956,16 +946,9 @@ async def async_request_eb_openai_chat_completions_multi_turn(
outputs.append(output)

if not output.success:
session_end = time.perf_counter()
metrics = SessionMetrics(
session_no=request_func_input.no,
session_e2e_time=session_end - session_start,
pure_llm_time=llm_time,
input_tokens=input_tokens,
output_tokens=output_tokens,
tool_calls=tool_call_count,
)
return outputs, metrics
print(f"[SESSION STOP] {round_input.no} tool loop request failed, stop session")
session_stopped = True
break

input_tokens += output.prompt_tokens
output_tokens += output.output_tokens
Expand All @@ -987,6 +970,10 @@ async def async_request_eb_openai_chat_completions_multi_turn(
else:
print(f"Warning {prompt_no} exceed max_loop={max_loop}, force stop tool loop")

if session_stopped:
# 工具调用失败，中断整个session
break

else:
# 无tools
# history.append(
Expand Down
110 changes: 63 additions & 47 deletions benchmarks/benchmark_serving.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import os
import random
import time
import uuid
import warnings
from argparse import ArgumentParser as FlexibleArgumentParser
from collections.abc import AsyncGenerator, Iterable
Expand Down Expand Up @@ -63,12 +64,12 @@ class BenchmarkMetrics:
request_goodput: float
output_throughput: float
total_token_throughput: float
# 全局聚合解码速度(过滤 burst 与 preemption 后)
s_decode_clean: float # tok/s
n_itls_total: int # 全部 itl 样本数
n_itls_burst: int # itl < 1ms 的数量
n_itls_preempt: int # itl > 500ms 的数量
n_itls_clean: int # 1ms <= itl <= 500ms 的数量
# 解码速度(过滤 TPOT<1ms 的请求后)
s_decode_filtered_mean: float # tok/s
s_decode_filtered_median: float # tok/s
n_decode_total: int # 参与统计的总请求数
n_decode_filtered: int # 被过滤的请求数(TPOT<1ms)
n_decode_reliable: int # 可信请求数(TPOT>=1ms)
mean_s_decode: float
median_s_decode: float
std_s_decode: float
Expand Down Expand Up @@ -266,10 +267,6 @@ def calculate_metrics(
else:
s_decodes.append(0)
completed += 1
else:
actual_output_lens.append(0)
input_lens.append(0)
infer_input_lens.append(0)

if goodput_config_dict:
valid_metrics = []
Expand All @@ -296,20 +293,25 @@ def calculate_metrics(
stacklevel=2,
)

# === Cleaned ITL aggregation ===
BURST_THRESHOLD_S = 0.001 # 1 ms
PREEMPT_THRESHOLD_S = 0.5 # 500 ms
all_itls_flat: list[float] = []
# === 解码速度过滤：TPOT < 1ms 的请求视为不可信(引擎批量flush伪象) ===
MIN_TPOT_S = 0.001 # 1ms
reliable_s_decodes = []
n_decode_total = 0
n_decode_filtered = 0
for o in outputs:
if o.success:
all_itls_flat.extend(o.itl)
_arr = np.asarray(all_itls_flat, dtype=float) if all_itls_flat else np.empty(0)
n_itls_total = int(_arr.size)
n_itls_burst = int((_arr < BURST_THRESHOLD_S).sum())
n_itls_preempt = int((_arr > PREEMPT_THRESHOLD_S).sum())
_clean = _arr[(_arr >= BURST_THRESHOLD_S) & (_arr <= PREEMPT_THRESHOLD_S)]
n_itls_clean = int(_clean.size)
s_decode_clean = float(_clean.size / _clean.sum()) if _clean.sum() > 0 else 0.0
if not o.success or o.output_tokens <= 1:
continue
decode_time = sum(o.itl) if o.itl else 0
if decode_time <= 0:
continue
n_decode_total += 1
tokens = o.output_tokens - 1
tpot = decode_time / tokens
if tpot >= MIN_TPOT_S:
reliable_s_decodes.append(tokens / decode_time)
else:
n_decode_filtered += 1
n_decode_reliable = len(reliable_s_decodes)

metrics = BenchmarkMetrics(
completed=completed,
Expand All @@ -319,6 +321,11 @@ def calculate_metrics(
request_goodput=good_completed / dur_s,
output_throughput=sum(actual_output_lens) / dur_s,
total_token_throughput=(total_input + sum(actual_output_lens)) / dur_s,
s_decode_filtered_mean=float(np.mean(reliable_s_decodes)) if reliable_s_decodes else 0.0,
s_decode_filtered_median=float(np.median(reliable_s_decodes)) if reliable_s_decodes else 0.0,
n_decode_total=n_decode_total,
n_decode_filtered=n_decode_filtered,
n_decode_reliable=n_decode_reliable,
mean_s_decode=np.mean(s_decodes or 0) * 1, # ttfts is empty if streaming is not supported by backend
std_s_decode=np.std(s_decodes or 0) * 1,
median_s_decode=np.median(s_decodes or 0) * 1,
Expand Down Expand Up @@ -371,11 +378,6 @@ def calculate_metrics(
std_res_ttft_ms=np.std(res_ttfts or 0) * 1000,
median_res_ttft_ms=np.median(res_ttfts or 0) * 1000,
percentiles_res_ttft_ms=[(p, np.percentile(res_ttfts or 0, p) * 1000) for p in selected_percentiles],
s_decode_clean=s_decode_clean,
n_itls_total=n_itls_total,
n_itls_burst=n_itls_burst,
n_itls_preempt=n_itls_preempt,
n_itls_clean=n_itls_clean,
)

return metrics, actual_output_lens
Expand Down Expand Up @@ -439,11 +441,12 @@ async def benchmark(
# warmup短输出：取128和hyper_parameters中max_tokens的最小值
warmup_output_len = min(128, hyper_parameters.get("max_tokens", 128))
warmup_hyper = {k: v for k, v in hyper_parameters.items() if k != "max_tokens"}
warmup_request_id = f"warmup_{uuid.uuid4().hex[:8]}"
test_input = RequestFuncInput(
model=model_id,
model_name=model_name,
prompt=test_prompt,
no=test_no,
no=warmup_request_id,
prompt_len=0,
history_QA=test_history_QA,
hyper_parameters=warmup_hyper,
Expand All @@ -460,6 +463,8 @@ async def benchmark(
tokenizer_model=args.tokenizer_model,
tokenizer_path=args.tokenizer_path,
stream=args.stream,
session_id=f"warmup-{uuid.uuid4().hex}",
turn_idx=0,
)

if args.warmup:
Expand Down Expand Up @@ -772,7 +777,7 @@ async def limited_request_func_per_ip(req_input, semaphore, pbar):
"total_token_throughput": metrics.total_token_throughput,
"input_lens": [output.prompt_len for output in outputs],
"infer_input_lens": [output.prompt_tokens for output in outputs],
"output_lens": actual_output_lens,
"output_lens": [output.output_tokens for output in outputs],
"ttfts": [output.ttft for output in outputs],
"res_ttfts": [output.res_ttft for output in outputs],
"itls": [output.itl for output in outputs],
Expand All @@ -781,11 +786,11 @@ async def limited_request_func_per_ip(req_input, semaphore, pbar):
"reasoning_contents": [output.reasoning_content for output in outputs],
"errors": [output.error for output in outputs],
"metrics": [output.metrics for output in outputs],
"s_decode_clean": metrics.s_decode_clean,
"n_itls_total": metrics.n_itls_total,
"n_itls_burst": metrics.n_itls_burst,
"n_itls_preempt": metrics.n_itls_preempt,
"n_itls_clean": metrics.n_itls_clean,
"s_decode_filtered_mean": metrics.s_decode_filtered_mean,
"s_decode_filtered_median": metrics.s_decode_filtered_median,
"n_decode_total": metrics.n_decode_total,
"n_decode_filtered": metrics.n_decode_filtered,
"n_decode_reliable": metrics.n_decode_reliable,
}

def process_one_metric(
Expand Down Expand Up @@ -934,25 +939,21 @@ def process_one_length(
print("{:<40} {:<10.2f}".format(f"P{p_word} {metric_name}:", value))
result[f"p{p_word}_{metric_attribute_name}"] = value

print("{s:{c}^{n}}".format(s="解码速度 (ITL全局聚合)", n=50, c="-"))
_tot = max(metrics.n_itls_total, 1)
print("{:<40} {:<10d}".format("Total ITLs:", metrics.n_itls_total))
print(
"{:<40} {:<10d} ({:.2f}%)".format(
"ITL < 1ms (burst):", metrics.n_itls_burst, 100 * metrics.n_itls_burst / _tot
)
)
print("{s:{c}^{n}}".format(s="解码速度 (过滤TPOT<1ms)", n=50, c="-"))
_n_total = max(metrics.n_decode_total, 1)
print("{:<40} {:<10d}".format("Total requests:", metrics.n_decode_total))
print(
"{:<40} {:<10d} ({:.2f}%)".format(
"ITL > 500ms (preempt):", metrics.n_itls_preempt, 100 * metrics.n_itls_preempt / _tot
"Filtered (TPOT<1ms):", metrics.n_decode_filtered, 100 * metrics.n_decode_filtered / _n_total
)
)
print(
"{:<40} {:<10d} ({:.2f}%)".format(
"ITL clean [1ms,500ms]:", metrics.n_itls_clean, 100 * metrics.n_itls_clean / _tot
"Reliable (TPOT>=1ms):", metrics.n_decode_reliable, 100 * metrics.n_decode_reliable / _n_total
)
)
print("{:<40} {:<10.2f}".format("Decode speed (clean, tok/s):", metrics.s_decode_clean))
print("{:<40} {:<10.2f}".format("Mean Decode (tok/s):", metrics.s_decode_filtered_mean))
print("{:<40} {:<10.2f}".format("Median Decode (tok/s):", metrics.s_decode_filtered_median))
process_one_length("s_decode", "Decode", "解码速度(tok/s)")
process_one_metric("ttft", "TTFT", "Time to First Token")
process_one_metric("s_ttft", "S_TTFT", "Infer Time to First Token")
Expand Down Expand Up @@ -1057,7 +1058,7 @@ def benchmark_metrics(
"output_throughput": metrics.output_throughput,
"total_token_throughput": metrics.total_token_throughput,
"input_lens": [output.prompt_len for output in outputs],
"output_lens": actual_output_lens,
"output_lens": [output.output_tokens for output in outputs],
"ttfts": [output.ttft for output in outputs],
"itls": [output.itl for output in outputs],
"input_texts": ["" for input in input_requests],
Expand Down Expand Up @@ -1131,6 +1132,21 @@ def process_one_length(
print("{:<40} {:<10.2f}".format(f"P{p_word} {metric_name}:", value))
result[f"p{p_word}_{metric_attribute_name}"] = value

print("{s:{c}^{n}}".format(s="解码速度 (过滤TPOT<1ms)", n=50, c="-"))
_n_total = max(metrics.n_decode_total, 1)
print("{:<40} {:<10d}".format("Total requests:", metrics.n_decode_total))
print(
"{:<40} {:<10d} ({:.2f}%)".format(
"Filtered (TPOT<1ms):", metrics.n_decode_filtered, 100 * metrics.n_decode_filtered / _n_total
)
)
print(
"{:<40} {:<10d} ({:.2f}%)".format(
"Reliable (TPOT>=1ms):", metrics.n_decode_reliable, 100 * metrics.n_decode_reliable / _n_total
)
)
print("{:<40} {:<10.2f}".format("Mean Decode (tok/s):", metrics.s_decode_filtered_mean))
print("{:<40} {:<10.2f}".format("Median Decode (tok/s):", metrics.s_decode_filtered_median))
process_one_length("s_decode", "Decode", "解码速度(tok/s)")
process_one_metric("ttft", "TTFT", "Time to First Token")
process_one_metric("s_ttft", "S_TTFT", "Infer Time to First Token")
Expand Down
Loading