diff --git a/.buildkite/test-nightly.yml b/.buildkite/test-nightly.yml index 1e311c351..c6ece9189 100644 --- a/.buildkite/test-nightly.yml +++ b/.buildkite/test-nightly.yml @@ -225,6 +225,93 @@ steps: mountPath: /dev/shm - name: hf-cache mountPath: /root/.cache/huggingface + env: + - name: HF_HOME + value: /root/.cache/huggingface + - name: HF_TOKEN + valueFrom: + secretKeyRef: + name: hf-token-secret + key: token + nodeSelector: + node.kubernetes.io/instance-type: gpu-h100-sxm + volumes: + - name: devshm + emptyDir: + medium: Memory + - name: hf-cache + hostPath: + path: /mnt/hf-cache + type: DirectoryOrCreate + + - label: ":full_moon: GEBench Accuracy Test with H100" + key: nightly-gebench-accuracy + timeout_in_minutes: 60 + depends_on: upload-nightly-pipeline + if: build.env("NIGHTLY") == "1" || build.pull_request.labels includes "nightly-test" + commands: + - export VLLM_WORKER_MULTIPROC_METHOD=spawn + - pytest -s -v tests/e2e/accuracy/test_gebench_h100_smoke.py --run-level advanced_model --gebench-model Qwen/Qwen-Image-2512 --accuracy-judge-model QuantTrio/Qwen3-VL-30B-A3B-Instruct-AWQ --accuracy-gpu 0 --gebench-port 8093 --accuracy-workers 1 + - buildkite-agent artifact upload "tests/e2e/accuracy/artifacts/gebench_qwen-image-2512/summary*.json" + agents: + queue: "mithril-h100-pool" + plugins: + - kubernetes: + podSpec: + containers: + - image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT + resources: + limits: + nvidia.com/gpu: 1 + volumeMounts: + - name: devshm + mountPath: /dev/shm + - name: hf-cache + mountPath: /root/.cache/huggingface + env: + - name: HF_HOME + value: /root/.cache/huggingface + - name: HF_TOKEN + valueFrom: + secretKeyRef: + name: hf-token-secret + key: token + nodeSelector: + node.kubernetes.io/instance-type: gpu-h100-sxm + volumes: + - name: devshm + emptyDir: + medium: Memory + - name: hf-cache + hostPath: + path: /mnt/hf-cache + type: DirectoryOrCreate + + - label: ":full_moon: GEdit-Bench Accuracy Test with H100" + key: nightly-gedit-bench-accuracy + timeout_in_minutes: 60 + depends_on: upload-nightly-pipeline + if: build.env("NIGHTLY") == "1" || build.pull_request.labels includes "nightly-test" + commands: + - export VLLM_WORKER_MULTIPROC_METHOD=spawn + - pytest -s -v tests/e2e/accuracy/test_gedit_bench_h100_smoke.py --run-level advanced_model --gedit-model Qwen/Qwen-Image-Edit --accuracy-judge-model QuantTrio/Qwen3-VL-30B-A3B-Instruct-AWQ --accuracy-gpu 0 --gedit-port 8093 --gedit-samples-per-group 20 --accuracy-workers 1 + - buildkite-agent artifact upload "tests/e2e/accuracy/artifacts/gedit_scores_qwen-image-edit/qwen-image-edit_all_all_vie_score_*.csv" + - buildkite-agent artifact upload "tests/e2e/accuracy/artifacts/gedit_scores_qwen-image-edit/qwen-image-edit_all_all_summary_*.json" + agents: + queue: "mithril-h100-pool" + plugins: + - kubernetes: + podSpec: + containers: + - image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT + resources: + limits: + nvidia.com/gpu: 1 + volumeMounts: + - name: devshm + mountPath: /dev/shm + - name: hf-cache + mountPath: /root/.cache/huggingface env: - name: HF_HOME value: /root/.cache/huggingface diff --git a/benchmarks/README.md b/benchmarks/README.md index 5ff270b7f..68ffd40ef 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -15,6 +15,8 @@ This document explains the benchmark architecture across all benchmark assets in ## Directory layout - `benchmarks/build_dataset/` — dataset prep utilities (e.g., SeedTTS top100). - `benchmarks//vllm_omni/` — vLLM-Omni pipeline benchmarks, logs, outputs. +- `benchmarks/accuracy/` — accuracy benchmark integrations that adapt external + benchmark suites to vLLM-Omni serving and evaluation flows. - Add new tasks under `benchmarks//...` with the same pattern: `transformers/`, `vllm_omni/`, task-specific README, and (optionally) dataset prep notes. ## Reference workflows diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 000000000..79917e818 --- /dev/null +++ b/benchmarks/__init__.py @@ -0,0 +1 @@ +"""Benchmark helpers and runnable benchmark entrypoints.""" diff --git a/benchmarks/accuracy/README.md b/benchmarks/accuracy/README.md new file mode 100644 index 000000000..0d73215b6 --- /dev/null +++ b/benchmarks/accuracy/README.md @@ -0,0 +1,27 @@ +# Accuracy Benchmarks + +This directory hosts accuracy benchmark integrations that run entirely through a +local `vllm-omni serve` deployment. + +Current integrations: + +- `text_to_image/`: GEBench generation + local judge scoring flow. +- `image_to_image/`: GEdit-Bench generation + local VIEScore-style scoring flow. + +Design notes: + +- Generation is executed through the OpenAI-compatible endpoints exposed by + `vllm-omni serve`. +- Evaluation is also executed through a local OpenAI-compatible judge model + served by `vllm-omni`. +- Both generation and judge requests accept either `http://host:port` or + `http://host:port/v1`. +- Output directory layout intentionally stays close to the upstream repos. + +Test guidance: + +- Local static/self-checks live in `tests/benchmarks/test_accuracy_bench_utils.py`. +- End-to-end generation/evaluation should be validated in a remote GPU + environment. In the current repo marker system there is `L4` but no `L5` + marker, so benchmark smoke tests should be wired as `advanced_model + + benchmark + L4` when GPU capacity is available. diff --git a/benchmarks/accuracy/__init__.py b/benchmarks/accuracy/__init__.py new file mode 100644 index 000000000..029ec71da --- /dev/null +++ b/benchmarks/accuracy/__init__.py @@ -0,0 +1 @@ +"""Accuracy benchmark integrations for vLLM-Omni.""" diff --git a/benchmarks/accuracy/common.py b/benchmarks/accuracy/common.py new file mode 100644 index 000000000..bce0b2359 --- /dev/null +++ b/benchmarks/accuracy/common.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +import base64 +import io +import json +from pathlib import Path +from typing import Any + +import requests +from PIL import Image + + +def ensure_dir(path: Path) -> Path: + path.mkdir(parents=True, exist_ok=True) + return path + + +def load_json(path: Path) -> dict[str, Any]: + with path.open("r", encoding="utf-8") as handle: + return json.load(handle) + + +def write_json(path: Path, payload: dict[str, Any]) -> None: + ensure_dir(path.parent) + with path.open("w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2, ensure_ascii=False) + + +def save_image(path: Path, image: Image.Image) -> None: + ensure_dir(path.parent) + image.save(path) + + +def find_first_image(folder: Path, stem: str | None = None) -> Path | None: + patterns = [f"{stem}.*"] if stem else ["*.png", "*.jpg", "*.jpeg", "*.webp"] + for pattern in patterns: + for candidate in sorted(folder.glob(pattern)): + if candidate.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}: + return candidate + return None + + +def extract_json_object(raw_text: str) -> dict[str, Any]: + raw_text = raw_text.strip() + delimiter = "||V^=^V||" + if raw_text.count(delimiter) >= 2: + start = raw_text.find(delimiter) + len(delimiter) + end = raw_text.rfind(delimiter) + raw_text = raw_text[start:end].strip() + + start = raw_text.find("{") + end = raw_text.rfind("}") + if start == -1 or end == -1 or end < start: + raise ValueError(f"Could not find JSON object in: {raw_text[:200]}") + return json.loads(raw_text[start : end + 1]) + + +def build_openai_url(base_url: str, api_path: str) -> str: + base = base_url.rstrip("/") + normalized_path = api_path if api_path.startswith("/") else f"/{api_path}" + if base.endswith(normalized_path): + return base + if base.endswith("/v1"): + return f"{base}{normalized_path}" + return f"{base}/v1{normalized_path}" + + +def pil_to_base64(image: Image.Image, image_format: str = "PNG") -> str: + buffer = io.BytesIO() + image.save(buffer, format=image_format) + return base64.b64encode(buffer.getvalue()).decode("utf-8") + + +def pil_to_data_url(image: Image.Image, image_format: str = "PNG") -> str: + return f"data:image/{image_format.lower()};base64,{pil_to_base64(image, image_format=image_format)}" + + +def decode_base64_image(encoded: str) -> Image.Image: + image = Image.open(io.BytesIO(base64.b64decode(encoded))) + image.load() + return image.convert("RGB") + + +def pil_to_png_bytes(image: Image.Image) -> bytes: + buffer = io.BytesIO() + image.save(buffer, format="PNG") + return buffer.getvalue() + + +class VllmOmniImageClient: + """Thin OpenAI-compatible image client for vLLM-Omni serving.""" + + def __init__(self, base_url: str, api_key: str = "EMPTY", timeout: int = 600): + self.base_url = base_url.rstrip("/") + self.api_key = api_key + self.timeout = timeout + + @property + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + } + + def generate_text_to_image( + self, + *, + model: str, + prompt: str, + width: int, + height: int, + num_inference_steps: int = 20, + guidance_scale: float | None = None, + seed: int | None = None, + output_compression: int | None = None, + ) -> Image.Image: + payload: dict[str, Any] = { + "model": model, + "prompt": prompt, + "n": 1, + "size": f"{width}x{height}", + "response_format": "b64_json", + "num_inference_steps": num_inference_steps, + } + if guidance_scale is not None: + payload["guidance_scale"] = guidance_scale + if seed is not None: + payload["seed"] = seed + if output_compression is not None: + payload["output_compression"] = output_compression + + response = requests.post( + build_openai_url(self.base_url, "/images/generations"), + json=payload, + headers=self._headers, + timeout=self.timeout, + ) + response.raise_for_status() + return decode_base64_image(response.json()["data"][0]["b64_json"]) + + def generate_image_edit( + self, + *, + model: str, + prompt: str, + images: Image.Image | list[Image.Image], + width: int, + height: int, + num_inference_steps: int = 20, + guidance_scale: float | None = None, + seed: int | None = None, + negative_prompt: str | None = None, + output_compression: int | None = None, + ) -> Image.Image: + if not isinstance(images, list): + images = [images] + data: dict[str, Any] = { + "model": model, + "prompt": prompt, + "n": 1, + "size": f"{width}x{height}", + "response_format": "b64_json", + "num_inference_steps": str(num_inference_steps), + } + if guidance_scale is not None: + data["guidance_scale"] = str(guidance_scale) + if seed is not None: + data["seed"] = str(seed) + if negative_prompt: + data["negative_prompt"] = negative_prompt + if output_compression is not None: + data["output_compression"] = str(output_compression) + + files = [ + ( + "image[]" if len(images) > 1 else "image", + (f"image_{index}.png", pil_to_png_bytes(image), "image/png"), + ) + for index, image in enumerate(images) + ] + + edit_paths = ["/images/edits", "/images/edit"] + last_response: requests.Response | None = None + for api_path in edit_paths: + response = requests.post( + build_openai_url(self.base_url, api_path), + data=data, + files=files, + headers={"Authorization": f"Bearer {self.api_key}"}, + timeout=self.timeout, + ) + last_response = response + if response.status_code != 404: + response.raise_for_status() + return decode_base64_image(response.json()["data"][0]["b64_json"]) + + assert last_response is not None + last_response.raise_for_status() + raise ValueError("No image payload returned from image edit endpoint") diff --git a/benchmarks/accuracy/image_to_image/README.md b/benchmarks/accuracy/image_to_image/README.md new file mode 100644 index 000000000..ee1d58f10 --- /dev/null +++ b/benchmarks/accuracy/image_to_image/README.md @@ -0,0 +1,103 @@ +# GEdit-Bench on vLLM-Omni + +This integration adapts the upstream `stepfun-ai/Step1X-Edit/GEdit-Bench` +evaluation flow into `vllm-omni/benchmarks/accuracy/image_to_image`. + +Upstream mapping: + +- `run_gedit_score.py` -> `run_gedit_bench.py evaluate` +- `calculate_statistics.py` -> `run_gedit_bench.py summarize` +- upstream output layout under `results//fullset/...` is preserved + +What changed: + +- The upstream repo mainly ships evaluation scripts. This integration adds a + generation runner that uses the local `vllm-omni` OpenAI-compatible + `/v1/images/edits` endpoint to produce benchmark outputs in the expected + directory structure. +- The evaluator keeps the same VIEScore-style decomposition: + - `semantics_score` + - `quality_score` + - `overall_score = sqrt(semantics_score * quality_score)` +- Judge calls are routed to a local OpenAI-compatible model served by + `vllm-omni`, not a remote provider. + +Dataset: + +- Default `--dataset-ref` is `stepfun-ai/GEdit-Bench` +- You can also pass a local dataset directory previously saved with + Hugging Face `datasets` + +Example usage: + +```bash +python benchmarks/accuracy/image_to_image/run_gedit_bench.py generate \ + --output-root benchmarks/accuracy/image_to_image/results \ + --base-url http://127.0.0.1:8000 \ + --model Qwen/Qwen-Image-Edit \ + --model-name qwen_image_edit \ + --dataset-ref stepfun-ai/GEdit-Bench \ + --task-type all \ + --instruction-language en +``` + +```bash +python benchmarks/accuracy/image_to_image/run_gedit_bench.py evaluate \ + --output-root benchmarks/accuracy/image_to_image/results \ + --model-name qwen_image_edit \ + --save-dir benchmarks/accuracy/image_to_image/scores \ + --dataset-ref stepfun-ai/GEdit-Bench \ + --judge-base-url http://127.0.0.1:8000 \ + --judge-model Qwen/Qwen2.5-VL-7B-Instruct \ + --judge-api-key EMPTY +``` + +```bash +python benchmarks/accuracy/image_to_image/run_gedit_bench.py summarize \ + --csv-path benchmarks/accuracy/image_to_image/scores/qwen_image_edit_all_all_vie_score.csv \ + --language en +``` + +Example summary output: + +```json +{ + "language": "all", + "languages": { + "en": { + "overall": {"count": 110, "Q_SC": 6.58, "Q_PQ": 5.89, "Q_O": 5.86}, + "intersection": {"count": 78, "Q_SC": 6.50, "Q_PQ": 5.66, "Q_O": 5.65} + }, + "cn": { + "overall": {"count": 110, "Q_SC": 6.90, "Q_PQ": 5.78, "Q_O": 6.11}, + "intersection": {"count": 63, "Q_SC": 7.22, "Q_PQ": 5.59, "Q_O": 6.28} + } + } +} +``` + +Example generated images to inspect: + +- `benchmarks/accuracy/image_to_image/results/qwen_image_edit/fullset/background_change/en/.png` +- `benchmarks/accuracy/image_to_image/results/qwen_image_edit/fullset/text_change/en/.png` +- `benchmarks/accuracy/image_to_image/results/qwen_image_edit/fullset/subject-replace/cn/.png` + +Example score artifacts to inspect together with the images: + +- `benchmarks/accuracy/image_to_image/scores/qwen_image_edit_all_all_vie_score.csv` +- `benchmarks/accuracy/image_to_image/scores/qwen_image_edit_all_all_summary.json` + +What to expect: + +- `Q_SC` measures instruction following and content preservation. +- `Q_PQ` measures image naturalness and artifact quality. +- `Q_O` is the combined overall score; higher is better. +- `overall.count` is the number of evaluated samples for that language, while `intersection.count` is the subset with `intersection_exist == True`. + +Notes: + +- This flow requires the optional Hugging Face `datasets` package. +- `generate` writes `generation_manifest.json` with local output coverage. +- The current repo marker set exposes `L4` but not `L5`, so if you promote an + end-to-end smoke test into CI, use the existing `advanced_model`, `benchmark`, + and `L4` markers or introduce a new repo-wide marker explicitly first. diff --git a/benchmarks/accuracy/image_to_image/__init__.py b/benchmarks/accuracy/image_to_image/__init__.py new file mode 100644 index 000000000..1b9b2d44d --- /dev/null +++ b/benchmarks/accuracy/image_to_image/__init__.py @@ -0,0 +1 @@ +"""GEdit-Bench integration for vLLM-Omni.""" diff --git a/benchmarks/accuracy/image_to_image/gedit_bench.py b/benchmarks/accuracy/image_to_image/gedit_bench.py new file mode 100644 index 000000000..adb072c0a --- /dev/null +++ b/benchmarks/accuracy/image_to_image/gedit_bench.py @@ -0,0 +1,718 @@ +from __future__ import annotations + +import argparse +import csv +import json +import logging +import math +import statistics +from collections import defaultdict +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timezone +from io import BytesIO +from pathlib import Path +from typing import Any + +import requests +from PIL import Image +from tqdm.auto import tqdm + +from benchmarks.accuracy.common import ( + VllmOmniImageClient, + build_openai_url, + extract_json_object, + write_json, +) + +GROUPS = [ + "background_change", + "color_alter", + "material_alter", + "motion_change", + "ps_human", + "style_change", + "subject-add", + "subject-remove", + "subject-replace", + "text_change", + "tone_transfer", +] +DEFAULT_SAMPLES_PER_GROUP = 10 +logger = logging.getLogger(__name__) + + +def infer_model_name(model: str) -> str: + normalized = model.rstrip("/\\") + name = Path(normalized).name + return name or normalized + + +def resolve_model_name(*, model_name: str | None, model: str | None = None, output_root: Path | None = None) -> str: + if model_name: + return model_name + if model: + return infer_model_name(model) + if output_root is not None: + candidates = sorted(path.name for path in output_root.iterdir() if path.is_dir()) + if len(candidates) == 1: + return candidates[0] + if not candidates: + raise ValueError(f"Could not infer model-name from empty output root: {output_root}") + raise ValueError( + f"Could not infer model-name from output root {output_root}; multiple candidates found: {candidates}" + ) + raise ValueError("model-name is required when it cannot be inferred from model or output-root") + + +def parse_score_payload(raw_text: str) -> dict[str, Any]: + try: + parsed = extract_json_object(raw_text) + except Exception: + stripped = raw_text.strip() + if stripped.startswith("[") and stripped.endswith("]"): + parsed = {"score": json.loads(stripped), "reasoning": ""} + else: + raise + score = parsed.get("score", []) + if not isinstance(score, list): + score = [score] + parsed["score"] = [int(value) for value in score] + return parsed + + +def _utc_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + + +def summarize_generated_records(records: list[dict[str, Any]]) -> dict[str, Any]: + by_task: dict[str, list[dict[str, Any]]] = defaultdict(list) + by_language: dict[str, list[dict[str, Any]]] = defaultdict(list) + for record in records: + by_task[record["task_type"]].append(record) + by_language[record["instruction_language"]].append(record) + + return { + "count": len(records), + "by_task": { + task_type: { + "count": len(rows), + "samples": sorted(row["key"] for row in rows), + } + for task_type, rows in sorted(by_task.items()) + }, + "by_language": { + language: { + "count": len(rows), + "samples": sorted(row["key"] for row in rows), + } + for language, rows in sorted(by_language.items()) + }, + } + + +def select_balanced_gedit_rows( + rows: list[dict[str, Any]], + *, + task_type: str = "all", + instruction_language: str = "all", + samples_per_group: int | None, +) -> list[dict[str, Any]]: + filtered_rows = [] + for row in rows: + if task_type != "all" and row["task_type"] != task_type: + continue + if instruction_language != "all" and row["instruction_language"] != instruction_language: + continue + filtered_rows.append(row) + + if samples_per_group is None: + return filtered_rows + + def _select_rows_for_group(group_rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + if instruction_language != "all": + return group_rows[:samples_per_group] + + rows_by_language: dict[str, list[dict[str, Any]]] = defaultdict(list) + for row in group_rows: + rows_by_language[str(row.get("instruction_language", "")).strip()].append(row) + + ordered_languages = ["en", "cn"] + per_language_quota = samples_per_group // max(len(ordered_languages), 1) + remainder = samples_per_group % max(len(ordered_languages), 1) + + selected_rows: list[dict[str, Any]] = [] + leftovers: list[dict[str, Any]] = [] + for index, language in enumerate(ordered_languages): + language_rows = rows_by_language.get(language, []) + quota = per_language_quota + (1 if index < remainder else 0) + selected_rows.extend(language_rows[:quota]) + leftovers.extend(language_rows[quota:]) + + if len(selected_rows) < samples_per_group: + selected_rows.extend(leftovers[: samples_per_group - len(selected_rows)]) + + return selected_rows + + if task_type != "all": + return _select_rows_for_group(filtered_rows) + + grouped_rows: dict[str, list[dict[str, Any]]] = defaultdict(list) + for row in filtered_rows: + grouped_rows[row["task_type"]].append(row) + + selected: list[dict[str, Any]] = [] + for group in GROUPS: + selected.extend(_select_rows_for_group(grouped_rows.get(group, []))) + return selected + + +def summarize_gedit_rows(rows: list[dict[str, Any]], language: str = "all") -> dict[str, Any]: + return summarize_gedit_rows_with_backbone(rows, language=language) + + +def _mean_or_none(values: list[float]) -> float | None: + return statistics.fmean(values) if values else None + + +def _to_q_metrics(section: dict[str, float | None]) -> dict[str, float | None]: + return { + "count": section.get("count"), + "Q_SC": section.get("avg_semantics"), + "Q_PQ": section.get("avg_quality"), + "Q_O": section.get("avg_overall"), + } + + +def _summarize_gedit_rows_single_language( + rows: list[dict[str, Any]], + *, + language: str, +) -> dict[str, Any]: + filtered_rows = [row for row in rows if str(row.get("instruction_language", "")).strip() == language] + + def _to_bool(value: Any) -> bool: + if isinstance(value, bool): + return value + return str(value).strip().lower() in {"1", "true", "yes"} + + per_group: dict[str, dict[str, float | None]] = {} + intersection_group: dict[str, dict[str, float | None]] = {} + + for group in GROUPS: + group_rows = [row for row in filtered_rows if row.get("task_type") == group] + semantics = [float(row["semantics_score"]) for row in group_rows] + quality = [float(row["quality_score"]) for row in group_rows] + overall = [math.sqrt(float(row["semantics_score"]) * float(row["quality_score"])) for row in group_rows] + + per_group[group] = { + "count": len(group_rows), + "avg_semantics": _mean_or_none(semantics), + "avg_quality": _mean_or_none(quality), + "avg_overall": _mean_or_none(overall), + } + + intersection_rows = [row for row in group_rows if _to_bool(row.get("intersection_exist", False))] + intersection_semantics = [float(row["semantics_score"]) for row in intersection_rows] + intersection_quality = [float(row["quality_score"]) for row in intersection_rows] + intersection_overall = [ + math.sqrt(float(row["semantics_score"]) * float(row["quality_score"])) for row in intersection_rows + ] + intersection_group[group] = { + "count": len(intersection_rows), + "avg_semantics": _mean_or_none(intersection_semantics), + "avg_quality": _mean_or_none(intersection_quality), + "avg_overall": _mean_or_none(intersection_overall), + } + + overall_section = { + "count": len(filtered_rows), + "avg_semantics": _mean_or_none( + [score["avg_semantics"] for score in per_group.values() if score["avg_semantics"] is not None] + ), + "avg_quality": _mean_or_none( + [score["avg_quality"] for score in per_group.values() if score["avg_quality"] is not None] + ), + "avg_overall": _mean_or_none( + [score["avg_overall"] for score in per_group.values() if score["avg_overall"] is not None] + ), + } + intersection_section = { + "count": sum(score["count"] for score in intersection_group.values()), + "avg_semantics": _mean_or_none( + [score["avg_semantics"] for score in intersection_group.values() if score["avg_semantics"] is not None] + ), + "avg_quality": _mean_or_none( + [score["avg_quality"] for score in intersection_group.values() if score["avg_quality"] is not None] + ), + "avg_overall": _mean_or_none( + [score["avg_overall"] for score in intersection_group.values() if score["avg_overall"] is not None] + ), + } + + return { + "language": language, + "by_group": {group: _to_q_metrics(section) for group, section in per_group.items()}, + "overall": _to_q_metrics(overall_section), + "intersection": _to_q_metrics(intersection_section), + } + + +def summarize_gedit_rows_with_backbone( + rows: list[dict[str, Any]], + *, + language: str = "all", +) -> dict[str, Any]: + if language == "all": + return { + "language": "all", + "languages": { + single_language: _summarize_gedit_rows_single_language( + rows, + language=single_language, + ) + for single_language in ["en", "cn"] + }, + } + + return _summarize_gedit_rows_single_language(rows, language=language) + + +def _require_datasets(): + try: + from datasets import load_dataset, load_from_disk + except ImportError as exc: + raise ImportError("GEdit-Bench requires the optional `datasets` package.") from exc + return load_dataset, load_from_disk + + +def _load_gedit_dataset(dataset_ref: str): + load_dataset, load_from_disk = _require_datasets() + dataset_path = Path(dataset_ref) + if dataset_path.exists(): + if (dataset_path / "state.json").exists() and (dataset_path / "dataset_info.json").exists(): + return load_from_disk(str(dataset_path)) + return load_dataset(str(dataset_path)) + return load_dataset(dataset_ref) + + +def _resolve_gedit_split(dataset_obj: Any) -> Any: + if isinstance(dataset_obj, dict): + if "train" in dataset_obj: + return dataset_obj["train"] + return dataset_obj + try: + return dataset_obj["train"] + except Exception: + return dataset_obj + + +def _to_pil_image(value: Any) -> Image.Image: + if isinstance(value, Image.Image): + return value.convert("RGB") + if isinstance(value, dict) and "bytes" in value: + image = Image.open(BytesIO(value["bytes"])) + image.load() + return image.convert("RGB") + raise TypeError(f"Unsupported image payload type: {type(value)!r}") + + +class LocalVIEScorer: + def __init__(self, *, base_url: str, api_key: str, model: str, timeout: int = 600): + self.base_url = base_url.rstrip("/") + self.api_key = api_key + self.model = model + self.timeout = timeout + self.sc_prompt = ( + "You are evaluating image editing quality.\n" + "Two images are provided: the source image and the edited image.\n" + 'Return JSON only in the format {"score": [edit_success, content_preservation], ' + '"reasoning": "..."}.\n' + "Each score must be an integer from 0 to 10.\n" + "Editing instruction: " + ) + self.pq_prompt = ( + "You are evaluating image quality.\n" + 'Return JSON only in the format {"score": [naturalness, artifact_free], "reasoning": "..."}.\n' + "Each score must be an integer from 0 to 10." + ) + + def _request(self, prompt: str, images: list[Image.Image]) -> dict[str, Any]: + from benchmarks.accuracy.common import pil_to_data_url + + content: list[dict[str, Any]] = [{"type": "text", "text": prompt}] + for image in images: + content.append({"type": "image_url", "image_url": {"url": pil_to_data_url(image)}}) + + response = requests.post( + build_openai_url(self.base_url, "/chat/completions"), + json={ + "model": self.model, + "messages": [{"role": "user", "content": content}], + "temperature": 0, + }, + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + }, + timeout=self.timeout, + ) + response.raise_for_status() + message_content = response.json()["choices"][0]["message"]["content"] + if isinstance(message_content, list): + text = "\n".join(part.get("text", "") for part in message_content if part.get("type") == "text") + else: + text = str(message_content) + return parse_score_payload(text) + + def evaluate(self, source_image: Image.Image, edited_image: Image.Image, instruction: str) -> dict[str, float]: + sc_payload = self._request(self.sc_prompt.replace("", instruction), [source_image, edited_image]) + pq_payload = self._request(self.pq_prompt, [edited_image]) + semantics = float(min(sc_payload["score"])) if sc_payload["score"] else 0.0 + quality = float(min(pq_payload["score"])) if pq_payload["score"] else 0.0 + overall = math.sqrt(semantics * quality) + return { + "semantics_score": semantics, + "quality_score": quality, + "overall_score": overall, + } + + +class GEditBenchRunner: + def __init__( + self, + *, + dataset_ref: str, + output_root: Path, + base_url: str, + model: str, + api_key: str = "EMPTY", + width: int = 512, + height: int = 512, + num_inference_steps: int = 20, + guidance_scale: float | None = None, + seed: int | None = 42, + ): + self.dataset_ref = dataset_ref + self.output_root = output_root + self.model = model + self.width = width + self.height = height + self.num_inference_steps = num_inference_steps + self.guidance_scale = guidance_scale + self.seed = seed + self.client = VllmOmniImageClient(base_url=base_url, api_key=api_key) + + def generate( + self, + *, + model_name: str, + task_type: str = "all", + instruction_language: str = "all", + workers: int = 1, + max_samples: int | None = None, + samples_per_group: int | None = None, + ) -> list[dict[str, Any]]: + dataset = _resolve_gedit_split(_load_gedit_dataset(self.dataset_ref)) + rows = select_balanced_gedit_rows( + list(dataset), + task_type=task_type, + instruction_language=instruction_language, + samples_per_group=samples_per_group, + ) + if max_samples is not None: + rows = rows[:max_samples] + + outputs: list[dict[str, Any]] = [] + total = len(rows) + if workers <= 1: + with tqdm(total=total, desc="GEdit generate", unit="sample") as progress: + for item in rows: + result = self._safe_generate_one(model_name, item) + if result: + outputs.append(result) + progress.update(1) + return outputs + + with tqdm(total=total, desc="GEdit generate", unit="sample") as progress: + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [executor.submit(self._safe_generate_one, model_name, item) for item in rows] + for future in as_completed(futures): + result = future.result() + if result: + outputs.append(result) + progress.update(1) + return outputs + + def _safe_generate_one(self, model_name: str, item: dict[str, Any]) -> dict[str, Any] | None: + try: + return self._generate_one(model_name, item) + except Exception: + logger.exception("Failed to generate GEdit-Bench sample %s", item.get("key", "")) + return None + + def _generate_one(self, model_name: str, item: dict[str, Any]) -> dict[str, Any] | None: + output_path = ( + self.output_root + / model_name + / "fullset" + / item["task_type"] + / item["instruction_language"] + / f"{item['key']}.png" + ) + output_path.parent.mkdir(parents=True, exist_ok=True) + if output_path.exists(): + return { + "task_type": item["task_type"], + "instruction_language": item["instruction_language"], + "key": item["key"], + "output_path": str(output_path), + } + + source_image = _to_pil_image(item["input_image_raw"]) + edited_image = self.client.generate_image_edit( + model=self.model, + prompt=item["instruction"], + images=source_image, + width=self.width, + height=self.height, + num_inference_steps=self.num_inference_steps, + guidance_scale=self.guidance_scale, + seed=self.seed, + ) + edited_image.save(output_path) + return { + "task_type": item["task_type"], + "instruction_language": item["instruction_language"], + "key": item["key"], + "output_path": str(output_path), + } + + +class GEditBenchEvaluator: + def __init__(self, *, dataset_ref: str, output_root: Path, scorer: LocalVIEScorer): + self.dataset_ref = dataset_ref + self.output_root = output_root + self.scorer = scorer + + def evaluate( + self, + *, + model_name: str, + save_dir: Path, + task_type: str = "all", + instruction_language: str = "all", + workers: int = 1, + max_samples: int | None = None, + samples_per_group: int | None = None, + ) -> dict[str, Any]: + dataset = _resolve_gedit_split(_load_gedit_dataset(self.dataset_ref)) + rows = select_balanced_gedit_rows( + list(dataset), + task_type=task_type, + instruction_language=instruction_language, + samples_per_group=samples_per_group, + ) + if max_samples is not None: + rows = rows[:max_samples] + + results: list[dict[str, Any]] = [] + total = len(rows) + if workers <= 1: + with tqdm(total=total, desc="GEdit evaluate", unit="sample") as progress: + for item in rows: + result = self._safe_evaluate_one(model_name, item) + if result: + results.append(result) + progress.update(1) + else: + with tqdm(total=total, desc="GEdit evaluate", unit="sample") as progress: + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [executor.submit(self._safe_evaluate_one, model_name, item) for item in rows] + for future in as_completed(futures): + result = future.result() + if result: + results.append(result) + progress.update(1) + + save_dir.mkdir(parents=True, exist_ok=True) + base_name = f"{model_name}_{task_type}_{instruction_language}" + timestamp = _utc_timestamp() + csv_path = save_dir / f"{base_name}_vie_score.csv" + timestamped_csv_path = save_dir / f"{base_name}_vie_score_{timestamp}.csv" + + def _write_csv(path: Path) -> None: + with path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter( + handle, + fieldnames=[ + "key", + "task_type", + "edited_image", + "instruction", + "semantics_score", + "quality_score", + "overall_score", + "intersection_exist", + "instruction_language", + ], + ) + writer.writeheader() + for row in results: + writer.writerow(row) + + _write_csv(csv_path) + _write_csv(timestamped_csv_path) + + summary = summarize_gedit_rows_with_backbone( + results, + language=instruction_language, + ) + summary_path = save_dir / f"{base_name}_summary.json" + timestamped_summary_path = save_dir / f"{base_name}_summary_{timestamp}.json" + write_json(summary_path, summary) + write_json(timestamped_summary_path, summary) + return { + "results": results, + "summary": summary, + "csv_path": str(csv_path), + "summary_path": str(summary_path), + "timestamped_csv_path": str(timestamped_csv_path), + "timestamped_summary_path": str(timestamped_summary_path), + } + + def _safe_evaluate_one(self, model_name: str, item: dict[str, Any]) -> dict[str, Any] | None: + try: + return self._evaluate_one(model_name, item) + except Exception: + logger.exception("Failed to evaluate GEdit-Bench sample %s", item.get("key", "")) + return None + + def _evaluate_one(self, model_name: str, item: dict[str, Any]) -> dict[str, Any] | None: + edited_image_path = ( + self.output_root + / model_name + / "fullset" + / item["task_type"] + / item["instruction_language"] + / f"{item['key']}.png" + ) + if not edited_image_path.exists(): + return None + + source_image = _to_pil_image(item["input_image_raw"]) + edited_image = Image.open(edited_image_path).convert("RGB") + scores = self.scorer.evaluate(source_image, edited_image, item["instruction"]) + return { + "key": item["key"], + "task_type": item["task_type"], + "edited_image": str(edited_image_path), + "instruction": item["instruction"], + "semantics_score": scores["semantics_score"], + "quality_score": scores["quality_score"], + "overall_score": scores["overall_score"], + "intersection_exist": item.get("Intersection_exist", False), + "instruction_language": item["instruction_language"], + } + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Run the GEdit-Bench integration against a local vLLM-Omni server.") + subparsers = parser.add_subparsers(dest="command", required=True) + + generate = subparsers.add_parser("generate") + generate.add_argument("--dataset-ref", type=str, default="stepfun-ai/GEdit-Bench") + generate.add_argument("--output-root", type=Path, required=True) + generate.add_argument("--base-url", type=str, required=True) + generate.add_argument("--model", type=str, required=True) + generate.add_argument("--model-name", type=str, default=None) + generate.add_argument("--api-key", type=str, default="EMPTY") + generate.add_argument("--task-type", choices=["all", *GROUPS], default="all") + generate.add_argument("--instruction-language", choices=["all", "en", "cn"], default="all") + generate.add_argument("--width", type=int, default=512) + generate.add_argument("--height", type=int, default=512) + generate.add_argument("--num-inference-steps", type=int, default=20) + generate.add_argument("--guidance-scale", type=float, default=None) + generate.add_argument("--seed", type=int, default=42) + generate.add_argument("--workers", type=int, default=1) + generate.add_argument("--max-samples", type=int, default=None) + generate.add_argument("--samples-per-group", type=int, default=None) + + evaluate = subparsers.add_parser("evaluate") + evaluate.add_argument("--dataset-ref", type=str, default="stepfun-ai/GEdit-Bench") + evaluate.add_argument("--output-root", type=Path, required=True) + evaluate.add_argument("--model-name", type=str, default=None) + evaluate.add_argument("--save-dir", type=Path, required=True) + evaluate.add_argument("--task-type", choices=["all", *GROUPS], default="all") + evaluate.add_argument("--instruction-language", choices=["all", "en", "cn"], default="all") + evaluate.add_argument("--judge-base-url", type=str, required=True) + evaluate.add_argument("--judge-model", type=str, required=True) + evaluate.add_argument("--judge-api-key", type=str, default="EMPTY") + evaluate.add_argument("--workers", type=int, default=1) + evaluate.add_argument("--max-samples", type=int, default=None) + evaluate.add_argument("--samples-per-group", type=int, default=None) + + summarize = subparsers.add_parser("summarize") + summarize.add_argument("--csv-path", type=Path, required=True) + summarize.add_argument("--language", choices=["all", "en", "cn"], default="all") + + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + if args.command == "generate": + model_name = resolve_model_name(model_name=args.model_name, model=args.model) + runner = GEditBenchRunner( + dataset_ref=args.dataset_ref, + output_root=args.output_root, + base_url=args.base_url, + model=args.model, + api_key=args.api_key, + width=args.width, + height=args.height, + num_inference_steps=args.num_inference_steps, + guidance_scale=args.guidance_scale, + seed=args.seed, + ) + records = runner.generate( + model_name=model_name, + task_type=args.task_type, + instruction_language=args.instruction_language, + workers=args.workers, + max_samples=args.max_samples, + samples_per_group=args.samples_per_group, + ) + payload = {"records": records, "summary": summarize_generated_records(records)} + write_json(args.output_root / model_name / "generation_manifest.json", payload) + return 0 + + if args.command == "evaluate": + model_name = resolve_model_name(model_name=args.model_name, output_root=args.output_root) + scorer = LocalVIEScorer( + base_url=args.judge_base_url, + api_key=args.judge_api_key, + model=args.judge_model, + ) + evaluator = GEditBenchEvaluator(dataset_ref=args.dataset_ref, output_root=args.output_root, scorer=scorer) + evaluator.evaluate( + model_name=model_name, + save_dir=args.save_dir, + task_type=args.task_type, + instruction_language=args.instruction_language, + workers=args.workers, + max_samples=args.max_samples, + samples_per_group=args.samples_per_group, + ) + return 0 + + if args.command == "summarize": + with args.csv_path.open("r", encoding="utf-8", newline="") as handle: + rows = list(csv.DictReader(handle)) + summary = summarize_gedit_rows_with_backbone(rows, language=args.language) + print(json.dumps(summary, indent=2, ensure_ascii=False)) + return 0 + + parser.error(f"Unknown command: {args.command}") + return 1 diff --git a/benchmarks/accuracy/image_to_image/run_gedit_bench.py b/benchmarks/accuracy/image_to_image/run_gedit_bench.py new file mode 100644 index 000000000..4ce10c184 --- /dev/null +++ b/benchmarks/accuracy/image_to_image/run_gedit_bench.py @@ -0,0 +1,13 @@ +# ruff: noqa: E402, I001 +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[3] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from benchmarks.accuracy.image_to_image.gedit_bench import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/accuracy/text_to_image/README.md b/benchmarks/accuracy/text_to_image/README.md new file mode 100644 index 000000000..79641e8fc --- /dev/null +++ b/benchmarks/accuracy/text_to_image/README.md @@ -0,0 +1,103 @@ +# GEBench on vLLM-Omni + +This integration adapts the upstream `stepfun-ai/GEBench` scripts into +`vllm-omni/benchmarks/accuracy/text_to_image`. + +Upstream mapping: + +- `scripts/generate.py` -> `run_gebench.py generate` +- `scripts/evaluate.py` -> `run_gebench.py evaluate` +- upstream prompt / judge logic -> `gbench.py` + +What changed: + +- Generation calls the local OpenAI-compatible `vllm-omni` endpoints: + - `/v1/images/generations` for text-only frame generation + - `/v1/images/edits` for image-conditioned GUI transition generation +- Evaluation still keeps the GEBench scoring dimensions: + - `goal` + - `logic` + - `cons` + - `ui` + - `qual` +- Judge calls are also routed to a local OpenAI-compatible model served by + `vllm-omni` instead of a remote service. + +Expected dataset layout: + +- Clone the benchmark dataset from Hugging Face into a local directory: + +```bash +git clone https://huggingface.co/datasets/stepfun-ai/GEBench /path/to/GEBench +``` + +Example usage: + +```bash +python benchmarks/accuracy/text_to_image/run_gebench.py generate \ + --dataset-root /path/to/GEBench \ + --output-root benchmarks/accuracy/text_to_image/outputs \ + --base-url http://127.0.0.1:8000 \ + --model Tongyi-MAI/Z-Image-Turbo \ + --data-type type3 +``` + +```bash +python benchmarks/accuracy/text_to_image/run_gebench.py evaluate \ + --dataset-root /path/to/GEBench \ + --output-root benchmarks/accuracy/text_to_image/outputs \ + --data-type type3 \ + --judge-base-url http://127.0.0.1:8000 \ + --judge-model Qwen/Qwen2.5-VL-7B-Instruct \ + --judge-api-key EMPTY +``` + +```bash +python benchmarks/accuracy/text_to_image/run_gebench.py summarize \ + --output-root benchmarks/accuracy/text_to_image/outputs +``` + +Example summary output: + +```json +{ + "generation": { + "count": 20, + "by_type": { + "type3": {"count": 10}, + "type4": {"count": 10} + } + }, + "evaluation": { + "count": 20, + "overall_mean": 0.52, + "by_type": { + "type3": {"count": 10, "overall_mean": 0.50, "overall_mean_100": 50.0}, + "type4": {"count": 10, "overall_mean": 0.54, "overall_mean_100": 54.0} + } + } +} +``` + +Example generated images to inspect: + +- `benchmarks/accuracy/text_to_image/outputs/03_trajectory_text_fictionalapp///frame0.png` +- `benchmarks/accuracy/text_to_image/outputs/03_trajectory_text_fictionalapp///frame5.png` +- `benchmarks/accuracy/text_to_image/outputs/04_trajectory_text_realapp///frame0.png` +- `benchmarks/accuracy/text_to_image/outputs/04_trajectory_text_realapp///frame5.png` + +What to expect: + +- `overall_mean` is normalized to `0.0 ~ 1.0`; higher is better. +- `frame0.png` is the initial GUI frame; `frame5.png` is the final trajectory frame most often used for quick inspection. +- For full debugging, inspect the whole `frame0.png` ... `frame5.png` sequence for one sample directory. + +Notes: + +- GEBench upstream leaves type3/type4 generation unfinished. This integration + fills that gap with a trajectory runner that generates `frame0.png` followed + by `frame1.png` ... `frame5.png`. +- Type1/2/5 require an image-edit capable model exposed through + `vllm-omni serve`. +- `summarize` will report both generated coverage and any existing evaluation + summary files. diff --git a/benchmarks/accuracy/text_to_image/__init__.py b/benchmarks/accuracy/text_to_image/__init__.py new file mode 100644 index 000000000..e1f7b2bb6 --- /dev/null +++ b/benchmarks/accuracy/text_to_image/__init__.py @@ -0,0 +1 @@ +"""GEBench integration for vLLM-Omni.""" diff --git a/benchmarks/accuracy/text_to_image/gbench.py b/benchmarks/accuracy/text_to_image/gbench.py new file mode 100644 index 000000000..2ea02130d --- /dev/null +++ b/benchmarks/accuracy/text_to_image/gbench.py @@ -0,0 +1,927 @@ +from __future__ import annotations + +import argparse +import json +import statistics +from collections import defaultdict +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import requests +from PIL import Image + +from benchmarks.accuracy.common import ( + VllmOmniImageClient, + build_openai_url, + ensure_dir, + extract_json_object, + find_first_image, + load_json, + pil_to_data_url, + save_image, + write_json, +) + +TYPE_TO_FOLDER = { + "type1": "01_single_step", + "type2": "02_multi_step", + "type3": "03_trajectory_text_fictionalapp", + "type4": "04_trajectory_text_realapp", + "type5": "05_grounding_data", +} +SCORE_KEYS = ("goal", "logic", "cons", "ui", "qual") +DEFAULT_SAMPLES_PER_TYPE = 10 + + +def _utc_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + + +def _write_json_with_timestamp(path: Path, payload: dict[str, Any]) -> Path: + write_json(path, payload) + timestamped_path = path.with_name(f"{path.stem}_{_utc_timestamp()}{path.suffix}") + write_json(timestamped_path, payload) + return timestamped_path + + +@dataclass(frozen=True) +class GEBenchSampleSpec: + sample_path: Path + metadata: dict[str, Any] + sample_name: str + lang_device: str + + +def summarize_generated_records(records: list[dict[str, Any]]) -> dict[str, Any]: + by_type: dict[str, list[dict[str, Any]]] = defaultdict(list) + for record in records: + by_type[record["data_type"]].append(record) + + return { + "count": len(records), + "by_type": { + data_type: { + "count": len(rows), + } + for data_type, rows in sorted(by_type.items()) + }, + } + + +def summarize_gebench_results(results: list[dict[str, Any]]) -> dict[str, Any]: + by_type: dict[str, list[dict[str, Any]]] = defaultdict(list) + for result in results: + by_type[result["data_type"]].append(result) + + summary: dict[str, Any] = { + "count": len(results), + "overall_mean": statistics.fmean(r["overall"] for r in results) if results else 0.0, + "by_type": {}, + } + for data_type, rows in by_type.items(): + score_means: dict[str, float] = {} + all_score_keys = {key for row in rows for key in row.get("scores", {}).keys()} + for score_key in all_score_keys: + values = [row["scores"][score_key] for row in rows if score_key in row.get("scores", {})] + score_means[score_key] = statistics.fmean(values) if values else 0.0 + overall_mean = statistics.fmean(row["overall"] for row in rows) + summary["by_type"][data_type] = { + "count": len(rows), + "overall_mean": overall_mean, + "overall_mean_100": overall_mean * 100.0, + "score_means": score_means, + } + return summary + + +def select_balanced_gebench_samples( + sample_paths_by_type: dict[str, list[Any]], + *, + samples_per_type: int | None, +) -> dict[str, list[Any]]: + if samples_per_type is None: + return {data_type: list(paths) for data_type, paths in sample_paths_by_type.items()} + return {data_type: list(paths)[:samples_per_type] for data_type, paths in sample_paths_by_type.items()} + + +def collect_gebench_generation_summary(output_root: Path) -> dict[str, Any]: + records: list[dict[str, Any]] = [] + for data_type, folder_name in TYPE_TO_FOLDER.items(): + type_root = output_root / folder_name + if not type_root.exists(): + continue + for lang_dir in sorted(path for path in type_root.iterdir() if path.is_dir()): + for sample_dir in sorted(path for path in lang_dir.iterdir() if path.is_dir()): + expected = sample_dir / "frame5.png" if data_type in {"type2", "type3", "type4"} else None + if expected is None: + expected = find_first_image(sample_dir) + elif not expected.exists(): + expected = None + if expected is None: + continue + records.append( + { + "data_type": data_type, + "sample_name": f"{lang_dir.name}/{sample_dir.name}", + "output_path": str(expected), + } + ) + return summarize_generated_records(records) + + +def _normalize_score_key(key: str) -> str: + mapping = { + "goal": "goal", + "logic": "logic", + "cons": "cons", + "consistency": "cons", + "ui": "ui", + "qual": "qual", + "quality": "qual", + } + return mapping.get(key.lower(), key.lower()) + + +def _normalize_scores(raw_scores: dict[str, Any]) -> dict[str, int]: + scores: dict[str, int] = {} + for key, value in raw_scores.items(): + normalized = _normalize_score_key(key) + if normalized not in SCORE_KEYS: + continue + scalar = value.get("s", 0) if isinstance(value, dict) else value + try: + scores[normalized] = int(scalar) + except (TypeError, ValueError): + scores[normalized] = 0 + for key in SCORE_KEYS: + scores.setdefault(key, 0) + return scores + + +def _compute_overall(scores: dict[str, int]) -> float: + return sum(scores.values()) / (len(SCORE_KEYS) * 5.0) + + +def _iter_sample_paths(dataset_root: Path, data_type: str) -> list[Path]: + data_dir = dataset_root / TYPE_TO_FOLDER[data_type] + if not data_dir.exists(): + raise FileNotFoundError(f"Dataset path not found: {data_dir}") + + samples: list[Path] = [] + for lang_dir in sorted(path for path in data_dir.iterdir() if path.is_dir()): + for child in sorted(lang_dir.iterdir()): + if child.is_dir(): + samples.append(child) + elif child.suffix.lower() == ".json": + samples.append(child) + return samples + + +def _sample_name_from_metadata(metadata: dict[str, Any], sample_path: Path, item_index: int | None = None) -> str: + sample_id = metadata.get("id") or metadata.get("sample_id") or metadata.get("name") + if sample_id: + return str(sample_id) + if item_index is not None: + return f"{sample_path.stem}_{item_index:04d}" + return sample_path.stem if sample_path.is_file() else sample_path.name + + +def _expand_sample_path(sample_path: Path) -> list[GEBenchSampleSpec]: + if sample_path.is_dir(): + metadata = _load_metadata(sample_path) + return [ + GEBenchSampleSpec( + sample_path=sample_path, + metadata=metadata, + sample_name=_sample_name(sample_path), + lang_device=_lang_device(sample_path, metadata), + ) + ] + + payload = load_json(sample_path) + if isinstance(payload, dict): + return [ + GEBenchSampleSpec( + sample_path=sample_path, + metadata=payload, + sample_name=_sample_name_from_metadata(payload, sample_path), + lang_device=_lang_device(sample_path, payload), + ) + ] + + if isinstance(payload, list): + specs: list[GEBenchSampleSpec] = [] + for item_index, item in enumerate(payload): + if not isinstance(item, dict): + continue + specs.append( + GEBenchSampleSpec( + sample_path=sample_path, + metadata=item, + sample_name=_sample_name_from_metadata(item, sample_path, item_index), + lang_device=_lang_device(sample_path, item), + ) + ) + return specs + + raise TypeError(f"Unsupported metadata payload type for sample {sample_path}: {type(payload)!r}") + + +def _iter_sample_specs(dataset_root: Path, data_type: str) -> list[GEBenchSampleSpec]: + sample_specs: list[GEBenchSampleSpec] = [] + for sample_path in _iter_sample_paths(dataset_root, data_type): + sample_specs.extend(_expand_sample_path(sample_path)) + return sample_specs + + +def _load_metadata(sample_path: Path) -> dict[str, Any]: + if sample_path.is_file(): + return load_json(sample_path) + for candidate in ("meta_data.json", "metadata.json"): + meta_path = sample_path / candidate + if meta_path.exists(): + return load_json(meta_path) + raise FileNotFoundError(f"Metadata not found for sample: {sample_path}") + + +def _sample_name(sample_path: Path) -> str: + return sample_path.stem if sample_path.is_file() else sample_path.name + + +def _lang_device(sample_path: Path, metadata: dict[str, Any]) -> str: + return str(metadata.get("lang_device") or sample_path.parent.name) + + +def _resolve_referenced_image( + *, + metadata: dict[str, Any], + sample_path: Path, + dataset_root: Path, + data_type: str, +) -> Image.Image | None: + for key in ("image", "input_image", "initial_image", "reference_image"): + image_ref = metadata.get(key) + if not image_ref: + continue + candidate = dataset_root / TYPE_TO_FOLDER[data_type] / str(image_ref) + if candidate.exists(): + image = Image.open(candidate) + image.load() + return image.convert("RGB") + if sample_path.is_dir(): + local_image = find_first_image(sample_path) + if local_image: + image = Image.open(local_image) + image.load() + return image.convert("RGB") + return None + + +def _trajectory_steps(metadata: dict[str, Any]) -> list[dict[str, Any]]: + for key in ("trajectory", "steps", "frames"): + value = metadata.get(key) + if isinstance(value, list): + return [step for step in value if isinstance(step, dict)] + extracted: list[dict[str, Any]] = [] + for index in range(1, 6): + value = metadata.get(f"step{index}") or metadata.get(str(index)) + if isinstance(value, dict): + extracted.append(value) + return extracted + + +def _text_or_default(value: Any, default: str = "") -> str: + return str(value).strip() if value is not None else default + + +def _type1_prompt(metadata: dict[str, Any]) -> str: + caption = _text_or_default(metadata.get("caption") or metadata.get("instruction"), "Transform the reference GUI.") + return ( + "Using the reference GUI screenshot, generate the next GUI state after the requested interaction.\n\n" + f"Requested change:\n{caption}\n\n" + "Requirements:\n" + "- Preserve layout, visual identity, and unrelated regions.\n" + "- Only apply the requested state change.\n" + "- Keep all text and controls readable.\n" + ) + + +def _type2_prompt(goal: str, step_num: int) -> str: + return ( + "Generate the next GUI state for a multi-step task.\n\n" + f"Overall goal: {goal}\n" + f"Current progress step: {step_num}/5\n\n" + "Requirements:\n" + "- The change should be incremental and plausible.\n" + "- Preserve layout and visual identity.\n" + "- Make text/buttons readable.\n" + ) + + +def _type34_initial_prompt(metadata: dict[str, Any], first_step: dict[str, Any]) -> str: + app_name = _text_or_default(metadata.get("app_name"), "App") + final_goal = _text_or_default(metadata.get("final_goal") or metadata.get("instruction"), "Complete the task.") + visual_description = _text_or_default( + metadata.get("visual_description") or first_step.get("visual_description") or first_step.get("description"), + "A clean product-quality app home screen.", + ) + return ( + "Generate the first GUI frame for a task trajectory.\n\n" + f"App name: {app_name}\n" + f"Final goal: {final_goal}\n" + f"Visual description:\n{visual_description}\n\n" + "Requirements:\n" + "- Generate a production-looking UI screenshot only.\n" + "- Keep the layout coherent and readable.\n" + ) + + +def _type34_next_prompt(step_num: int, step_info: dict[str, Any]) -> str: + action = _text_or_default(step_info.get("action") or step_info.get("instruction"), "Continue the task.") + visual_description = _text_or_default( + step_info.get("visual_description") or step_info.get("description"), + "Reflect the expected next GUI state.", + ) + return ( + "Using the previous frame as reference, generate the next GUI frame.\n\n" + f"Step {step_num} action: {action}\n" + f"Expected visual state:\n{visual_description}\n\n" + "Requirements:\n" + "- Only change UI regions affected by this action.\n" + "- Preserve persistent bars, layout, and style.\n" + "- Keep text and icons readable.\n" + ) + + +def _type5_prompt(metadata: dict[str, Any]) -> str: + grounding = metadata.get("grounding") or {} + explanation = _text_or_default( + metadata.get("grounding_explanation") or grounding.get("effect") or grounding.get("description"), + "Predict the immediate GUI reaction to the indicated target.", + ) + return ( + "Using the reference GUI screenshot, predict the immediate GUI state after the grounded interaction.\n\n" + f"Expected effect: {explanation}\n" + f"Grounding metadata: {json.dumps(grounding, ensure_ascii=False)}\n\n" + "Requirements:\n" + "- Apply only the interaction-triggered change.\n" + "- Preserve unrelated regions.\n" + "- Keep the UI realistic and readable.\n" + ) + + +def _make_storyboard_image( + frames: list[Image.Image], + *, + columns: int = 3, + background_color: tuple[int, int, int] = (255, 255, 255), +) -> Image.Image: + if not frames: + raise ValueError("Expected at least one frame to build a storyboard image.") + + normalized = [frame.convert("RGB") for frame in frames] + frame_width = max(frame.width for frame in normalized) + frame_height = max(frame.height for frame in normalized) + rows = (len(normalized) + columns - 1) // columns + storyboard = Image.new("RGB", (frame_width * columns, frame_height * rows), color=background_color) + + for index, frame in enumerate(normalized): + x_offset = (index % columns) * frame_width + y_offset = (index // columns) * frame_height + if frame.size != (frame_width, frame_height): + frame = frame.resize((frame_width, frame_height)) + storyboard.paste(frame, (x_offset, y_offset)) + return storyboard + + +def _trajectory_judge_payload(frames: list[Image.Image]) -> tuple[str, list[Image.Image]]: + storyboard = _make_storyboard_image(frames, columns=3) + prompt_suffix = ( + "The attached image is a storyboard containing six frames arranged left-to-right, " + "top-to-bottom as frame0, frame1, frame2, frame3, frame4, frame5." + ) + return prompt_suffix, [storyboard] + + +class LocalJudgeClient: + def __init__(self, base_url: str, api_key: str, model: str, timeout: int = 600): + self.base_url = base_url.rstrip("/") + self.api_key = api_key + self.model = model + self.timeout = timeout + + def _build_scoring_prompt(self, task_prompt: str) -> str: + return ( + "You are an expert evaluator for GUI image editing and GUI trajectory generation.\n" + "Evaluate whether the generated image(s) satisfy the task.\n\n" + "Score these five dimensions from 0 to 5:\n" + "- goal: whether the user goal is completed correctly\n" + "- logic: whether the transition/state change is logically correct\n" + "- cons: whether unrelated regions remain consistent\n" + "- ui: whether the UI layout/components remain realistic and coherent\n" + "- qual: whether the images are visually clear and artifact-free\n\n" + "Return JSON only. Do not add any prose outside JSON.\n" + "Use exactly this schema:\n" + "{\n" + ' "goal": 0,\n' + ' "logic": 0,\n' + ' "cons": 0,\n' + ' "ui": 0,\n' + ' "qual": 0,\n' + ' "reasoning": "short explanation"\n' + "}\n\n" + "Scoring task:\n" + f"{task_prompt}" + ) + + def _request_text(self, prompt: str, images: list[Image.Image]) -> str: + content: list[dict[str, Any]] = [{"type": "text", "text": prompt}] + for image in images: + content.append({"type": "image_url", "image_url": {"url": pil_to_data_url(image)}}) + + response = requests.post( + build_openai_url(self.base_url, "/chat/completions"), + json={ + "model": self.model, + "messages": [{"role": "user", "content": content}], + "temperature": 0, + }, + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + }, + timeout=self.timeout, + ) + response.raise_for_status() + message_content = response.json()["choices"][0]["message"]["content"] + if isinstance(message_content, list): + return "\n".join(part.get("text", "") for part in message_content if part.get("type") == "text") + return str(message_content) + + def evaluate(self, *, prompt: str, images: list[Image.Image]) -> dict[str, Any]: + primary_prompt = self._build_scoring_prompt(prompt) + text = self._request_text(primary_prompt, images) + try: + return extract_json_object(text) + except ValueError: + retry_prompt = ( + self._build_scoring_prompt(prompt) + "\n\nYour previous response was not valid JSON. " + "Return only the JSON object with integer scores." + ) + retry_text = self._request_text(retry_prompt, images) + try: + return extract_json_object(retry_text) + except ValueError: + return { + "goal": 0, + "logic": 0, + "cons": 0, + "ui": 0, + "qual": 0, + "reasoning": retry_text.strip() or text.strip() or "Judge response was not valid JSON.", + } + + +class GEBenchRunner: + def __init__( + self, + *, + dataset_root: Path, + output_root: Path, + base_url: str, + model: str, + api_key: str = "EMPTY", + width: int = 768, + height: int = 576, + num_inference_steps: int = 8, + output_compression: int | None = 98, + guidance_scale: float | None = None, + seed: int | None = 42, + ): + self.dataset_root = dataset_root + self.output_root = output_root + self.model = model + self.width = width + self.height = height + self.num_inference_steps = num_inference_steps + self.output_compression = output_compression + self.guidance_scale = guidance_scale + self.seed = seed + self.client = VllmOmniImageClient(base_url=base_url, api_key=api_key) + + def generate( + self, + *, + data_type: str, + workers: int = 1, + max_samples: int | None = None, + samples_per_type: int | None = None, + ) -> list[dict[str, Any]]: + sample_specs = select_balanced_gebench_samples( + {data_type: _iter_sample_specs(self.dataset_root, data_type)}, + samples_per_type=samples_per_type, + )[data_type] + if max_samples is not None: + sample_specs = sample_specs[:max_samples] + + results: list[dict[str, Any]] = [] + if workers <= 1: + for sample_spec in sample_specs: + result = self._generate_one(data_type, sample_spec) + if result: + results.append(result) + return results + + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [executor.submit(self._generate_one, data_type, sample_spec) for sample_spec in sample_specs] + for future in as_completed(futures): + result = future.result() + if result: + results.append(result) + return results + + def _generate_one(self, data_type: str, sample_spec: GEBenchSampleSpec) -> dict[str, Any] | None: + sample_path = sample_spec.sample_path + metadata = sample_spec.metadata + lang_device = sample_spec.lang_device + sample_name = sample_spec.sample_name + output_dir = ensure_dir(self.output_root / TYPE_TO_FOLDER[data_type] / lang_device / sample_name) + + if data_type == "type1": + output_path = output_dir / "generated.png" + if output_path.exists(): + return { + "data_type": data_type, + "sample_name": f"{lang_device}/{sample_name}", + "output_path": str(output_path), + } + source = _resolve_referenced_image( + metadata=metadata, sample_path=sample_path, dataset_root=self.dataset_root, data_type=data_type + ) + if source is None: + return None + generated = self.client.generate_image_edit( + model=self.model, + prompt=_type1_prompt(metadata), + images=source, + width=self.width, + height=self.height, + num_inference_steps=self.num_inference_steps, + output_compression=self.output_compression, + guidance_scale=self.guidance_scale, + seed=self.seed, + ) + save_image(output_path, generated) + return { + "data_type": data_type, + "sample_name": f"{lang_device}/{sample_name}", + "output_path": str(output_path), + } + + if data_type == "type2": + goal = _text_or_default(metadata.get("question") or metadata.get("caption"), "Complete the task.") + source = _resolve_referenced_image( + metadata=metadata, sample_path=sample_path, dataset_root=self.dataset_root, data_type=data_type + ) + if source is None: + return None + frame0_path = output_dir / "frame0.png" + if not frame0_path.exists(): + save_image(frame0_path, source) + previous = source + for step_num in range(1, 6): + frame_path = output_dir / f"frame{step_num}.png" + if frame_path.exists(): + previous = Image.open(frame_path).convert("RGB") + continue + generated = self.client.generate_image_edit( + model=self.model, + prompt=_type2_prompt(goal, step_num), + images=previous, + width=self.width, + height=self.height, + num_inference_steps=self.num_inference_steps, + output_compression=self.output_compression, + guidance_scale=self.guidance_scale, + seed=self.seed, + ) + save_image(frame_path, generated) + previous = generated + output_path = output_dir / "frame5.png" + return { + "data_type": data_type, + "sample_name": f"{lang_device}/{sample_name}", + "output_path": str(output_path), + } + + if data_type in {"type3", "type4"}: + steps = _trajectory_steps(metadata) + frame0_path = output_dir / "frame0.png" + if frame0_path.exists(): + previous = Image.open(frame0_path).convert("RGB") + else: + previous = self.client.generate_text_to_image( + model=self.model, + prompt=_type34_initial_prompt(metadata, steps[0] if steps else {}), + width=self.width, + height=self.height, + num_inference_steps=self.num_inference_steps, + output_compression=self.output_compression, + guidance_scale=self.guidance_scale, + seed=self.seed, + ) + save_image(frame0_path, previous) + + for step_num in range(1, 6): + frame_path = output_dir / f"frame{step_num}.png" + if frame_path.exists(): + previous = Image.open(frame_path).convert("RGB") + continue + step_info = steps[step_num - 1] if step_num - 1 < len(steps) else {} + generated = self.client.generate_image_edit( + model=self.model, + prompt=_type34_next_prompt(step_num, step_info), + images=previous, + width=self.width, + height=self.height, + num_inference_steps=self.num_inference_steps, + output_compression=self.output_compression, + guidance_scale=self.guidance_scale, + seed=self.seed, + ) + save_image(frame_path, generated) + previous = generated + output_path = output_dir / "frame5.png" + return { + "data_type": data_type, + "sample_name": f"{lang_device}/{sample_name}", + "output_path": str(output_path), + } + + if data_type == "type5": + output_path = output_dir / "generated.png" + if output_path.exists(): + return { + "data_type": data_type, + "sample_name": f"{lang_device}/{sample_name}", + "output_path": str(output_path), + } + source = _resolve_referenced_image( + metadata=metadata, sample_path=sample_path, dataset_root=self.dataset_root, data_type=data_type + ) + if source is None: + return None + generated = self.client.generate_image_edit( + model=self.model, + prompt=_type5_prompt(metadata), + images=source, + width=self.width, + height=self.height, + num_inference_steps=self.num_inference_steps, + output_compression=self.output_compression, + guidance_scale=self.guidance_scale, + seed=self.seed, + ) + save_image(output_path, generated) + return { + "data_type": data_type, + "sample_name": f"{lang_device}/{sample_name}", + "output_path": str(output_path), + } + + raise ValueError(f"Unsupported data type: {data_type}") + + +class GEBenchEvaluator: + def __init__(self, *, dataset_root: Path, output_root: Path, judge: LocalJudgeClient): + self.dataset_root = dataset_root + self.output_root = output_root + self.judge = judge + + def evaluate( + self, + *, + data_type: str, + workers: int = 1, + max_samples: int | None = None, + samples_per_type: int | None = None, + ) -> dict[str, Any]: + output_type_dir = self.output_root / TYPE_TO_FOLDER[data_type] + sample_specs_by_name = { + (spec.lang_device, spec.sample_name): spec for spec in _iter_sample_specs(self.dataset_root, data_type) + } + if not output_type_dir.exists(): + payload = {"data_type": data_type, "results": [], "summary": summarize_gebench_results([])} + write_json(self.output_root / "evaluations" / f"{data_type}.json", payload) + return payload + sample_dirs = [ + sample_dir + for lang_dir in sorted(path for path in output_type_dir.iterdir() if path.is_dir()) + for sample_dir in sorted(path for path in lang_dir.iterdir() if path.is_dir()) + if (lang_dir.name, sample_dir.name) in sample_specs_by_name + ] + sample_dirs = select_balanced_gebench_samples( + {data_type: sample_dirs}, + samples_per_type=samples_per_type, + )[data_type] + if max_samples is not None: + sample_dirs = sample_dirs[:max_samples] + results: list[dict[str, Any]] = [] + if workers <= 1: + for sample_dir in sample_dirs: + result = self._evaluate_one( + data_type, + sample_dir, + sample_specs_by_name[(sample_dir.parent.name, sample_dir.name)], + ) + if result: + results.append(result) + else: + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [ + executor.submit( + self._evaluate_one, + data_type, + sample_dir, + sample_specs_by_name[(sample_dir.parent.name, sample_dir.name)], + ) + for sample_dir in sample_dirs + ] + for future in as_completed(futures): + result = future.result() + if result: + results.append(result) + + payload = {"data_type": data_type, "results": results, "summary": summarize_gebench_results(results)} + write_json(self.output_root / "evaluations" / f"{data_type}.json", payload) + return payload + + def _evaluate_one(self, data_type: str, sample_dir: Path, sample_spec: GEBenchSampleSpec) -> dict[str, Any] | None: + lang_device = sample_dir.parent.name + sample_name = sample_dir.name + dataset_sample = sample_spec.sample_path + metadata = sample_spec.metadata + + if data_type == "type1": + source = _resolve_referenced_image( + metadata=metadata, sample_path=dataset_sample, dataset_root=self.dataset_root, data_type=data_type + ) + generated_path = find_first_image(sample_dir) + if source is None or generated_path is None: + return None + generated = Image.open(generated_path).convert("RGB") + raw_scores = self.judge.evaluate(prompt=_type1_prompt(metadata), images=[source, generated]) + elif data_type == "type2": + frames = [Image.open(sample_dir / f"frame{i}.png").convert("RGB") for i in range(6)] + goal = _text_or_default(metadata.get("question") or metadata.get("caption"), "Complete the task.") + prompt_suffix, judge_images = _trajectory_judge_payload(frames) + raw_scores = self.judge.evaluate( + prompt=f"Evaluate a six-frame GUI trajectory.\nTask: {goal}\n{prompt_suffix}", + images=judge_images, + ) + elif data_type in {"type3", "type4"}: + frames = [Image.open(sample_dir / f"frame{i}.png").convert("RGB") for i in range(6)] + instruction = _text_or_default(metadata.get("instruction") or metadata.get("caption"), "Complete the task.") + prompt_suffix, judge_images = _trajectory_judge_payload(frames) + raw_scores = self.judge.evaluate( + prompt=f"Evaluate a six-frame GUI trajectory.\nInstruction: {instruction}\n{prompt_suffix}", + images=judge_images, + ) + elif data_type == "type5": + source = _resolve_referenced_image( + metadata=metadata, sample_path=dataset_sample, dataset_root=self.dataset_root, data_type=data_type + ) + generated_path = find_first_image(sample_dir) + if source is None or generated_path is None: + return None + generated = Image.open(generated_path).convert("RGB") + raw_scores = self.judge.evaluate(prompt=_type5_prompt(metadata), images=[source, generated]) + else: + raise ValueError(f"Unsupported data type: {data_type}") + + scores = _normalize_scores(raw_scores) + return { + "sample_name": f"{lang_device}/{sample_name}", + "data_type": data_type, + "scores": scores, + "overall": _compute_overall(scores), + "raw_scores": raw_scores, + } + + +def _data_types_arg(value: str) -> list[str]: + return list(TYPE_TO_FOLDER.keys()) if value == "all" else [value] + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Run local GEBench generation and scoring against vLLM-Omni.") + subparsers = parser.add_subparsers(dest="command", required=True) + + generate = subparsers.add_parser("generate") + generate.add_argument("--dataset-root", type=Path, required=True) + generate.add_argument("--output-root", type=Path, required=True) + generate.add_argument("--base-url", type=str, required=True) + generate.add_argument("--model", type=str, required=True) + generate.add_argument("--data-type", choices=["all", *TYPE_TO_FOLDER.keys()], default="all") + generate.add_argument("--api-key", type=str, default="EMPTY") + generate.add_argument("--width", type=int, default=768) + generate.add_argument("--height", type=int, default=576) + generate.add_argument("--num-inference-steps", type=int, default=8) + generate.add_argument("--output-compression", type=int, default=98) + generate.add_argument("--guidance-scale", type=float, default=None) + generate.add_argument("--seed", type=int, default=42) + generate.add_argument("--workers", type=int, default=1) + generate.add_argument("--max-samples", type=int, default=None) + generate.add_argument("--samples-per-type", type=int, default=None) + + evaluate = subparsers.add_parser("evaluate") + evaluate.add_argument("--dataset-root", type=Path, required=True) + evaluate.add_argument("--output-root", type=Path, required=True) + evaluate.add_argument("--data-type", choices=["all", *TYPE_TO_FOLDER.keys()], default="all") + evaluate.add_argument("--judge-base-url", type=str, required=True) + evaluate.add_argument("--judge-model", type=str, required=True) + evaluate.add_argument("--judge-api-key", type=str, default="EMPTY") + evaluate.add_argument("--workers", type=int, default=1) + evaluate.add_argument("--max-samples", type=int, default=None) + evaluate.add_argument("--samples-per-type", type=int, default=None) + + summarize = subparsers.add_parser("summarize") + summarize.add_argument("--output-root", type=Path, required=True) + + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + if args.command == "generate": + runner = GEBenchRunner( + dataset_root=args.dataset_root, + output_root=args.output_root, + base_url=args.base_url, + model=args.model, + api_key=args.api_key, + width=args.width, + height=args.height, + num_inference_steps=args.num_inference_steps, + output_compression=args.output_compression, + guidance_scale=args.guidance_scale, + seed=args.seed, + ) + records: list[dict[str, Any]] = [] + for data_type in _data_types_arg(args.data_type): + records.extend( + runner.generate( + data_type=data_type, + workers=args.workers, + max_samples=args.max_samples, + samples_per_type=args.samples_per_type, + ) + ) + payload = {"records": records, "summary": summarize_generated_records(records)} + write_json(args.output_root / "generation_manifest.json", payload) + return 0 + + if args.command == "evaluate": + judge = LocalJudgeClient( + base_url=args.judge_base_url, + api_key=args.judge_api_key, + model=args.judge_model, + ) + evaluator = GEBenchEvaluator(dataset_root=args.dataset_root, output_root=args.output_root, judge=judge) + combined_results: list[dict[str, Any]] = [] + for data_type in _data_types_arg(args.data_type): + payload = evaluator.evaluate( + data_type=data_type, + workers=args.workers, + max_samples=args.max_samples, + samples_per_type=args.samples_per_type, + ) + combined_results.extend(payload["results"]) + _write_json_with_timestamp( + args.output_root / "evaluations" / "summary.json", + {"summary": summarize_gebench_results(combined_results)}, + ) + return 0 + + if args.command == "summarize": + generation_summary = collect_gebench_generation_summary(args.output_root) + evaluation_dir = args.output_root / "evaluations" + result_records: list[dict[str, Any]] = [] + if evaluation_dir.exists(): + for file_path in sorted(evaluation_dir.glob("type*.json")): + payload = load_json(file_path) + result_records.extend(payload.get("results", [])) + payload: dict[str, Any] = {"generation": generation_summary} + if result_records: + payload["evaluation"] = summarize_gebench_results(result_records) + _write_json_with_timestamp(args.output_root / "summary.json", payload) + print(json.dumps(payload, indent=2, ensure_ascii=False)) + return 0 + + parser.error(f"Unknown command: {args.command}") + return 1 diff --git a/benchmarks/accuracy/text_to_image/run_gebench.py b/benchmarks/accuracy/text_to_image/run_gebench.py new file mode 100644 index 000000000..554c968aa --- /dev/null +++ b/benchmarks/accuracy/text_to_image/run_gebench.py @@ -0,0 +1,13 @@ +# ruff: noqa: E402, I001 +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[3] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from benchmarks.accuracy.text_to_image.gbench import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml index 3fd987d99..aedcaf132 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,6 +41,7 @@ dev = [ "pytest-asyncio>=0.21.0", "pytest-cov>=4.0.0", "pytest-mock>=3.10.0", + "datasets>=2.14.0", "mypy==1.11.1", "pre-commit==4.0.1", "openai-whisper>=20250625", @@ -211,3 +212,5 @@ extend-ignore-identifiers-re = [ ] [tool.typos.default.extend-words] ue = "ue" +semantics = "semantics" +fullset = "fullset" diff --git a/tests/benchmarks/test_accuracy_bench_utils.py b/tests/benchmarks/test_accuracy_bench_utils.py new file mode 100644 index 000000000..a0479fb1b --- /dev/null +++ b/tests/benchmarks/test_accuracy_bench_utils.py @@ -0,0 +1,685 @@ +# ruff: noqa: E402, I001 +import math +import sys +from pathlib import Path + +import pytest +from PIL import Image + +REPO_ROOT = Path(__file__).resolve().parents[2] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from benchmarks.accuracy.common import VllmOmniImageClient +from benchmarks.accuracy.image_to_image.gedit_bench import ( + GROUPS as GEDIT_GROUPS, + GEditBenchEvaluator, + GEditBenchRunner, + _load_gedit_dataset, + _resolve_gedit_split, + infer_model_name, + resolve_model_name, + select_balanced_gedit_rows, + parse_score_payload, + summarize_generated_records as summarize_gedit_generated_records, + summarize_gedit_rows, + summarize_gedit_rows_with_backbone, +) +from benchmarks.accuracy.text_to_image.gbench import ( + _expand_sample_path, + _trajectory_judge_payload, + _write_json_with_timestamp, + LocalJudgeClient, + GEBenchEvaluator, + TYPE_TO_FOLDER, + select_balanced_gebench_samples, + summarize_generated_records as summarize_gebench_generated_records, + summarize_gebench_results, +) + +pytestmark = [pytest.mark.core_model, pytest.mark.diffusion, pytest.mark.cpu] + + +def test_summarize_gebench_generated_records_groups_by_type(): + records = [ + {"data_type": "type1", "sample_name": "english_phone/folder_1", "output_path": "a.png"}, + {"data_type": "type1", "sample_name": "english_phone/folder_2", "output_path": "b.png"}, + {"data_type": "type2", "sample_name": "english_phone/folder_3", "output_path": "c.png"}, + ] + + summary = summarize_gebench_generated_records(records) + + assert summary["count"] == 3 + assert summary["by_type"]["type1"]["count"] == 2 + assert summary["by_type"]["type2"]["count"] == 1 + assert "samples" not in summary["by_type"]["type1"] + + +def test_summarize_gebench_results_computes_type_and_global_means(): + results = [ + {"data_type": "type1", "overall": 0.8, "scores": {"goal": 5, "logic": 4}}, + {"data_type": "type1", "overall": 0.6, "scores": {"goal": 3, "logic": 4}}, + {"data_type": "type2", "overall": 0.5, "scores": {"goal": 2, "logic": 3}}, + ] + + summary = summarize_gebench_results(results) + + assert math.isclose(summary["overall_mean"], (0.8 + 0.6 + 0.5) / 3) + assert math.isclose(summary["by_type"]["type1"]["overall_mean"], 0.7) + assert math.isclose(summary["by_type"]["type2"]["overall_mean"], 0.5) + assert math.isclose(summary["by_type"]["type1"]["score_means"]["goal"], 4.0) + + +def test_write_json_with_timestamp_writes_stable_and_timestamped_files(monkeypatch, tmp_path: Path): + monkeypatch.setattr( + "benchmarks.accuracy.text_to_image.gbench._utc_timestamp", + lambda: "20260325T130000Z", + ) + + timestamped_path = _write_json_with_timestamp(tmp_path / "summary.json", {"ok": True}) + + assert (tmp_path / "summary.json").exists() + assert timestamped_path == tmp_path / "summary_20260325T130000Z.json" + assert timestamped_path.exists() + + +def test_select_balanced_gebench_samples_limits_each_type_independently(): + sample_paths_by_type = { + "type1": [Path(f"/tmp/type1_{idx}") for idx in range(12)], + "type2": [Path(f"/tmp/type2_{idx}") for idx in range(8)], + "type3": [Path(f"/tmp/type3_{idx}") for idx in range(15)], + } + + selected = select_balanced_gebench_samples(sample_paths_by_type, samples_per_type=10) + + assert len(selected["type1"]) == 10 + assert len(selected["type2"]) == 8 + assert len(selected["type3"]) == 10 + assert selected["type1"][0].name == "type1_0" + assert selected["type3"][-1].name == "type3_9" + + +def test_expand_sample_path_flattens_json_list_samples(tmp_path: Path): + sample_path = tmp_path / "trajectories.json" + sample_path.write_text( + """ +[ + {"id": "sample_a", "lang_device": "english_phone", "instruction": "do a"}, + {"id": "sample_b", "lang_device": "english_phone", "instruction": "do b"} +] +""".strip(), + encoding="utf-8", + ) + + specs = _expand_sample_path(sample_path) + + assert len(specs) == 2 + assert specs[0].sample_name == "sample_a" + assert specs[1].sample_name == "sample_b" + assert specs[0].lang_device == "english_phone" + + +def test_gebench_evaluate_skips_missing_output_folder(tmp_path: Path): + dataset_type_root = tmp_path / TYPE_TO_FOLDER["type3"] / "english_phone" + sample_dir = dataset_type_root / "sample_a" + sample_dir.mkdir(parents=True) + (sample_dir / "meta_data.json").write_text("{}", encoding="utf-8") + + judge = LocalJudgeClient(base_url="http://127.0.0.1:8094", api_key="EMPTY", model="judge") + evaluator = GEBenchEvaluator(dataset_root=tmp_path, output_root=tmp_path / "outputs", judge=judge) + + payload = evaluator.evaluate(data_type="type3") + + assert payload["results"] == [] + assert payload["summary"]["count"] == 0 + + +def test_local_judge_client_retries_when_first_response_is_not_json(monkeypatch): + responses = iter( + [ + "The image looks like a GUI screenshot with several controls.", + '{"goal": 4, "logic": 4, "cons": 5, "ui": 4, "qual": 4, "reasoning": "mostly correct"}', + ] + ) + + def fake_request_text(self, prompt, images): + return next(responses) + + monkeypatch.setattr(LocalJudgeClient, "_request_text", fake_request_text) + + judge = LocalJudgeClient(base_url="http://127.0.0.1:8094", api_key="EMPTY", model="judge") + result = judge.evaluate(prompt="Evaluate this GUI trajectory.", images=[Image.new("RGB", (2, 2), color="white")]) + + assert result["goal"] == 4 + assert result["cons"] == 5 + + +def test_local_judge_client_returns_zero_scores_when_retry_is_still_invalid(monkeypatch): + responses = iter( + [ + "not json", + "still not json", + ] + ) + + def fake_request_text(self, prompt, images): + return next(responses) + + monkeypatch.setattr(LocalJudgeClient, "_request_text", fake_request_text) + + judge = LocalJudgeClient(base_url="http://127.0.0.1:8094", api_key="EMPTY", model="judge") + result = judge.evaluate(prompt="Evaluate this GUI trajectory.", images=[Image.new("RGB", (2, 2), color="white")]) + + assert result["goal"] == 0 + assert result["logic"] == 0 + assert result["cons"] == 0 + assert result["ui"] == 0 + assert result["qual"] == 0 + assert result["reasoning"] == "still not json" + + +def test_trajectory_judge_payload_collapses_six_frames_into_single_storyboard(): + frames = [Image.new("RGB", (8, 6), color=(idx * 10, idx * 10, idx * 10)) for idx in range(6)] + + prompt_suffix, judge_images = _trajectory_judge_payload(frames) + + assert "frame0" in prompt_suffix + assert len(judge_images) == 1 + assert judge_images[0].size == (24, 12) + + +def test_image_edit_client_uses_openai_image_edit_endpoint(monkeypatch): + captured = {} + + class FakeResponse: + status_code = 200 + + def raise_for_status(self): + return None + + def json(self): + return { + "data": [ + { + "b64_json": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aY0cAAAAASUVORK5CYII=" + } + ] + } + + def fake_post(url, data=None, files=None, headers=None, timeout=None, **kwargs): + captured["url"] = url + captured["data"] = data + captured["files"] = files + captured["headers"] = headers + captured["timeout"] = timeout + return FakeResponse() + + monkeypatch.setattr("benchmarks.accuracy.common.requests.post", fake_post) + + client = VllmOmniImageClient(base_url="http://127.0.0.1:8093", api_key="EMPTY") + image = Image.new("RGB", (2, 2), color="white") + output = client.generate_image_edit( + model="Qwen/Qwen-Image-Edit", + prompt="edit this image", + images=image, + width=512, + height=512, + ) + + assert output.size == (1, 1) + assert captured["url"] == "http://127.0.0.1:8093/v1/images/edits" + assert captured["data"]["prompt"] == "edit this image" + assert captured["data"]["size"] == "512x512" + assert captured["files"][0][0] == "image" + + +def test_text_to_image_client_forwards_output_compression(monkeypatch): + captured = {} + + class FakeResponse: + status_code = 200 + + def raise_for_status(self): + return None + + def json(self): + return { + "data": [ + { + "b64_json": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aY0cAAAAASUVORK5CYII=" + } + ] + } + + def fake_post(url, json=None, headers=None, timeout=None, **kwargs): + captured["url"] = url + captured["json"] = json + return FakeResponse() + + monkeypatch.setattr("benchmarks.accuracy.common.requests.post", fake_post) + + client = VllmOmniImageClient(base_url="http://127.0.0.1:8093", api_key="EMPTY") + output = client.generate_text_to_image( + model="Qwen/Qwen-Image", + prompt="generate a gui", + width=768, + height=576, + num_inference_steps=8, + output_compression=98, + ) + + assert output.size == (1, 1) + assert captured["url"] == "http://127.0.0.1:8093/v1/images/generations" + assert captured["json"]["size"] == "768x576" + assert captured["json"]["num_inference_steps"] == 8 + assert captured["json"]["output_compression"] == 98 + + +def test_parse_score_payload_handles_raw_json_and_delimited_json(): + raw = '{"score": [7, 8], "reasoning": "ok"}' + wrapped = 'prefix ||V^=^V|| {"score": [6], "reasoning": "fine"} ||V^=^V|| suffix' + + assert parse_score_payload(raw)["score"] == [7, 8] + assert parse_score_payload(wrapped)["score"] == [6] + + +def test_summarize_gedit_generated_records_groups_by_task_and_language(): + records = [] + for group in GEDIT_GROUPS[:2]: + records.append( + { + "task_type": group, + "instruction_language": "en", + "key": f"{group}_en", + "output_path": f"{group}_en.png", + } + ) + records.append( + { + "task_type": group, + "instruction_language": "cn", + "key": f"{group}_cn", + "output_path": f"{group}_cn.png", + } + ) + + summary = summarize_gedit_generated_records(records) + + assert summary["count"] == 4 + assert summary["by_task"][GEDIT_GROUPS[0]]["count"] == 2 + assert summary["by_language"]["en"]["count"] == 2 + assert summary["by_language"]["cn"]["samples"] == [ + f"{GEDIT_GROUPS[0]}_cn", + f"{GEDIT_GROUPS[1]}_cn", + ] + + +def test_select_balanced_gedit_rows_limits_each_group_independently(): + rows = [] + for idx in range(12): + rows.append( + { + "task_type": "background_change", + "instruction_language": "en", + "key": f"background_change_{idx}", + } + ) + for idx in range(7): + rows.append( + { + "task_type": "color_alter", + "instruction_language": "en", + "key": f"color_alter_{idx}", + } + ) + + selected = select_balanced_gedit_rows( + rows, + task_type="all", + instruction_language="en", + samples_per_group=10, + ) + + selected_background = [row for row in selected if row["task_type"] == "background_change"] + selected_color = [row for row in selected if row["task_type"] == "color_alter"] + + assert len(selected_background) == 10 + assert len(selected_color) == 7 + assert selected_background[0]["key"] == "background_change_0" + assert selected_background[-1]["key"] == "background_change_9" + + +def test_select_balanced_gedit_rows_balances_languages_when_all_requested(): + rows = [] + for idx in range(10): + rows.append( + { + "task_type": "background_change", + "instruction_language": "cn", + "key": f"background_change_cn_{idx}", + } + ) + for idx in range(10): + rows.append( + { + "task_type": "background_change", + "instruction_language": "en", + "key": f"background_change_en_{idx}", + } + ) + + selected = select_balanced_gedit_rows( + rows, + task_type="all", + instruction_language="all", + samples_per_group=10, + ) + + selected_background = [row for row in selected if row["task_type"] == "background_change"] + + assert len(selected_background) == 10 + assert sum(1 for row in selected_background if row["instruction_language"] == "en") == 5 + assert sum(1 for row in selected_background if row["instruction_language"] == "cn") == 5 + + +def test_infer_model_name_uses_last_path_segment(): + assert infer_model_name("/workspace/models/Qwen/Qwen-Image-Edit") == "Qwen-Image-Edit" + + +def test_resolve_model_name_prefers_explicit_value_then_model_then_output_root(tmp_path: Path): + assert ( + resolve_model_name( + model_name="explicit_name", + model="/workspace/models/Qwen/Qwen-Image-Edit", + ) + == "explicit_name" + ) + assert ( + resolve_model_name( + model_name=None, + model="/workspace/models/Qwen/Qwen-Image-Edit", + ) + == "Qwen-Image-Edit" + ) + + output_root = tmp_path / "results" + (output_root / "qwen_image_edit").mkdir(parents=True) + assert resolve_model_name(model_name=None, output_root=output_root) == "qwen_image_edit" + + +def test_resolve_gedit_split_accepts_dataset_dict_like_input(): + train_rows = [{"key": "a"}] + dataset = {"train": train_rows} + + assert _resolve_gedit_split(dataset) == train_rows + + +def test_resolve_gedit_split_accepts_dataset_like_input(): + rows = [{"key": "a"}] + + assert _resolve_gedit_split(rows) == rows + + +def test_load_gedit_dataset_uses_load_from_disk_for_saved_dataset(monkeypatch, tmp_path: Path): + (tmp_path / "state.json").write_text("{}", encoding="utf-8") + (tmp_path / "dataset_info.json").write_text("{}", encoding="utf-8") + captured = {} + + def fake_load_dataset(path): + captured["load_dataset"] = path + return "load_dataset" + + def fake_load_from_disk(path): + captured["load_from_disk"] = path + return "load_from_disk" + + monkeypatch.setattr( + "benchmarks.accuracy.image_to_image.gedit_bench._require_datasets", + lambda: (fake_load_dataset, fake_load_from_disk), + ) + + result = _load_gedit_dataset(str(tmp_path)) + + assert result == "load_from_disk" + assert captured["load_from_disk"] == str(tmp_path) + assert "load_dataset" not in captured + + +def test_load_gedit_dataset_uses_load_dataset_for_local_snapshot_path(monkeypatch, tmp_path: Path): + (tmp_path / "README.md").write_text("dataset repo snapshot", encoding="utf-8") + captured = {} + + def fake_load_dataset(path): + captured["load_dataset"] = path + return "load_dataset" + + def fake_load_from_disk(path): + captured["load_from_disk"] = path + return "load_from_disk" + + monkeypatch.setattr( + "benchmarks.accuracy.image_to_image.gedit_bench._require_datasets", + lambda: (fake_load_dataset, fake_load_from_disk), + ) + + result = _load_gedit_dataset(str(tmp_path)) + + assert result == "load_dataset" + assert captured["load_dataset"] == str(tmp_path) + assert "load_from_disk" not in captured + + +def test_gedit_runner_generate_skips_failed_samples(monkeypatch, tmp_path: Path): + rows = [ + {"key": "ok", "task_type": "background_change", "instruction_language": "en"}, + {"key": "bad", "task_type": "background_change", "instruction_language": "en"}, + ] + + monkeypatch.setattr("benchmarks.accuracy.image_to_image.gedit_bench._load_gedit_dataset", lambda ref: rows) + runner = GEditBenchRunner( + dataset_ref="dataset", + output_root=tmp_path, + base_url="http://127.0.0.1:8093", + model="model", + ) + + def fake_generate_one(self, model_name, item): + if item["key"] == "bad": + raise RuntimeError("boom") + return { + "key": item["key"], + "task_type": item["task_type"], + "instruction_language": item["instruction_language"], + } + + monkeypatch.setattr(GEditBenchRunner, "_generate_one", fake_generate_one) + + outputs = runner.generate(model_name="demo", workers=1) + + assert outputs == [{"key": "ok", "task_type": "background_change", "instruction_language": "en"}] + + +def test_gedit_runner_uses_tqdm_progress(monkeypatch, tmp_path: Path): + rows = [ + {"key": "one", "task_type": "background_change", "instruction_language": "en"}, + {"key": "two", "task_type": "background_change", "instruction_language": "en"}, + ] + updates = [] + + monkeypatch.setattr("benchmarks.accuracy.image_to_image.gedit_bench._load_gedit_dataset", lambda ref: rows) + runner = GEditBenchRunner( + dataset_ref="dataset", + output_root=tmp_path, + base_url="http://127.0.0.1:8093", + model="model", + ) + + def fake_generate_one(self, model_name, item): + return { + "key": item["key"], + "task_type": item["task_type"], + "instruction_language": item["instruction_language"], + } + + monkeypatch.setattr(GEditBenchRunner, "_generate_one", fake_generate_one) + + class FakeTqdm: + def __init__(self, total, desc, unit): + self.total = total + self.desc = desc + self.unit = unit + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def update(self, value): + updates.append(value) + + monkeypatch.setattr("benchmarks.accuracy.image_to_image.gedit_bench.tqdm", FakeTqdm) + + runner.generate(model_name="demo", workers=1) + + assert updates == [1, 1] + + +def test_gedit_evaluator_skips_failed_samples(monkeypatch, tmp_path: Path): + rows = [ + {"key": "ok", "task_type": "background_change", "instruction_language": "en"}, + {"key": "bad", "task_type": "background_change", "instruction_language": "en"}, + ] + + monkeypatch.setattr("benchmarks.accuracy.image_to_image.gedit_bench._load_gedit_dataset", lambda ref: rows) + evaluator = GEditBenchEvaluator(dataset_ref="dataset", output_root=tmp_path / "results", scorer=object()) + + def fake_evaluate_one(self, model_name, item): + if item["key"] == "bad": + raise RuntimeError("boom") + return { + "key": item["key"], + "task_type": item["task_type"], + "edited_image": "ok.png", + "instruction": "edit", + "semantics_score": 8.0, + "quality_score": 7.0, + "overall_score": math.sqrt(56.0), + "intersection_exist": True, + "instruction_language": item["instruction_language"], + } + + monkeypatch.setattr(GEditBenchEvaluator, "_evaluate_one", fake_evaluate_one) + monkeypatch.setattr( + "benchmarks.accuracy.image_to_image.gedit_bench._utc_timestamp", + lambda: "20260325T120000Z", + ) + + payload = evaluator.evaluate( + model_name="demo", + save_dir=tmp_path / "scores", + instruction_language="en", + workers=1, + ) + + assert len(payload["results"]) == 1 + assert payload["results"][0]["key"] == "ok" + assert payload["summary"]["overall"]["count"] == 1 + assert Path(payload["csv_path"]).name == "demo_all_en_vie_score.csv" + assert Path(payload["summary_path"]).name == "demo_all_en_summary.json" + assert Path(payload["timestamped_csv_path"]).name == "demo_all_en_vie_score_20260325T120000Z.csv" + assert Path(payload["timestamped_summary_path"]).name == "demo_all_en_summary_20260325T120000Z.json" + assert Path(payload["timestamped_csv_path"]).exists() + assert Path(payload["timestamped_summary_path"]).exists() + + +def test_summarize_gedit_rows_computes_group_and_intersection_means(): + rows = [] + for group in GEDIT_GROUPS: + rows.append( + { + "task_type": group, + "instruction_language": "en", + "semantics_score": 8.0, + "quality_score": 9.0, + "intersection_exist": True, + } + ) + rows.append( + { + "task_type": group, + "instruction_language": "en", + "semantics_score": 6.0, + "quality_score": 4.0, + "intersection_exist": False, + } + ) + + summary = summarize_gedit_rows(rows, language="en") + + expected_overall = (math.sqrt(8.0 * 9.0) + math.sqrt(6.0 * 4.0)) / 2 + assert math.isclose(summary["overall"]["Q_SC"], 7.0) + assert math.isclose(summary["overall"]["Q_PQ"], 6.5) + assert math.isclose(summary["overall"]["Q_O"], expected_overall) + assert math.isclose(summary["intersection"]["Q_SC"], 8.0) + + +def test_summarize_gedit_rows_uses_macro_average_across_groups(): + rows = [] + for idx in range(10): + rows.append( + { + "task_type": "background_change", + "instruction_language": "en", + "semantics_score": 10.0, + "quality_score": 10.0, + "intersection_exist": True, + } + ) + for group in GEDIT_GROUPS[1:]: + rows.append( + { + "task_type": group, + "instruction_language": "en", + "semantics_score": 1.0, + "quality_score": 1.0, + "intersection_exist": True, + } + ) + + summary = summarize_gedit_rows_with_backbone(rows, language="en") + + expected_macro = (10.0 + 10.0 * 1.0) / 11 + assert math.isclose(summary["overall"]["Q_SC"], expected_macro) + assert math.isclose(summary["overall"]["Q_O"], expected_macro) + assert math.isclose(summary["by_group"]["background_change"]["Q_SC"], 10.0) + + +def test_summarize_gedit_rows_with_all_language_splits_en_and_cn(): + rows = [] + for group in GEDIT_GROUPS: + rows.append( + { + "task_type": group, + "instruction_language": "en", + "semantics_score": 8.0, + "quality_score": 6.0, + "intersection_exist": True, + } + ) + rows.append( + { + "task_type": group, + "instruction_language": "cn", + "semantics_score": 4.0, + "quality_score": 2.0, + "intersection_exist": True, + } + ) + + summary = summarize_gedit_rows_with_backbone(rows, language="all") + + assert set(summary["languages"]) == {"en", "cn"} + assert math.isclose(summary["languages"]["en"]["overall"]["Q_SC"], 8.0) + assert math.isclose(summary["languages"]["en"]["overall"]["Q_PQ"], 6.0) + assert math.isclose(summary["languages"]["cn"]["overall"]["Q_O"], math.sqrt(8.0)) diff --git a/tests/conftest.py b/tests/conftest.py index 0d90f639d..49b18fdae 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -54,6 +54,8 @@ class OmniServerParams(NamedTuple): port: int | None = None stage_config_path: str | None = None server_args: list[str] | None = None + env_dict: dict[str, str] | None = None + use_omni: bool = True def assert_image_valid(image: Path | Image.Image, *, width: int | None = None, height: int | None = None): @@ -1076,6 +1078,7 @@ class OmniServer: *, port: int | None = None, env_dict: dict[str, str] | None = None, + use_omni: bool = True, ) -> None: _run_pre_test_cleanup(enable_force=True) _run_post_test_cleanup(enable_force=True) @@ -1083,6 +1086,7 @@ class OmniServer: self.model = model self.serve_args = serve_args self.env_dict = env_dict + self.use_omni = use_omni self.proc: subprocess.Popen | None = None self.host = "127.0.0.1" if port is None: @@ -1103,12 +1107,14 @@ class OmniServer: "vllm_omni.entrypoints.cli.main", "serve", self.model, - "--omni", "--host", self.host, "--port", str(self.port), - ] + self.serve_args + ] + if self.use_omni: + cmd.append("--omni") + cmd += self.serve_args print(f"Launching OmniServer with: {' '.join(cmd)}") self.proc = subprocess.Popen( @@ -1237,19 +1243,35 @@ def omni_server(request: pytest.FixtureRequest, run_level: str, model_prefix: st stage_config_path = params.stage_config_path if run_level == "advanced_model" and stage_config_path is not None: with open(stage_config_path, encoding="utf-8") as f: - _cfg = yaml.safe_load(f) or {} - _stage_ids = [s["stage_id"] for s in _cfg.get("stage_args", []) if "stage_id" in s] + cfg = yaml.safe_load(f) or {} + stage_ids = [stage["stage_id"] for stage in cfg.get("stage_args", []) if "stage_id" in stage] stage_config_path = modify_stage_config( stage_config_path, - deletes={"stage_args": {sid: ["engine_args.load_format"] for sid in _stage_ids}}, + deletes={"stage_args": {stage_id: ["engine_args.load_format"] for stage_id in stage_ids}}, ) server_args = params.server_args or [] - server_args = ["--stage-init-timeout", "120", *server_args] + if params.use_omni: + server_args = ["--stage-init-timeout", "120", *server_args] if stage_config_path is not None: server_args += ["--stage-configs-path", stage_config_path] - with OmniServer(model, server_args, port=port) if port else OmniServer(model, server_args) as server: + with ( + OmniServer( + model, + server_args, + port=port, + env_dict=params.env_dict, + use_omni=params.use_omni, + ) + if port + else OmniServer( + model, + server_args, + env_dict=params.env_dict, + use_omni=params.use_omni, + ) + ) as server: print("OmniServer started successfully") yield server print("OmniServer stopping...") diff --git a/tests/e2e/accuracy/conftest.py b/tests/e2e/accuracy/conftest.py new file mode 100644 index 000000000..96ac20146 --- /dev/null +++ b/tests/e2e/accuracy/conftest.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +import os +import shutil +import subprocess +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path + +import pytest +import torch + +from tests.conftest import OmniServer, OmniServerParams + + +def pytest_addoption(parser): + group = parser.getgroup("accuracy-e2e") + group.addoption("--gebench-root", action="store", default=None, help="Local GEBench dataset root") + group.addoption("--gedit-root", action="store", default=None, help="Local GEdit-Bench dataset root") + group.addoption( + "--gebench-model", action="store", default="Qwen/Qwen-Image-2512", help="Generate model for GEBench smoke" + ) + group.addoption( + "--gedit-model", action="store", default="Qwen/Qwen-Image-Edit", help="Generate model for GEdit-Bench smoke" + ) + group.addoption( + "--accuracy-judge-model", + action="store", + default="QuantTrio/Qwen3-VL-30B-A3B-Instruct-AWQ", + help="Judge model path", + ) + group.addoption("--accuracy-gpu", action="store", default="0", help="Single GPU id used sequentially") + group.addoption("--gebench-port", action="store", type=int, default=8093, help="Generate port for GEBench") + group.addoption("--gedit-port", action="store", type=int, default=8093, help="Generate port for GEdit-Bench") + group.addoption( + "--gebench-samples-per-type", + action="store", + type=int, + default=10, + help="Balanced sample count per GEBench type", + ) + group.addoption( + "--gedit-samples-per-group", + action="store", + type=int, + default=20, + help="Balanced sample count per GEdit task group", + ) + group.addoption("--accuracy-workers", action="store", type=int, default=1, help="Worker count for accuracy benches") + + +def _hf_cache_root() -> Path: + return Path(os.environ.get("HF_HOME", "/root/.cache/huggingface")) + + +def _dataset_cache_dirs(dataset_id: str) -> list[Path]: + cache_root = _hf_cache_root() / "hub" / f"datasets--{dataset_id.replace('/', '--')}" / "snapshots" + if not cache_root.exists(): + return [] + return sorted( + (path for path in cache_root.iterdir() if path.is_dir()), key=lambda path: path.stat().st_mtime, reverse=True + ) + + +def _ensure_dataset_snapshot(dataset_id: str) -> Path: + candidates = _dataset_cache_dirs(dataset_id) + if candidates: + return candidates[0] + + subprocess.run( + ["huggingface-cli", "download", "--repo-type", "dataset", dataset_id], + check=True, + ) + candidates = _dataset_cache_dirs(dataset_id) + if not candidates: + raise FileNotFoundError( + f"Dataset {dataset_id} was downloaded but no snapshot was found under {_hf_cache_root()}" + ) + return candidates[0] + + +def _resolve_dataset_root(request: pytest.FixtureRequest, option_name: str, dataset_id: str) -> Path: + value = request.config.getoption(option_name) + if value: + path = Path(value) + if not path.exists(): + pytest.skip(f"Dataset path does not exist: {path}") + return path + return _ensure_dataset_snapshot(dataset_id) + + +@dataclass +class AccuracyServerConfig: + generate_params: OmniServerParams + judge_params: OmniServerParams + run_level: str + model_prefix: str + + @contextmanager + def generate_server(self): + params = self.generate_params + model = self.model_prefix + params.model + server_args = params.server_args or [] + if params.use_omni: + server_args = ["--stage-init-timeout", "120", *server_args] + with OmniServer( + model, + server_args, + port=params.port, + env_dict=params.env_dict, + use_omni=params.use_omni, + ) as server: + yield server + + @contextmanager + def judge_server(self): + params = self.judge_params + model = self.model_prefix + params.model + server_args = params.server_args or [] + with OmniServer( + model, + server_args, + port=params.port, + env_dict=params.env_dict, + use_omni=params.use_omni, + ) as server: + yield server + + +@pytest.fixture(scope="session") +def gebench_dataset_root(request: pytest.FixtureRequest) -> Path: + return _resolve_dataset_root(request, "gebench_root", "stepfun-ai/GEBench") + + +@pytest.fixture(scope="session") +def gedit_dataset_root(request: pytest.FixtureRequest) -> Path: + return _resolve_dataset_root(request, "gedit_root", "stepfun-ai/GEdit-Bench") + + +@pytest.fixture(scope="session") +def accuracy_workers(request: pytest.FixtureRequest) -> int: + return int(request.config.getoption("accuracy_workers")) + + +@pytest.fixture(scope="session") +def gebench_samples_per_type(request: pytest.FixtureRequest) -> int: + return int(request.config.getoption("gebench_samples_per_type")) + + +@pytest.fixture(scope="session") +def gedit_samples_per_group(request: pytest.FixtureRequest) -> int: + return int(request.config.getoption("gedit_samples_per_group")) + + +@pytest.fixture(scope="session") +def accuracy_artifact_root() -> Path: + root = Path(__file__).resolve().parent / "artifacts" + root.mkdir(parents=True, exist_ok=True) + return root + + +def reset_artifact_dir(path: Path) -> Path: + if path.exists(): + shutil.rmtree(path) + path.mkdir(parents=True, exist_ok=True) + return path + + +def infer_model_label(model: str) -> str: + label = Path(model.rstrip("/\\")).name or "model" + return "".join(char if char.isalnum() or char in {"-", "_"} else "_" for char in label) + + +def _build_accuracy_server_config( + *, + generate_model: str, + judge_model: str, + shared_gpu: str, + port: int, + run_level: str, + model_prefix: str, +) -> AccuracyServerConfig: + if torch.cuda.device_count() < 1: + pytest.skip("Need at least 1 CUDA GPU for accuracy benchmark smoke tests.") + + if not generate_model: + pytest.skip("No generate model configured for accuracy benchmark test.") + generate_server_args = ["--num-gpus", "1"] + judge_server_args = [ + "--max-model-len", + "32768", + "--gpu-memory-utilization", + "0.8", + ] + + judge_env = {"CUDA_VISIBLE_DEVICES": shared_gpu} + + return AccuracyServerConfig( + generate_params=OmniServerParams( + model=generate_model, + port=port, + server_args=generate_server_args, + env_dict={"CUDA_VISIBLE_DEVICES": shared_gpu}, + use_omni=True, + ), + judge_params=OmniServerParams( + model=judge_model, + port=port, + server_args=judge_server_args, + env_dict=judge_env, + use_omni=False, + ), + run_level=run_level, + model_prefix=model_prefix, + ) + + +@pytest.fixture +def gebench_accuracy_servers( + request: pytest.FixtureRequest, + run_level: str, + model_prefix: str, +) -> AccuracyServerConfig: + return _build_accuracy_server_config( + generate_model=request.config.getoption("gebench_model"), + judge_model=request.config.getoption("accuracy_judge_model"), + shared_gpu=str(request.config.getoption("accuracy_gpu")), + port=int(request.config.getoption("gebench_port")), + run_level=run_level, + model_prefix=model_prefix, + ) + + +@pytest.fixture +def gedit_accuracy_servers( + request: pytest.FixtureRequest, + run_level: str, + model_prefix: str, +) -> AccuracyServerConfig: + return _build_accuracy_server_config( + generate_model=request.config.getoption("gedit_model"), + judge_model=request.config.getoption("accuracy_judge_model"), + shared_gpu=str(request.config.getoption("accuracy_gpu")), + port=int(request.config.getoption("gedit_port")), + run_level=run_level, + model_prefix=model_prefix, + ) diff --git a/tests/e2e/accuracy/test_gebench_h100_smoke.py b/tests/e2e/accuracy/test_gebench_h100_smoke.py new file mode 100644 index 000000000..b4b831871 --- /dev/null +++ b/tests/e2e/accuracy/test_gebench_h100_smoke.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from benchmarks.accuracy.text_to_image.gbench import main as gbench_main +from tests.e2e.accuracy.conftest import infer_model_label, reset_artifact_dir +from tests.utils import hardware_test + + +@pytest.mark.advanced_model +@pytest.mark.benchmark +@pytest.mark.diffusion +@hardware_test(res={"cuda": "H100"}, num_cards=1) +def test_gebench_h100_smoke( + gebench_accuracy_servers, + accuracy_artifact_root: Path, + gebench_dataset_root: Path, + accuracy_workers: int, +) -> None: + model_label = infer_model_label(gebench_accuracy_servers.generate_params.model).lower() + output_root = reset_artifact_dir(accuracy_artifact_root / f"gebench_{model_label}") + + with gebench_accuracy_servers.generate_server() as generate_server: + for data_type in ("type3", "type4"): + assert ( + gbench_main( + [ + "generate", + "--dataset-root", + str(gebench_dataset_root), + "--output-root", + str(output_root), + "--base-url", + f"http://{generate_server.host}:{generate_server.port}", + "--model", + generate_server.model, + "--data-type", + data_type, + "--width", + "768", + "--height", + "576", + "--output-compression", + "98", + "--num-inference-steps", + "8", + "--workers", + str(accuracy_workers), + ] + ) + == 0 + ) + + with gebench_accuracy_servers.judge_server() as judge_server: + for data_type in ("type3", "type4"): + assert ( + gbench_main( + [ + "evaluate", + "--dataset-root", + str(gebench_dataset_root), + "--output-root", + str(output_root), + "--data-type", + data_type, + "--judge-base-url", + f"http://{judge_server.host}:{judge_server.port}", + "--judge-model", + judge_server.model, + "--judge-api-key", + "EMPTY", + "--workers", + str(accuracy_workers), + ] + ) + == 0 + ) + + assert gbench_main(["summarize", "--output-root", str(output_root)]) == 0 + + summary = json.loads((output_root / "summary.json").read_text(encoding="utf-8")) + assert "generation" in summary + assert "evaluation" in summary + + for data_type in ("type3", "type4"): + assert data_type in summary["generation"]["by_type"] + assert summary["generation"]["by_type"][data_type]["count"] > 0 + assert data_type in summary["evaluation"]["by_type"] + assert summary["evaluation"]["by_type"][data_type]["count"] > 0 + + assert summary["evaluation"]["overall_mean"] >= 0.45 + assert summary["evaluation"]["by_type"]["type3"]["overall_mean"] >= 0.45 + assert summary["evaluation"]["by_type"]["type4"]["overall_mean"] >= 0.45 diff --git a/tests/e2e/accuracy/test_gedit_bench_h100_smoke.py b/tests/e2e/accuracy/test_gedit_bench_h100_smoke.py new file mode 100644 index 000000000..80359fb13 --- /dev/null +++ b/tests/e2e/accuracy/test_gedit_bench_h100_smoke.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from benchmarks.accuracy.image_to_image.gedit_bench import GROUPS +from benchmarks.accuracy.image_to_image.gedit_bench import main as gedit_main +from tests.e2e.accuracy.conftest import infer_model_label, reset_artifact_dir +from tests.utils import hardware_test + + +@pytest.mark.advanced_model +@pytest.mark.benchmark +@pytest.mark.diffusion +@hardware_test(res={"cuda": "H100"}, num_cards=1) +def test_gedit_bench_h100_smoke( + gedit_accuracy_servers, + accuracy_artifact_root: Path, + gedit_dataset_root: Path, + gedit_samples_per_group: int, + accuracy_workers: int, +) -> None: + model_label = infer_model_label(gedit_accuracy_servers.generate_params.model).lower() + output_root = reset_artifact_dir(accuracy_artifact_root / f"gedit_results_{model_label}") + score_root = reset_artifact_dir(accuracy_artifact_root / f"gedit_scores_{model_label}") + model_name = model_label + + with gedit_accuracy_servers.generate_server() as generate_server: + assert ( + gedit_main( + [ + "generate", + "--dataset-ref", + str(gedit_dataset_root), + "--output-root", + str(output_root), + "--base-url", + f"http://{generate_server.host}:{generate_server.port}", + "--model", + generate_server.model, + "--model-name", + model_name, + "--task-type", + "all", + "--instruction-language", + "all", + "--samples-per-group", + str(gedit_samples_per_group), + "--workers", + str(accuracy_workers), + ] + ) + == 0 + ) + + with gedit_accuracy_servers.judge_server() as judge_server: + assert ( + gedit_main( + [ + "evaluate", + "--dataset-ref", + str(gedit_dataset_root), + "--output-root", + str(output_root), + "--model-name", + model_name, + "--save-dir", + str(score_root), + "--task-type", + "all", + "--instruction-language", + "all", + "--judge-base-url", + f"http://{judge_server.host}:{judge_server.port}", + "--judge-model", + judge_server.model, + "--judge-api-key", + "EMPTY", + "--samples-per-group", + str(gedit_samples_per_group), + "--workers", + str(accuracy_workers), + ] + ) + == 0 + ) + + csv_path = score_root / f"{model_name}_all_all_vie_score.csv" + assert gedit_main(["summarize", "--csv-path", str(csv_path), "--language", "all"]) == 0 + + summary_path = score_root / f"{model_name}_all_all_summary.json" + summary = json.loads(summary_path.read_text(encoding="utf-8")) + assert set(summary["languages"]) == {"en", "cn"} + + for language in ["en", "cn"]: + language_summary = summary["languages"][language] + assert language_summary["overall"]["count"] is not None + assert language_summary["intersection"]["count"] is not None + assert language_summary["overall"]["Q_SC"] is not None + assert language_summary["overall"]["Q_PQ"] is not None + assert language_summary["overall"]["Q_O"] is not None + + for group in GROUPS: + group_summary = language_summary["by_group"][group] + assert set(group_summary) == {"count", "Q_SC", "Q_PQ", "Q_O"} + + assert summary["languages"]["en"]["overall"]["Q_SC"] >= 7.1 + assert summary["languages"]["en"]["overall"]["Q_PQ"] >= 5.9 + assert summary["languages"]["en"]["overall"]["Q_O"] >= 6.3 + assert summary["languages"]["cn"]["overall"]["Q_SC"] >= 7.0 + assert summary["languages"]["cn"]["overall"]["Q_PQ"] >= 5.8 + assert summary["languages"]["cn"]["overall"]["Q_O"] >= 6.2