312 lines
11 KiB
Python
312 lines
11 KiB
Python
|
#!/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()
|