多模态模型评估：视觉理解与跨模态推理

多模态评估维度
├── 视觉感知
│   ├── 图像识别（VQA, Image Captioning）
│   ├── 细粒度理解（OCR, 属性识别）
│   └── 空间推理（位置关系, 3D 理解）
├── 跨模态推理
│   ├── 图文推理（图→文推理）
│   ├── 文图推理（文→图检索/生成）
│   └── 多模态链式推理
├── 多模态对话
│   ├── 多轮图像对话
│   └── 视频问答
└── 生成质量
    ├── 图文一致性
    ├── 视觉质量
    └── 创意与忠实度

class VQAEvaluator:
    """VQA 评估器"""
    
    def __init__(self, eval_mode: str = "vqa_accuracy"):
        self.eval_mode = eval_mode

    def evaluate(self, predictions: list[dict]) -> dict:
        """
        predictions: [{"question_id": int, "answer": str, "gt_answers": [str, ...]}]
        """
        if self.eval_mode == "vqa_accuracy":
            return self._vqa_accuracy(predictions)
        elif self.eval_mode == "exact_match":
            return self._exact_match(predictions)

    def _vqa_accuracy(self, predictions: list[dict]) -> dict:
        """
        标准 VQA 准确率：
        对每个问题，如果至少 3/10 的标注者给出了相同答案，则算正确
        简化版：答案出现在 GT 答案列表中即算正确
        """
        correct = 0
        for pred in predictions:
            gt = [a.strip().lower() for a in pred["gt_answers"]]
            ans = pred["answer"].strip().lower()
            # VQA 标准的 soft accuracy
            count = gt.count(ans)
            min_count = 1  # 简化：至少1个匹配
            if count >= min_count:
                correct += min(1, count / 3.0)
        
        accuracy = correct / len(predictions) if predictions else 0
        return {"vqa_accuracy": accuracy, "total": len(predictions)}

    def _exact_match(self, predictions: list[dict]) -> dict:
        correct = 0
        for pred in predictions:
            gt = [a.strip().lower() for a in pred["gt_answers"]]
            ans = pred["answer"].strip().lower()
            if ans in gt:
                correct += 1
        return {"exact_match": correct / len(predictions) if predictions else 0}


class FineGrainedVQAEvaluator(VQAEvaluator):
    """细粒度 VQA 评估：按问题类型分桶"""
    
    QUESTION_TYPES = {
        "object": "图中有什么物体？",
        "count": "图中有几个XX？",
        "color": "XX是什么颜色的？",
        "spatial": "XX在YY的哪个位置？",
        "attribute": "XX有什么特征？",
        "relation": "XX和YY是什么关系？",
        "scene": "这是什么场景？",
        "ocr": "图中的文字写了什么？",
    }

    def evaluate_by_type(self, predictions: list[dict]) -> dict:
        from collections import defaultdict
        by_type = defaultdict(list)
        for pred in predictions:
            by_type[pred.get("question_type", "unknown")].append(pred)
        
        results = {}
        for q_type, preds in by_type.items():
            results[q_type] = {
                "count": len(preds),
                "accuracy": self._exact_match(preds)["exact_match"],
            }
        return results

class CaptioningEvaluator:
    """图像描述评估"""
    
    def __init__(self):
        self.metrics = {}

    def evaluate(self, prediction: str, references: list[str]) -> dict:
        results = {}
        
        # CIDEr-D: 专为图像描述设计的指标
        results["cider"] = self._cider(prediction, references)
        
        # BLEU-4
        results["bleu4"] = self._bleu(prediction, references, n=4)
        
        # METEOR
        results["meteor"] = self._meteor(prediction, references)
        
        # ROUGE-L
        results["rouge_l"] = self._rouge_l(prediction, references)
        
        # CLIPScore: 基于 CLIP 的图文匹配度
        results["clip_score"] = self._clip_score(prediction, references)
        
        return results

    def _cider(self, pred: str, refs: list[str]) -> float:
        """CIDEr: 共识评估，基于 TF-IDF 加权的 n-gram 重叠"""
        # 简化实现，实际使用 pycocoevalcap
        from pycocoevalcap.cider.cider import Cider
        cider_scorer = Cider()
        gts = {0: refs}
        res = {0: [pred]}
        score, _ = cider_scorer.compute_score(gts, res)
        return score

    def _bleu(self, pred: str, refs: list[str], n: int = 4) -> float:
        from pycocoevalcap.bleu.bleu import Bleu
        bleu_scorer = Bleu(n)
        gts = {0: refs}
        res = {0: [pred]}
        score, _ = bleu_scorer.compute_score(gts, res)
        return score[n-1]

    def _meteor(self, pred: str, refs: list[str]) -> float:
        from pycocoevalcap.meteor.meteor import Meteor
        meteor_scorer = Meteor()
        gts = {0: refs}
        res = {0: [pred]}
        score, _ = meteor_scorer.compute_score(gts, res)
        return score

    def _rouge_l(self, pred: str, refs: list[str]) -> float:
        from pycocoevalcap.rouge.rouge import Rouge
        rouge_scorer = Rouge()
        gts = {0: refs}
        res = {0: [pred]}
        score, _ = rouge_scorer.compute_score(gts, res)
        return score

    def _clip_score(self, pred: str, refs: list[str], image_path: str = None) -> float:
        """CLIPScore: 使用 CLIP 计算图文匹配度"""
        from transformers import CLIPProcessor, CLIPModel
        from PIL import Image
        import torch
        
        model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
        processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
        
        image = Image.open(image_path)
        inputs = processor(text=[pred], images=image, return_tensors="pt", padding=True)
        with torch.no_grad():
            outputs = model(**inputs)
            # 归一化的图文相似度
            score = outputs.logits_per_image.item()
            # 归一化到 0-1
            return min(max(score / 100, 0), 1)

class OCREvaluator:
    """OCR 能力评估"""
    
    def evaluate(self, prediction: str, ground_truth: str) -> dict:
        results = {}
        
        # 字符级准确率
        results["char_accuracy"] = self._char_accuracy(prediction, ground_truth)
        
        # 词级准确率
        results["word_accuracy"] = self._word_accuracy(prediction, ground_truth)
        
        # 编辑距离
        results["edit_distance"] = self._edit_distance(prediction, ground_truth)
        
        # 归一化编辑距离
        max_len = max(len(prediction), len(ground_truth), 1)
        results["normalized_edit_distance"] = results["edit_distance"] / max_len
        
        # ANLS (Average Normalized Levenshtein Similarity)
        results["anls"] = 1 - results["normalized_edit_distance"]
        
        return results

    def _char_accuracy(self, pred: str, gt: str) -> float:
        """字符级准确率"""
        if not gt:
            return 1.0 if not pred else 0.0
        correct = sum(1 for p, g in zip(pred, gt) if p == g)
        # 加上长度差异惩罚
        correct += 0  # 多出或缺少的字符算错
        return correct / len(gt)

    def _word_accuracy(self, pred: str, gt: str) -> float:
        pred_words = pred.split()
        gt_words = gt.split()
        if not gt_words:
            return 1.0 if not pred_words else 0.0
        correct = sum(1 for p, g in zip(pred_words, gt_words) if p == g)
        return correct / len(gt_words)

    def _edit_distance(self, s1: str, s2: str) -> int:
        """Levenshtein 编辑距离"""
        if len(s1) < len(s2):
            return self._edit_distance(s2, s1)
        if len(s2) == 0:
            return len(s1)
        previous_row = range(len(s2) + 1)
        for i, c1 in enumerate(s1):
            current_row = [i + 1]
            for j, c2 in enumerate(s2):
                insertions = previous_row[j + 1] + 1
                deletions = current_row[j] + 1
                substitutions = previous_row[j] + (c1 != c2)
                current_row.append(min(insertions, deletions, substitutions))
            previous_row = current_row
        return previous_row[-1]

class CrossModalReasoningEvaluator:
    """跨模态推理评估"""
    
    TASK_TYPES = [
        "visual_entailment",    # 视觉蕴含：图→文 是否支持
        "visual_reasoning",     # 视觉推理：基于图的逻辑推理
        "image_text_matching",  # 图文匹配
        "visual_commonsense",   # 视觉常识推理
        "multimodal_cot",       # 多模态链式推理
    ]

    def evaluate_visual_entailment(self, predictions: list[dict]) -> dict:
        """
        视觉蕴含：判断文本假设是否被图像支持
        标签: entailment / neutral / contradiction
        """
        from sklearn.metrics import classification_report, accuracy_score
        labels = ["entailment", "neutral", "contradiction"]
        y_true = [p["gt_label"] for p in predictions]
        y_pred = [p["pred_label"] for p in predictions]
        
        report = classification_report(y_true, y_pred, 
                                       labels=labels, output_dict=True)
        return {
            "accuracy": accuracy_score(y_true, y_pred),
            "per_class": {l: report[l] for l in labels},
        }

    def evaluate_image_text_matching(self, predictions: list[dict]) -> dict:
        """
        图文匹配：给定图片和多个文本，选择最匹配的
        """
        correct = 0
        for pred in predictions:
            if pred["pred_match"] == pred["gt_match"]:
                correct += 1
        return {"accuracy": correct / len(predictions) if predictions else 0}

    def evaluate_multimodal_cot(self, predictions: list[dict]) -> dict:
        """
        多模态链式推理：评估推理步骤和最终答案
        """
        results = []
        for pred in predictions:
            # 评估最终答案
            answer_correct = self._check_answer(pred["pred_answer"], pred["gt_answer"])
            # 评估推理步骤（使用 LLM-as-Judge）
            reasoning_score = self._evaluate_reasoning(
                pred["image_description"],
                pred["reasoning_steps"],
                pred["pred_answer"]
            )
            results.append({
                "answer_correct": answer_correct,
                "reasoning_score": reasoning_score,
            })
        
        return {
            "answer_accuracy": sum(r["answer_correct"] for r in results) / len(results),
            "avg_reasoning_score": sum(r["reasoning_score"] for r in results) / len(results),
        }

    def _check_answer(self, pred: str, gt: str) -> bool:
        pred_clean = pred.strip().lower()
        gt_clean = gt.strip().lower()
        return gt_clean in pred_clean or pred_clean == gt_clean

    def _evaluate_reasoning(self, image_desc: str, reasoning: str, answer: str) -> float:
        """使用 LLM 评估推理质量"""
        prompt = f"""请评估以下多模态推理的质量（0-10分）：
图像描述：{image_desc}
推理过程：{reasoning}
最终答案：{answer}

评估维度：
- 推理是否基于图像信息
- 逻辑是否连贯
- 是否有跳步或错误
请输出一个数字（0-10）。"""
        # 调用 LLM 评估
        score = call_llm("gpt-4o", prompt)
        try:
            return float(score.strip()) / 10.0
        except ValueError:
            return 0.5

class CrossModalConsistency:
    """图文一致性评估"""
    
    def __init__(self):
        from transformers import CLIPModel, CLIPProcessor
        self.clip_model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
        self.clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")

    def clip_similarity(self, image_path: str, text: str) -> float:
        """CLIP 图文相似度"""
        from PIL import Image
        import torch
        image = Image.open(image_path)
        inputs = self.clip_processor(text=[text], images=image, return_tensors="pt")
        with torch.no_grad():
            outputs = self.clip_model(**inputs)
            # 归一化余弦相似度
            img_feat = outputs.image_embeds
            text_feat = outputs.text_embeds
            sim = torch.nn.functional.cosine_similarity(img_feat, text_feat).item()
            return sim

    def detailed_consistency_check(self, image_path: str, text_description: str) -> dict:
        """
        细粒度一致性检查：将文本描述拆分为原子陈述，
        逐一检查图像是否支持
        """
        # 1. 将描述拆分为原子陈述
        statements = self._split_into_statements(text_description)
        
        # 2. 对每个陈述，使用 VLM 检查
        results = []
        for stmt in statements:
            check_result = self._check_statement_with_vlm(image_path, stmt)
            results.append({
                "statement": stmt,
                "supported": check_result["supported"],
                "confidence": check_result["confidence"],
            })
        
        supported = sum(1 for r in results if r["supported"])
        return {
            "total_statements": len(results),
            "supported_count": supported,
            "consistency_rate": supported / len(results) if results else 0,
            "details": results,
        }

    def _split_into_statements(self, description: str) -> list[str]:
        """将描述拆分为原子陈述"""
        prompt = f"""将以下描述拆分为独立的原子陈述，每行一个：
{description}
只输出陈述列表，不要其他内容。"""
        result = call_llm("gpt-4o", prompt)
        return [line.strip() for line in result.strip().split("\n") if line.strip()]

    def _check_statement_with_vlm(self, image_path: str, statement: str) -> dict:
        """使用 VLM 检查图像是否支持某个陈述"""
        prompt = f"""请看这张图片，判断以下陈述是否被图片内容支持：
陈述："{statement}"
请输出 JSON：{{"supported": true/false, "confidence": 0-1, "reason": "<简短理由>"}}"""
        # 调用多模态模型
        result = call_vlm("gpt-4o", image_path, prompt)
        import json
        try:
            return json.loads(result)
        except json.JSONDecodeError:
            return {"supported": False, "confidence": 0, "reason": "parse error"}

class VideoUnderstandingEvaluator:
    """视频理解评估"""
    
    def evaluate_temporal_reasoning(self, predictions: list[dict]) -> dict:
        """时序推理评估"""
        correct = 0
        for pred in predictions:
            # 时序问题：什么先发生？什么后发生？因果推理
            if pred["pred_answer"].strip().lower() == pred["gt_answer"].strip().lower():
                correct += 1
        return {"temporal_accuracy": correct / len(predictions) if predictions else 0}

    def evaluate_video_qa(self, predictions: list[dict]) -> dict:
        """视频问答评估"""
        from collections import defaultdict
        by_duration = defaultdict(list)
        for pred in predictions:
            duration_bucket = self._duration_bucket(pred.get("video_duration", 0))
            correct = pred["pred_answer"].strip().lower() == pred["gt_answer"].strip().lower()
            by_duration[duration_bucket].append(correct)
        
        results = {}
        for bucket, corrects in by_duration.items():
            results[bucket] = {
                "count": len(corrects),
                "accuracy": sum(corrects) / len(corrects),
            }
        return results

    def _duration_bucket(self, duration: float) -> str:
        if duration < 60: return "short (<1min)"
        elif duration < 300: return "medium (1-5min)"
        elif duration < 600: return "long (5-10min)"
        else: return "very_long (>10min)"

class MultimodalEvalSuite:
    """多模态综合评估套件"""
    
    def __init__(self, config: dict):
        self.vqa_eval = VQAEvaluator()
        self.caption_eval = CaptioningEvaluator()
        self.ocr_eval = OCREvaluator()
        self.reasoning_eval = CrossModalReasoningEvaluator()
        self.consistency_eval = CrossModalConsistency()
        self.video_eval = VideoUnderstandingEvaluator()
        self.results_history = []

    def run_full_evaluation(self, model, test_suite: dict) -> dict:
        """运行完整评估"""
        report = {
            "model_name": model.name,
            "timestamp": datetime.now().isoformat(),
            "categories": {},
        }

        # 1. 视觉感知
        if "vqa" in test_suite:
            report["categories"]["vqa"] = self.vqa_eval.evaluate(test_suite["vqa"])
        
        if "captioning" in test_suite:
            caption_scores = []
            for item in test_suite["captioning"]:
                score = self.caption_eval.evaluate(item["pred"], item["refs"])
                caption_scores.append(score)
            report["categories"]["captioning"] = self._aggregate_scores(caption_scores)

        # 2. OCR
        if "ocr" in test_suite:
            ocr_scores = [self.ocr_eval.evaluate(p["pred"], p["gt"]) 
                          for p in test_suite["ocr"]]
            report["categories"]["ocr"] = self._aggregate_scores(ocr_scores)

        # 3. 跨模态推理
        if "visual_entailment" in test_suite:
            report["categories"]["visual_entailment"] = \
                self.reasoning_eval.evaluate_visual_entailment(test_suite["visual_entailment"])

        # 4. 视频理解
        if "video_qa" in test_suite:
            report["categories"]["video_qa"] = \
                self.video_eval.evaluate_video_qa(test_suite["video_qa"])

        # 5. 综合评分
        report["overall"] = self._compute_overall(report["categories"])
        self.results_history.append(report)
        return report

    def _compute_overall(self, categories: dict) -> dict:
        """计算综合评分"""
        weights = {
            "vqa": 0.20,
            "captioning": 0.15,
            "ocr": 0.10,
            "visual_entailment": 0.25,
            "video_qa": 0.15,
        }
        # 还有 15% 给自定义维度
        score = 0
        total_weight = 0
        for cat, weight in weights.items():
            if cat in categories:
                cat_score = self._extract_score(cat, categories[cat])
                score += cat_score * weight
                total_weight += weight
        return {
            "weighted_score": round(score / total_weight, 4) if total_weight > 0 else 0,
            "categories_evaluated": list(categories.keys()),
        }