diff --git a/.github/workflows/validate-notebooks.yml b/.github/workflows/validate-notebooks.yml index 716a00d9..b189dc4c 100644 --- a/.github/workflows/validate-notebooks.yml +++ b/.github/workflows/validate-notebooks.yml @@ -97,5 +97,6 @@ jobs: shell: bash run: | mapfile -d '' notebooks < <(git ls-files -z -- '*.ipynb') + (( ${#notebooks[@]} )) || exit 0 # Catch syntax and runtime-invalid constructs without enforcing style on legacy notebooks. ruff check --output-format=github --select=E9,F63,F7,F82 -- "${notebooks[@]}" diff --git a/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_fd_droid_posttrain.sh b/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_fd_droid_posttrain.sh index 02b393be..336f3b46 100644 --- a/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_fd_droid_posttrain.sh +++ b/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_fd_droid_posttrain.sh @@ -101,6 +101,13 @@ fi TRAILING_ARGS=(-- "${TAIL_OVERRIDES[@]}") TORCHRUN_ARGS=(--nproc_per_node="${NPROC_PER_NODE:-8}" --master_port="${MASTER_PORT:-50012}") +if [[ -n "${NNODES:-}" || -n "${NODE_RANK:-}" || -n "${MASTER_ADDR:-}" ]]; then + # Fail fast with an actionable message instead of torchrun's opaque + # rendezvous error later. + : "${NNODES:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" + : "${NODE_RANK:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" + : "${MASTER_ADDR:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" +fi [[ -n "${NNODES:-}" ]] && TORCHRUN_ARGS+=(--nnodes="$NNODES") [[ -n "${NODE_RANK:-}" ]] && TORCHRUN_ARGS+=(--node_rank="$NODE_RANK") [[ -n "${MASTER_ADDR:-}" ]] && TORCHRUN_ARGS+=(--master_addr="$MASTER_ADDR") diff --git a/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_droid_nano.sh b/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_droid_nano.sh index 9389fdf9..c9495444 100755 --- a/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_droid_nano.sh +++ b/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_droid_nano.sh @@ -71,6 +71,13 @@ fi TORCHRUN_ARGS=(--nproc_per_node="${NPROC_PER_NODE:-8}") TORCHRUN_ARGS+=(--master_port="${MASTER_PORT:-50012}") +if [[ -n "${NNODES:-}" || -n "${NODE_RANK:-}" || -n "${MASTER_ADDR:-}" ]]; then + # Fail fast with an actionable message instead of torchrun's opaque + # rendezvous error later. + : "${NNODES:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" + : "${NODE_RANK:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" + : "${MASTER_ADDR:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" +fi [[ -n "${NNODES:-}" ]] && TORCHRUN_ARGS+=(--nnodes="$NNODES") [[ -n "${NODE_RANK:-}" ]] && TORCHRUN_ARGS+=(--node_rank="$NODE_RANK") [[ -n "${MASTER_ADDR:-}" ]] && TORCHRUN_ARGS+=(--master_addr="$MASTER_ADDR") diff --git a/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_10_nano.sh b/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_10_nano.sh index 550d195a..21eda944 100755 --- a/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_10_nano.sh +++ b/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_10_nano.sh @@ -61,6 +61,13 @@ fi TORCHRUN_ARGS=(--nproc_per_node="${NPROC_PER_NODE:-8}") TORCHRUN_ARGS+=(--master_port="${MASTER_PORT:-50012}") +if [[ -n "${NNODES:-}" || -n "${NODE_RANK:-}" || -n "${MASTER_ADDR:-}" ]]; then + # Fail fast with an actionable message instead of torchrun's opaque + # rendezvous error later. + : "${NNODES:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" + : "${NODE_RANK:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" + : "${MASTER_ADDR:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" +fi [[ -n "${NNODES:-}" ]] && TORCHRUN_ARGS+=(--nnodes="$NNODES") [[ -n "${NODE_RANK:-}" ]] && TORCHRUN_ARGS+=(--node_rank="$NODE_RANK") [[ -n "${MASTER_ADDR:-}" ]] && TORCHRUN_ARGS+=(--master_addr="$MASTER_ADDR") diff --git a/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_all_nano.sh b/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_all_nano.sh index fb410e07..d139c458 100755 --- a/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_all_nano.sh +++ b/cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_all_nano.sh @@ -68,6 +68,13 @@ fi TORCHRUN_ARGS=(--nproc_per_node="${NPROC_PER_NODE:-8}") TORCHRUN_ARGS+=(--master_port="${MASTER_PORT:-50012}") +if [[ -n "${NNODES:-}" || -n "${NODE_RANK:-}" || -n "${MASTER_ADDR:-}" ]]; then + # Fail fast with an actionable message instead of torchrun's opaque + # rendezvous error later. + : "${NNODES:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" + : "${NODE_RANK:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" + : "${MASTER_ADDR:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}" +fi [[ -n "${NNODES:-}" ]] && TORCHRUN_ARGS+=(--nnodes="$NNODES") [[ -n "${NODE_RANK:-}" ]] && TORCHRUN_ARGS+=(--node_rank="$NODE_RANK") [[ -n "${MASTER_ADDR:-}" ]] && TORCHRUN_ARGS+=(--master_addr="$MASTER_ADDR") diff --git a/cookbooks/cosmos3/generator/transfer/preview_helpers.py b/cookbooks/cosmos3/generator/transfer/preview_helpers.py index 8f3bd82e..2fdb241f 100644 --- a/cookbooks/cosmos3/generator/transfer/preview_helpers.py +++ b/cookbooks/cosmos3/generator/transfer/preview_helpers.py @@ -71,27 +71,36 @@ def load_transfer_spec(control: str) -> dict: def make_preview(src: Path, crf: int = 28) -> Path: preview = src.with_name(f"{src.stem}_preview.mp4") if not preview.exists() or preview.stat().st_mtime < src.stat().st_mtime: - subprocess.run( - [ - _ffmpeg_exe(), - "-y", - "-loglevel", - "error", - "-i", - str(src), - "-c:v", - "libx264", - "-crf", - str(crf), - "-preset", - "veryfast", - "-an", - "-pix_fmt", - "yuv420p", - str(preview), - ], - check=True, - ) + # Encode beside the target (keeping the .mp4 suffix so ffmpeg infers + # the container), then swap atomically so an interrupted cell never + # leaves a half-written video behind under the final name. + partial = src.with_name(f"{src.stem}_preview.partial.mp4") + try: + subprocess.run( + [ + _ffmpeg_exe(), + "-y", + "-loglevel", + "error", + "-i", + str(src), + "-c:v", + "libx264", + "-crf", + str(crf), + "-preset", + "veryfast", + "-an", + "-pix_fmt", + "yuv420p", + str(partial), + ], + check=True, + ) + os.replace(partial, preview) + except BaseException: + partial.unlink(missing_ok=True) + raise return preview diff --git a/evaluation/cosmos3/generator/paibench_c/run_paibench_c.sh b/evaluation/cosmos3/generator/paibench_c/run_paibench_c.sh index 793aedbb..1aaab7bd 100755 --- a/evaluation/cosmos3/generator/paibench_c/run_paibench_c.sh +++ b/evaluation/cosmos3/generator/paibench_c/run_paibench_c.sh @@ -10,11 +10,11 @@ # automatically on first run. # # --- Quick start -------------------------------------------------------------- -# # Smoke-test: 1 task, edge, Cosmos3-Nano +# # Default demo: 4 tasks, edge control, Cosmos3-Nano # bash run_paibench_c.sh # -# # Demo with 4 tasks (default) -# bash run_paibench_c.sh +# # Minimal smoke: 1 task, edge control +# PAIBENCH_C_NUM_SAMPLES=1 bash run_paibench_c.sh # # # Full 600-task run, all modalities, Cosmos3-Super # PAIBENCH_C_NUM_SAMPLES=600 \ @@ -704,6 +704,16 @@ unset MPLBACKEND # prevent Jupyter's inline backend from leaking into subproces _ckpt_slug="${PAIBENCH_C_CHECKPOINT//\//-}" +# Evaluation shards one torchrun worker per visible GPU. Recompute the real CUDA +# device list here instead of trusting COSMOS3_NUM_GPUS, so bumping that knob +# past what CUDA_VISIBLE_DEVICES exports cannot oversubscribe workers below. +IFS=',' read -r -a _eval_gpu_ids <<< "${CUDA_VISIBLE_DEVICES}" +_max_eval_gpus=${#_eval_gpu_ids[@]} +if (( _max_eval_gpus == 0 )); then + log " WARNING: CUDA_VISIBLE_DEVICES exposes no GPUs; capping evaluation at 1 GPU" + _max_eval_gpus=1 +fi + # Use a robust python for printing results - prefer cosmos3 venv if functional, # fall back to system python3 (only stdlib used). _py_for_summary() { @@ -720,6 +730,10 @@ for _mod in $PAIBENCH_C_MODALITIES; do _videos_dir="$_videos_parent/videos" _metrics_out="$_videos_parent/metrics.json" _eval_ngpu="$(int_min "$PAIBENCH_C_NUM_SAMPLES" "$COSMOS3_NUM_GPUS")" + if (( _eval_ngpu > _max_eval_gpus )); then + log " WARNING: clamping evaluation GPUs $_eval_ngpu -> $_max_eval_gpus (visible: $CUDA_VISIBLE_DEVICES)" + _eval_ngpu=$_max_eval_gpus + fi # Verify videos exist before launching torchrun to surface errors early. if [[ ! -d "$_videos_dir" ]]; then diff --git a/evaluation/cosmos3/generator/paibench_g/merge_result_selfcheck.py b/evaluation/cosmos3/generator/paibench_g/merge_result_selfcheck.py new file mode 100644 index 00000000..bceb9516 --- /dev/null +++ b/evaluation/cosmos3/generator/paibench_g/merge_result_selfcheck.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Self-check for run_motion_smoothness_sharded.write_merged_result (CPU-only). + +Covers the merge-step contract, including the partial-results behaviour added so +permanently failed videos no longer discard cleanly computed scores: + +1. complete run -> overall mean, no missing list in the payload; +2. failed video -> partial mean plus result-file ``missing_videos`` key; +3. strict mode -> default call still raises when gaps exist; +4. foreign records -> records outside the requested set are always rejected. + +Run: python merge_result_selfcheck.py (exit 0 == pass) +""" + +from __future__ import annotations + +import json +import math +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from run_motion_smoothness_sharded import write_merged_result + + +def main() -> int: + with tempfile.TemporaryDirectory() as tmp_str: + tmp = Path(tmp_str) + + # --- 1. complete run -------------------------------------------------- + d1 = tmp / "complete" + d1.mkdir() + r1 = d1 / "result.json" + # Absolute canonical temp paths: Path(...).absolute() leaves them intact, + # matching how collect_saved_records keys records on every platform. + vids = [str(tmp / f"v{i}.mp4") for i in range(3)] + (d1 / "worker_00.json").write_text(json.dumps([ + {"video_path": vids[0], "video_results": 0.5}, + {"video_path": vids[1], "video_results": 0.25}, + ])) + (d1 / "worker_01.json").write_text(json.dumps([ + {"video_path": vids[2], "video_results": 0.25}, + ])) + score, missing = write_merged_result(output_dir=d1, result_file=r1, videos=vids) + assert math.isclose(score, 1 / 3, rel_tol=1e-12), score + assert missing == [], missing + assert "missing_videos" not in json.loads(r1.read_text()) + + # --- 2. permanently failed video -> partial payload -------------------- + d2 = tmp / "partial" + d2.mkdir() + r2 = d2 / "result.json" + vids2 = [str(tmp / f"w{i}.mp4") for i in range(2)] + (d2 / "worker_00.json").write_text(json.dumps([ + {"video_path": vids2[0], "video_results": 0.75}, + ])) + score_b, missing_b = write_merged_result( + output_dir=d2, result_file=r2, videos=vids2, require_complete=False + ) + assert score_b == 0.75 and missing_b == [vids2[1]], (score_b, missing_b) + payload = json.loads(r2.read_text()) + assert payload["missing_videos"] == [vids2[1]] + assert len(payload["motion_smoothness"][1]) == 1 + + # --- 3. strict mode (default) rejects gaps ----------------------------- + try: + write_merged_result(output_dir=d2, result_file=r2, videos=vids2) + except RuntimeError: + pass + else: + raise AssertionError("strict merge accepted missing results") + + # --- 4. records outside the requested set are rejected ------------------ + d4 = tmp / "unexpected" + d4.mkdir() + (d4 / "worker_00.json").write_text(json.dumps([ + {"video_path": str(tmp / "stray.mp4"), "video_results": 0.0}, + ])) + try: + write_merged_result( + output_dir=d4, result_file=d4 / "result.json", + videos=[str(tmp / "kept.mp4")], + ) + except RuntimeError as exc: + assert "outside the input manifest" in str(exc), exc + else: + raise AssertionError("unexpected record accepted") + + print("merge_result_selfcheck: PASS (complete/partial/strict/unexpected)") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/evaluation/cosmos3/generator/paibench_g/run_motion_smoothness_sharded.py b/evaluation/cosmos3/generator/paibench_g/run_motion_smoothness_sharded.py index c5cabea5..95b569ae 100644 --- a/evaluation/cosmos3/generator/paibench_g/run_motion_smoothness_sharded.py +++ b/evaluation/cosmos3/generator/paibench_g/run_motion_smoothness_sharded.py @@ -15,6 +15,9 @@ RUNNER_VERSION = 2 +# Quick second attempt before recording a video as permanently errored; AMT +# scoring failures (decoder hiccups, transient CUDA errors) are often transient. +VIDEO_RETRIES = 1 def parse_args() -> argparse.Namespace: @@ -156,32 +159,50 @@ def worker_main( continue if video_path is None: break - try: - score = float(motion.motion_score(video_path)) + final_error = "" + for _attempt in range(1 + VIDEO_RETRIES): + try: + score = float(motion.motion_score(video_path)) + except Exception: + final_error = traceback.format_exc() + continue records.append({"video_path": video_path, "video_results": score}) write_json_atomic(output_path, records) status_queue.put(("done", worker_id, video_path, score)) - except Exception: - status_queue.put(("error", worker_id, video_path, traceback.format_exc())) + break + else: + status_queue.put(("error", worker_id, video_path, final_error)) status_queue.put(("exit", worker_id, physical_gpu, "")) except Exception: status_queue.put(("fatal", worker_id, physical_gpu, traceback.format_exc())) raise -def write_merged_result(*, output_dir: Path, result_file: Path, videos: list[str]) -> float: +def write_merged_result( + *, + output_dir: Path, + result_file: Path, + videos: list[str], + require_complete: bool = True, +) -> tuple[float, list[str]]: saved = collect_saved_records(output_dir) target = set(videos) unexpected = set(saved) - target if unexpected: raise RuntimeError(f"found {len(unexpected)} results outside the input manifest") missing = [video_path for video_path in videos if video_path not in saved] - if missing: + if missing and require_complete: raise RuntimeError(f"missing {len(missing)} motion-smoothness results") - ordered = [saved[video_path] for video_path in videos] + scored = [video_path for video_path in videos if video_path in saved] + if not scored: + raise RuntimeError("no motion-smoothness results were computed") + ordered = [saved[video_path] for video_path in scored] mean_score = fmean(record["video_results"] for record in ordered) - write_json_atomic(result_file, {"motion_smoothness": [mean_score, ordered]}) - return mean_score + payload: dict[str, object] = {"motion_smoothness": [mean_score, ordered]} + if missing: + payload["missing_videos"] = missing + write_json_atomic(result_file, payload) + return mean_score, missing def main() -> None: @@ -248,6 +269,7 @@ def main() -> None: flush=True, ) + run_failed = False if pending: context = mp.get_context("spawn") task_queue = context.Queue() @@ -314,8 +336,13 @@ def main() -> None: for process in workers: process.join() bad_exit_codes = [process.exitcode for process in workers if process.exitcode != 0] - if bad_exit_codes or failures: - raise RuntimeError(f"worker failures={len(failures)} exit_codes={bad_exit_codes}") + run_failed = bool(bad_exit_codes or failures) + if run_failed: + print( + f"workers finished with failures={len(failures)} " + f"exit_codes={bad_exit_codes}; merging partial results", + flush=True, + ) except BaseException: for process in workers: if process.is_alive(): @@ -329,11 +356,21 @@ def main() -> None: task_queue.close() status_queue.close() - mean_score = write_merged_result( + # Persist whatever scored cleanly even when workers failed mid-run; the + # missing entries stay discoverable via the result file's "missing_videos". + mean_score, missing = write_merged_result( output_dir=output_dir, result_file=result_file, videos=videos, + require_complete=not run_failed, ) + if run_failed: + print( + f"INCOMPLETE run: scored={len(videos) - len(missing)}/{len(videos)}; " + f"partial results: {result_file} -- re-invoke the runner to rescore missing videos", + flush=True, + ) + sys.exit(1) print(f"motion_smoothness {mean_score:.12f}", flush=True) print(f"saved {result_file}", flush=True) diff --git a/evaluation/cosmos3/generator/unigenbench/query_core_selfcheck.py b/evaluation/cosmos3/generator/unigenbench/query_core_selfcheck.py new file mode 100644 index 00000000..3fe5d186 --- /dev/null +++ b/evaluation/cosmos3/generator/unigenbench/query_core_selfcheck.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""Self-check for GatewayVLM.query_core streaming fixes (no network needed). + +Validates that: +- chunks carrying an empty ``choices`` list (terminal usage-only frames) are + skipped instead of raising IndexError mid-stream; +- frames with ``delta.content is None`` are likewise ignored; +- surrounding text is concatenated intact. + +Run: python query_core_selfcheck.py (exit 0 == pass) + +If loguru/tqdm/openai are not installed, minimal stand-ins are injected so this +can validate query_core's logic with plain stdlib only. +""" + +from __future__ import annotations + +import sys +import types +from pathlib import Path + +for _name in ("loguru", "tqdm", "openai"): + if _name in sys.modules: + continue + try: + __import__(_name) + except ImportError: + _stub = types.ModuleType(_name) + if _name == "loguru": + _stub.logger = types.SimpleNamespace(warning=lambda *_a, **_k: None) + elif _name == "tqdm": + def _fake_pbar(*_a, **_k): + return types.SimpleNamespace(update=lambda *_a: None, close=lambda: None) + _stub.tqdm = _fake_pbar + else: + class _StubAsyncOpenAI: # never contacted; instance replaced below + def __init__(self, **_kwargs): + pass + _stub.AsyncOpenAI = _StubAsyncOpenAI + sys.modules[_name] = _stub + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from utils import GatewayVLM + + +class _Chunk: + def __init__(self, content: object = "__empty__"): + # content="__empty__" models a gateway usage-only frame (no choices); + # content=None models a normal choice whose delta carries no text. + self.choices = ( + [] + if content == "__empty__" + else [types.SimpleNamespace(delta=types.SimpleNamespace(content=content))] + ) + + +class _Stream: + async def __aiter__(self): + yield _Chunk("hel") + yield _Chunk("__empty__") # would crash the pre-fix code here + yield _Chunk("lo wor") + yield _Chunk(None) # ignored by design + yield _Chunk("__empty__") # terminal usage-only frame + yield _Chunk("ld") + + +class _Completions: + async def create(self, **_kwargs): + return _Stream() + + +def main() -> int: + vlm = GatewayVLM( + gateway_url="http://127.0.0.1:1", # unreachable; client replaced below + gateway_api="unused", + model_str="unit-test", + num_concurrency=1, + ) + vlm.client = types.SimpleNamespace( + chat=types.SimpleNamespace(completions=_Completions()) + ) + results = vlm.query([{"model": "unit-test"}]) + vlm.close() + + assert results == ["hello world"], f"unexpected stream result: {results!r}" + print("query_core_selfcheck: PASS (empty-choices and None-content chunks skipped)") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/evaluation/cosmos3/generator/unigenbench/ugb_scorer.py b/evaluation/cosmos3/generator/unigenbench/ugb_scorer.py index 3550adc6..2d204884 100644 --- a/evaluation/cosmos3/generator/unigenbench/ugb_scorer.py +++ b/evaluation/cosmos3/generator/unigenbench/ugb_scorer.py @@ -229,6 +229,13 @@ def parse_vlm_response(text: str, testpoints: list[str]) -> Optional[tuple[list[ return analyses, verdicts +def _as_native_list(value: Any) -> list: + """Accept both native lists and legacy str(repr(list)) result payloads.""" + if isinstance(value, list): + return value + return list(ast.literal_eval(value)) + + def compute_final_metrics(score_breakdown: dict, verbose: bool = True) -> dict: big_class_stats = defaultdict(lambda: [0, 0]) small_class_stats = defaultdict(lambda: [0, 0]) @@ -238,8 +245,8 @@ def compute_final_metrics(score_breakdown: dict, verbose: bool = True) -> dict: if verbose: print(f"Skipping {name}: no result") continue - testpoints = ast.literal_eval(result["testpoints"]) - verdicts = ast.literal_eval(result["verdicts"]) + testpoints = _as_native_list(result["testpoints"]) + verdicts = _as_native_list(result["verdicts"]) for testpoint, verdict in zip(testpoints, verdicts): big_class = testpoint.split("-", 1)[0].strip() if "-" in testpoint else testpoint.strip() @@ -392,11 +399,18 @@ def main() -> None: gateway_api=args.gateway_api, model_str=args.modelstr, num_concurrency=args.num_concurrency, + num_max_retry=args.vlm_max_retry, + timeout=args.vlm_timeout, ) score_breakdown: dict[str, Any] = {} + failed_samples: list[str] = [] while samples_todo: + # Samples whose retries are exhausted are recorded instead of being + # silently dropped, so partial runs stay auditable against total. + expired = [s for s in samples_todo if s["try_num"] >= args.max_retry] + failed_samples.extend(osp.basename(s["image_path"]) for s in expired) samples_todo = [s for s in samples_todo if s["try_num"] < args.max_retry] if not samples_todo: break @@ -433,10 +447,12 @@ def main() -> None: samples_error.append(si) else: analyses, verdicts = parsed + # Store native lists; legacy result files carry str(list) blobs, + # which compute_final_metrics still accepts defensively. score_breakdown[name] = { - "testpoints": str(testpoints), - "analyses": str(analyses), - "verdicts": str(verdicts), + "testpoints": list(testpoints), + "analyses": list(analyses), + "verdicts": [bool(v) for v in verdicts], "raw_output": response, } @@ -454,6 +470,7 @@ def main() -> None: }, "judge_model": args.modelstr, "success_count": f"{len(score_breakdown)}/{samples_total}", + "failed_samples": sorted(set(failed_samples)), "breakdown": score_breakdown, } @@ -473,6 +490,8 @@ def parse_arguments() -> argparse.Namespace: parser.add_argument("--num_concurrency", type=int, default=128) parser.add_argument("--batch_size", type=int, default=1170) parser.add_argument("--max_retry", type=int, default=3) + parser.add_argument("--vlm_timeout", type=int, default=300) + parser.add_argument("--vlm_max_retry", type=int, default=3) return parser.parse_args() diff --git a/evaluation/cosmos3/generator/unigenbench/utils.py b/evaluation/cosmos3/generator/unigenbench/utils.py index 8ec64aba..02d1d874 100644 --- a/evaluation/cosmos3/generator/unigenbench/utils.py +++ b/evaluation/cosmos3/generator/unigenbench/utils.py @@ -57,13 +57,19 @@ def __init__( async def query_core(self, request: dict) -> str: for attempt in range(self.num_max_retry): + if attempt > 0: + # Back off briefly before re-issuing so a rate-limited gateway + # is not hammered back-to-back. + await asyncio.sleep(min(2**attempt, 30)) try: async def stream_response(): stream = await self.client.chat.completions.create(**request) result = "" async for chunk in stream: - if chunk.choices[0].delta.content is not None: + # Some gateways emit a terminal usage-only chunk with an + # empty choices list; skipping it keeps the scored text. + if chunk.choices and chunk.choices[0].delta.content is not None: result += chunk.choices[0].delta.content return result @@ -71,15 +77,13 @@ async def stream_response(): except KeyboardInterrupt: raise except asyncio.TimeoutError: - if attempt == 0 or attempt == self.num_max_retry - 1: - logger.warning( - f"VLM API for {request['model']} timeout (attempt {attempt + 1}/{self.num_max_retry}): exceeded {self.timeout}s" - ) + logger.warning( + f"VLM API for {request['model']} timeout (attempt {attempt + 1}/{self.num_max_retry}): exceeded {self.timeout}s" + ) except Exception as e: - if attempt == 0 or attempt == self.num_max_retry - 1: - logger.warning( - f"VLM API for {request['model']} error (attempt {attempt + 1}/{self.num_max_retry}): {type(e).__name__}: {e}" - ) + logger.warning( + f"VLM API for {request['model']} error (attempt {attempt + 1}/{self.num_max_retry}): {type(e).__name__}: {e}" + ) logger.warning("VLM query failed after max retries") return "" @@ -137,5 +141,14 @@ def build_request( return request def close(self): + # Shut down the underlying HTTP pool before stopping the loop so idle + # connections do not linger until interpreter GC (ResourceWarnings). + closer = getattr(self.client, "close", None) + if closer is not None: + future = asyncio.run_coroutine_threadsafe(closer(), self.loop) + try: + future.result(timeout=10) + except Exception as exc: # best-effort teardown + logger.warning(f"VLM client close failed: {type(exc).__name__}: {exc}") self.loop.call_soon_threadsafe(self.loop.stop) self.thread.join()