File size: 5,403 Bytes
0dd9a1c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """
数据清洗脚本：从 07_oracle_results_nothink + 05_delta_frame_mapping 生成 RL 训练数据
过滤策略：
1. 非多选题（answer 不符合 A/A.xxx 格式）→ 丢弃
2. oracle_results 为空（推理失败）→ 丢弃
3. 单 pattern 条目（1个delta）→ 直接保留
4. 多 pattern 条目（2+个delta）→ 过滤掉含异常 oracle_answer 的 pattern
异常定义：oracle_answer 不是单个字母 A-E
"""
import json
from pathlib import Path
DATA_DIR = Path(__file__).resolve().parent
ORACLE_FILE = DATA_DIR / "07_oracle_results_nothink.json"
DELTA_FILE = DATA_DIR / "05_delta_frame_mapping.json"
OUTPUT_FILE = DATA_DIR / "08_rl_training_data.json"
VALID_ANSWERS = set("ABCDE")
def is_multiple_choice(answer: str) -> bool:
"""判断是否为标准多选题格式：A / A. xxx / A xxx"""
ans = answer.strip()
return (len(ans) == 1 and ans in VALID_ANSWERS) or \
(len(ans) >= 2 and ans[0] in VALID_ANSWERS and ans[1] in ". ")
def is_valid_oracle_answer(answer: str) -> bool:
"""oracle_answer 是否为合法的单字母选项"""
return answer.strip() in VALID_ANSWERS
def main():
# 加载 delta frame mapping
with open(DELTA_FILE) as f:
delta_data = json.load(f)
delta_map = {m["problem_id"]: m for m in delta_data["mappings"]}
print(f"[Delta] 加载 {len(delta_map)} 条")
# 加载 oracle results，同时构建 oracle 查找表
with open(ORACLE_FILE) as f:
oracle_data = json.load(f)
oracle_results = oracle_data["results"]
print(f"[Oracle] 加载 {len(oracle_results)} 条")
# 构建查找表：(problem_id, mask_delta_id) -> oracle entry
oracle_map = {}
for r in oracle_results:
pid = r["problem_id"]
for o in r.get("oracle_results", []):
oracle_map[(pid, o["mask_delta_id"])] = o
stats = {
"total_problems": len(oracle_results),
"skipped_not_mc": 0,
"skipped_empty_oracle": 0,
"skipped_no_delta": 0,
"output_problems": 0,
"single_pattern_kept": 0,
"multi_pattern_kept": 0,
"multi_pattern_abnormal_removed": 0,
"total_mask_patterns": 0,
"oracle_correct_count": 0,
}
training_data = []
for problem in oracle_results:
pid = problem["problem_id"]
# 过滤1：非多选题
if not is_multiple_choice(problem["gt_answer"]):
stats["skipped_not_mc"] += 1
continue
# 过滤2：oracle_results 为空
if not problem["oracle_results"]:
stats["skipped_empty_oracle"] += 1
continue
# 过滤3：没有 delta mapping
delta_item = delta_map.get(pid)
if delta_item is None:
stats["skipped_no_delta"] += 1
continue
delta_mask_patterns = delta_item.get("mask_patterns", [])
n_patterns = len(delta_mask_patterns)
# 构建 mask_patterns
mask_patterns = []
for p in delta_mask_patterns:
mid = p["mask_delta_id"]
oracle_entry = oracle_map.get((pid, mid))
oracle_correct = oracle_entry["oracle_correct"] if oracle_entry else None
# 多 pattern 时，过滤掉异常 oracle_answer 的 pattern
if n_patterns > 1 and oracle_entry is not None:
if not is_valid_oracle_answer(oracle_entry["oracle_answer"]):
stats["multi_pattern_abnormal_removed"] += 1
continue
pattern = {
"mask_delta_id": mid,
"zero_frame_indices": p["zero_frame_indices"],
"remaining_frames": p["remaining_frames"],
"mask_ratio": p["mask_ratio"],
"oracle_correct": oracle_correct,
}
mask_patterns.append(pattern)
stats["total_mask_patterns"] += 1
if oracle_correct:
stats["oracle_correct_count"] += 1
# 过滤掉清洗后 mask_patterns 为空的条目
if not mask_patterns:
continue
entry = {
"problem_id": pid,
"video": problem["video_path"],
"question": problem["question"],
"answer": problem["gt_answer"],
"n_sampled_frames": problem["n_sampled_frames"],
"mask_patterns": mask_patterns,
}
training_data.append(entry)
stats["output_problems"] += 1
if n_patterns == 1:
stats["single_pattern_kept"] += 1
else:
stats["multi_pattern_kept"] += 1
# 写出
n_total_p = stats["total_mask_patterns"]
output = {
"metadata": {
"total_entries": len(training_data),
"total_mask_patterns": n_total_p,
"oracle_correct_count": stats["oracle_correct_count"],
"oracle_correct_rate": round(stats["oracle_correct_count"] / n_total_p, 3) if n_total_p else 0,
"oracle_source": ORACLE_FILE.name,
"delta_source": DELTA_FILE.name,
},
"data": training_data,
}
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
json.dump(output, f, ensure_ascii=False, indent=2)
print(f"\n{'='*50}")
print(f"清洗完成，输出到 {OUTPUT_FILE.name}")
print(f"{'='*50}")
for k, v in stats.items():
print(f" {k}: {v}")
if __name__ == "__main__":
main()
|