pratik-250620/MultiModal-Coherence-AI

1from __future__ import annotations2 3from typing import Any, Dict, Optional4 5 6# Map weakest metric -> human meaning + fix suggestions7METRIC_HINTS = {8    "st_i": {9        "dominant_failure_mode": "text_image_misalignment",10        "suggested_fix": [11            "Make the visual plan more specific: add concrete objects, setting, lighting, and camera cues.",12            "Ensure primary_entities appear in visual_attributes (e.g., 'bus', 'station', 'crowd').",13            "Avoid abstract captions; rewrite into a visualizable scene.",14        ],15    },16    "st_a": {17        "dominant_failure_mode": "text_audio_misalignment",18        "suggested_fix": [19            "Strengthen audio_intent + audio_elements: include distinct sound sources (rain, wind, traffic, birds).",20            "Add timing/texture words: 'distant', 'foreground', 'soft', 'rhythmic', 'echo'.",21            "Avoid silent/ambiguous scenes unless the prompt implies quiet.",22        ],23    },24    "si_a": {25        "dominant_failure_mode": "image_audio_misalignment",26        "suggested_fix": [27            "Align audio sources with visible scene elements (city -> traffic/hum, beach -> waves/seagulls).",28            "Remove conflicting audio elements (e.g., birds in neon city street).",29            "Add must_include constraints tying audio cues to visual objects.",30        ],31    },32    "msci": {33        "dominant_failure_mode": "global_cross_modal_incoherence",34        "suggested_fix": [35            "Regenerate the unified plan with stronger must_include/must_avoid constraints.",36            "Use prompt decomposition: scene -> visual -> audio subplans, then merge.",37            "If repeated failure: retry generation with tighter constraints (regeneration policy).",38        ],39    },40}41 42 43def diagnose_run(44    *,45    prompt: str,46    plan: Optional[Dict[str, Any]],47    narrative: Optional[Dict[str, Any]],48    scores: Dict[str, float],49    classification: Dict[str, Any],50    drift: Optional[Dict[str, Any]] = None,51) -> Dict[str, Any]:52    """53    Produces a compact, human-readable diagnostic block for bundle.json.54    """55    weakest = None56    if isinstance(classification, dict):57        weakest = classification.get("weakest_metric")58 59    hint = METRIC_HINTS.get(weakest, None)60 61    score_flags = []62    for key in ["msci", "st_i", "st_a", "si_a"]:63        value = scores.get(key)64        if value is not None and value < 0:65            score_flags.append(f"{key}<0")66 67    drift_flags = []68    if isinstance(drift, dict):69        for key in ["visual_drift", "audio_drift", "global_drift"]:70            if drift.get(key) is True:71                drift_flags.append(key)72 73    diagnostics = {74        "weakest_metric": weakest,75        "dominant_failure_mode": (hint["dominant_failure_mode"] if hint else "unknown"),76        "suggested_fix": (77            hint["suggested_fix"]78            if hint79            else ["Inspect plan + outputs; no heuristic available."]80        ),81        "evidence": {82            "score_flags": score_flags,83            "drift_flags": drift_flags,84        },85        "notes": {86            "prompt_summary": (prompt[:220] + "...")87            if len(prompt) > 22088            else prompt,89            "plan_domain": (plan.get("domain") if isinstance(plan, dict) else None),90            "plan_scene_summary": (91                plan.get("scene_summary") if isinstance(plan, dict) else None92            ),93        },94    }95 96    if isinstance(classification, dict) and classification.get("label") == "HIGH_COHERENCE":97        diagnostics["dominant_failure_mode"] = "none_high_coherence"98        diagnostics["suggested_fix"] = [99            "Optional: improve the weakest metric slightly by tightening constraints for that modality.",100            "Run multi-seed stability to ensure coherence is consistent across random seeds.",101        ]102 103    return diagnostics104