import json import os def load_jsonl(path): """ Loads a JSONL file into a dictionary keyed by 'doc_id'. """ data = {} try: with open(path, 'r', encoding='utf-8') as f: for i, line in enumerate(f): try: record = json.loads(line) # Use 'doc_id' as the unique key for comparison if 'doc_id' not in record: print(f"Warning: Missing 'doc_id' in {path} at line {i + 1}. Skipping.") continue data[record['doc_id']] = record except json.JSONDecodeError: print(f"Error: Could not parse JSON in {path} at line {i + 1}. Skipping.") except FileNotFoundError: print(f"Error: File not found at {path}") return None except Exception as e: print(f"Error opening file {path}: {e}") return None print(f"Successfully loaded {len(data)} records from {path}.") return data def find_failure_cases_textvqa(ours_output_path, baseline_output_path, score_key="exact_match"): """ Compares two output files and finds items where 'ours' is wrong (0.0) and 'baseline' is right (1.0). Args: ours_output_path (str): Path to your model's output JSONL. baseline_output_path (str): Path to the baseline model's output JSONL. score_key (str): The key in the JSON to find the score dictionary. Default is "vstar_overall_acc". Returns: list: A list of dictionaries, each containing details of a failure case. """ failure_cases = [] # Load the baseline results into a dictionary for fast lookup baseline_results = load_jsonl(baseline_output_path) if baseline_results is None: return [] print(f"\nProcessing {ours_output_path} against baseline...") try: with open(ours_output_path, 'r', encoding='utf-8') as f: for i, line in enumerate(f): try: our_record = json.loads(line) if 'doc_id' not in our_record: print(f"Warning: Missing 'doc_id' in 'ours' file at line {i + 1}. Skipping.") continue doc_id = our_record['doc_id'] # Check if this doc_id exists in the baseline if doc_id not in baseline_results: print(f"Warning: doc_id {doc_id} from 'ours' file not found in baseline. Skipping.") continue baseline_record = baseline_results[doc_id] # Check if score keys exist before accessing if score_key not in our_record : print(f"Warning: Missing score key '{score_key}' in 'ours' file for doc_id {doc_id}. Skipping.") continue if score_key not in baseline_record: print( f"Warning: Missing score key '{score_key}' in baseline file for doc_id {doc_id}. Skipping.") continue # Get the scores if type(our_record[score_key]) is not dict or type(baseline_record[score_key]) is not dict: our_score = our_record[score_key] baseline_score = baseline_record[score_key] else: our_score = our_record[score_key]['score'] baseline_score = baseline_record[score_key]['score'] # The core logic: Find where we failed and baseline succeeded if our_score < 0.3 and baseline_score > 0.7: # Format a clean dictionary for case study failure_info = { "doc_id": doc_id, "question": our_record['input'], "ground_truth": baseline_record['filtered_resps'][0], "our_prediction": our_record['filtered_resps'][0],#[score_key].get('prediction', 'N/A'), "baseline_prediction": baseline_record['filtered_resps'][0], "full_our_record": our_record # Optional: include for full context } failure_cases.append(failure_info) except json.JSONDecodeError: print(f"Error: Could not parse JSON in 'ours' file at line {i + 1}. Skipping.") except KeyError: # This catches if a sub-key like 'score' is missing print(f"Error: Missing expected key structure in 'ours' file at line {i + 1}. Skipping.") except FileNotFoundError: print(f"Error: File not found at {ours_output_path}") return [] except Exception as e: print(f"Error opening file {ours_output_path}: {e}") return [] print(f"\n--- Analysis Complete ---") print(f"Found {len(failure_cases)} failure cases where 'ours' was wrong and 'baseline' was right.") return failure_cases def find_failure_cases_vstar(ours_output_path, baseline_output_path, score_key="vstar_overall_acc"): """ Compares two output files and finds items where 'ours' is wrong (0.0) and 'baseline' is right (1.0). Args: ours_output_path (str): Path to your model's output JSONL. baseline_output_path (str): Path to the baseline model's output JSONL. score_key (str): The key in the JSON to find the score dictionary. Default is "vstar_overall_acc". Returns: list: A list of dictionaries, each containing details of a failure case. """ failure_cases = [] # Load the baseline results into a dictionary for fast lookup baseline_results = load_jsonl(baseline_output_path) if baseline_results is None: return [] print(f"\nProcessing {ours_output_path} against baseline...") try: with open(ours_output_path, 'r', encoding='utf-8') as f: for i, line in enumerate(f): try: our_record = json.loads(line) if 'doc_id' not in our_record: print(f"Warning: Missing 'doc_id' in 'ours' file at line {i + 1}. Skipping.") continue doc_id = our_record['doc_id'] # Check if this doc_id exists in the baseline if doc_id not in baseline_results: print(f"Warning: doc_id {doc_id} from 'ours' file not found in baseline. Skipping.") continue baseline_record = baseline_results[doc_id] # Check if score keys exist before accessing if score_key not in our_record or 'score' not in our_record[score_key]: print(f"Warning: Missing score key '{score_key}' in 'ours' file for doc_id {doc_id}. Skipping.") continue if score_key not in baseline_record or 'score' not in baseline_record[score_key]: print( f"Warning: Missing score key '{score_key}' in baseline file for doc_id {doc_id}. Skipping.") continue # Get the scores our_score = our_record[score_key]['score'] baseline_score = baseline_record[score_key]['score'] # The core logic: Find where we failed and baseline succeeded if our_score == 0.0 and baseline_score == 1.0: # Format a clean dictionary for case study failure_info = { "doc_id": doc_id, "question": our_record.get('doc', {}).get('text', 'N/A'), "ground_truth": our_record.get('target', 'N/A'), "our_prediction": our_record[score_key].get('prediction', 'N/A'), "baseline_prediction": baseline_record[score_key].get('prediction', 'N/A'), "full_our_record": our_record # Optional: include for full context } failure_cases.append(failure_info) except json.JSONDecodeError: print(f"Error: Could not parse JSON in 'ours' file at line {i + 1}. Skipping.") except KeyError: # This catches if a sub-key like 'score' is missing print(f"Error: Missing expected key structure in 'ours' file at line {i + 1}. Skipping.") except FileNotFoundError: print(f"Error: File not found at {ours_output_path}") return [] except Exception as e: print(f"Error opening file {ours_output_path}: {e}") return [] print(f"\n--- Analysis Complete ---") print(f"Found {len(failure_cases)} failure cases where 'ours' was wrong and 'baseline' was right.") return failure_cases # --- Example Usage --- if __name__ == "__main__": # --- 1. Create dummy data files for the example --- # This example has 3 cases: # 1. We fail (B), baseline succeeds (A). -> This should be found. # 2. We succeed (A), baseline fails (C). -> This should be ignored. # 3. Both succeed (A). -> This should be ignored. # # Define file paths # baseline_file = "./logs/liuhaotian__llava-v1.5-7b/20250902_190647_samples_textvqa_val.jsonl" # ours_file = "./logs/checkpoints__llava-v1.5-7b-roi-K15T3-152k-v1bf16Mheads-bg0p05fg0p2-filled/20251113_113923_samples_textvqa_val.jsonl" # # #failures = find_failure_cases_textvqa(ours_file, baseline_file, 'exact_match') # failures = find_failure_cases_textvqa(baseline_file, ours_file , 'exact_match') #baseline_file = "./logs/liuhaotian__llava-v1.5-7b/20250902_201850_samples_vstar_bench.jsonl" #ours_file = "./logs/checkpoints__llava-v1.5-7b-roi-K15T3-152k-v1bf16Mheads-bg0p05fg0p2-filled/20251113_130922_samples_vstar_bench.jsonl" # baseline_file = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251113_135313_samples_vstar_bench.jsonl" # ours_file = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K16T3-152k-v1bf16Mheads-twiginit-filled/20251113_134244_samples_vstar_bench.jsonl" # # failures = find_failure_cases_vstar(baseline_file, ours_file) # failure_ids = [case['doc_id'] for case in failures] ###### TextVQA Failure Case Analysis #baseline_file_textvqa = "logs/output__qwen2_5vl-7b-roi-K18T3-185k-v1bf16-denoisedv2-twiginit-filled/20251209_181751_samples_textvqa_val.jsonl" baseline_file_textvqa = "logs/Qwen__Qwen3-VL-4B-Instruct/20260122_115842_samples_textvqa_val.jsonl" ours_file_textvqa = "logs/output__qwen3vl-4b-roi-K24T3-185k-v1bf16-twiginit/20260122_120855_samples_textvqa_val.jsonl" failures_baseline_textvqa = find_failure_cases_textvqa(baseline_file_textvqa, ours_file_textvqa, 'exact_match') failure_ids_baseline_textvqa = [case['doc_id'] for case in failures_baseline_textvqa] failures_ours_textvqa = find_failure_cases_textvqa(ours_file_textvqa, baseline_file_textvqa, 'exact_match') failure_ids_ours_textvqa = [case['doc_id'] for case in failures_ours_textvqa] failure_ids_ours_textvqa_dict = {case['doc_id']: case for case in failures_ours_textvqa} ### print some results for i, case in enumerate(failures_ours_textvqa[:10]): print(f"\nCase {i + 1} (doc_id: {case['doc_id']})") print(f" Question: {case['question']}") print(f" Ground Truth: {case['ground_truth']}") print(f" Our (Wrong) Prediction: {case['our_prediction']}") print(f" Baseline (Correct) Prediction: {case['baseline_prediction']}") # ###### InfoVQA Failure Case Analysis # baseline_file_infovqa = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251204_093838_samples_infovqa_val.jsonl" # ours_file_infovqa = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K18T3-185k-v1bf16-denoisedv2-twiginit-filled/20251215_131620_samples_infovqa_val.jsonl" # # failures_baseline_infovqa = find_failure_cases_textvqa(baseline_file_infovqa, ours_file_infovqa, 'anls') # failure_ids_baseline_infovqa = [case['doc_id'] for case in failures_baseline_infovqa] # # failures_ours_infovqa = find_failure_cases_textvqa(ours_file_infovqa, baseline_file_infovqa, 'anls') # failure_ids_ours_infovqa = [case['doc_id'] for case in failures_ours_infovqa] # # ##### DocVQA Failure Case Analysis # baseline_file_docvqa = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251204_095001_samples_docvqa_val.jsonl" # ours_file_docvqa = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K18T3-134k-gatedv2PB-wGTwRoI-twiginit-filled/20251210_143202_samples_docvqa_val.jsonl" # # failures_baseline_docvqa = find_failure_cases_textvqa(baseline_file_docvqa, ours_file_docvqa, 'anls') # failure_ids_baseline_docvqa = [case['doc_id'] for case in failures_baseline_docvqa] # failures_ours_docvqa = find_failure_cases_textvqa(ours_file_docvqa, baseline_file_docvqa, 'anls') # failure_ids_ours_docvqa = [case['doc_id'] for case in failures_ours_docvqa] # # ##### ChartQA Failure Case Analysis # baseline_file_chartqa = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251204_094359_samples_chartqa.jsonl" # ours_file_chartqa = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K18T3-134k-gatedv2PB-wGTwRoI-twiginit-filled/20251210_142453_samples_chartqa.jsonl" # # failures_baseline_chartqa = find_failure_cases_textvqa(baseline_file_chartqa, ours_file_chartqa, 'relaxed_overall') # failure_ids_baseline_chartqa = [case['doc_id'] for case in failures_baseline_chartqa] # failures_ours_chartqa = find_failure_cases_textvqa(ours_file_chartqa, baseline_file_chartqa, 'relaxed_overall') # failure_ids_ours_chartqa = [case['doc_id'] for case in failures_ours_chartqa] # # # ###### OCRBench Failure Case Analysis # baseline_file_ocrbench = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251204_094800_samples_ocrbench.jsonl" # ours_file_ocrbench = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K18T3-134k-gatedv2PB-wGTwRoI-twiginit-filled/20251210_142936_samples_ocrbench.jsonl" # failures_baseline_ocrbench = find_failure_cases_textvqa(baseline_file_ocrbench, ours_file_ocrbench, 'ocrbench_accuracy') # failure_ids_baseline_ocrbench = [case['doc_id'] for case in failures_baseline_ocrbench] # failures_ours_ocrbench = find_failure_cases_textvqa(ours_file_ocrbench, baseline_file_ocrbench, 'ocrbench_accuracy') # failure_ids_ours_ocrbench = [case['doc_id'] for case in failures_ours_ocrbench] # pass # if failure_ids_ours: # print("\n--- Failure Cases ---") # for i, case in enumerate(failure_ids_ours): # print(f"\nCase {i + 1} (doc_id: {case['doc_id']})") # print(f" Question: {case['question']}") # print(f" Ground Truth: {case['ground_truth']}") # print(f" Our (Wrong) Prediction: {case['our_prediction']}") # print(f" Baseline (Correct) Prediction: {case['baseline_prediction']}")