) for a message.
+ """
+ return jinja_env.from_string(_message_template).render(
+ role=message["role"],
+ content=message["content"],
+ variant=message.get("variant", None),
+ )
+
+
+jinja_env.globals["message_to_html"] = message_to_html
+
+
+_report_template = """
+
+
+
+
+
+ {% if metrics %}
+
Metrics
+
+
+ | Metric |
+ Value |
+
+
+ | Score |
+ {{ score | float | round(3) }} |
+
+ {% for name, value in metrics.items() %}
+
+ | {{ name }} |
+ {{ value }} |
+
+ {% endfor %}
+
+ {% endif %}
+
Examples
+ {% for html in htmls %}
+ {{ html | safe }}
+
+ {% endfor %}
+
+
+"""
+
+
+def make_report(eval_result: EvalResult) -> str:
+ """
+ Create a standalone HTML report from an EvalResult.
+ """
+ return jinja_env.from_string(_report_template).render(
+ score=eval_result.score,
+ metrics=eval_result.metrics,
+ htmls=eval_result.htmls,
+ )
+
+
+def make_report_from_example_htmls(htmls: List[str]):
+ """
+ Create a standalone HTML report from a list of example htmls
+ """
+ return jinja_env.from_string(_report_template).render(
+ score=None, metrics={}, htmls=htmls
+ )
+
+
+def download_dataset(path: str, url: str) -> None:
+ """Download a dataset from URL to path."""
+ logger.info("Downloading dataset from %s", url)
+ try:
+ response = requests.get(url, stream=True)
+ response.raise_for_status()
+
+ total_size = int(response.headers.get("content-length", 0))
+ block_size = 8192
+
+ with open(path, "wb") as f, tqdm(
+ desc="Downloading",
+ total=total_size,
+ unit="iB",
+ unit_scale=True,
+ unit_divisor=1024,
+ leave=False,
+ ) as progress_bar:
+ for data in response.iter_content(block_size):
+ size = f.write(data)
+ progress_bar.update(size)
+
+ logger.debug("Dataset saved to %s", path)
+ except requests.RequestException as e:
+ raise RuntimeError(f"Failed to download dataset: {e}") from e
+
+
+def set_ulimit(target_soft_limit: int = 65535) -> None:
+ """Set the file descriptor limit for parallel requests."""
+ resource_type = resource.RLIMIT_NOFILE
+ current_soft, current_hard = resource.getrlimit(resource_type)
+
+ if current_soft < target_soft_limit:
+ try:
+ resource.setrlimit(resource_type, (target_soft_limit, current_hard))
+ except ValueError as e:
+ logger.debug("Could not set RLIMIT_NOFILE: %s", e)
diff --git a/sgl-model-gateway/e2e_test/infra/simple_eval_mmlu.py b/sgl-model-gateway/e2e_test/infra/simple_eval_mmlu.py
new file mode 100644
index 000000000..80ec65cee
--- /dev/null
+++ b/sgl-model-gateway/e2e_test/infra/simple_eval_mmlu.py
@@ -0,0 +1,126 @@
+# Adapted from https://github.com/openai/simple-evals/
+"""
+MMLU Evaluation - Measuring Massive Multitask Language Understanding
+Dan Hendrycks et al. https://arxiv.org/abs/2009.03300
+"""
+
+from __future__ import annotations
+
+import random
+import re
+from typing import TYPE_CHECKING
+
+import pandas
+
+from . import simple_eval_common as common
+from .simple_eval_common import (
+ ANSWER_PATTERN_MULTICHOICE,
+ HTML_JINJA,
+ Eval,
+ EvalResult,
+ SingleEvalResult,
+ format_multichoice_question,
+)
+
+if TYPE_CHECKING:
+ from .simple_eval_common import SamplerBase
+
+SUBJECT_TO_CATEGORY = {
+ "abstract_algebra": "stem",
+ "anatomy": "other",
+ "astronomy": "stem",
+ "business_ethics": "other",
+ "clinical_knowledge": "other",
+ "college_biology": "stem",
+ "college_chemistry": "stem",
+ "college_computer_science": "stem",
+ "college_mathematics": "stem",
+ "college_medicine": "other",
+ "college_physics": "stem",
+ "computer_security": "stem",
+ "conceptual_physics": "stem",
+ "econometrics": "social_sciences",
+ "electrical_engineering": "stem",
+ "elementary_mathematics": "stem",
+ "formal_logic": "humanities",
+ "global_facts": "other",
+ "high_school_biology": "stem",
+ "high_school_chemistry": "stem",
+ "high_school_computer_science": "stem",
+ "high_school_european_history": "humanities",
+ "high_school_geography": "social_sciences",
+ "high_school_government_and_politics": "social_sciences",
+ "high_school_macroeconomics": "social_sciences",
+ "high_school_mathematics": "stem",
+ "high_school_microeconomics": "social_sciences",
+ "high_school_physics": "stem",
+ "high_school_psychology": "social_sciences",
+ "high_school_statistics": "stem",
+ "high_school_us_history": "humanities",
+ "high_school_world_history": "humanities",
+ "human_aging": "other",
+ "human_sexuality": "social_sciences",
+ "international_law": "humanities",
+ "jurisprudence": "humanities",
+ "logical_fallacies": "humanities",
+ "machine_learning": "stem",
+ "management": "other",
+ "marketing": "other",
+ "medical_genetics": "other",
+ "miscellaneous": "other",
+ "moral_disputes": "humanities",
+ "moral_scenarios": "humanities",
+ "nutrition": "other",
+ "philosophy": "humanities",
+ "prehistory": "humanities",
+ "professional_accounting": "other",
+ "professional_law": "humanities",
+ "professional_medicine": "other",
+ "professional_psychology": "social_sciences",
+ "public_relations": "social_sciences",
+ "security_studies": "social_sciences",
+ "sociology": "social_sciences",
+ "us_foreign_policy": "social_sciences",
+ "virology": "other",
+ "world_religions": "humanities",
+}
+
+
+class MMLUEval(Eval):
+ """MMLU benchmark evaluation."""
+
+ def __init__(self, filename: str, num_examples: int | None, num_threads: int):
+ df = pandas.read_csv(filename)
+ examples = [row.to_dict() for _, row in df.iterrows()]
+ if num_examples:
+ examples = random.Random(0).sample(examples, num_examples)
+ self.examples = examples
+ self.num_threads = num_threads
+
+ def __call__(self, sampler: "SamplerBase") -> EvalResult:
+ def fn(row: dict) -> SingleEvalResult:
+ prompt_messages = [
+ sampler._pack_message(
+ content=format_multichoice_question(row), role="user"
+ )
+ ]
+ response_text = sampler(prompt_messages)
+ response_text = response_text or ""
+ match = re.search(ANSWER_PATTERN_MULTICHOICE, response_text)
+ extracted_answer = match.group(1) if match else None
+ score = 1.0 if extracted_answer == row["Answer"] else 0.0
+ html = common.jinja_env.from_string(HTML_JINJA).render(
+ prompt_messages=prompt_messages,
+ next_message=dict(content=response_text, role="assistant"),
+ score=score,
+ correct_answer=row["Answer"],
+ extracted_answer=extracted_answer,
+ )
+ convo = prompt_messages + [dict(content=response_text, role="assistant")]
+ category = SUBJECT_TO_CATEGORY.get(row["Subject"], "other")
+ return SingleEvalResult(
+ html=html, score=score, metrics={category: score}, convo=convo
+ )
+
+ results = common.map_with_progress(fn, self.examples, self.num_threads)
+ return common.aggregate_results(results)
diff --git a/sgl-model-gateway/e2e_test/pyproject.toml b/sgl-model-gateway/e2e_test/pyproject.toml
new file mode 100644
index 000000000..00fb6a1d6
--- /dev/null
+++ b/sgl-model-gateway/e2e_test/pyproject.toml
@@ -0,0 +1,30 @@
+[project]
+name = "sgl-model-gateway-e2e-tests"
+version = "0.1.0"
+description = "E2E tests for sgl-model-gateway"
+requires-python = ">=3.9"
+
+dependencies = [
+ "grpcio",
+ "grpcio-health-checking",
+ "httpx",
+ "openai",
+ "pytest",
+ "pytest-rerunfailures",
+]
+
+[project.optional-dependencies]
+dev = [
+ "ruff",
+]
+
+[tool.pytest.ini_options]
+testpaths = ["."]
+markers = [
+ "e2e: mark test as end-to-end test requiring GPU workers",
+ "slow: mark test as slow-running",
+]
+addopts = "-v -s"
+# Explicitly disable live log to avoid "---- live log ----" dividers
+# We configure logging manually in conftest.py
+log_cli = false
diff --git a/sgl-model-gateway/e2e_test/router/test_mmlu.py b/sgl-model-gateway/e2e_test/router/test_mmlu.py
new file mode 100644
index 000000000..13d491e44
--- /dev/null
+++ b/sgl-model-gateway/e2e_test/router/test_mmlu.py
@@ -0,0 +1,78 @@
+"""MMLU evaluation tests for router functionality.
+
+Tests the router's ability to handle MMLU benchmark evaluations across
+different backend configurations (gRPC and HTTP workers).
+
+Usage:
+ # Run with gRPC backend only
+ pytest e2e_test/router/test_mmlu.py -v
+
+ # Run with specific backend
+ pytest e2e_test/router/test_mmlu.py -v -k "grpc"
+"""
+
+from __future__ import annotations
+
+import logging
+from types import SimpleNamespace
+
+import pytest
+from infra import run_eval
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.e2e
+@pytest.mark.parametrize("setup_backend", ["grpc", "http"], indirect=True)
+class TestMMLU:
+ """MMLU evaluation tests using local workers (gRPC and HTTP)."""
+
+ def test_mmlu_basic(self, setup_backend):
+ """Basic MMLU evaluation with score threshold.
+
+ Runs MMLU evaluation with 64 examples and validates that
+ accuracy meets minimum threshold (>= 0.65).
+
+ Note: setup_backend fixture already waits for workers to be ready.
+ """
+ backend, model, client = setup_backend
+ base_url = str(client.base_url).rstrip("/v1")
+
+ args = SimpleNamespace(
+ base_url=base_url,
+ model=model,
+ eval_name="mmlu",
+ num_examples=64,
+ num_threads=32,
+ temperature=0.1,
+ )
+ metrics = run_eval(args)
+
+ assert (
+ metrics["score"] >= 0.65
+ ), f"MMLU score {metrics['score']:.2f} below threshold 0.65"
+ logger.info("MMLU score: %.2f (threshold: 0.65)", metrics["score"])
+
+ def test_mmlu_extended(self, setup_backend):
+ """Extended MMLU evaluation with more examples.
+
+ Runs MMLU with 128 examples for more statistically
+ significant results.
+ """
+ backend, model, client = setup_backend
+ base_url = str(client.base_url).rstrip("/v1")
+
+ args = SimpleNamespace(
+ base_url=base_url,
+ model=model,
+ eval_name="mmlu",
+ num_examples=128,
+ num_threads=64,
+ temperature=0.1,
+ )
+ metrics = run_eval(args)
+
+ assert (
+ metrics["score"] >= 0.65
+ ), f"MMLU score {metrics['score']:.2f} below threshold 0.65"
+ logger.info("MMLU extended score: %.2f (threshold: 0.65)", metrics["score"])
diff --git a/sgl-model-gateway/e2e_test/router/test_pd_mmlu.py b/sgl-model-gateway/e2e_test/router/test_pd_mmlu.py
new file mode 100644
index 000000000..880b73756
--- /dev/null
+++ b/sgl-model-gateway/e2e_test/router/test_pd_mmlu.py
@@ -0,0 +1,61 @@
+"""MMLU evaluation tests for PD (Prefill-Decode) disaggregated routing.
+
+PD disaggregation separates prefill and decode phases across different
+workers for improved throughput and resource utilization.
+
+Requirements:
+ - sgl_kernel package
+ - GPUs: num_prefill + num_decode (default: 2 GPUs for 1+1)
+ - Optional: InfiniBand for high-performance transfers
+
+Configuration via markers:
+ @pytest.mark.model("model-id") # Override default model
+ @pytest.mark.pd(num_prefill=2, num_decode=2) # Custom worker counts
+
+Usage:
+ # Basic (1 prefill + 1 decode)
+ pytest e2e_test/router/test_pd_mmlu.py -v
+
+ # Run specific test
+ pytest e2e_test/router/test_pd_mmlu.py::TestPDMMLU::test_pd_mmlu_basic -v
+"""
+
+from __future__ import annotations
+
+import logging
+from types import SimpleNamespace
+
+import pytest
+from infra import run_eval
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.e2e
+@pytest.mark.parametrize("setup_backend", ["pd"], indirect=True)
+class TestPDMMLU:
+ """MMLU evaluation tests using PD disaggregated routing."""
+
+ def test_pd_mmlu_basic(self, setup_backend):
+ """Basic MMLU evaluation with PD disaggregation.
+
+ Runs MMLU with 1 prefill + 1 decode worker and validates
+ accuracy meets threshold (>= 0.65).
+ """
+ backend, model, client = setup_backend
+ base_url = str(client.base_url).rstrip("/v1")
+
+ args = SimpleNamespace(
+ base_url=base_url,
+ model=model,
+ eval_name="mmlu",
+ num_examples=64,
+ num_threads=32,
+ temperature=0.1,
+ )
+ metrics = run_eval(args)
+
+ assert (
+ metrics["score"] >= 0.65
+ ), f"PD MMLU score {metrics['score']:.2f} below threshold 0.65"
+ logger.info("PD MMLU score: %.2f (threshold: 0.65)", metrics["score"])
diff --git a/sgl-model-gateway/e2e_test/utils.py b/sgl-model-gateway/e2e_test/utils.py
index 9a8640528..7b1bbc23e 100644
--- a/sgl-model-gateway/e2e_test/utils.py
+++ b/sgl-model-gateway/e2e_test/utils.py
@@ -182,49 +182,6 @@ def get_tokenizer_from_processor(processor):
return processor.tokenizer
-# =============================================================================
-# Pytest Utilities
-# =============================================================================
-
-
-def pytest_retry(max_retries: int = 3):
- """Decorator for pytest test functions with retry support.
-
- Args:
- max_retries: Maximum number of retry attempts
-
- Example:
- @pytest_retry(max_retries=3)
- def test_flaky_operation():
- # Test that might occasionally fail
- pass
- """
- import functools
-
- def decorator(func):
- @functools.wraps(func)
- def wrapper(*args, **kwargs):
- last_exception = None
- for attempt in range(max_retries + 1):
- try:
- return func(*args, **kwargs)
- except Exception as e:
- last_exception = e
- if attempt < max_retries:
- logger.info(
- "Test %s failed on attempt %d/%d, retrying...",
- func.__name__,
- attempt + 1,
- max_retries + 1,
- )
- continue
- raise last_exception
-
- return wrapper
-
- return decorator
-
-
# =============================================================================
# Environment Utilities
# =============================================================================