diff --git a/.agents/exploration/interleavethinker-fastvideo-integration.md b/.agents/exploration/interleavethinker-fastvideo-integration.md index 3bb71a612..d34066439 100644 --- a/.agents/exploration/interleavethinker-fastvideo-integration.md +++ b/.agents/exploration/interleavethinker-fastvideo-integration.md @@ -2994,3 +2994,55 @@ Next recommended integration step: `scripts/interleave_thinker/evaluate_traces.py` and/or a small HTML/contact sheet report. The prompt-set runner now produces the trace and summary artifacts those tools can consume. + +## Stage 12 Execution: Trace Evaluation Reports + +Status: implemented locally, pending commit/push and pushed-branch Modal +validation. + +Goal: + +- Add a trace-level evaluator for the prompt-set output produced by + `fastvideo interleave-eval`. +- Support both machine-readable aggregate metrics and a lightweight HTML report + with final-image thumbnails. + +Implemented changes: + +- Added `fastvideo/entrypoints/interleave/trace_eval.py`. + - Discovers traces from: + - prompt-set output directories; + - `summary.json` files; + - explicit `trace.json` files. + - Computes: + - trace count, success count, success rate; + - total/average attempts; + - total/average retry attempts; + - traces with final images; + - total/average inference time when attempt timing is present; + - failure-reason counts; + - category-level success rates from `prompt_set_metadata.category`; + - per-trace rows with sample id, prompt, final image, and failure fields. + - Writes JSON metrics and an optional HTML report. +- Added `scripts/interleave_thinker/evaluate_traces.py`. + - CLI shape: + `python scripts/interleave_thinker/evaluate_traces.py + [--output metrics.json] [--html-output + report.html]`. +- Exported trace-evaluation helpers from `fastvideo.entrypoints.interleave`. +- Updated `examples/interleave/README.md` with post-run evaluation commands. +- Added `tests/local_tests/test_interleave_trace_eval.py` for: + - evaluating traces discovered through `summary.json`; + - directory discovery; + - HTML report generation; + - script-level JSON and HTML output. + +Validation so far: + +- Local `python -m py_compile` passed for changed Python files. +- Local `git diff --check` passed. +- Local pre-commit with `PRE_COMMIT_HOME=/tmp/fastvideo-pre-commit-cache` + passed yapf, ruff, codespell, filename, and suggestion hooks. +- Local mypy still fails for the known hyphenated worktree path issue: + `fastvideo-interleavethinker is not a valid Python package name`. + Modal mypy remains the authoritative type gate. diff --git a/examples/interleave/README.md b/examples/interleave/README.md index 366f4e0d8..76590c7b7 100644 --- a/examples/interleave/README.md +++ b/examples/interleave/README.md @@ -124,6 +124,20 @@ loader accepts JSONL rows with `id`, `prompt` or `instruction`, optional `initial_image_path`, and arbitrary metadata fields. Use `--resume` to skip rows whose trace already exists and `--limit N` for small smoke runs. +Evaluate saved traces after a prompt-set run: + +```bash +python scripts/interleave_thinker/evaluate_traces.py \ + outputs/interleave_eval \ + --output outputs/interleave_eval/metrics.json \ + --html-output outputs/interleave_eval/report.html +``` + +The evaluator accepts a prompt-set output directory, a `summary.json`, or one +or more explicit `trace.json` files. It reports success rate, attempts, retry +counts, failure reasons, inference timing when present, and an optional HTML +table with final-image thumbnails. + ## Nano Banana / Gemini API Backends The InterleaveThinker RL config can use closed-source Google models through diff --git a/fastvideo/entrypoints/interleave/__init__.py b/fastvideo/entrypoints/interleave/__init__.py index 2ed43313a..1c6d6c815 100644 --- a/fastvideo/entrypoints/interleave/__init__.py +++ b/fastvideo/entrypoints/interleave/__init__.py @@ -55,6 +55,16 @@ from fastvideo.entrypoints.interleave.trace import ( save_trace, trace_to_dict, ) +from fastvideo.entrypoints.interleave.trace_eval import ( + InterleaveTraceEvaluationSummary, + InterleaveTraceMetrics, + discover_interleave_trace_paths, + evaluate_interleave_traces, + interleave_trace_evaluation_to_dict, + load_interleave_trace_metrics, + write_interleave_trace_evaluation, + write_interleave_trace_html_report, +) __all__ = [ "AcceptAllCritic", @@ -80,13 +90,19 @@ __all__ = [ "InterleaveThinkerCriticProvider", "InterleaveThinkerPlannerProvider", "InterleaveTrace", + "InterleaveTraceEvaluationSummary", + "InterleaveTraceMetrics", "PlannedInterleaveStep", "PlannerInput", "PlannerProvider", "SinglePromptPlanner", "build_app", + "discover_interleave_trace_paths", + "evaluate_interleave_traces", + "interleave_trace_evaluation_to_dict", "load_interleave_prompt_set", "load_interleave_run_config", + "load_interleave_trace_metrics", "prompt_set_summary_to_dict", "resolve_interleave_instruction", "run_interleave_prompt_set", @@ -95,4 +111,6 @@ __all__ = [ "save_prompt_set_summary", "save_trace", "trace_to_dict", + "write_interleave_trace_evaluation", + "write_interleave_trace_html_report", ] diff --git a/fastvideo/entrypoints/interleave/trace_eval.py b/fastvideo/entrypoints/interleave/trace_eval.py new file mode 100644 index 000000000..6cb2e906f --- /dev/null +++ b/fastvideo/entrypoints/interleave/trace_eval.py @@ -0,0 +1,464 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Trace-level evaluation helpers for Interleave prompt-set outputs.""" + +from __future__ import annotations + +import html +import json +from collections import Counter +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, cast + + +@dataclass(frozen=True) +class InterleaveTraceMetrics: + trace_path: str + instruction: str + success: bool + attempts: int + steps: int + retry_attempts: int + failed_step_index: int | None = None + failure_reason: str | None = None + final_image_path: str | None = None + final_prompt: str | None = None + total_inference_time_s: float | None = None + prompt_set_id: str | None = None + prompt_set_index: int | None = None + metadata: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class InterleaveTraceEvaluationSummary: + input_paths: list[str] + num_traces: int + num_success: int + success_rate: float + total_attempts: int + average_attempts: float + total_retry_attempts: int + average_retry_attempts: float + traces_with_final_image: int + total_inference_time_s: float | None + average_inference_time_s: float | None + failure_reasons: dict[str, int] + success_by_category: dict[str, dict[str, float]] + traces: list[InterleaveTraceMetrics] + + +def discover_interleave_trace_paths(paths: Sequence[str | Path]) -> list[Path]: + """Discover trace JSON files from trace files, summaries, or output dirs.""" + + if not paths: + raise ValueError("At least one trace, summary, or output directory is required") + + discovered: list[Path] = [] + for raw_path in paths: + path = Path(raw_path) + if not path.exists(): + raise FileNotFoundError(f"Trace input not found: {path}") + if path.is_dir(): + summary_path = path / "summary.json" + if summary_path.is_file(): + discovered.extend(_trace_paths_from_summary(summary_path)) + else: + discovered.extend(sorted(path.rglob("trace.json"))) + continue + if path.name == "summary.json": + discovered.extend(_trace_paths_from_summary(path)) + continue + discovered.append(path) + + unique: list[Path] = [] + seen: set[Path] = set() + for trace_path in discovered: + resolved = trace_path.resolve() + if resolved in seen: + continue + seen.add(resolved) + unique.append(trace_path) + if not unique: + raise ValueError(f"No trace files found in inputs: {[str(path) for path in paths]}") + return unique + + +def evaluate_interleave_traces(paths: Sequence[str | Path]) -> InterleaveTraceEvaluationSummary: + """Evaluate saved Interleave traces and return aggregate metrics.""" + + trace_paths = discover_interleave_trace_paths(paths) + traces = [load_interleave_trace_metrics(path) for path in trace_paths] + num_traces = len(traces) + num_success = sum(1 for trace in traces if trace.success) + total_attempts = sum(trace.attempts for trace in traces) + total_retry_attempts = sum(trace.retry_attempts for trace in traces) + inference_times = [trace.total_inference_time_s for trace in traces if trace.total_inference_time_s is not None] + + return InterleaveTraceEvaluationSummary( + input_paths=[str(path) for path in paths], + num_traces=num_traces, + num_success=num_success, + success_rate=(num_success / num_traces if num_traces else 0.0), + total_attempts=total_attempts, + average_attempts=(total_attempts / num_traces if num_traces else 0.0), + total_retry_attempts=total_retry_attempts, + average_retry_attempts=(total_retry_attempts / num_traces if num_traces else 0.0), + traces_with_final_image=sum(1 for trace in traces if trace.final_image_path), + total_inference_time_s=(sum(inference_times) if inference_times else None), + average_inference_time_s=((sum(inference_times) / len(inference_times)) if inference_times else None), + failure_reasons=_failure_reason_counts(traces), + success_by_category=_success_by_category(traces), + traces=traces, + ) + + +def load_interleave_trace_metrics(path: str | Path) -> InterleaveTraceMetrics: + trace_path = Path(path) + payload = _load_json_mapping(trace_path) + attempts = _mapping_list(payload.get("attempts")) + metadata = _string_mapping(payload.get("metadata")) + final_image = _optional_mapping(payload.get("final_image")) + final_image_path = _string_value(final_image.get("file_path")) if final_image is not None else None + final_prompt = _string_value(final_image.get("prompt")) if final_image is not None else None + total_time = _sum_attempt_inference_time(attempts) + + return InterleaveTraceMetrics( + trace_path=str(trace_path), + instruction=_string_value(payload.get("instruction")) or "", + success=bool(payload.get("success")), + attempts=len(attempts), + steps=_count_steps(attempts), + retry_attempts=sum(1 for attempt in attempts if _int_value(attempt.get("attempt_index")) not in (None, 0)), + failed_step_index=_int_value(metadata.get("failed_step_index")), + failure_reason=_failure_reason(payload, attempts, metadata), + final_image_path=final_image_path, + final_prompt=final_prompt, + total_inference_time_s=total_time, + prompt_set_id=_string_value(metadata.get("prompt_set_id")), + prompt_set_index=_int_value(metadata.get("prompt_set_index")), + metadata=dict(metadata), + ) + + +def interleave_trace_evaluation_to_dict(summary: InterleaveTraceEvaluationSummary) -> dict[str, Any]: + return { + "input_paths": list(summary.input_paths), + "num_traces": summary.num_traces, + "num_success": summary.num_success, + "success_rate": summary.success_rate, + "total_attempts": summary.total_attempts, + "average_attempts": summary.average_attempts, + "total_retry_attempts": summary.total_retry_attempts, + "average_retry_attempts": summary.average_retry_attempts, + "traces_with_final_image": summary.traces_with_final_image, + "total_inference_time_s": summary.total_inference_time_s, + "average_inference_time_s": summary.average_inference_time_s, + "failure_reasons": dict(summary.failure_reasons), + "success_by_category": dict(summary.success_by_category), + "traces": [_trace_metrics_to_dict(trace) for trace in summary.traces], + } + + +def write_interleave_trace_evaluation( + summary: InterleaveTraceEvaluationSummary, + output_path: str | Path, +) -> None: + path = Path(output_path) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + interleave_trace_evaluation_to_dict(summary), + indent=2, + sort_keys=True, + ) + "\n", + encoding="utf-8", + ) + + +def write_interleave_trace_html_report( + summary: InterleaveTraceEvaluationSummary, + output_path: str | Path, + *, + title: str = "Interleave Trace Evaluation", +) -> None: + path = Path(output_path) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(_render_html_report(summary, path.parent, title=title), encoding="utf-8") + + +def _trace_paths_from_summary(summary_path: Path) -> list[Path]: + payload = _load_json_mapping(summary_path) + results = _mapping_list(payload.get("results")) + trace_paths: list[Path] = [] + for result in results: + raw_trace_path = _string_value(result.get("trace_path")) + if not raw_trace_path: + continue + candidate = Path(raw_trace_path) + if not candidate.is_absolute() and not candidate.exists(): + candidate = summary_path.parent / candidate + if candidate.is_file(): + trace_paths.append(candidate) + return trace_paths + + +def _load_json_mapping(path: Path) -> Mapping[str, Any]: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, Mapping): + raise ValueError(f"{path} must contain a JSON object") + return cast(Mapping[str, Any], payload) + + +def _mapping_list(value: Any) -> list[Mapping[str, Any]]: + if not isinstance(value, list): + return [] + rows: list[Mapping[str, Any]] = [] + for item in value: + if isinstance(item, Mapping): + rows.append(cast(Mapping[str, Any], item)) + return rows + + +def _optional_mapping(value: Any) -> Mapping[str, Any] | None: + if isinstance(value, Mapping): + return cast(Mapping[str, Any], value) + return None + + +def _string_mapping(value: Any) -> Mapping[str, Any]: + if isinstance(value, Mapping): + return cast(Mapping[str, Any], value) + return {} + + +def _string_value(value: Any) -> str | None: + if isinstance(value, str) and value: + return value + return None + + +def _int_value(value: Any) -> int | None: + if isinstance(value, bool): + return None + if isinstance(value, int): + return value + return None + + +def _float_value(value: Any) -> float | None: + if isinstance(value, bool): + return None + if isinstance(value, int | float): + return float(value) + return None + + +def _count_steps(attempts: Sequence[Mapping[str, Any]]) -> int: + step_indices = {_int_value(attempt.get("step_index")) for attempt in attempts} + step_indices.discard(None) + return len(step_indices) + + +def _sum_attempt_inference_time(attempts: Sequence[Mapping[str, Any]]) -> float | None: + total = 0.0 + found = False + for attempt in attempts: + generated = _optional_mapping(attempt.get("generated")) + if generated is None: + continue + value = _float_value(generated.get("inference_time_s")) + if value is None: + continue + total += value + found = True + return total if found else None + + +def _failure_reason( + payload: Mapping[str, Any], + attempts: Sequence[Mapping[str, Any]], + metadata: Mapping[str, Any], +) -> str | None: + if bool(payload.get("success")): + return None + explicit_error = _string_value(metadata.get("error")) + if explicit_error: + return explicit_error + for attempt in reversed(attempts): + decision = _optional_mapping(attempt.get("decision")) + if decision is None: + continue + reason = _string_value(decision.get("reason")) + if reason: + return reason + failed_step = _int_value(metadata.get("failed_step_index")) + if failed_step is not None: + return f"failed_step_{failed_step}" + return "unknown" + + +def _failure_reason_counts(traces: Sequence[InterleaveTraceMetrics]) -> dict[str, int]: + counts: Counter[str] = Counter() + for trace in traces: + if trace.success: + continue + counts[trace.failure_reason or "unknown"] += 1 + return dict(sorted(counts.items())) + + +def _success_by_category(traces: Sequence[InterleaveTraceMetrics]) -> dict[str, dict[str, float]]: + grouped: dict[str, list[InterleaveTraceMetrics]] = {} + for trace in traces: + category = _metadata_category(trace.metadata) + if category is None: + continue + grouped.setdefault(category, []).append(trace) + + result: dict[str, dict[str, float]] = {} + for category, category_traces in sorted(grouped.items()): + total = len(category_traces) + success = sum(1 for trace in category_traces if trace.success) + result[category] = { + "num_traces": float(total), + "num_success": float(success), + "success_rate": success / total if total else 0.0, + } + return result + + +def _metadata_category(metadata: Mapping[str, Any]) -> str | None: + prompt_metadata = _optional_mapping(metadata.get("prompt_set_metadata")) + if prompt_metadata is None: + return None + return _string_value(prompt_metadata.get("category")) + + +def _trace_metrics_to_dict(trace: InterleaveTraceMetrics) -> dict[str, Any]: + return { + "trace_path": trace.trace_path, + "instruction": trace.instruction, + "success": trace.success, + "attempts": trace.attempts, + "steps": trace.steps, + "retry_attempts": trace.retry_attempts, + "failed_step_index": trace.failed_step_index, + "failure_reason": trace.failure_reason, + "final_image_path": trace.final_image_path, + "final_prompt": trace.final_prompt, + "total_inference_time_s": trace.total_inference_time_s, + "prompt_set_id": trace.prompt_set_id, + "prompt_set_index": trace.prompt_set_index, + "metadata": dict(trace.metadata), + } + + +def _render_html_report( + summary: InterleaveTraceEvaluationSummary, + html_dir: Path, + *, + title: str, +) -> str: + rows = "\n".join(_render_trace_row(trace, html_dir) for trace in summary.traces) + failure_rows = "\n".join(f"
  • {html.escape(reason)}: {count}
    • " + for reason, count in sorted(summary.failure_reasons.items())) + if not failure_rows: + failure_rows = "
  • None
    • " + return f""" + + + + {html.escape(title)} + + + +

    {html.escape(title)}

    +
    +
    Traces: {summary.num_traces}
    +
    Success: {summary.num_success}
    +
    Success rate: {summary.success_rate:.4f}
    +
    Avg attempts: {summary.average_attempts:.2f}
    +
    Avg retries: {summary.average_retry_attempts:.2f}
    +
    +

    Failure Reasons

    + +

    Traces

    + + + + + + + + + + + + +{rows} + +
    SampleStatusAttemptsInstructionFinal imageTrace
    + + +""" + + +def _render_trace_row(trace: InterleaveTraceMetrics, html_dir: Path) -> str: + sample = trace.prompt_set_id or Path(trace.trace_path).parent.name + status_class = "ok" if trace.success else "fail" + status_text = "success" if trace.success else f"failed: {trace.failure_reason or 'unknown'}" + image_html = _image_html(trace.final_image_path, html_dir) + trace_link = _path_link(trace.trace_path, html_dir) + return (" " + f"{html.escape(sample)}" + f"{html.escape(status_text)}" + f"{trace.attempts} ({trace.retry_attempts} retries)" + f"{html.escape(trace.instruction)}" + f"{image_html}" + f"{trace_link}" + "") + + +def _image_html(image_path: str | None, html_dir: Path) -> str: + if not image_path: + return "" + path = Path(image_path) + href = _relative_or_raw_path(path, html_dir) + return f"\"final" + + +def _path_link(raw_path: str, html_dir: Path) -> str: + href = _relative_or_raw_path(Path(raw_path), html_dir) + return f"trace" + + +def _relative_or_raw_path(path: Path, base_dir: Path) -> str: + try: + return str(path.resolve().relative_to(base_dir.resolve())) + except ValueError: + try: + return str(path.resolve().relative_to(Path.cwd().resolve())) + except ValueError: + return str(path) + + +__all__ = [ + "InterleaveTraceEvaluationSummary", + "InterleaveTraceMetrics", + "discover_interleave_trace_paths", + "evaluate_interleave_traces", + "interleave_trace_evaluation_to_dict", + "load_interleave_trace_metrics", + "write_interleave_trace_evaluation", + "write_interleave_trace_html_report", +] diff --git a/scripts/interleave_thinker/evaluate_traces.py b/scripts/interleave_thinker/evaluate_traces.py new file mode 100644 index 000000000..ac47e27eb --- /dev/null +++ b/scripts/interleave_thinker/evaluate_traces.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: Apache-2.0 +"""Evaluate saved Interleave prompt-set traces.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from fastvideo.entrypoints.interleave.trace_eval import ( + evaluate_interleave_traces, + interleave_trace_evaluation_to_dict, + write_interleave_trace_evaluation, + write_interleave_trace_html_report, +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Evaluate Interleave trace.json files or prompt-set output dirs.") + parser.add_argument( + "paths", + nargs="+", + help="Trace file, summary.json, or prompt-set output directory. Multiple inputs are allowed.", + ) + parser.add_argument( + "--output", + type=str, + default=None, + help="Write aggregate metrics JSON to this path. Defaults to stdout.", + ) + parser.add_argument( + "--html-output", + type=str, + default=None, + help="Optional HTML report path with per-trace rows and final-image thumbnails.", + ) + parser.add_argument( + "--title", + type=str, + default="Interleave Trace Evaluation", + help="Title for the optional HTML report.", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + summary = evaluate_interleave_traces([Path(path) for path in args.paths]) + if args.output: + write_interleave_trace_evaluation(summary, args.output) + print(f"Wrote metrics: {args.output}") + else: + print(json.dumps(interleave_trace_evaluation_to_dict(summary), indent=2, sort_keys=True)) + if args.html_output: + write_interleave_trace_html_report(summary, args.html_output, title=args.title) + print(f"Wrote HTML report: {args.html_output}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/local_tests/test_interleave_trace_eval.py b/tests/local_tests/test_interleave_trace_eval.py new file mode 100644 index 000000000..3289802be --- /dev/null +++ b/tests/local_tests/test_interleave_trace_eval.py @@ -0,0 +1,208 @@ +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +from fastvideo.entrypoints.interleave import ( + discover_interleave_trace_paths, + evaluate_interleave_traces, + interleave_trace_evaluation_to_dict, + write_interleave_trace_html_report, +) + + +def test_evaluate_interleave_traces_from_summary(tmp_path: Path) -> None: + output_dir = _write_trace_fixture(tmp_path) + + summary = evaluate_interleave_traces([output_dir / "summary.json"]) + + assert summary.num_traces == 2 + assert summary.num_success == 1 + assert summary.success_rate == 0.5 + assert summary.total_attempts == 3 + assert summary.average_attempts == 1.5 + assert summary.total_retry_attempts == 1 + assert summary.traces_with_final_image == 1 + assert summary.total_inference_time_s == 1.5 + assert summary.failure_reasons == {"critic rejected final attempt": 1} + assert summary.success_by_category["product"] == { + "num_traces": 1.0, + "num_success": 1.0, + "success_rate": 1.0, + } + payload = interleave_trace_evaluation_to_dict(summary) + assert payload["traces"][0]["prompt_set_id"] == "mug" + assert payload["traces"][1]["failure_reason"] == "critic rejected final attempt" + + +def test_discover_interleave_trace_paths_from_directory(tmp_path: Path) -> None: + output_dir = _write_trace_fixture(tmp_path) + + paths = discover_interleave_trace_paths([output_dir]) + + assert [path.name for path in paths] == ["trace.json", "trace.json"] + assert {path.parent.name for path in paths} == {"mug", "poster"} + + +def test_write_interleave_trace_html_report(tmp_path: Path) -> None: + output_dir = _write_trace_fixture(tmp_path) + summary = evaluate_interleave_traces([output_dir]) + html_path = tmp_path / "report.html" + + write_interleave_trace_html_report(summary, html_path, title="Smoke Report") + + html_text = html_path.read_text(encoding="utf-8") + assert "Smoke Report" in html_text + assert "mug" in html_text + assert "critic rejected final attempt" in html_text + assert " None: + output_dir = _write_trace_fixture(tmp_path) + metrics_path = tmp_path / "metrics.json" + html_path = tmp_path / "report.html" + + result = subprocess.run( + [ + sys.executable, + "scripts/interleave_thinker/evaluate_traces.py", + str(output_dir), + "--output", + str(metrics_path), + "--html-output", + str(html_path), + ], + check=True, + capture_output=True, + text=True, + ) + + assert "Wrote metrics" in result.stdout + assert json.loads(metrics_path.read_text(encoding="utf-8"))["num_traces"] == 2 + assert html_path.exists() + + +def _write_trace_fixture(tmp_path: Path) -> Path: + output_dir = tmp_path / "eval" + image_path = output_dir / "mug" / "final.png" + image_path.parent.mkdir(parents=True, exist_ok=True) + image_path.write_bytes(b"fake-image") + _write_json( + output_dir / "mug" / "trace.json", + { + "instruction": "draw a mug", + "success": True, + "metadata": { + "prompt_set_id": "mug", + "prompt_set_index": 0, + "prompt_set_metadata": { + "category": "product", + }, + }, + "final_image": { + "prompt": "refined mug", + "file_path": str(image_path), + "inference_time_s": 0.4, + "metadata": {}, + }, + "attempts": [ + { + "step_index": 0, + "attempt_index": 0, + "prompt": "draw a mug", + "generated": { + "prompt": "draw a mug", + "file_path": str(output_dir / "mug" / "attempt0.png"), + "inference_time_s": 0.5, + "metadata": {}, + }, + "decision": { + "success": False, + "refine_prompt": "refined mug", + "reason": "needs refinement", + "metadata": {}, + }, + }, + { + "step_index": 0, + "attempt_index": 1, + "prompt": "refined mug", + "generated": { + "prompt": "refined mug", + "file_path": str(image_path), + "inference_time_s": 0.7, + "metadata": {}, + }, + "decision": { + "success": True, + "refine_prompt": None, + "reason": None, + "metadata": {}, + }, + }, + ], + }, + ) + _write_json( + output_dir / "poster" / "trace.json", + { + "instruction": "draw a poster", + "success": False, + "metadata": { + "prompt_set_id": "poster", + "prompt_set_index": 1, + "failed_step_index": 0, + "prompt_set_metadata": { + "category": "poster", + }, + }, + "final_image": None, + "attempts": [ + { + "step_index": 0, + "attempt_index": 0, + "prompt": "draw a poster", + "generated": { + "prompt": "draw a poster", + "file_path": str(output_dir / "poster" / "attempt0.png"), + "inference_time_s": 0.3, + "metadata": {}, + }, + "decision": { + "success": False, + "refine_prompt": None, + "reason": "critic rejected final attempt", + "metadata": {}, + }, + }, + ], + }, + ) + _write_json( + output_dir / "summary.json", + { + "num_samples": 2, + "num_success": 1, + "results": [ + { + "sample_id": "mug", + "trace_path": str(output_dir / "mug" / "trace.json"), + }, + { + "sample_id": "poster", + "trace_path": str(output_dir / "poster" / "trace.json"), + }, + ], + }, + ) + return output_dir + + +def _write_json(path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")