from __future__ import annotations import hashlib import json from pathlib import Path from typing import Any from pptx.presentation import Presentation as PptxPresentation from .pptx_functions import PptxEditor from .presentbench.scoring import ( score_presentbench, score_presentbench_slide, ) from .presentbench.spec_builder import build_presentbench_eval_spec from .pptx_extraction import PptxExtractionService from .reward_metrics import clamp_reward from .reward_models import ( EvalSpec, IntermediateSlideRewardResult, RewardResult, ExtractedSlide, SourcePack, TaskConstraints, to_serializable, ) from .reward_prompts import build_task_spec from .slidesgenbench.quizbank_service import QuizBankGenerationService from .slidesgenbench.quantitative_judge import ( QuantitativeQuizJudgeService, ) from .slidesgenbench.scoring import score_slidesgenbench from .slidesgenbench.spec_builder import build_slidesgenbench_eval_spec from .slidesgenbench.rendered_aesthetics import ( compute_intermediate_rendered_aesthetics_score, ) from .slidesgenbench.text_layout import compute_slide_text_layout_scores SPEC_VERSION = "1.0" INTERMEDIATE_SLIDESGENBENCH_AESTHETIC_WEIGHT = 0.20 INTERMEDIATE_SLIDESGENBENCH_TEXT_LAYOUT_WEIGHT = 0.35 DEFAULT_REWARD_KERNEL_CONFIG: dict[str, Any] = { "branch_weights": {"presentbench": 0.45, "slidesgenbench": 0.55}, } def _reward_details_payload( result: RewardResult | IntermediateSlideRewardResult, ) -> dict[str, Any]: return to_serializable(result) def _write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") def _persist_eval_spec(cache_dir: str, eval_spec: EvalSpec) -> dict[str, str]: root = Path(cache_dir) / eval_spec.spec_hash root.mkdir(parents=True, exist_ok=True) task_spec_path = root / "task_spec.json" presentbench_path = root / "presentbench_eval_spec.json" slidesgenbench_path = root / "slidesgenbench_eval_spec.json" scoring_config_path = root / "scoring_config.json" eval_spec_path = root / "eval_spec.json" _write_json(task_spec_path, to_serializable(eval_spec.task_spec)) _write_json(presentbench_path, to_serializable(eval_spec.presentbench)) _write_json(slidesgenbench_path, to_serializable(eval_spec.slidesgenbench)) _write_json(scoring_config_path, eval_spec.scoring_config) _write_json(eval_spec_path, to_serializable(eval_spec)) return { "cache_root": str(root), "task_spec": str(task_spec_path), "presentbench_eval_spec": str(presentbench_path), "slidesgenbench_eval_spec": str(slidesgenbench_path), "scoring_config": str(scoring_config_path), "eval_spec": str(eval_spec_path), } def build_eval_spec( prompt: str, source_pack: SourcePack, task_constraints: TaskConstraints | None = None, *, quiz_bank_service: QuizBankGenerationService, cache_dir: str | None = None, mode: str = "eval", ) -> EvalSpec: if not source_pack.task_id: raise ValueError("source_pack.task_id is required") if not source_pack.documents: raise ValueError("source_pack.documents must not be empty") task_spec = build_task_spec(prompt, source_pack, task_constraints) slidesgenbench = build_slidesgenbench_eval_spec( task_spec, source_pack, quiz_bank_service=quiz_bank_service, mode=mode, ) presentbench = build_presentbench_eval_spec(task_spec, mode=mode) payload = { "task_spec": to_serializable(task_spec), "presentbench": to_serializable(presentbench), "slidesgenbench": to_serializable(slidesgenbench), "scoring_config": {**DEFAULT_REWARD_KERNEL_CONFIG, "mode": mode}, "spec_version": SPEC_VERSION, } spec_hash = hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() eval_spec = EvalSpec( task_spec=task_spec, presentbench=presentbench, slidesgenbench=slidesgenbench, scoring_config={**DEFAULT_REWARD_KERNEL_CONFIG, "mode": mode}, spec_version=SPEC_VERSION, spec_hash=spec_hash, ) if cache_dir: artifact_paths = _persist_eval_spec(cache_dir, eval_spec) eval_spec.task_spec.metadata["cache_artifacts"] = artifact_paths return eval_spec def _reward_result_for_failure( eval_spec: EvalSpec, *, error: Exception, mode: str, ) -> RewardResult: return RewardResult( reward_total=0.0, reward_breakdown={"R_total": 0.0, "R_pb": 0.0, "R_sg": 0.0}, hard_caps={ "C_fidelity_critical": 1.0, "C_blankness": 1.0, "C_hard": 0.0, }, soft_penalties={}, checklist_results=[], quiz_results=[], aesthetics_results={}, artifacts={}, metadata={ "task_id": eval_spec.task_spec.task_id, "spec_hash": eval_spec.spec_hash, "spec_version": eval_spec.spec_version, "mode": mode, "used_mllm": False, "failure_counts": {"inspection": 1}, "error": str(error), }, ) def evaluate_presentation( eval_spec: EvalSpec, presentation: PptxEditor | PptxPresentation | str, *, use_mllm: bool = False, judge: Any | None = None, render_service: Any | None = None, inspection_service: PptxExtractionService | None = None, aesthetics_service: Any | None = None, quantitative_quiz_judge_service: QuantitativeQuizJudgeService, mode: str = "eval", ) -> RewardResult: del judge inspection_service = inspection_service or PptxExtractionService() try: presentation_extraction = inspection_service.inspect_presentation(presentation) except Exception as error: return _reward_result_for_failure(eval_spec, error=error, mode=mode) rendered_presentation = None render_error: str | None = None if render_service and hasattr(render_service, "render_presentation"): try: rendered_presentation = render_service.render_presentation(presentation) except Exception as error: render_error = str(error) presentbench_result = score_presentbench( eval_spec.task_spec, presentation_extraction, eval_spec.presentbench, aesthetics_service=aesthetics_service, ) slidesgenbench_result = score_slidesgenbench( eval_spec.task_spec, presentation_extraction, eval_spec.slidesgenbench, quantitative_quiz_judge_service=quantitative_quiz_judge_service, rendered_presentation=rendered_presentation, ) branch_weights = eval_spec.scoring_config["branch_weights"] reward_total = clamp_reward( branch_weights["presentbench"] * presentbench_result.reward_total + branch_weights["slidesgenbench"] * slidesgenbench_result.reward_total ) return RewardResult( reward_total=reward_total, reward_breakdown={ "R_total": reward_total, "R_pb": presentbench_result.reward_total, "R_sg": slidesgenbench_result.reward_total, **presentbench_result.reward_breakdown, **slidesgenbench_result.reward_breakdown, }, hard_caps=presentbench_result.hard_caps, soft_penalties=presentbench_result.soft_penalties, checklist_results=presentbench_result.checklist_results, quiz_results=slidesgenbench_result.quiz_results, aesthetics_results=( slidesgenbench_result.aesthetics_results if slidesgenbench_result.aesthetics_results else presentbench_result.aesthetics_results ), artifacts={ "presentation_digest": presentation_extraction.metadata.get( "presentation_digest" ), "rendered_presentation": to_serializable(rendered_presentation) if rendered_presentation is not None else None, }, metadata={ "task_id": eval_spec.task_spec.task_id, "spec_hash": eval_spec.spec_hash, "presentbench_spec_hash": eval_spec.presentbench.spec_hash, "slidesgenbench_spec_hash": eval_spec.slidesgenbench.spec_hash, "spec_version": eval_spec.spec_version, "mode": mode, "slide_count": presentation_extraction.slide_count, "presentbench_item_count": len(eval_spec.presentbench.checklist), "slidesgenbench_question_count": len(eval_spec.slidesgenbench.quiz_bank), "judge_call_count": 0, "failure_counts": { "inspection": 0, "judge": 0, "render": 0 if render_error is None else 1, }, "used_mllm": use_mllm, "inspection_mode": presentation_extraction.metadata.get("inspection_mode"), "presentation_digest": presentation_extraction.metadata.get( "presentation_digest" ), "render_backend": getattr(rendered_presentation, "backend", None), "render_error": render_error, }, ) def evaluate_slide( eval_spec: EvalSpec, slide_index: int, *, presentation: PptxEditor | PptxPresentation | None = None, slide_extraction: ExtractedSlide | None = None, use_mllm: bool = False, judge: Any | None = None, render_service: Any | None = None, inspection_service: PptxExtractionService | None = None, aesthetics_service: Any | None = None, previous_slide_extractions: list[ExtractedSlide] | None = None, mode: str = "eval", ) -> IntermediateSlideRewardResult: del judge, aesthetics_service inspection_service = inspection_service or PptxExtractionService() try: if slide_extraction is None: if presentation is None: raise ValueError("presentation or slide_extraction is required") slide_extraction = inspection_service.inspect_slide( slide_index, presentation=presentation, ) except Exception as error: return IntermediateSlideRewardResult( slide_index=slide_index, reward_total=clamp_reward(0.0), reward_breakdown={ "R_slide": 0.0, "S_prompt_alignment": 0.0, "S_local_completeness": 0.0, "S_local_correctness": 0.0, "S_local_fidelity": 0.0, "S_local_usability": 0.0, }, hard_caps={ "C_slide_fidelity_critical": 1.0, "C_slide_blankness": 1.0, "C_slide_hard": 0.0, }, metadata={ "slide_index": slide_index, "spec_hash": eval_spec.spec_hash, "mode": mode, "error": str(error), }, ) presentbench_result = score_presentbench_slide( eval_spec.task_spec, eval_spec.presentbench, slide_index, slide_extraction, previous_slide_extractions=previous_slide_extractions, use_mllm=use_mllm, mode=mode, ) rendered_aesthetics: dict[str, Any] = {} text_layout_scores = compute_slide_text_layout_scores( slide_extraction, config=eval_spec.slidesgenbench.scoring_config.get("text_layout_config"), weights=eval_spec.slidesgenbench.scoring_config.get("text_layout_weights"), ) render_error: str | None = None if ( presentation is not None and render_service and hasattr(render_service, "render_presentation") ): try: rendered_presentation = render_service.render_presentation(presentation) rendered_slide_map = { slide.slide_index: slide for slide in rendered_presentation.slide_images } current_rendered_slide = rendered_slide_map.get(slide_index) if current_rendered_slide is None: raise ValueError( f"missing rendered slide for slide_index {slide_index}" ) previous_slide = None previous_rendered_slide = None if slide_index > 1: previous_rendered_slide = rendered_slide_map.get(slide_index - 1) if previous_rendered_slide is None: raise ValueError( f"missing rendered previous slide for slide_index {slide_index - 1}" ) if previous_slide_extractions: previous_slide = previous_slide_extractions[-1] else: previous_slide = inspection_service.inspect_slide( slide_index - 1, presentation=presentation, ) rendered_aesthetics = compute_intermediate_rendered_aesthetics_score( current_slide=slide_extraction, current_rendered_slide=current_rendered_slide, previous_slide=previous_slide, previous_rendered_slide=previous_rendered_slide, slide_width_in=float( slide_extraction.metadata.get("slide_width_in") or 0.0 ), slide_height_in=float( slide_extraction.metadata.get("slide_height_in") or 0.0 ), metric_weights=eval_spec.slidesgenbench.scoring_config.get( "aesthetic_weights" ), harmony_config=eval_spec.slidesgenbench.scoring_config.get( "harmony_config" ), rhythm_config=eval_spec.slidesgenbench.scoring_config.get( "rhythm_config" ), ) except Exception as error: render_error = str(error) s_slide_aesthetic = float(rendered_aesthetics.get("aesthetic", 0.0)) s_slide_text_layout = float(text_layout_scores.get("text_layout", 0.0)) aesthetic_weight = INTERMEDIATE_SLIDESGENBENCH_AESTHETIC_WEIGHT text_layout_weight = INTERMEDIATE_SLIDESGENBENCH_TEXT_LAYOUT_WEIGHT presentbench_weight = max(0.0, 1.0 - aesthetic_weight - text_layout_weight) reward_pre_cap = ( presentbench_weight * presentbench_result.reward_total + aesthetic_weight * s_slide_aesthetic + text_layout_weight * s_slide_text_layout ) text_layout_hard_cap = float(text_layout_scores.get("hard_cap", 1.0)) reward_total = clamp_reward(reward_pre_cap * text_layout_hard_cap) reward_breakdown = { **presentbench_result.reward_breakdown, "R_slide_presentbench": presentbench_result.reward_total, "R_slide_pre_text_layout_cap": reward_pre_cap, "S_slide_aesthetic": s_slide_aesthetic, "S_slide_harmony": float(rendered_aesthetics.get("harmony", 0.0)), "S_slide_engagement": float(rendered_aesthetics.get("engagement", 0.0)), "S_slide_usability_rendered": float(rendered_aesthetics.get("usability", 0.0)), "S_slide_rhythm": float(rendered_aesthetics.get("rhythm", 0.0)), "S_slide_text_layout": s_slide_text_layout, "S_slide_text_bounds": float(text_layout_scores.get("text_bounds", 0.0)), "S_slide_text_density": float(text_layout_scores.get("text_density", 0.0)), "S_slide_text_overlap": float(text_layout_scores.get("text_overlap", 0.0)), "C_slide_text_layout_hard": text_layout_hard_cap, "R_slide": reward_total, } metadata = { **presentbench_result.metadata, "render_backend": rendered_aesthetics.get("backend"), "render_error": render_error, "used_rendered_slide_aesthetics": bool(rendered_aesthetics), "intermediate_slidesgenbench_aesthetic_weight": aesthetic_weight, "intermediate_slidesgenbench_text_layout_weight": text_layout_weight, "text_layout": text_layout_scores, } return IntermediateSlideRewardResult( slide_index=presentbench_result.slide_index, reward_total=reward_total, reward_breakdown=reward_breakdown, hard_caps=presentbench_result.hard_caps, soft_penalties=presentbench_result.soft_penalties, checklist_results=presentbench_result.checklist_results, aesthetics_results=rendered_aesthetics, artifacts=presentbench_result.artifacts, metadata=metadata, ) def compute_presentation_reward( prompt: str, source_pack: SourcePack, presentation: PptxEditor | PptxPresentation | str, *, task_constraints: TaskConstraints | None = None, use_mllm: bool = False, judge: Any | None = None, render_service: Any | None = None, inspection_service: PptxExtractionService | None = None, aesthetics_service: Any | None = None, quiz_bank_service: QuizBankGenerationService, quantitative_quiz_judge_service: QuantitativeQuizJudgeService, cache_dir: str | None = None, mode: str = "eval", ) -> RewardResult: eval_spec = build_eval_spec( prompt, source_pack, task_constraints, cache_dir=cache_dir, mode=mode, quiz_bank_service=quiz_bank_service, ) return evaluate_presentation( eval_spec, presentation, use_mllm=use_mllm, judge=judge, render_service=render_service, inspection_service=inspection_service, aesthetics_service=aesthetics_service, quantitative_quiz_judge_service=quantitative_quiz_judge_service, mode=mode, ) def compute_intermediate_slide_reward( prompt: str, source_pack: SourcePack, *, slide_index: int, presentation: PptxEditor | PptxPresentation | None = None, slide_extraction: ExtractedSlide | None = None, task_constraints: TaskConstraints | None = None, use_mllm: bool = False, judge: Any | None = None, render_service: Any | None = None, inspection_service: PptxExtractionService | None = None, aesthetics_service: Any | None = None, previous_slide_extractions: list[ExtractedSlide] | None = None, quiz_bank_service: QuizBankGenerationService, cache_dir: str | None = None, mode: str = "eval", ) -> IntermediateSlideRewardResult: eval_spec = build_eval_spec( prompt, source_pack, task_constraints, cache_dir=cache_dir, mode=mode, quiz_bank_service=quiz_bank_service, ) return evaluate_slide( eval_spec, slide_index, presentation=presentation, slide_extraction=slide_extraction, use_mllm=use_mllm, judge=judge, render_service=render_service, inspection_service=inspection_service, aesthetics_service=aesthetics_service, previous_slide_extractions=previous_slide_extractions, mode=mode, ) __all__ = [ "build_eval_spec", "compute_intermediate_slide_reward", "compute_presentation_reward", "evaluate_presentation", "evaluate_slide", ]