#!/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()