eval/eval_masked_obj_generative.py

"""
遮挡物体幻觉基准评估 - 生成任务
==================================

【中文说明】
本脚本用于评估模型在物体被遮挡后的生成任务中的幻觉率。

评估逻辑:
    - 问题: "详细描述这张图片"
    - 图像处理: 某个物体已被遮挡（涂黑）
    - 正确: 生成的描述中不包含被遮挡的物体
    - 幻觉: 生成的描述中仍然提到被遮挡的物体（明明看不见却说看见了）

评估指标:
    - Accuracy (准确率): 没有提及被遮挡物体的样本比例
    - Hallucination Rate (幻觉率): 提及被遮挡物体的样本比例

核心思想:
    如果一个物体被遮挡了，模型不应该在描述中提到它。
    如果仍然提到，说明模型产生了幻觉。

使用方法:
    python eval_masked_obj_generative.py \\
        --inference_file model_outputs.jsonl \\
        --save_file evaluation_results.json

输入格式 (inference_file):
    {
        "image_id": "1154_masked_person",  # 图像ID包含被遮挡物体信息
        "text": "A dog is sitting on a bench.",  # 模型生成的描述
        ...
    }

输出格式 (save_file):
    {
        "metrics": {
            "total_samples": 100,
            "correct_count": 85,
            "hallucination_count": 15,
            "accuracy": 0.85,
            "hallucination_rate": 0.15
        },
        "detailed_results": [...]
    }

【English Documentation】
Evaluation script for Masked Object Hallucination Benchmark - Generative Task

For each sample:
- Question: "Describe this image in detail."
- The masked object has been removed from the image
- Correct: Caption does NOT mention the masked object
- Hallucination: Caption mentions the masked object (which is not visible)

Metrics:
- Accuracy: percentage of samples where model does NOT mention the masked object
- Hallucination Rate: percentage of samples where model mentions the masked object
"""

import argparse
import json
import os
import re
import sys
from argparse import Namespace

import nltk
from nltk.stem import WordNetLemmatizer

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..")))

from llava.eval.utils.utils import read_json

# Initialize lemmatizer
lemmatizer = WordNetLemmatizer()

# COCO 常见物体的同义词映射表
# 用于检测模型是否用同义词提到了被遮挡的物体
# 例如: "puppy" 是 "dog" 的同义词
OBJECT_SYNONYMS = {
    "person": ["person", "people", "man", "woman", "boy", "girl", "child", "kid", "adult", "human", "individual", "pedestrian", "passenger", "player", "rider", "skier", "snowboarder", "surfer", "skater"],
    "bicycle": ["bicycle", "bike", "cycle"],
    "car": ["car", "automobile", "vehicle", "sedan", "suv"],
    "motorcycle": ["motorcycle", "motorbike", "bike"],
    "airplane": ["airplane", "plane", "aircraft", "jet", "aeroplane"],
    "bus": ["bus", "coach", "shuttle"],
    "train": ["train", "locomotive", "subway", "metro", "rail"],
    "truck": ["truck", "lorry", "pickup"],
    "boat": ["boat", "ship", "vessel", "yacht", "canoe", "kayak", "sailboat"],
    "traffic light": ["traffic light", "traffic signal", "stoplight"],
    "fire hydrant": ["fire hydrant", "hydrant"],
    "stop sign": ["stop sign"],
    "parking meter": ["parking meter", "meter"],
    "bench": ["bench", "seat"],
    "bird": ["bird", "sparrow", "pigeon", "crow", "eagle", "seagull"],
    "cat": ["cat", "kitten", "feline"],
    "dog": ["dog", "puppy", "canine", "hound"],
    "horse": ["horse", "pony", "stallion", "mare", "equine"],
    "sheep": ["sheep", "lamb"],
    "cow": ["cow", "cattle", "bull", "calf", "bovine"],
    "elephant": ["elephant"],
    "bear": ["bear"],
    "zebra": ["zebra"],
    "giraffe": ["giraffe"],
    "backpack": ["backpack", "bag", "rucksack", "knapsack"],
    "umbrella": ["umbrella", "parasol"],
    "handbag": ["handbag", "purse", "bag"],
    "tie": ["tie", "necktie", "bowtie"],
    "suitcase": ["suitcase", "luggage", "baggage"],
    "frisbee": ["frisbee", "disc", "flying disc"],
    "skis": ["skis", "ski"],
    "snowboard": ["snowboard"],
    "sports ball": ["sports ball", "ball", "soccer ball", "basketball", "football", "tennis ball", "baseball"],
    "kite": ["kite"],
    "baseball bat": ["baseball bat", "bat"],
    "baseball glove": ["baseball glove", "glove", "mitt"],
    "skateboard": ["skateboard", "board"],
    "surfboard": ["surfboard", "board"],
    "tennis racket": ["tennis racket", "racket", "racquet"],
    "bottle": ["bottle", "water bottle"],
    "wine glass": ["wine glass", "glass", "wineglass"],
    "cup": ["cup", "mug", "glass"],
    "fork": ["fork"],
    "knife": ["knife"],
    "spoon": ["spoon"],
    "bowl": ["bowl"],
    "banana": ["banana"],
    "apple": ["apple"],
    "sandwich": ["sandwich"],
    "orange": ["orange"],
    "broccoli": ["broccoli"],
    "carrot": ["carrot"],
    "hot dog": ["hot dog", "hotdog"],
    "pizza": ["pizza"],
    "donut": ["donut", "doughnut"],
    "cake": ["cake"],
    "chair": ["chair", "seat", "stool"],
    "couch": ["couch", "sofa", "loveseat"],
    "potted plant": ["potted plant", "plant", "houseplant"],
    "bed": ["bed"],
    "dining table": ["dining table", "table", "desk"],
    "toilet": ["toilet", "restroom", "bathroom"],
    "tv": ["tv", "television", "monitor", "screen"],
    "laptop": ["laptop", "notebook", "computer"],
    "mouse": ["mouse", "computer mouse"],
    "remote": ["remote", "remote control"],
    "keyboard": ["keyboard"],
    "cell phone": ["cell phone", "phone", "cellphone", "mobile phone", "smartphone", "mobile"],
    "microwave": ["microwave", "microwave oven"],
    "oven": ["oven", "stove"],
    "toaster": ["toaster"],
    "sink": ["sink", "basin"],
    "refrigerator": ["refrigerator", "fridge"],
    "book": ["book", "books"],
    "clock": ["clock", "watch"],
    "vase": ["vase", "flower vase"],
    "scissors": ["scissors"],
    "teddy bear": ["teddy bear", "teddy", "stuffed bear"],
    "hair drier": ["hair drier", "hair dryer", "dryer"],
    "toothbrush": ["toothbrush"],
}


def extract_masked_object(image_id: str) -> str:
    """
    从图像ID中提取被遮挡的物体名称
    
    命名规则: {image_id}_masked_{object_name}
    
    Examples:
        "1154_masked_person" → "person"
        "1159258_masked_dining_table" → "dining table"
    
    Args:
        image_id: 图像ID（包含被遮挡物体信息）
    
    Returns:
        str: 被遮挡的物体名称
    """
    match = re.search(r"_masked_(.+)$", image_id)
    if match:
        return match.group(1).replace("_", " ")
    return ""


def get_synonyms(obj: str) -> list:
    """
    获取物体的所有同义词
    
    从 OBJECT_SYNONYMS 映射表中查找物体的同义词列表。
    
    Args:
        obj: 物体名称
    
    Returns:
        list[str]: 同义词列表（包含物体本身）
    
    Example:
        >>> get_synonyms("dog")
        ["dog", "puppy", "canine", "hound"]
    """
    obj_lower = obj.lower().strip()
    
    # Direct match
    if obj_lower in OBJECT_SYNONYMS:
        return OBJECT_SYNONYMS[obj_lower]
    
    # Check if obj is a synonym of any key
    for key, synonyms in OBJECT_SYNONYMS.items():
        if obj_lower in synonyms:
            return synonyms
    
    # Return just the object itself if no synonyms found
    return [obj_lower]


def caption_contains_object(caption: str, target_object: str) -> tuple[bool, list[str]]:
    """
    检查描述中是否包含目标物体或其同义词
    
    检测策略:
        1. 分词并词形还原（dogs → dog）
        2. 检查单词匹配（包括同义词）
        3. 检查多词短语匹配（如 "dining table", "cell phone"）
    
    Args:
        caption: 模型生成的描述文本
        target_object: 被遮挡的目标物体
    
    Returns:
        tuple[bool, list[str]]:
            - bool: 是否包含目标物体
            - list[str]: 匹配到的词列表
    
    Example:
        >>> caption_contains_object("A puppy is running", "dog")
        (True, ["puppy"])  # puppy 是 dog 的同义词
    """
    caption_lower = caption.lower()
    
    # Tokenize and lemmatize caption
    try:
        words = nltk.word_tokenize(caption_lower)
        lemmatized_words = [lemmatizer.lemmatize(w) for w in words]
    except Exception:
        # Fallback to simple split if nltk fails
        words = caption_lower.split()
        lemmatized_words = words
    
    # Get synonyms for target object
    synonyms = get_synonyms(target_object)
    
    matched_words = []
    
    # Check single words
    for word in lemmatized_words:
        word_lemma = lemmatizer.lemmatize(word)
        for synonym in synonyms:
            synonym_lemma = lemmatizer.lemmatize(synonym)
            if word_lemma == synonym_lemma or word == synonym:
                matched_words.append(word)
    
    # Check multi-word synonyms (e.g., "dining table", "cell phone")
    caption_text = " ".join(lemmatized_words)
    for synonym in synonyms:
        if " " in synonym:  # Multi-word synonym
            synonym_parts = synonym.split()
            synonym_lemmatized = " ".join([lemmatizer.lemmatize(p) for p in synonym_parts])
            if synonym_lemmatized in caption_text or synonym in caption_lower:
                matched_words.append(synonym)
    
    # Remove duplicates
    matched_words = list(set(matched_words))
    
    return len(matched_words) > 0, matched_words


def evaluate_generative(inference_file: str, save_file: str = None) -> dict:
    """
    评估生成任务结果
    
    遍历所有推理结果，检查模型是否在描述中提到了被遮挡的物体。
    
    Args:
        inference_file: 推理结果文件路径（JSONL 格式）
        save_file: 评估结果保存路径（可选）
    
    Returns:
        dict: 包含评估指标和详细结果的字典
            {
                "metrics": {统计指标},
                "detailed_results": [{每个样本的评估结果}]
            }
    
    处理流程:
        1. 从 image_id 提取被遮挡的物体
        2. 从生成的描述中提取名词
        3. 检查是否包含被遮挡物体（含同义词）
        4. 统计幻觉率和准确率
    """
    results = read_json(inference_file)
    
    total = 0
    correct = 0  # Did NOT mention masked object (no hallucination)
    hallucinated = 0  # Mentioned masked object (hallucination)
    
    detailed_results = []
    
    for item in results:
        # Get image_id and caption
        if "question_id" in item:
            image_id = item["question_id"]
        elif "image_id" in item:
            image_id = item["image_id"]
        else:
            continue
        
        if "text" in item:
            caption = item["text"]
        elif "answer" in item:
            caption = item["answer"]
        elif "caption" in item:
            caption = item["caption"]
        else:
            continue
        
        masked_object = extract_masked_object(str(image_id))
        if not masked_object:
            continue
        
        total += 1
        
        # Check if caption contains the masked object
        contains_object, matched_words = caption_contains_object(caption, masked_object)
        
        if contains_object:
            hallucinated += 1
            status = "hallucination"
        else:
            correct += 1
            status = "correct"
        
        detailed_results.append({
            "image_id": image_id,
            "masked_object": masked_object,
            "caption": caption,
            "status": status,
            "matched_words": matched_words,
        })
    
    # Calculate metrics
    accuracy = correct / total if total > 0 else 0
    hallucination_rate = hallucinated / total if total > 0 else 0
    
    metrics = {
        "total_samples": total,
        "correct_count": correct,
        "hallucination_count": hallucinated,
        "accuracy": accuracy,
        "hallucination_rate": hallucination_rate,
        "accuracy_percent": accuracy * 100,
        "hallucination_rate_percent": hallucination_rate * 100,
    }
    
    output = {
        "metrics": metrics,
        "detailed_results": detailed_results,
    }
    
    # Save results if path provided
    if save_file:
        os.makedirs(os.path.dirname(save_file), exist_ok=True)
        with open(save_file, "w") as f:
            json.dump(output, f, indent=2)
        
        # Also save a summary file
        summary_file = save_file.replace(".json", "_summary.json")
        with open(summary_file, "w") as f:
            json.dump(metrics, f, indent=2)
    
    return output


def print_metrics(metrics: dict) -> None:
    """
    打印评估指标
    
    以格式化的方式输出评估结果到控制台。
    
    Args:
        metrics: 评估指标字典
    """
    print("\n" + "=" * 60)
    print("Masked Object Hallucination Benchmark - Generative Task")
    print("=" * 60)
    print(f"Total Samples:        {metrics['total_samples']}")
    print(f"Correct (No Hall.):   {metrics['correct_count']}")
    print(f"Hallucination:        {metrics['hallucination_count']}")
    print("-" * 60)
    print(f"Accuracy:             {metrics['accuracy_percent']:.2f}%")
    print(f"Hallucination Rate:   {metrics['hallucination_rate_percent']:.2f}%")
    print("=" * 60 + "\n")


def parse_args() -> Namespace:
    parser = argparse.ArgumentParser(description="Evaluate Masked Object Hallucination Benchmark - Generative Task")
    parser.add_argument("--inference_file", type=str, required=True, help="Path to inference results (.jsonl)")
    parser.add_argument("--save_file", type=str, default=None, help="Path to save evaluation results (.json)")
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()
    
    output = evaluate_generative(args.inference_file, args.save_file)
    print_metrics(output["metrics"])