312 lines
11 KiB
Python
Executable File
312 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
评估脚本：计算自动提取 vs Gold Standard的准确率
|
||
"""
|
||
|
||
import json
|
||
from dataclasses import dataclass
|
||
from typing import List, Dict, Tuple
|
||
from pathlib import Path
|
||
|
||
|
||
@dataclass
|
||
class EvaluationMetrics:
|
||
"""评估指标"""
|
||
precision: float
|
||
recall: float
|
||
f1: float
|
||
tp: int # True Positive
|
||
fp: int # False Positive
|
||
fn: int # False Negative
|
||
|
||
def __str__(self):
|
||
return (
|
||
f"Precision: {self.precision:.3f}\n"
|
||
f"Recall: {self.recall:.3f}\n"
|
||
f"F1-Score: {self.f1:.3f}\n"
|
||
f"TP: {self.tp}, FP: {self.fp}, FN: {self.fn}"
|
||
)
|
||
|
||
|
||
class Evaluator:
|
||
"""评估器"""
|
||
|
||
def __init__(self, gold_path: str, pred_path: str):
|
||
"""
|
||
Args:
|
||
gold_path: Gold Standard文件路径
|
||
pred_path: 自动提取结果文件路径
|
||
"""
|
||
with open(gold_path, 'r', encoding='utf-8') as f:
|
||
self.gold = json.load(f)
|
||
with open(pred_path, 'r', encoding='utf-8') as f:
|
||
self.pred = json.load(f)
|
||
|
||
self.errors = {
|
||
"entity": {"fp": [], "fn": [], "type_mismatch": []},
|
||
"relation": {"fp": [], "fn": [], "type_mismatch": []}
|
||
}
|
||
|
||
def _normalize_name(self, name: str) -> str:
|
||
"""标准化名称用于比较"""
|
||
return name.lower().strip().replace(" ", "_")
|
||
|
||
def match_entity(self, gold_ent: Dict, pred_ent: Dict) -> Tuple[bool, str]:
|
||
"""
|
||
判断两个实体是否匹配
|
||
|
||
Returns:
|
||
(是否匹配, 匹配类型)
|
||
匹配类型: "exact"(完全匹配), "name"(名称匹配), "id"(ID匹配), "none"(不匹配)
|
||
"""
|
||
# 标准1: ID完全匹配（最严格）
|
||
if gold_ent.get("id") and pred_ent.get("id"):
|
||
if gold_ent["id"] == pred_ent["id"]:
|
||
return True, "id"
|
||
|
||
# 标准2: 名称+类型匹配
|
||
gold_name = self._normalize_name(gold_ent.get("name", ""))
|
||
pred_name = self._normalize_name(pred_ent.get("name", ""))
|
||
|
||
if gold_name == pred_name:
|
||
if gold_ent.get("type") == pred_ent.get("type"):
|
||
return True, "exact"
|
||
else:
|
||
return False, "type_mismatch"
|
||
|
||
# 标准3: 名称包含关系（宽松）
|
||
if gold_name in pred_name or pred_name in gold_name:
|
||
if gold_ent.get("type") == pred_ent.get("type"):
|
||
return True, "name"
|
||
|
||
return False, "none"
|
||
|
||
def match_relation(self, gold_rel: Dict, pred_rel: Dict) -> Tuple[bool, str]:
|
||
"""
|
||
判断两个关系是否匹配
|
||
|
||
关系匹配需要: source匹配 + target匹配 + type相同
|
||
"""
|
||
# 检查source是否匹配
|
||
source_match = self._match_relation_endpoint(
|
||
gold_rel.get("source", ""),
|
||
pred_rel.get("source", "")
|
||
)
|
||
|
||
# 检查target是否匹配
|
||
target_match = self._match_relation_endpoint(
|
||
gold_rel.get("target", ""),
|
||
pred_rel.get("target", "")
|
||
)
|
||
|
||
# 检查type是否匹配
|
||
type_match = gold_rel.get("type") == pred_rel.get("type")
|
||
|
||
if source_match and target_match and type_match:
|
||
return True, "exact"
|
||
elif source_match and target_match:
|
||
return False, "type_mismatch"
|
||
else:
|
||
return False, "none"
|
||
|
||
def _match_relation_endpoint(self, gold_id: str, pred_id: str) -> bool:
|
||
"""匹配关系的端点（source或target）"""
|
||
# 直接匹配
|
||
if gold_id == pred_id:
|
||
return True
|
||
|
||
# 名称标准化后匹配
|
||
return self._normalize_name(gold_id) == self._normalize_name(pred_id)
|
||
|
||
def evaluate_entities(self) -> EvaluationMetrics:
|
||
"""评估实体提取"""
|
||
gold_ents = self.gold.get("entities", [])
|
||
pred_ents = self.pred.get("entities", [])
|
||
|
||
tp = fp = fn = 0
|
||
|
||
# 计算TP和FP
|
||
matched_gold_indices = set()
|
||
for pred_ent in pred_ents:
|
||
matched = False
|
||
for i, gold_ent in enumerate(gold_ents):
|
||
is_match, match_type = self.match_entity(gold_ent, pred_ent)
|
||
if is_match:
|
||
tp += 1
|
||
matched_gold_indices.add(i)
|
||
matched = True
|
||
break
|
||
|
||
if not matched:
|
||
fp += 1
|
||
self.errors["entity"]["fp"].append({
|
||
"pred": pred_ent,
|
||
"reason": "未找到匹配的gold实体"
|
||
})
|
||
|
||
# 计算FN
|
||
for i, gold_ent in enumerate(gold_ents):
|
||
if i not in matched_gold_indices:
|
||
fn += 1
|
||
self.errors["entity"]["fn"].append({
|
||
"gold": gold_ent,
|
||
"reason": "未提取到"
|
||
})
|
||
|
||
# 计算指标
|
||
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
|
||
recall = tp / (tp + fn) if (tp + fn) > 0 else 0
|
||
f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
|
||
|
||
return EvaluationMetrics(precision, recall, f1, tp, fp, fn)
|
||
|
||
def evaluate_relations(self) -> EvaluationMetrics:
|
||
"""评估关系提取"""
|
||
gold_rels = self.gold.get("relations", [])
|
||
pred_rels = self.pred.get("relations", [])
|
||
|
||
tp = fp = fn = 0
|
||
|
||
# 计算TP和FP
|
||
matched_gold_indices = set()
|
||
for pred_rel in pred_rels:
|
||
matched = False
|
||
for i, gold_rel in enumerate(gold_rels):
|
||
is_match, match_type = self.match_relation(gold_rel, pred_rel)
|
||
if is_match:
|
||
tp += 1
|
||
matched_gold_indices.add(i)
|
||
matched = True
|
||
break
|
||
|
||
if not matched:
|
||
fp += 1
|
||
self.errors["relation"]["fp"].append({
|
||
"pred": pred_rel,
|
||
"reason": "未找到匹配的gold关系"
|
||
})
|
||
|
||
# 计算FN
|
||
for i, gold_rel in enumerate(gold_rels):
|
||
if i not in matched_gold_indices:
|
||
fn += 1
|
||
self.errors["relation"]["fn"].append({
|
||
"gold": gold_rel,
|
||
"reason": "未提取到"
|
||
})
|
||
|
||
# 计算指标
|
||
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
|
||
recall = tp / (tp + fn) if (tp + fn) > 0 else 0
|
||
f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
|
||
|
||
return EvaluationMetrics(precision, recall, f1, tp, fp, fn)
|
||
|
||
def print_detailed_errors(self, max_errors: int = 5):
|
||
"""打印详细错误分析"""
|
||
print("\n" + "="*60)
|
||
print("详细错误分析")
|
||
print("="*60)
|
||
|
||
# 实体错误
|
||
print("\n【实体错误】")
|
||
|
||
if self.errors["entity"]["fp"]:
|
||
print(f"\n假阳性 (FP): {len(self.errors['entity']['fp'])}个")
|
||
print("提取了但不应该存在的实体:")
|
||
for i, error in enumerate(self.errors["entity"]["fp"][:max_errors]):
|
||
pred = error["pred"]
|
||
print(f" {i+1}. {pred.get('name')} ({pred.get('type')})")
|
||
print(f" 原因: {error['reason']}")
|
||
|
||
if self.errors["entity"]["fn"]:
|
||
print(f"\n假阴性 (FN): {len(self.errors['entity']['fn'])}个")
|
||
print("应该提取但未提取的实体:")
|
||
for i, error in enumerate(self.errors["entity"]["fn"][:max_errors]):
|
||
gold = error["gold"]
|
||
print(f" {i+1}. {gold.get('name')} ({gold.get('type')})")
|
||
print(f" ID: {gold.get('id')}")
|
||
|
||
# 关系错误
|
||
print("\n【关系错误】")
|
||
|
||
if self.errors["relation"]["fp"]:
|
||
print(f"\n假阳性 (FP): {len(self.errors['relation']['fp'])}个")
|
||
print("提取了但不应该存在的关系:")
|
||
for i, error in enumerate(self.errors["relation"]["fp"][:max_errors]):
|
||
pred = error["pred"]
|
||
print(f" {i+1}. {pred.get('source')} --[{pred.get('type')}]--> {pred.get('target')}")
|
||
print(f" 原因: {error['reason']}")
|
||
|
||
if self.errors["relation"]["fn"]:
|
||
print(f"\n假阴性 (FN): {len(self.errors['relation']['fn'])}个")
|
||
print("应该提取但未提取的关系:")
|
||
for i, error in enumerate(self.errors["relation"]["fn"][:max_errors]):
|
||
gold = error["gold"]
|
||
print(f" {i+1}. {gold.get('source')} --[{gold.get('type')}]--> {gold.get('target')}")
|
||
|
||
def generate_report(self, output_path: str = None):
|
||
"""生成完整评估报告"""
|
||
lines = []
|
||
lines.append("="*60)
|
||
lines.append("GeneFamily-RAG 提取质量评估报告")
|
||
lines.append("="*60)
|
||
lines.append(f"Gold Standard: {self.gold.get('doc_id', 'unknown')}")
|
||
lines.append(f"Prediction: {self.pred.get('doc_id', 'unknown')}")
|
||
lines.append("")
|
||
|
||
# 实体评估
|
||
ent_metrics = self.evaluate_entities()
|
||
lines.append("【实体提取】")
|
||
lines.append(str(ent_metrics))
|
||
lines.append("")
|
||
|
||
# 关系评估
|
||
rel_metrics = self.evaluate_relations()
|
||
lines.append("【关系提取】")
|
||
lines.append(str(rel_metrics))
|
||
lines.append("")
|
||
|
||
# 综合评分
|
||
overall_f1 = (ent_metrics.f1 + rel_metrics.f1) / 2
|
||
lines.append("="*60)
|
||
lines.append(f"综合F1-Score: {overall_f1:.3f}")
|
||
lines.append("="*60)
|
||
|
||
report = "\n".join(lines)
|
||
|
||
if output_path:
|
||
with open(output_path, 'w', encoding='utf-8') as f:
|
||
f.write(report)
|
||
print(f"\n报告已保存: {output_path}")
|
||
|
||
return report
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description='评估实体关系提取质量')
|
||
parser.add_argument('--gold', '-g', required=True, help='Gold Standard文件路径')
|
||
parser.add_argument('--pred', '-p', required=True, help='自动提取结果文件路径')
|
||
parser.add_argument('--output', '-o', help='评估报告输出路径')
|
||
parser.add_argument('--verbose', '-v', action='store_true', help='显示详细错误')
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 运行评估
|
||
evaluator = Evaluator(args.gold, args.pred)
|
||
|
||
# 打印报告
|
||
report = evaluator.generate_report(args.output)
|
||
print(report)
|
||
|
||
# 详细错误分析
|
||
if args.verbose:
|
||
evaluator.print_detailed_errors()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|