[AMD] Add AMD CI registration (1-gpu unit test) to nightly CI. (#16941)

This commit is contained in:
Michael
2026-01-13 23:41:45 -08:00
committed by GitHub
parent 030496eb06
commit b025cff441
39 changed files with 3701 additions and 2180 deletions
@@ -0,0 +1,250 @@
"""MI35x DeepSeek-R1 GSM8K Completion Evaluation Test (8-GPU)
Tests DeepSeek-R1-0528 with basic configuration using few-shot completion
benchmark on MI35x.
Registry: nightly-amd-accuracy-8-gpu-mi35x-deepseek-r1 suite
"""
import ast
import os
# Set HF cache for MI35x
os.environ.setdefault("HF_HOME", "/data2/models/huggingface")
os.environ.setdefault("HF_HUB_CACHE", "/data2/models/huggingface/hub")
import re
import time
import unittest
from dataclasses import dataclass
from typing import List, Optional, Tuple
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - MI35x DeepSeek-R1-0528 accuracy test (~30 min for basic only)
register_amd_ci(
est_time=1800,
suite="nightly-amd-accuracy-8-gpu-mi35x-deepseek-r1",
nightly=True,
)
INVALID = -9999999
@dataclass
class ModelConfig:
"""Configuration for a model to test."""
model_path: str
tp_size: int = 8
accuracy_threshold: float = 0.50
other_args: Optional[List[str]] = None
env_vars: Optional[dict] = None
timeout: Optional[int] = None
variant: Optional[str] = None
def __post_init__(self):
if self.other_args is None:
self.other_args = []
if self.env_vars is None:
self.env_vars = {}
def get_display_name(self) -> str:
if self.variant:
return f"{self.model_path} ({self.variant})"
return self.model_path
# DeepSeek-R1 models for MI35x - only basic variant for faster CI
# MTP, DP, and TC variants removed to reduce test time from ~2h to ~30min
MI35X_DEEPSEEK_R1_MODELS = [
# DeepSeek-R1-0528 basic
ModelConfig(
model_path="deepseek-ai/DeepSeek-R1-0528",
tp_size=8,
accuracy_threshold=0.93,
timeout=3600,
variant="basic",
other_args=[
"--attention-backend",
"aiter",
"--chunked-prefill-size",
"131072",
"--disable-radix-cache",
"--mem-fraction-static",
"0.85",
"--trust-remote-code",
"--watchdog-timeout",
"1200", # 20 minutes for weight loading
],
env_vars={"SGLANG_USE_AITER": "1"},
),
]
def get_one_example(lines, i, include_answer):
"""Format a single GSM8K example."""
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
"""Get k few-shot examples for prompting."""
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
"""Extract numerical answer from response."""
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(
base_url: str,
num_questions: int = 200,
num_shots: int = 5,
parallel: int = 64,
) -> Tuple[float, float, float]:
"""Run GSM8K few-shot completion benchmark."""
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
assert all(l != INVALID for l in labels)
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
invalid = np.mean(np.array(preds) == INVALID)
return float(acc), float(invalid), float(latency)
class TestDeepSeekR1EvalMI35x(unittest.TestCase):
"""DeepSeek-R1 GSM8K Completion Evaluation Test for AMD MI35x."""
@classmethod
def setUpClass(cls):
cls.models = MI35X_DEEPSEEK_R1_MODELS
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
def test_deepseek_r1_accuracy(self):
"""Test DeepSeek-R1 models with GSM8K completion benchmark."""
all_results = []
summary = "### DeepSeek-R1 Models (MI35x)\n\n"
summary += "| Model | Variant | TP | Accuracy | Threshold | Status |\n"
summary += "| ----- | ------- | -- | -------- | --------- | ------ |\n"
for config in self.models:
display_name = config.get_display_name()
with self.subTest(model=display_name):
print(f"\n{'='*60}")
print(f"Testing: {display_name}")
print(f"{'='*60}")
env = os.environ.copy()
for key, value in config.env_vars.items():
env[key] = value
other_args = list(config.other_args)
other_args.extend(["--tp", str(config.tp_size)])
timeout = config.timeout or DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH
try:
process = popen_launch_server(
model=config.model_path,
base_url=self.base_url,
timeout=timeout,
other_args=other_args,
env=env,
)
try:
acc, invalid, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= config.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
all_results.append(
{
"model": display_name,
"accuracy": acc,
"passed": passed,
}
)
summary += f"| {config.model_path} | {config.variant or 'N/A'} | {config.tp_size} | {acc:.3f} | {config.accuracy_threshold} | {status} |\n"
finally:
kill_process_tree(process.pid)
except Exception as e:
summary += f"| {config.model_path} | {config.variant or 'N/A'} | {config.tp_size} | N/A | {config.accuracy_threshold} | ❌ ERROR |\n"
all_results.append(
{
"model": display_name,
"accuracy": None,
"passed": False,
"error": str(e),
}
)
if is_in_ci():
write_github_step_summary(summary)
failed = [r for r in all_results if not r["passed"]]
if failed:
raise AssertionError(f"Failed models: {[r['model'] for r in failed]}")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,300 @@
"""MI35x DeepSeek-R1-MXFP4 GSM8K Completion Evaluation Test (8-GPU)
Tests DeepSeek-R1-MXFP4 quantized model with multiple configurations
(basic, MTP, DP, TC) using few-shot completion benchmark on MI35x.
Registry: nightly-amd-8-gpu-mi35x-deepseek-r1-mxfp4 suite
"""
import ast
import os
# Set HF cache for MI35x
os.environ.setdefault("HF_HOME", "/data2/models/huggingface")
os.environ.setdefault("HF_HUB_CACHE", "/data2/models/huggingface/hub")
import re
import time
import unittest
from dataclasses import dataclass
from typing import List, Optional, Tuple
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - MI35x DeepSeek-R1-MXFP4 accuracy tests (~120 min)
register_amd_ci(
est_time=7200, suite="nightly-amd-8-gpu-mi35x-deepseek-r1-mxfp4", nightly=True
)
INVALID = -9999999
# Model path configuration for MI35x DeepSeek-R1-MXFP4
# Priority: 1) env var, 2) local path, 3) HuggingFace model ID
DEEPSEEK_R1_MXFP4_LOCAL_PATH = "/data2/models/amd-DeepSeek-R1-MXFP4-Preview"
DEEPSEEK_R1_MXFP4_HF_MODEL_ID = "amd/DeepSeek-R1-MXFP4-Preview"
def get_model_path() -> str:
"""Get effective model path: env var > local path > HF model ID."""
env_path = os.environ.get("DEEPSEEK_R1_MXFP4_MODEL_PATH")
if env_path:
return env_path
if os.path.exists(DEEPSEEK_R1_MXFP4_LOCAL_PATH):
return DEEPSEEK_R1_MXFP4_LOCAL_PATH
return DEEPSEEK_R1_MXFP4_HF_MODEL_ID
@dataclass
class ModelConfig:
"""Configuration for a model to test."""
model_path: str
tp_size: int = 8
accuracy_threshold: float = 0.50
other_args: Optional[List[str]] = None
env_vars: Optional[dict] = None
timeout: Optional[int] = None
variant: Optional[str] = None
def __post_init__(self):
if self.other_args is None:
self.other_args = []
if self.env_vars is None:
self.env_vars = {}
def get_display_name(self) -> str:
if self.variant:
return f"{self.model_path} ({self.variant})"
return self.model_path
def get_mxfp4_models() -> List[ModelConfig]:
"""Get DeepSeek-R1-MXFP4 model configurations for MI35x."""
model_path = get_model_path()
return [
# DeepSeek-R1-MXFP4 basic
ModelConfig(
model_path=model_path,
tp_size=8,
accuracy_threshold=0.93,
timeout=3600,
variant="basic",
other_args=[
"--attention-backend",
"aiter",
"--chunked-prefill-size",
"131072",
"--disable-radix-cache",
"--mem-fraction-static",
"0.85",
"--trust-remote-code",
],
env_vars={"SGLANG_USE_AITER": "1"},
),
# DeepSeek-R1-MXFP4 with MTP (EAGLE)
ModelConfig(
model_path=model_path,
tp_size=8,
accuracy_threshold=0.93,
timeout=3600,
variant="MTP",
other_args=[
"--chunked-prefill-size",
"131072",
"--speculative-algorithm",
"EAGLE",
"--speculative-num-steps",
"3",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"4",
"--mem-fraction-static",
"0.7",
"--trust-remote-code",
],
env_vars={"SGLANG_USE_AITER": "1"},
),
# Note: DP and TC variants are not supported for MXFP4 on MI35x
]
def get_one_example(lines, i, include_answer):
"""Format a single GSM8K example."""
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
"""Get k few-shot examples for prompting."""
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
"""Extract numerical answer from response."""
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(
base_url: str,
num_questions: int = 200,
num_shots: int = 5,
parallel: int = 64,
) -> Tuple[float, float, float]:
"""Run GSM8K few-shot completion benchmark."""
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
assert all(l != INVALID for l in labels)
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
invalid = np.mean(np.array(preds) == INVALID)
return float(acc), float(invalid), float(latency)
class TestDeepSeekR1MXFP4EvalMI35x(unittest.TestCase):
"""DeepSeek-R1-MXFP4 GSM8K Completion Evaluation Test for AMD MI35x."""
@classmethod
def setUpClass(cls):
cls.models = get_mxfp4_models()
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
def test_deepseek_r1_mxfp4_accuracy(self):
"""Test DeepSeek-R1-MXFP4 models with GSM8K completion benchmark."""
# Check if model exists
model_path = get_model_path()
is_local_path = model_path.startswith("/")
if is_local_path and not os.path.exists(model_path):
print(f"\n⏭️ SKIPPING: Local model not found at {model_path}")
self.skipTest(f"Local model not found at {model_path}")
return
if is_local_path:
print(f"📁 Using local model: {model_path}")
else:
print(f"📥 Using HuggingFace model: {model_path}")
all_results = []
summary = "### DeepSeek-R1-MXFP4 Models (MI35x)\n\n"
summary += "| Model | Variant | TP | Accuracy | Threshold | Status |\n"
summary += "| ----- | ------- | -- | -------- | --------- | ------ |\n"
for config in self.models:
display_name = config.get_display_name()
with self.subTest(model=display_name):
print(f"\n{'='*60}")
print(f"Testing: {display_name}")
print(f"{'='*60}")
env = os.environ.copy()
for key, value in config.env_vars.items():
env[key] = value
other_args = list(config.other_args)
other_args.extend(["--tp", str(config.tp_size)])
timeout = config.timeout or DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH
try:
process = popen_launch_server(
model=config.model_path,
base_url=self.base_url,
timeout=timeout,
other_args=other_args,
env=env,
)
try:
acc, invalid, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= config.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
all_results.append(
{
"model": display_name,
"accuracy": acc,
"passed": passed,
}
)
summary += f"| {config.model_path} | {config.variant or 'N/A'} | {config.tp_size} | {acc:.3f} | {config.accuracy_threshold} | {status} |\n"
finally:
kill_process_tree(process.pid)
except Exception as e:
summary += f"| {config.model_path} | {config.variant or 'N/A'} | {config.tp_size} | N/A | {config.accuracy_threshold} | ❌ ERROR |\n"
all_results.append(
{
"model": display_name,
"accuracy": None,
"passed": False,
"error": str(e),
}
)
if is_in_ci():
write_github_step_summary(summary)
failed = [r for r in all_results if not r["passed"]]
if failed:
raise AssertionError(f"Failed models: {[r['model'] for r in failed]}")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,254 @@
"""MI35x GPT-OSS GSM8K Completion Evaluation Test (8-GPU)
Tests GPT-OSS models (openai/gpt-oss-20b, openai/gpt-oss-120b) using
few-shot completion benchmark on MI35x.
Note: MI35x uses openai/* paths, not lmsys/* paths like MI300X.
Registry: nightly-amd-8-gpu-mi35x suite
"""
import ast
import os
# Set HF cache for MI35x
os.environ.setdefault("HF_HOME", "/data2/models/huggingface")
os.environ.setdefault("HF_HUB_CACHE", "/data2/models/huggingface/hub")
import re
import time
import unittest
from dataclasses import dataclass
from typing import List, Optional, Tuple
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - MI35x GPT-OSS accuracy tests (~30 min)
register_amd_ci(est_time=1800, suite="nightly-amd-8-gpu-mi35x", nightly=True)
INVALID = -9999999
@dataclass
class ModelConfig:
"""Configuration for a model to test."""
model_path: str
tp_size: int = 8
accuracy_threshold: float = 0.50
other_args: Optional[List[str]] = None
env_vars: Optional[dict] = None
timeout: Optional[int] = None
def __post_init__(self):
if self.other_args is None:
self.other_args = []
if self.env_vars is None:
self.env_vars = {}
# GPT-OSS models for MI35x (different paths from MI300X)
MI35X_GPT_OSS_MODELS = [
ModelConfig(
model_path="openai/gpt-oss-20b",
tp_size=8,
accuracy_threshold=0.47,
other_args=[
"--chunked-prefill-size",
"130172",
"--max-running-requests",
"128",
"--mem-fraction-static",
"0.85",
"--attention-backend",
"triton",
"--trust-remote-code",
],
env_vars={"SGLANG_USE_AITER": "1"},
),
ModelConfig(
model_path="openai/gpt-oss-120b",
tp_size=8,
accuracy_threshold=0.79,
timeout=900,
other_args=[
"--chunked-prefill-size",
"130172",
"--max-running-requests",
"128",
"--mem-fraction-static",
"0.85",
"--attention-backend",
"triton",
"--trust-remote-code",
],
env_vars={"SGLANG_USE_AITER": "1"},
),
]
def get_one_example(lines, i, include_answer):
"""Format a single GSM8K example."""
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
"""Get k few-shot examples for prompting."""
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
"""Extract numerical answer from response."""
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(
base_url: str,
num_questions: int = 200,
num_shots: int = 5,
parallel: int = 64,
) -> Tuple[float, float, float]:
"""Run GSM8K few-shot completion benchmark."""
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
assert all(l != INVALID for l in labels)
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
invalid = np.mean(np.array(preds) == INVALID)
return float(acc), float(invalid), float(latency)
class TestGptOssEvalMI35x(unittest.TestCase):
"""GPT-OSS GSM8K Completion Evaluation Test for AMD MI35x."""
@classmethod
def setUpClass(cls):
cls.models = MI35X_GPT_OSS_MODELS
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
def test_gpt_oss_accuracy(self):
"""Test GPT-OSS models with GSM8K completion benchmark."""
all_results = []
summary = "### GPT-OSS Models (MI35x)\n\n"
summary += "| Model | TP | Accuracy | Threshold | Status |\n"
summary += "| ----- | -- | -------- | --------- | ------ |\n"
for config in self.models:
with self.subTest(model=config.model_path):
print(f"\n{'='*60}")
print(f"Testing: {config.model_path}")
print(f"{'='*60}")
env = os.environ.copy()
for key, value in config.env_vars.items():
env[key] = value
other_args = list(config.other_args)
other_args.extend(["--tp", str(config.tp_size)])
timeout = config.timeout or DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH
try:
process = popen_launch_server(
model=config.model_path,
base_url=self.base_url,
timeout=timeout,
other_args=other_args,
env=env,
)
try:
acc, invalid, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= config.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
all_results.append(
{
"model": config.model_path,
"accuracy": acc,
"passed": passed,
}
)
summary += f"| {config.model_path} | {config.tp_size} | {acc:.3f} | {config.accuracy_threshold} | {status} |\n"
finally:
kill_process_tree(process.pid)
except Exception as e:
summary += f"| {config.model_path} | {config.tp_size} | N/A | {config.accuracy_threshold} | ❌ ERROR |\n"
all_results.append(
{
"model": config.model_path,
"accuracy": None,
"passed": False,
"error": str(e),
}
)
if is_in_ci():
write_github_step_summary(summary)
failed = [r for r in all_results if not r["passed"]]
if failed:
raise AssertionError(f"Failed models: {[r['model'] for r in failed]}")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,162 @@
"""MI35x GROK1-INT4 GSM8K Completion Evaluation Test (8-GPU)
Tests Grok-1 INT4 (W4A8KV8) model using few-shot completion benchmark on MI35x.
Registry: nightly-amd-accuracy-8-gpu-mi35x-grok1-int4 suite
"""
import ast
import os
import re
import time
import unittest
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - GROK1-INT4 accuracy tests on MI35x (~25 min)
register_amd_ci(
est_time=1500, suite="nightly-amd-accuracy-8-gpu-mi35x-grok1-int4", nightly=True
)
INVALID = -9999999
GROK1_INT4_MODEL_PATH = os.environ.get("GROK1_INT4_MODEL_PATH", "amd/grok-1-W4A8KV8")
GROK1_TOKENIZER_PATH = os.environ.get("GROK1_TOKENIZER_PATH", "Xenova/grok-1-tokenizer")
def get_one_example(lines, i, include_answer):
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(base_url, num_questions=200, num_shots=5, parallel=64):
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
return float(acc), float(latency)
class TestGrok1INT4EvalMI35x(unittest.TestCase):
"""GROK1-INT4 GSM8K Completion Evaluation Test for MI35x."""
@classmethod
def setUpClass(cls):
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
cls.accuracy_threshold = 0.80
def test_grok1_int4_accuracy(self):
"""Test Grok-1 INT4 with GSM8K completion benchmark."""
env = os.environ.copy()
env["RCCL_MSCCL_ENABLE"] = "0"
env["SGLANG_USE_AITER"] = "1"
env["SGLANG_INT4_WEIGHT"] = "1"
other_args = [
"--tp",
"8",
"--quantization",
"fp8",
"--attention-backend",
"aiter",
"--mem-fraction-static",
"0.85",
"--tokenizer-path",
GROK1_TOKENIZER_PATH,
"--trust-remote-code",
]
process = popen_launch_server(
model=GROK1_INT4_MODEL_PATH,
base_url=self.base_url,
timeout=3600,
other_args=other_args,
env=env,
)
try:
acc, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= self.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
summary = f"### GROK1-INT4 (MI35x)\n\n"
summary += f"| Model | Accuracy | Threshold | Status |\n"
summary += f"| ----- | -------- | --------- | ------ |\n"
summary += f"| {GROK1_INT4_MODEL_PATH} | {acc:.3f} | {self.accuracy_threshold} | {status} |\n"
if is_in_ci():
write_github_step_summary(summary)
self.assertGreaterEqual(
acc,
self.accuracy_threshold,
f"Accuracy {acc:.3f} below threshold {self.accuracy_threshold}",
)
finally:
kill_process_tree(process.pid)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,164 @@
"""MI35x GROK2 GSM8K Completion Evaluation Test (8-GPU)
Tests Grok-2 model using few-shot completion benchmark on MI35x.
Registry: nightly-amd-accuracy-8-gpu-mi35x-grok2 suite
"""
import ast
import os
import re
import time
import unittest
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - GROK2 accuracy tests on MI35x (~25 min)
register_amd_ci(
est_time=1500, suite="nightly-amd-accuracy-8-gpu-mi35x-grok2", nightly=True
)
INVALID = -9999999
GROK2_MODEL_PATH = os.environ.get("GROK2_MODEL_PATH", "xai-org/grok-2")
GROK2_TOKENIZER_PATH = os.environ.get(
"GROK2_TOKENIZER_PATH", "alvarobartt/grok-2-tokenizer"
)
def get_one_example(lines, i, include_answer):
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(base_url, num_questions=200, num_shots=5, parallel=64):
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
return float(acc), float(latency)
class TestGrok2EvalMI35x(unittest.TestCase):
"""GROK2 GSM8K Completion Evaluation Test for MI35x."""
@classmethod
def setUpClass(cls):
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
cls.accuracy_threshold = 0.915
def test_grok2_accuracy(self):
"""Test Grok-2 with GSM8K completion benchmark."""
env = os.environ.copy()
env["RCCL_MSCCL_ENABLE"] = "0"
env["SGLANG_USE_AITER"] = "1"
env["SGLANG_INT4_WEIGHT"] = "0"
other_args = [
"--tp",
"8",
"--quantization",
"fp8",
"--attention-backend",
"aiter",
"--mem-fraction-static",
"0.85",
"--tokenizer-path",
GROK2_TOKENIZER_PATH,
"--trust-remote-code",
]
process = popen_launch_server(
model=GROK2_MODEL_PATH,
base_url=self.base_url,
timeout=3600,
other_args=other_args,
env=env,
)
try:
acc, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= self.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
summary = f"### GROK2 (MI35x)\n\n"
summary += f"| Model | Accuracy | Threshold | Status |\n"
summary += f"| ----- | -------- | --------- | ------ |\n"
summary += f"| {GROK2_MODEL_PATH} | {acc:.3f} | {self.accuracy_threshold} | {status} |\n"
if is_in_ci():
write_github_step_summary(summary)
self.assertGreaterEqual(
acc,
self.accuracy_threshold,
f"Accuracy {acc:.3f} below threshold {self.accuracy_threshold}",
)
finally:
kill_process_tree(process.pid)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,309 @@
"""AMD DeepSeek-R1 GSM8K Completion Evaluation Test (8-GPU)
Tests DeepSeek-R1-0528 with multiple configurations (basic, MTP, DP, TC)
using few-shot completion benchmark on MI300X.
Registry: nightly-amd-8-gpu-deepseek-r1 suite
"""
import ast
import os
import re
import time
import unittest
from dataclasses import dataclass
from typing import List, Optional, Tuple
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - DeepSeek-R1 accuracy tests (~120 min)
register_amd_ci(
est_time=7200, suite="nightly-amd-accuracy-8-gpu-deepseek-r1", nightly=True
)
INVALID = -9999999
@dataclass
class ModelConfig:
"""Configuration for a model to test."""
model_path: str
tp_size: int = 8
accuracy_threshold: float = 0.50
other_args: Optional[List[str]] = None
env_vars: Optional[dict] = None
timeout: Optional[int] = None
variant: Optional[str] = None
def __post_init__(self):
if self.other_args is None:
self.other_args = []
if self.env_vars is None:
self.env_vars = {}
def get_display_name(self) -> str:
if self.variant:
return f"{self.model_path} ({self.variant})"
return self.model_path
# DeepSeek-R1 models for MI300X
DEEPSEEK_R1_MODELS = [
# DeepSeek-R1-0528 basic
ModelConfig(
model_path="deepseek-ai/DeepSeek-R1-0528",
tp_size=8,
accuracy_threshold=0.93,
timeout=3600,
variant="basic",
other_args=[
"--attention-backend",
"aiter",
"--chunked-prefill-size",
"131072",
"--disable-radix-cache",
"--mem-fraction-static",
"0.85",
"--trust-remote-code",
],
env_vars={"SGLANG_USE_AITER": "1"},
),
# DeepSeek-R1-0528 with MTP (EAGLE)
ModelConfig(
model_path="deepseek-ai/DeepSeek-R1-0528",
tp_size=8,
accuracy_threshold=0.93,
timeout=3600,
variant="MTP",
other_args=[
"--chunked-prefill-size",
"131072",
"--speculative-algorithm",
"EAGLE",
"--speculative-num-steps",
"3",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"4",
"--mem-fraction-static",
"0.7",
"--trust-remote-code",
],
env_vars={"SGLANG_USE_AITER": "1"},
),
# DeepSeek-R1-0528 with DP attention
ModelConfig(
model_path="deepseek-ai/DeepSeek-R1-0528",
tp_size=8,
accuracy_threshold=0.93,
timeout=3600,
variant="DP",
other_args=[
"--chunked-prefill-size",
"131072",
"--dp-size",
"8",
"--enable-dp-attention",
"--mem-fraction-static",
"0.85",
"--trust-remote-code",
],
env_vars={
"SGLANG_USE_ROCM700A": "1",
"SGLANG_USE_AITER": "1",
},
),
# DeepSeek-R1-0528 with torch compile
ModelConfig(
model_path="deepseek-ai/DeepSeek-R1-0528",
tp_size=8,
accuracy_threshold=0.93,
timeout=7200,
variant="TC",
other_args=[
"--chunked-prefill-size",
"131072",
"--mem-fraction-static",
"0.70",
"--cuda-graph-max-bs",
"8",
"--enable-torch-compile",
"--disable-cuda-graph",
"--trust-remote-code",
],
env_vars={
"SGLANG_USE_ROCM700A": "1",
"SGLANG_USE_AITER": "1",
},
),
]
def get_one_example(lines, i, include_answer):
"""Format a single GSM8K example."""
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
"""Get k few-shot examples for prompting."""
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
"""Extract numerical answer from response."""
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(
base_url: str,
num_questions: int = 200,
num_shots: int = 5,
parallel: int = 64,
) -> Tuple[float, float, float]:
"""Run GSM8K few-shot completion benchmark."""
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
assert all(l != INVALID for l in labels)
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
invalid = np.mean(np.array(preds) == INVALID)
return float(acc), float(invalid), float(latency)
class TestDeepSeekR1EvalAMD(unittest.TestCase):
"""DeepSeek-R1 GSM8K Completion Evaluation Test for AMD MI300X."""
@classmethod
def setUpClass(cls):
cls.models = DEEPSEEK_R1_MODELS
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
def test_deepseek_r1_accuracy(self):
"""Test DeepSeek-R1 models with GSM8K completion benchmark."""
all_results = []
summary = "### DeepSeek-R1 Models (MI300X)\n\n"
summary += "| Model | Variant | TP | Accuracy | Threshold | Status |\n"
summary += "| ----- | ------- | -- | -------- | --------- | ------ |\n"
for config in self.models:
display_name = config.get_display_name()
with self.subTest(model=display_name):
print(f"\n{'='*60}")
print(f"Testing: {display_name}")
print(f"{'='*60}")
env = os.environ.copy()
for key, value in config.env_vars.items():
env[key] = value
other_args = list(config.other_args)
other_args.extend(["--tp", str(config.tp_size)])
timeout = config.timeout or DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH
try:
process = popen_launch_server(
model=config.model_path,
base_url=self.base_url,
timeout=timeout,
other_args=other_args,
env=env,
)
try:
acc, invalid, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= config.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
all_results.append(
{
"model": display_name,
"accuracy": acc,
"passed": passed,
}
)
summary += f"| {config.model_path} | {config.variant or 'N/A'} | {config.tp_size} | {acc:.3f} | {config.accuracy_threshold} | {status} |\n"
finally:
kill_process_tree(process.pid)
except Exception as e:
summary += f"| {config.model_path} | {config.variant or 'N/A'} | {config.tp_size} | N/A | {config.accuracy_threshold} | ❌ ERROR |\n"
all_results.append(
{
"model": display_name,
"accuracy": None,
"passed": False,
"error": str(e),
}
)
if is_in_ci():
write_github_step_summary(summary)
failed = [r for r in all_results if not r["passed"]]
if failed:
raise AssertionError(f"Failed models: {[r['model'] for r in failed]}")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,159 @@
"""AMD DeepSeek-V3.1 GSM8K Completion Evaluation Test (8-GPU)
Tests DeepSeek-V3.1 model using few-shot completion benchmark on MI300X.
Registry: nightly-amd-8-gpu-deepseek-v31 suite
"""
import ast
import os
import re
import time
import unittest
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - DeepSeek-V3.1 accuracy tests (~60 min)
register_amd_ci(
est_time=3600, suite="nightly-amd-accuracy-8-gpu-deepseek-v31", nightly=True
)
INVALID = -9999999
DEEPSEEK_V31_MODEL_PATH = os.environ.get(
"DEEPSEEK_V31_MODEL_PATH", "deepseek-ai/DeepSeek-V3-0324"
)
def get_one_example(lines, i, include_answer):
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(base_url, num_questions=200, num_shots=5, parallel=64):
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
return float(acc), float(latency)
class TestDeepSeekV31EvalAMD(unittest.TestCase):
"""DeepSeek-V3.1 GSM8K Completion Evaluation Test for AMD MI300X."""
@classmethod
def setUpClass(cls):
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
cls.accuracy_threshold = 0.90
def test_deepseek_v31_accuracy(self):
"""Test DeepSeek-V3.1 with GSM8K completion benchmark."""
env = os.environ.copy()
env["SGLANG_USE_AITER"] = "1"
other_args = [
"--tp",
"8",
"--attention-backend",
"aiter",
"--chunked-prefill-size",
"131072",
"--mem-fraction-static",
"0.85",
"--trust-remote-code",
]
process = popen_launch_server(
model=DEEPSEEK_V31_MODEL_PATH,
base_url=self.base_url,
timeout=3600,
other_args=other_args,
env=env,
)
try:
acc, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= self.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
summary = f"### DeepSeek-V3.1 (MI300X)\n\n"
summary += f"| Model | Accuracy | Threshold | Status |\n"
summary += f"| ----- | -------- | --------- | ------ |\n"
summary += f"| {DEEPSEEK_V31_MODEL_PATH} | {acc:.3f} | {self.accuracy_threshold} | {status} |\n"
if is_in_ci():
write_github_step_summary(summary)
self.assertGreaterEqual(
acc,
self.accuracy_threshold,
f"Accuracy {acc:.3f} below threshold {self.accuracy_threshold}",
)
finally:
kill_process_tree(process.pid)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,247 @@
"""AMD GPT-OSS GSM8K Completion Evaluation Test (8-GPU)
Tests GPT-OSS models (lmsys/gpt-oss-20b-bf16, lmsys/gpt-oss-120b-bf16) using
few-shot completion benchmark on MI300X.
Registry: nightly-amd-8-gpu suite
"""
import ast
import os
import re
import time
import unittest
from dataclasses import dataclass
from typing import List, Optional, Tuple
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - GPT-OSS accuracy tests (~30 min)
register_amd_ci(est_time=1800, suite="nightly-amd-accuracy-8-gpu-gpt-oss", nightly=True)
INVALID = -9999999
@dataclass
class ModelConfig:
"""Configuration for a model to test."""
model_path: str
tp_size: int = 8
accuracy_threshold: float = 0.50
other_args: Optional[List[str]] = None
env_vars: Optional[dict] = None
timeout: Optional[int] = None
def __post_init__(self):
if self.other_args is None:
self.other_args = []
if self.env_vars is None:
self.env_vars = {}
# GPT-OSS models for MI300X
GPT_OSS_MODELS = [
ModelConfig(
model_path="lmsys/gpt-oss-20b-bf16",
tp_size=8,
accuracy_threshold=0.47,
other_args=[
"--chunked-prefill-size",
"130172",
"--max-running-requests",
"128",
"--mem-fraction-static",
"0.85",
"--attention-backend",
"triton",
"--trust-remote-code",
],
env_vars={"SGLANG_USE_AITER": "0"},
),
ModelConfig(
model_path="lmsys/gpt-oss-120b-bf16",
tp_size=8,
accuracy_threshold=0.79,
timeout=900,
other_args=[
"--chunked-prefill-size",
"130172",
"--max-running-requests",
"128",
"--mem-fraction-static",
"0.85",
"--attention-backend",
"triton",
"--trust-remote-code",
],
env_vars={"SGLANG_USE_AITER": "0"},
),
]
def get_one_example(lines, i, include_answer):
"""Format a single GSM8K example."""
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
"""Get k few-shot examples for prompting."""
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
"""Extract numerical answer from response."""
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(
base_url: str,
num_questions: int = 200,
num_shots: int = 5,
parallel: int = 64,
) -> Tuple[float, float, float]:
"""Run GSM8K few-shot completion benchmark."""
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
assert all(l != INVALID for l in labels)
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
invalid = np.mean(np.array(preds) == INVALID)
return float(acc), float(invalid), float(latency)
class TestGptOssEvalAMD(unittest.TestCase):
"""GPT-OSS GSM8K Completion Evaluation Test for AMD MI300X."""
@classmethod
def setUpClass(cls):
cls.models = GPT_OSS_MODELS
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
def test_gpt_oss_accuracy(self):
"""Test GPT-OSS models with GSM8K completion benchmark."""
all_results = []
summary = "### GPT-OSS Models (MI300X)\n\n"
summary += "| Model | TP | Accuracy | Threshold | Status |\n"
summary += "| ----- | -- | -------- | --------- | ------ |\n"
for config in self.models:
with self.subTest(model=config.model_path):
print(f"\n{'='*60}")
print(f"Testing: {config.model_path}")
print(f"{'='*60}")
env = os.environ.copy()
for key, value in config.env_vars.items():
env[key] = value
other_args = list(config.other_args)
other_args.extend(["--tp", str(config.tp_size)])
timeout = config.timeout or DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH
try:
process = popen_launch_server(
model=config.model_path,
base_url=self.base_url,
timeout=timeout,
other_args=other_args,
env=env,
)
try:
acc, invalid, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= config.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
all_results.append(
{
"model": config.model_path,
"accuracy": acc,
"passed": passed,
}
)
summary += f"| {config.model_path} | {config.tp_size} | {acc:.3f} | {config.accuracy_threshold} | {status} |\n"
finally:
kill_process_tree(process.pid)
except Exception as e:
summary += f"| {config.model_path} | {config.tp_size} | N/A | {config.accuracy_threshold} | ❌ ERROR |\n"
all_results.append(
{
"model": config.model_path,
"accuracy": None,
"passed": False,
"error": str(e),
}
)
if is_in_ci():
write_github_step_summary(summary)
failed = [r for r in all_results if not r["passed"]]
if failed:
raise AssertionError(f"Failed models: {[r['model'] for r in failed]}")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,162 @@
"""AMD GROK1-FP8 GSM8K Completion Evaluation Test (8-GPU)
Tests Grok-1 FP8 model using few-shot completion benchmark on MI300X.
Registry: nightly-amd-8-gpu-grok1-fp8 suite
"""
import ast
import os
import re
import time
import unittest
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - GROK1-FP8 accuracy tests (~25 min)
register_amd_ci(
est_time=1500, suite="nightly-amd-accuracy-8-gpu-grok1-fp8", nightly=True
)
INVALID = -9999999
GROK1_FP8_MODEL_PATH = os.environ.get("GROK1_MODEL_PATH", "lmzheng/grok-1")
GROK1_TOKENIZER_PATH = os.environ.get("GROK1_TOKENIZER_PATH", "Xenova/grok-1-tokenizer")
def get_one_example(lines, i, include_answer):
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(base_url, num_questions=200, num_shots=5, parallel=64):
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
return float(acc), float(latency)
class TestGrok1FP8EvalAMD(unittest.TestCase):
"""GROK1-FP8 GSM8K Completion Evaluation Test for AMD MI300X."""
@classmethod
def setUpClass(cls):
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
cls.accuracy_threshold = 0.80
def test_grok1_fp8_accuracy(self):
"""Test Grok-1 FP8 with GSM8K completion benchmark."""
env = os.environ.copy()
env["RCCL_MSCCL_ENABLE"] = "0"
env["SGLANG_USE_AITER"] = "1"
env["SGLANG_INT4_WEIGHT"] = "0"
other_args = [
"--tp",
"8",
"--quantization",
"fp8",
"--attention-backend",
"aiter",
"--mem-fraction-static",
"0.85",
"--tokenizer-path",
GROK1_TOKENIZER_PATH,
"--trust-remote-code",
]
process = popen_launch_server(
model=GROK1_FP8_MODEL_PATH,
base_url=self.base_url,
timeout=3600,
other_args=other_args,
env=env,
)
try:
acc, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= self.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
summary = f"### GROK1-FP8 (MI300X)\n\n"
summary += f"| Model | Accuracy | Threshold | Status |\n"
summary += f"| ----- | -------- | --------- | ------ |\n"
summary += f"| {GROK1_FP8_MODEL_PATH} | {acc:.3f} | {self.accuracy_threshold} | {status} |\n"
if is_in_ci():
write_github_step_summary(summary)
self.assertGreaterEqual(
acc,
self.accuracy_threshold,
f"Accuracy {acc:.3f} below threshold {self.accuracy_threshold}",
)
finally:
kill_process_tree(process.pid)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,162 @@
"""AMD GROK1-INT4 GSM8K Completion Evaluation Test (8-GPU)
Tests Grok-1 INT4 (W4A8KV8) model using few-shot completion benchmark on MI300X.
Registry: nightly-amd-8-gpu-grok1-int4 suite
"""
import ast
import os
import re
import time
import unittest
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - GROK1-INT4 accuracy tests (~25 min)
register_amd_ci(
est_time=1500, suite="nightly-amd-accuracy-8-gpu-grok1-int4", nightly=True
)
INVALID = -9999999
GROK1_INT4_MODEL_PATH = os.environ.get("GROK1_INT4_MODEL_PATH", "amd/grok-1-W4A8KV8")
GROK1_TOKENIZER_PATH = os.environ.get("GROK1_TOKENIZER_PATH", "Xenova/grok-1-tokenizer")
def get_one_example(lines, i, include_answer):
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(base_url, num_questions=200, num_shots=5, parallel=64):
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
return float(acc), float(latency)
class TestGrok1INT4EvalAMD(unittest.TestCase):
"""GROK1-INT4 GSM8K Completion Evaluation Test for AMD MI300X."""
@classmethod
def setUpClass(cls):
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
cls.accuracy_threshold = 0.80
def test_grok1_int4_accuracy(self):
"""Test Grok-1 INT4 with GSM8K completion benchmark."""
env = os.environ.copy()
env["RCCL_MSCCL_ENABLE"] = "0"
env["SGLANG_USE_AITER"] = "1"
env["SGLANG_INT4_WEIGHT"] = "1"
other_args = [
"--tp",
"8",
"--quantization",
"fp8",
"--attention-backend",
"aiter",
"--mem-fraction-static",
"0.85",
"--tokenizer-path",
GROK1_TOKENIZER_PATH,
"--trust-remote-code",
]
process = popen_launch_server(
model=GROK1_INT4_MODEL_PATH,
base_url=self.base_url,
timeout=3600,
other_args=other_args,
env=env,
)
try:
acc, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= self.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
summary = f"### GROK1-INT4 (MI300X)\n\n"
summary += f"| Model | Accuracy | Threshold | Status |\n"
summary += f"| ----- | -------- | --------- | ------ |\n"
summary += f"| {GROK1_INT4_MODEL_PATH} | {acc:.3f} | {self.accuracy_threshold} | {status} |\n"
if is_in_ci():
write_github_step_summary(summary)
self.assertGreaterEqual(
acc,
self.accuracy_threshold,
f"Accuracy {acc:.3f} below threshold {self.accuracy_threshold}",
)
finally:
kill_process_tree(process.pid)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,162 @@
"""AMD GROK2 GSM8K Completion Evaluation Test (8-GPU)
Tests Grok-2 model using few-shot completion benchmark on MI300X.
Registry: nightly-amd-8-gpu-grok2 suite
"""
import ast
import os
import re
import time
import unittest
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# Register for AMD CI - GROK2 accuracy tests (~25 min)
register_amd_ci(est_time=1500, suite="nightly-amd-accuracy-8-gpu-grok2", nightly=True)
INVALID = -9999999
GROK2_MODEL_PATH = os.environ.get("GROK2_MODEL_PATH", "xai-org/grok-2")
GROK2_TOKENIZER_PATH = os.environ.get(
"GROK2_TOKENIZER_PATH", "alvarobartt/grok-2-tokenizer"
)
def get_one_example(lines, i, include_answer):
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(base_url, num_questions=200, num_shots=5, parallel=64):
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
return float(acc), float(latency)
class TestGrok2EvalAMD(unittest.TestCase):
"""GROK2 GSM8K Completion Evaluation Test for AMD MI300X."""
@classmethod
def setUpClass(cls):
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
cls.accuracy_threshold = 0.915
def test_grok2_accuracy(self):
"""Test Grok-2 with GSM8K completion benchmark."""
env = os.environ.copy()
env["RCCL_MSCCL_ENABLE"] = "0"
env["SGLANG_USE_AITER"] = "1"
env["SGLANG_INT4_WEIGHT"] = "0"
other_args = [
"--tp",
"8",
"--quantization",
"fp8",
"--attention-backend",
"aiter",
"--mem-fraction-static",
"0.85",
"--tokenizer-path",
GROK2_TOKENIZER_PATH,
"--trust-remote-code",
]
process = popen_launch_server(
model=GROK2_MODEL_PATH,
base_url=self.base_url,
timeout=3600,
other_args=other_args,
env=env,
)
try:
acc, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= self.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
summary = f"### GROK2 (MI300X)\n\n"
summary += f"| Model | Accuracy | Threshold | Status |\n"
summary += f"| ----- | -------- | --------- | ------ |\n"
summary += f"| {GROK2_MODEL_PATH} | {acc:.3f} | {self.accuracy_threshold} | {status} |\n"
if is_in_ci():
write_github_step_summary(summary)
self.assertGreaterEqual(
acc,
self.accuracy_threshold,
f"Accuracy {acc:.3f} below threshold {self.accuracy_threshold}",
)
finally:
kill_process_tree(process.pid)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,287 @@
"""AMD GROK GSM8K Completion Evaluation Test (8-GPU)
Tests GROK models (Grok-1 FP8, Grok-1 INT4, Grok-2) using
few-shot completion benchmark on MI300X.
Registry: nightly-amd-8-gpu-grok suite
"""
import ast
import os
import re
import time
import unittest
from dataclasses import dataclass
from typing import List, Optional, Tuple
import numpy as np
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
from sglang.utils import download_and_cache_file, read_jsonl
# DISABLED: Split into individual files for each model variant
# See: test_grok1_fp8_eval_amd.py, test_grok1_int4_eval_amd.py, test_grok2_eval_amd.py
register_amd_ci(
est_time=2700,
suite="nightly-amd-8-gpu-grok",
nightly=True,
disabled="Split into test_grok1_fp8_eval_amd.py, test_grok1_int4_eval_amd.py, test_grok2_eval_amd.py",
)
INVALID = -9999999
@dataclass
class ModelConfig:
"""Configuration for a model to test."""
model_path: str
tp_size: int = 8
accuracy_threshold: float = 0.50
other_args: Optional[List[str]] = None
env_vars: Optional[dict] = None
tokenizer_path: Optional[str] = None
timeout: Optional[int] = None
def __post_init__(self):
if self.other_args is None:
self.other_args = []
if self.env_vars is None:
self.env_vars = {}
# GROK models for MI300X
GROK_MODELS = [
# GROK1-FP8
ModelConfig(
model_path="lmzheng/grok-1",
tp_size=8,
accuracy_threshold=0.80,
timeout=3600,
tokenizer_path="Xenova/grok-1-tokenizer",
other_args=[
"--quantization",
"fp8",
"--attention-backend",
"aiter",
"--mem-fraction-static",
"0.85",
"--trust-remote-code",
],
env_vars={
"RCCL_MSCCL_ENABLE": "0",
"SGLANG_USE_AITER": "1",
"SGLANG_INT4_WEIGHT": "0",
},
),
# GROK1-INT4
ModelConfig(
model_path="amd/grok-1-W4A8KV8",
tp_size=8,
accuracy_threshold=0.80,
timeout=3600,
tokenizer_path="Xenova/grok-1-tokenizer",
other_args=[
"--quantization",
"fp8",
"--attention-backend",
"aiter",
"--mem-fraction-static",
"0.85",
"--trust-remote-code",
],
env_vars={
"RCCL_MSCCL_ENABLE": "0",
"SGLANG_USE_AITER": "1",
"SGLANG_INT4_WEIGHT": "1",
},
),
# GROK2
ModelConfig(
model_path="xai-org/grok-2",
tp_size=8,
accuracy_threshold=0.915,
timeout=3600,
tokenizer_path="alvarobartt/grok-2-tokenizer",
other_args=[
"--quantization",
"fp8",
"--attention-backend",
"aiter",
"--mem-fraction-static",
"0.85",
"--trust-remote-code",
],
env_vars={
"RCCL_MSCCL_ENABLE": "0",
"SGLANG_USE_AITER": "1",
"SGLANG_INT4_WEIGHT": "0",
},
),
]
def get_one_example(lines, i, include_answer):
"""Format a single GSM8K example."""
ret = "Question: " + lines[i]["question"] + "\nAnswer:"
if include_answer:
ret += " " + lines[i]["answer"]
return ret
def get_few_shot_examples(lines, k):
"""Get k few-shot examples for prompting."""
ret = ""
for i in range(k):
ret += get_one_example(lines, i, True) + "\n\n"
return ret
def get_answer_value(answer_str):
"""Extract numerical answer from response."""
answer_str = answer_str.replace(",", "")
numbers = re.findall(r"\d+", answer_str)
if len(numbers) < 1:
return INVALID
try:
return ast.literal_eval(numbers[-1])
except SyntaxError:
return INVALID
def run_gsm8k_benchmark(
base_url: str,
num_questions: int = 200,
num_shots: int = 5,
parallel: int = 64,
) -> Tuple[float, float, float]:
"""Run GSM8K few-shot completion benchmark."""
import sglang as sgl
from sglang.lang.backend.runtime_endpoint import RuntimeEndpoint
url = "https://raw.githubusercontent.com/openai/grade-school-math/master/grade_school_math/data/test.jsonl"
data_path = download_and_cache_file(url)
lines = list(read_jsonl(data_path))
few_shot_examples = get_few_shot_examples(lines, num_shots)
questions = []
labels = []
for i in range(len(lines[:num_questions])):
questions.append(get_one_example(lines, i, False))
labels.append(get_answer_value(lines[i]["answer"]))
assert all(l != INVALID for l in labels)
arguments = [{"question": q} for q in questions]
@sgl.function
def few_shot_gsm8k(s, question):
s += few_shot_examples + question
s += sgl.gen(
"answer", max_tokens=512, stop=["Question", "Assistant:", "<|separator|>"]
)
backend = RuntimeEndpoint(base_url)
sgl.set_default_backend(backend)
tic = time.perf_counter()
states = few_shot_gsm8k.run_batch(
arguments, temperature=0, num_threads=parallel, progress_bar=True
)
latency = time.perf_counter() - tic
preds = [get_answer_value(states[i]["answer"]) for i in range(len(states))]
acc = np.mean(np.array(preds) == np.array(labels))
invalid = np.mean(np.array(preds) == INVALID)
return float(acc), float(invalid), float(latency)
class TestGrokEvalAMD(unittest.TestCase):
"""GROK GSM8K Completion Evaluation Test for AMD MI300X."""
@classmethod
def setUpClass(cls):
cls.models = GROK_MODELS
cls.base_url = DEFAULT_URL_FOR_TEST
cls.num_questions = int(os.environ.get("GSM8K_NUM_QUESTIONS", "200"))
def test_grok_accuracy(self):
"""Test GROK models with GSM8K completion benchmark."""
all_results = []
summary = "### GROK Models (MI300X)\n\n"
summary += "| Model | TP | Accuracy | Threshold | Status |\n"
summary += "| ----- | -- | -------- | --------- | ------ |\n"
for config in self.models:
with self.subTest(model=config.model_path):
print(f"\n{'='*60}")
print(f"Testing: {config.model_path}")
print(f"{'='*60}")
env = os.environ.copy()
for key, value in config.env_vars.items():
env[key] = value
other_args = list(config.other_args)
other_args.extend(["--tp", str(config.tp_size)])
if config.tokenizer_path:
other_args.extend(["--tokenizer-path", config.tokenizer_path])
timeout = config.timeout or DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH
try:
process = popen_launch_server(
model=config.model_path,
base_url=self.base_url,
timeout=timeout,
other_args=other_args,
env=env,
)
try:
acc, invalid, latency = run_gsm8k_benchmark(
self.base_url, num_questions=self.num_questions
)
passed = acc >= config.accuracy_threshold
status = "✅ PASS" if passed else "❌ FAIL"
all_results.append(
{
"model": config.model_path,
"accuracy": acc,
"passed": passed,
}
)
summary += f"| {config.model_path} | {config.tp_size} | {acc:.3f} | {config.accuracy_threshold} | {status} |\n"
finally:
kill_process_tree(process.pid)
except Exception as e:
summary += f"| {config.model_path} | {config.tp_size} | N/A | {config.accuracy_threshold} | ❌ ERROR |\n"
all_results.append(
{
"model": config.model_path,
"accuracy": None,
"passed": False,
"error": str(e),
}
)
if is_in_ci():
write_github_step_summary(summary)
failed = [r for r in all_results if not r["passed"]]
if failed:
raise AssertionError(f"Failed models: {[r['model'] for r in failed]}")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,324 @@
"""
AMD VLM MMMU Evaluation Test - MI30x Only
This test evaluates Vision-Language Models (VLMs) on the MMMU benchmark on AMD GPUs.
Models are selected based on compatibility with AMD/ROCm platform.
VLMs tested here:
- Qwen2-VL series (Qwen2-VL-7B, Qwen2.5-VL-7B)
- InternVL2 series
- MiniCPM-v series
- deepseek-vl2-small
Note: Some VLMs from the Nvidia test are excluded due to AMD compatibility issues.
Note: This test runs only on MI30x runners (linux-mi325-gpu-2), not on MI35x.
Registry: nightly-amd-vlm suite (2-GPU VLM tests)
"""
import os
import time
import unittest
import warnings
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
write_results_to_json,
)
# Register for AMD CI - VLM MMMU evaluation tests (~120 min)
register_amd_ci(est_time=7200, suite="nightly-amd-accuracy-2-gpu-vlm", nightly=True)
# AMD-verified VLM models with conservative thresholds on 100 MMMU samples
# Format: (model_path, tp_size, accuracy_threshold, extra_args)
AMD_VLM_MODELS = [
# Qwen2-VL series - well supported on AMD
{
"model_path": "Qwen/Qwen2-VL-7B-Instruct",
"tp_size": 1,
"accuracy_threshold": 0.30,
"extra_args": ["--trust-remote-code"],
},
{
"model_path": "Qwen/Qwen2.5-VL-7B-Instruct",
"tp_size": 1,
"accuracy_threshold": 0.33,
"extra_args": ["--trust-remote-code"],
},
# InternVL2 - smaller model, good for testing
{
"model_path": "OpenGVLab/InternVL2_5-2B",
"tp_size": 1,
"accuracy_threshold": 0.29,
"extra_args": ["--trust-remote-code"],
},
# MiniCPM-v - lightweight VLM
{
"model_path": "openbmb/MiniCPM-v-2_6",
"tp_size": 1,
"accuracy_threshold": 0.25,
"extra_args": ["--trust-remote-code"],
},
# DeepSeek VL2 small - MoE VLM
{
"model_path": "deepseek-ai/deepseek-vl2-small",
"tp_size": 1,
"accuracy_threshold": 0.31,
"extra_args": ["--trust-remote-code"],
},
]
# Models that need special handling on AMD
TRITON_ATTENTION_MODELS = {
"deepseek-ai/deepseek-vl2-small", # MoE model
}
# Models known to fail on AMD - exclude from testing
AMD_FAILING_VLM_MODELS = {
# Add models here as they are discovered to fail
}
def get_active_models():
"""Get list of models to test, excluding known failures."""
return [m for m in AMD_VLM_MODELS if m["model_path"] not in AMD_FAILING_VLM_MODELS]
class TestNightlyVLMMmmuEvalAMD(unittest.TestCase):
"""AMD VLM MMMU Evaluation Test.
Tests Vision-Language Models on MMMU benchmark using AMD GPUs.
"""
@classmethod
def setUpClass(cls):
cls.models = get_active_models()
cls.base_url = DEFAULT_URL_FOR_TEST
def test_mmmu_vlm_models(self):
"""Test all configured VLM models on MMMU benchmark."""
warnings.filterwarnings(
"ignore", category=ResourceWarning, message="unclosed.*socket"
)
is_first = True
all_results = []
total_test_start = time.time()
print(f"\n{'='*60}")
print("AMD VLM MMMU Evaluation Test")
print(f"{'='*60}")
print(f"Benchmark: MMMU (100 samples)")
print(f"Models to test: {len(self.models)}")
for m in self.models:
print(f" - {m['model_path']} (TP={m['tp_size']})")
print(f"{'='*60}\n")
for model_config in self.models:
model_path = model_config["model_path"]
tp_size = model_config["tp_size"]
accuracy_threshold = model_config["accuracy_threshold"]
extra_args = model_config.get("extra_args", [])
error_message = None
with self.subTest(model=model_path):
print(f"\n{'='*60}")
print(f"Testing: {model_path} (TP={tp_size})")
print(f"{'='*60}")
model_start = time.time()
startup_time = None
eval_time = None
score = None
# Set AMD-specific environment variables
if model_path in TRITON_ATTENTION_MODELS:
os.environ["SGLANG_USE_AITER"] = "0"
else:
os.environ["SGLANG_USE_AITER"] = "1"
# Build launch args
other_args = list(extra_args)
other_args.extend(["--log-level-http", "warning"])
if tp_size > 1:
other_args.extend(["--tp", str(tp_size)])
# Launch server with timing
print(f"🚀 Launching server...")
server_start = time.time()
process = popen_launch_server(
model=model_path,
base_url=self.base_url,
other_args=other_args,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
)
startup_time = time.time() - server_start
print(f"⏱️ Server startup: {startup_time:.1f}s")
try:
args = SimpleNamespace(
base_url=self.base_url,
model=model_path,
eval_name="mmmu",
num_examples=100,
num_threads=64,
max_tokens=30,
)
# Run evaluation with timing
print(f"📊 Running MMMU evaluation (100 samples)...")
eval_start = time.time()
# Retry up to 3 times
metrics = None
for attempt in range(3):
try:
metrics = run_eval(args)
score = metrics["score"]
if score >= accuracy_threshold:
break
except Exception as e:
print(f" Attempt {attempt + 1} failed with error: {e}")
if attempt == 2:
raise
eval_time = time.time() - eval_start
total_time = time.time() - model_start
# Print results
print(f"\n📈 Results for {model_path}:")
print(
f" Score: {score:.3f} (threshold: {accuracy_threshold:.2f})"
)
print(f"\n⏱️ Runtime breakdown:")
print(f" Server startup: {startup_time:.1f}s")
print(f" Evaluation: {eval_time:.1f}s")
print(f" Total: {total_time:.1f}s")
passed = score >= accuracy_threshold
if passed:
print(f"\n Status: ✅ PASSED")
else:
print(f"\n Status: ❌ FAILED")
write_results_to_json(model_path, metrics, "w" if is_first else "a")
is_first = False
all_results.append(
{
"model": model_path,
"tp_size": tp_size,
"score": score,
"threshold": accuracy_threshold,
"startup_time": startup_time,
"eval_time": eval_time,
"total_time": total_time,
"passed": passed,
"error": None,
}
)
except Exception as e:
error_message = str(e)
total_time = time.time() - model_start
print(f"\n❌ Error evaluating {model_path}: {error_message}")
all_results.append(
{
"model": model_path,
"tp_size": tp_size,
"score": None,
"threshold": accuracy_threshold,
"startup_time": startup_time,
"eval_time": None,
"total_time": total_time,
"passed": False,
"error": error_message,
}
)
finally:
print(f"\n🛑 Stopping server...")
kill_process_tree(process.pid)
# Calculate total test runtime
total_test_time = time.time() - total_test_start
# Generate summary
self._check_results(all_results, total_test_time)
def _check_results(self, results, total_test_time):
"""Check results and generate summary."""
failed_models = []
passed_count = 0
failed_count = 0
summary = (
"| Model | TP | Score | Threshold | Startup | Eval | Total | Status |\n"
)
summary += (
"| ----- | -- | ----- | --------- | ------- | ---- | ----- | ------ |\n"
)
for result in results:
model = result["model"]
score = result["score"]
tp_size = result["tp_size"]
threshold = result["threshold"]
startup_time = result.get("startup_time")
eval_time = result.get("eval_time")
total_time = result.get("total_time")
error = result.get("error")
if error:
status = "❌ ERROR"
failed_count += 1
failed_models.append(f"- {model}: ERROR - {error[:100]}")
elif result["passed"]:
status = "✅ PASS"
passed_count += 1
else:
status = "❌ FAIL"
failed_count += 1
failed_models.append(
f"- {model}: score={score:.4f}, threshold={threshold:.4f}"
)
# Format values
score_str = f"{score:.3f}" if score is not None else "N/A"
startup_str = f"{startup_time:.0f}s" if startup_time is not None else "N/A"
eval_str = f"{eval_time:.0f}s" if eval_time is not None else "N/A"
total_str = f"{total_time:.0f}s" if total_time is not None else "N/A"
summary += f"| {model} | {tp_size} | {score_str} | {threshold:.2f} | {startup_str} | {eval_str} | {total_str} | {status} |\n"
print(f"\n{'='*60}")
print("SUMMARY - AMD VLM MMMU Evaluation")
print(f"{'='*60}")
print(summary)
print(f"\n📊 Final Statistics:")
print(f" Passed: {passed_count}")
print(f" Failed: {failed_count}")
print(
f"\n⏱️ Total test runtime: {total_test_time:.1f}s ({total_test_time/60:.1f} min)"
)
if is_in_ci():
write_github_step_summary(
f"### TestNightlyVLMMmmuEvalAMD\n{summary}\n\n"
f"**Total Runtime:** {total_test_time:.1f}s ({total_test_time/60:.1f} min)"
)
if failed_models:
failure_msg = "\n".join(failed_models)
raise AssertionError(f"The following models failed:\n{failure_msg}")
if __name__ == "__main__":
unittest.main()