metrics/graph_matching - 副本.py

import numpy as np
from rouge_score import rouge_scorer
from bert_score import score as score_bert
from nltk.translate.bleu_score import sentence_bleu
from nltk.translate.bleu_score import SmoothingFunction
from scipy.optimize import linear_sum_assignment
from spacy.tokenizer import Tokenizer
# from spacy.lang.en import English
from spacy.lang.zh import Chinese
import re
import networkx as nx
from sklearn import preprocessing
from sklearn.metrics import precision_score, recall_score, f1_score
import json

# 修改图
def modify_graph(original_graph):
    modified_graph = []
    for x in original_graph:
        # modified_graph.append([str(t).strip() for t in x])
        # 将所有元素转换为字符串，去除首尾的空格（无需转换大小写）
        modified_graph.append([str(t).strip() for t in x])
    return modified_graph

# 获取三元组匹配的F1分数
def get_triple_match_f1(gold_graphs, pred_graphs):
    # 1. 预处理：标准化格式，确保所有数据类型一致
    # 2. 使用MultiLabelBinarizer将三元组转换为二进制矩阵
    # 3. 计算精确率、召回率和F1
    new_gold_graphs = [modify_graph(graph) for graph in gold_graphs]
    new_pred_graphs = [modify_graph(graph) for graph in pred_graphs]
    new_gold_graphs_list = [[str(string) for string in sublist] for sublist in new_gold_graphs]
    new_pred_graphs_list = [[str(string) for string in sublist] for sublist in new_pred_graphs]
    # 获取所有类
    #First get all the classes by combining the triples in the pred_graphs and gold_graphs
    allclasses = new_pred_graphs_list + new_gold_graphs_list
    allclasses = [item for items in allclasses for item in items]
    allclasses = list(set(allclasses))

    lb = preprocessing.MultiLabelBinarizer(classes=allclasses)
    # 将三元组转换为二进制矩阵
    mcbin = lb.fit_transform(new_pred_graphs_list)
    mrbin = lb.fit_transform(new_gold_graphs_list)
    # 计算精确率、召回率和F1
    precision = precision_score(mrbin, mcbin, average='micro')
    recall = recall_score(mrbin, mcbin, average='micro')
    f1 = f1_score(mrbin, mcbin, average='micro')

    print('Full triple scores')
    print('-----------------------------------------------------------------')
    print('Precision: ' + str(precision) + ' Recall: ' + str(recall) + '\nF1: ' + str(f1))
    return f1

def get_triple_match_accuracy(pred_graph, gold_graph):
    """
    计算三元组匹配准确率:
    直接比较预测和真实三元组是否完全相同,不考虑图结构
    返回预测三元组中与真实三元组完全匹配的比例
    """
    pred = modify_graph(pred_graph)
    gold = modify_graph(gold_graph)    
    matchs = 0
    for x in pred:
        if x in gold:
            matchs += 1
    acc = matchs/len(pred)
    return acc

def get_graph_match_accuracy(pred_graphs, gold_graphs):
    """
    计算图匹配准确率:
    将三元组转换为有向图,考虑节点和边的连接关系
    通过图同构判断两个图的结构是否完全相同
    返回预测图与真实图完全同构的比例
    """
    matchs = 0
    for pred, gold in zip(pred_graphs, gold_graphs):
        g1 = nx.DiGraph()
        g2 = nx.DiGraph()

        for edge in gold:
            g1.add_node(str(edge[0]).strip(), label=str(edge[0]).strip())
            g1.add_node(str(edge[2]).strip(), label=str(edge[2]).strip())
            g1.add_edge(str(edge[0]).strip(), str(edge[2]).strip(), label=str(edge[1]).strip())

        for edge in pred:
            if len(edge) == 2:
                edge.append('NULL')
            elif len(edge) == 1:
                edge.append('NULL')
                edge.append('NULL')
            g2.add_node(str(edge[0]).strip(), label=str(edge[0]).strip())
            g2.add_node(str(edge[2]).strip(), label=str(edge[2]).strip())
            g2.add_edge(str(edge[0]).strip(), str(edge[2]).strip(), label=str(edge[1]).strip())

        if nx.is_isomorphic(g1, g2, edge_match=lambda x, y: x == y):
            matchs += 1
    acc = matchs/len(pred_graphs)
    return acc

def get_tokens(gold_edges, pred_edges):
    nlp = Chinese()  # 使用中文分词器
    # tokenizer = Tokenizer(nlp.vocab, infix_finditer=re.compile(r'''[;]''').finditer)
    tokenizer = Tokenizer(nlp.vocab, 
                         infix_finditer=re.compile(r'''[;|，|。]''').finditer,
                         prefix_search=re.compile(r'''^[【（\[［]''').search,
                         suffix_search=re.compile(r'''[】）\]］]$''').search)
    gold_tokens = []
    pred_tokens = []

    for i in range(len(gold_edges)):
        gold_tokens_edges = []
        pred_tokens_edges = []

        for sample in tokenizer.pipe(gold_edges[i]):
            gold_tokens_edges.append([j.text for j in sample])
        for sample in tokenizer.pipe(pred_edges[i]):
            pred_tokens_edges.append([j.text for j in sample])
        gold_tokens.append(gold_tokens_edges)
        pred_tokens.append(pred_tokens_edges)

    return gold_tokens, pred_tokens


def split_to_edges(graphs):
    processed_graphs = []
    for graph in graphs:
        #print(graph)
        processed_graphs.append([";".join(triple).strip() for triple in graph])
    return processed_graphs


def get_bert_score(all_gold_edges, all_pred_edges):
    references = []
    candidates = []

    ref_cand_index = {}
    for (gold_edges, pred_edges) in zip(all_gold_edges, all_pred_edges):
        for (i, gold_edge) in enumerate(gold_edges):
            for (j, pred_edge) in enumerate(pred_edges):
                references.append(gold_edge)
                candidates.append(pred_edge)
                ref_cand_index[(gold_edge, pred_edge)] = len(references) - 1

    _, _, bs_F1 = score_bert(cands=candidates, refs=references, lang='en', idf=False)
    print("Computed bert scores for all pairs")

    precisions, recalls, f1s = [], [], []
    for (gold_edges, pred_edges) in zip(all_gold_edges, all_pred_edges):
        score_matrix = np.zeros((len(gold_edges), len(pred_edges)))
        for (i, gold_edge) in enumerate(gold_edges):
            for (j, pred_edge) in enumerate(pred_edges):
                score_matrix[i][j] = bs_F1[ref_cand_index[(gold_edge, pred_edge)]]

        row_ind, col_ind = linear_sum_assignment(score_matrix, maximize=True)

        sample_precision = score_matrix[row_ind, col_ind].sum() / len(pred_edges)
        sample_recall = score_matrix[row_ind, col_ind].sum() / len(gold_edges)

        precisions.append(sample_precision)
        recalls.append(sample_recall)
        f1s.append(2 * sample_precision * sample_recall / (sample_precision + sample_recall))

    return np.array(precisions), np.array(recalls), np.array(f1s)


# Note: These graph matching metrics are computed by considering each graph as a set of edges and each edge as a
# sentence
def get_bleu_rouge(gold_tokens, pred_tokens, gold_sent, pred_sent):
    scorer_rouge = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rouge3', 'rougeL'], use_stemmer=True)

    precisions_bleu = []
    recalls_bleu = []
    f1s_bleu = []

    precisions_rouge = []
    recalls_rouge = []
    f1s_rouge = []

    for graph_idx in range(len(gold_tokens)):
        score_bleu = np.zeros((len(pred_tokens[graph_idx]), len(gold_tokens[graph_idx])))
        score_rouge = np.zeros((len(pred_tokens[graph_idx]), len(gold_tokens[graph_idx])))
        for p_idx in range(len(pred_tokens[graph_idx])):
            for g_idx in range(len(gold_tokens[graph_idx])):
                score_bleu[p_idx, g_idx] = sentence_bleu([gold_tokens[graph_idx][g_idx]], pred_tokens[graph_idx][p_idx], smoothing_function=SmoothingFunction().method1)
                # # 将token列表转换为字符串
                # gold_text = ' '.join(gold_tokens[graph_idx][g_idx])
                # pred_text = ' '.join(pred_tokens[graph_idx][p_idx])
                # score_rouge[p_idx, g_idx] = scorer_rouge.score(gold_text, pred_text)['rouge2'].precision
                
                score_rouge[p_idx, g_idx] = \
                    scorer_rouge.score(gold_sent[graph_idx][g_idx], pred_sent[graph_idx][p_idx])['rouge2'].fmeasure  # 使用F1而不是precision
                
        def _scores(cost_matrix):
            row_ind, col_ind = linear_sum_assignment(cost_matrix, maximize=True)
            precision = cost_matrix[row_ind, col_ind].sum() / cost_matrix.shape[0]
            recall = cost_matrix[row_ind, col_ind].sum() / cost_matrix.shape[1]
            f1 = (2 * precision * recall) / (precision + recall) if precision + recall > 0 else 0
            return precision, recall, f1

        precision_bleu, recall_bleu, f1_bleu = _scores(score_bleu)
        precisions_bleu.append(precision_bleu)
        recalls_bleu.append(recall_bleu)
        f1s_bleu.append(f1_bleu)

        precision_rouge, recall_rouge, f1_rouge = _scores(score_rouge)
        precisions_rouge.append(precision_rouge)
        recalls_rouge.append(recall_rouge)
        f1s_rouge.append(f1_rouge)

    return np.array(precisions_rouge), np.array(recalls_rouge), np.array(f1s_rouge), np.array(
        precisions_bleu), np.array(recalls_bleu), np.array(f1s_bleu)


def return_eq_node(node1, node2):
    return node1['label'] == node2['label']


def return_eq_edge(edge1, edge2):
    return edge1['label'] == edge2['label']

def get_ged(gold_graph, pred_graph=None):
    g1 = nx.DiGraph()
    g2 = nx.DiGraph()

    for edge in gold_graph:
        g1.add_node(str(edge[0]).strip(), label=str(edge[0]).strip())
        g1.add_node(str(edge[2]).strip(), label=str(edge[2]).strip())
        g1.add_edge(str(edge[0]).strip(), str(edge[2]).strip(), label=str(edge[1]).strip())

    # The upper bound is defined wrt the graph for which GED is the worst.
    # Since ExplaGraphs (by construction) allows a maximum of 8 edges, the worst GED = gold_nodes + gold_edges + 8 + 9.
    # This happens when the predicted graph is linear with 8 edges and 9 nodes.
    # In such a case, for GED to be the worst, we assume that all nodes and edges of the predicted graph are deleted and
    # then all nodes and edges of the gold graph are added.
    # Note that a stricter upper bound can be computed by considering some replacement operations but we ignore that for convenience
    normalizing_constant = g1.number_of_nodes() + g1.number_of_edges() + 30

    if pred_graph is None:
        return 1

    for edge in pred_graph:
        if len(edge) == 2:
            edge.append('NULL')
        elif len(edge) == 1:
            edge.append('NULL')
            edge.append('NULL')
        g2.add_node(str(edge[0]).strip(), label=str(edge[0]).strip())
        g2.add_node(str(edge[2]).strip(), label=str(edge[2]).strip())
        g2.add_edge(str(edge[0]).strip(), str(edge[2]).strip(), label=str(edge[1]).strip())

    ged = nx.graph_edit_distance(g1, g2, node_match=return_eq_node, edge_match=return_eq_edge)

    assert ged <= normalizing_constant

    return ged / normalizing_constant

def clean_and_validate_data(gold_graphs, pred_graphs):
    """清理和验证数据"""
    valid_gold = []
    valid_pred = []
    
    for gold, pred in zip(gold_graphs, pred_graphs):
        # 确保两者都不为空
        if gold and pred and isinstance(gold, list) and isinstance(pred, list):
            valid_gold.append(gold)
            valid_pred.append(pred)
            
    return valid_gold, valid_pred

def evaluate_triples(gold_graphs, pred_graphs):
    print("开始评估...")
    print("="*50)
    
    # 清理和验证数据
    gold_graphs, pred_graphs = clean_and_validate_data(gold_graphs, pred_graphs)
    
    if not gold_graphs or not pred_graphs:
        print("警告：没有有效的评估数据！")
        return None
        
    # 1. Triple Match F1
    try:
        triple_f1 = get_triple_match_f1(gold_graphs, pred_graphs)
        print(f"三元组匹配F1分数: {triple_f1:.4f}")
    except Exception as e:
        print(f"计算Triple Match F1出错: {str(e)}")
        triple_f1 = 0.0
    
    # 2. Graph Match Accuracy
    try:
        graph_acc = get_graph_match_accuracy(pred_graphs, gold_graphs)
        print(f"图匹配准确率: {graph_acc:.4f}")
    except Exception as e:
        print(f"计算Graph Match Accuracy出错: {str(e)}")
        graph_acc = 0.0
        
    # 防止除零错误
    def safe_divide(a, b):
        return a / b if b != 0 else 0.0
        
    # 修改BERT Score计算
    gold_edges = split_to_edges(gold_graphs)
    pred_edges = split_to_edges(pred_graphs)
    
    if any(gold_edges) and any(pred_edges):
        precisions_BS, recalls_BS, f1s_BS = get_bert_score(gold_edges, pred_edges)
        print(f"BERT Score:")
        print(f"- Precision: {safe_divide(precisions_BS.sum(), len(precisions_BS)):.4f}")
        print(f"- Recall: {safe_divide(recalls_BS.sum(), len(recalls_BS)):.4f}")
        print(f"- F1: {safe_divide(f1s_BS.sum(), len(f1s_BS)):.4f}")
    
    # ... 其余评估代码 ...

def load_data(gold_path, pred_path):
    try:
        # 加载真实数据
        with open(gold_path, 'r', encoding='utf-8') as f:
            gold_data = json.load(f)
        
        # 加载预测数据
        with open(pred_path, 'r', encoding='utf-8') as f:
            pred_data = json.load(f)
        
        # 数据验证
        if not isinstance(gold_data, list) or not isinstance(pred_data, list):
            print("警告：数据格式不正确，应为列表格式")
            return [], []
            
        # 提取并验证三元组列表
        gold_graphs = []
        pred_graphs = []
        
        for pred_item in pred_data:
            if not isinstance(pred_item, dict) or 'text' not in pred_item or 'triple_list' not in pred_item:
                continue
                
            pred_text = pred_item['text']
            for gold_item in gold_data:
                if gold_item.get('text') == pred_text:
                    if gold_item.get('triple_list') and pred_item.get('triple_list'):
                        gold_graphs.append(gold_item['triple_list'])
                        pred_graphs.append(pred_item['triple_list'])
                    break
        
        print(f"加载的数据数量: Gold={len(gold_graphs)}, Pred={len(pred_graphs)}")
        print("Gold样本:", gold_graphs[0] if gold_graphs else "空")
        print("Pred样本:", pred_graphs[0] if pred_graphs else "空")
        
        return gold_graphs, pred_graphs
        
    except Exception as e:
        print(f"加载数据时出错: {str(e)}")
        return [], []