split_word/create_pcl_match_dictionary.py

import os

from sympy.solvers.diophantine.diophantine import equivalent

script_dir = os.path.dirname(os.path.abspath(__file__))
import sys
sys.path.append(script_dir)
import pandas as pd
import jieba
import jieba.posseg as pseg
import re
import numpy as np
import json
import textdistance
import faiss
from remote_model import RemoteBGEModel


class PLCMatch:
    """通过关键词+语义相似度的方式，从用户输入中匹配PLC点位"""
    def __init__(self, project_id:int):
        # 水厂id
        self.project_id = str(project_id)
        # 路径
        self.script_dir = os.path.dirname(os.path.abspath(__file__)) # 脚本绝对路径
        # 水厂的词典根路径
        self.plc_dict_root_dir = os.path.join(self.script_dir, f'plc_dictionary/{self.project_id}_plc_dictionary')
        # 读取pcl点位文件，生成name-code映射字典
        self.name_2_code_dict = self.__read_pcl()

        # 加载用户自定义词典,添加到jieba词库
        user_dictionary_dir = os.path.join(self.script_dir, 'user_maintain_dictionary', 'jieba_words')
        user_dict_list = [os.path.join(user_dictionary_dir, _) for _ in os.listdir(user_dictionary_dir) if _.split('.')[-1] == 'txt']  # 用户词典
        self.user_dict_list = user_dict_list
        self.__load_user_dict()

        # 生成二级字典
        self.dict_level_2 = self.__make_level_two_dictionary()

        # 生成一级字典
        self.dict_level_1 = self.__make_level_one_dictionary()

        # 等价词映射表
        self.equivalent_wordmap_txt = os.path.join(self.script_dir,'user_maintain_dictionary','equivalent_words', 'equivalent_wordmap.txt')
        self.dict_equivalent_wordmap = self.__construct_equivalent_wordmap()

        # 生成知识库,PLC点位数据库中文字段
        # 加载bge-m3和bge-reranker远程模型
        self.plc_database_name_template_list = list(self.name_2_code_dict.keys())
        self.model = RemoteBGEModel('dev')
        self.knowledge = self.__load_faiss_database()


    def __load_faiss_database(self):
        """从本地加载向量数据库"""
        # 水厂的数据库字段知识库
        faiss_path = os.path.join(self.plc_dict_root_dir, f'{self.project_id}_knowledge.faiss')
        # 尝试从本地加载
        if os.path.exists(faiss_path):
            print('PLC点位查询功能从本地加载点位字段向量知识库...')
            return faiss.read_index(faiss_path)

        # 如果不存在就尝试重新创建
        # 首先，我们需要拿到数据库的点位名称，可以直接从name-code映射字典当中获取
        plc_database_name_template_list = self.plc_database_name_template_list
        # 调用远程embedding模型,one by one 地处理，远程模型通过配置参数进行归一化
        embeddings = [self.model.encode([temp], normalize=True)[0] for temp in plc_database_name_template_list]
        for _ in embeddings:
            if _ is None:
                raise RuntimeError('为plc数据库中文字段构建向量数据库时发生异常，embeddings不能存在None')
        # 要求embeddings是一个二维矩阵，类型为float32
        embeddings = np.array(embeddings, dtype=np.float32)
        # 创建 FAISS 索引
        dimension = embeddings[0].shape[0]
        local_faiss = faiss.IndexFlatIP(dimension)  # 建立内积索引
        local_faiss.add(embeddings)  # 添加索引
        # 保存未来使用
        faiss.write_index(local_faiss, faiss_path)
        return local_faiss


    def __read_pcl(self):
        """
        读取pcl文件,生成name2code词典
        :return:
        """
        # name-code映射词典路径
        dict_name2code_path = os.path.join(self.plc_dict_root_dir, f'{self.project_id}_dict_name_2_code.json')
        # 尝试从本地加载name-code映射字典
        if os.path.exists(dict_name2code_path):
            with open(dict_name2code_path, 'r', encoding='utf-8') as f:
                dict_name2code = json.load(f)
            return dict_name2code

        # 如果本地没有就重新生成
        # 检查点位文件是否存在
        pcl_file_path = os.path.join(self.plc_dict_root_dir, f'{self.project_id}_点位.xlsx') # 点位文件路径
        if not os.path.exists(pcl_file_path):
            raise FileNotFoundError(f'{pcl_file_path} does not exist')
        # 读点位
        points = pd.read_excel(pcl_file_path)
        # 列名称，name | code
        column_label_alias, column_label_code = points.columns.tolist()
        # 中英文匹配
        names = points.loc[:, column_label_alias].to_numpy()
        codes = points.loc[:, column_label_code].to_numpy()
        # 对齐命名规范, 按照中荷水厂命名风格，将1#UF或1#RO统一替换为UF1，RO1，将所有反渗透文字替换为RO，所有超滤文字替换为UF
        names = [s.replace('超滤','UF').replace('反渗透','RO') for s in names]
        names = [self.field_align(s) for s in names]
        # 名到英文的字典
        dict_name2code = dict(zip(names, codes))
        # name-code映射字典保存到本地文件
        with open(dict_name2code_path, 'w', encoding='utf-8') as f:
            json.dump(dict_name2code, f, ensure_ascii=False)
        return dict_name2code

    def __load_user_dict(self):
        """加载用户词典，添加到jieba词库"""
        # 删除
        jieba.del_word('反渗透')
        jieba.del_word('超滤')
        for user_dict_txt in self.user_dict_list:
            # 检查文件是否存在
            if not os.path.exists(user_dict_txt):
                raise FileNotFoundError(f'{user_dict_txt} does not exist')
            # 检查文件后缀名是否合法
            if os.path.splitext(user_dict_txt)[1] != '.txt':
                continue
            # 分词库加载用户字典
            jieba.load_userdict(user_dict_txt)

    def __construct_equivalent_wordmap(self):
        """构建等价词汇映射表，等价词汇的使用方式是将备查词的所有等效说法都纳入备查序列，从而保证了搜索的高召回率"""
        # 检查文件是否存在
        equivalent_wordmap_path = os.path.join(self.script_dir, 'user_maintain_dictionary','equivalent_words', 'dict_equivalent_wordmap.json')
        if os.path.exists(equivalent_wordmap_path):
            with open(equivalent_wordmap_path, 'r', encoding='utf-8') as f:
                equivalent_wordmap = json.load(f)
            return equivalent_wordmap
        # 如果本地不存在等价词典json文件，那么就尝试创建
        if not os.path.exists(self.equivalent_wordmap_txt):
            raise FileNotFoundError(f'{self.equivalent_wordmap_txt} does not exist')

        with open(self.equivalent_wordmap_txt, 'r', encoding='utf-8') as f:
            all_lines = [_.strip() for _ in f.readlines()]
        # 创建等价词汇映射表
        dict_equi_wordmap = {}
        for line in all_lines:
            split_list = line.split('=')
            for i in range(len(split_list)):
                dict_equi_wordmap[split_list[i]] = split_list
        with open(equivalent_wordmap_path, 'w', encoding='utf-8') as f:
            json.dump(dict_equi_wordmap,f,ensure_ascii=False)
        return dict_equi_wordmap

    def __make_level_two_dictionary(self):
        """创建二级字典，对点位所有字段进行正则匹配中文，将中文一样的字段聚合为同一个字典键值对，键为正则提取的中文字符"""
        group_dict = {}
        # 尝试从本地加载二级字典
        dict_level2_dict_path = os.path.join(self.plc_dict_root_dir, f'{self.project_id}_dict_level_2.json')
        if os.path.exists(dict_level2_dict_path):
            with open(dict_level2_dict_path, 'r', encoding='utf-8') as f:
                group_dict = json.load(f)
            return group_dict

        if self.name_2_code_dict is None:
            raise ValueError(f'name_2_code_dict is None', self.name_2_code_dict)
        data = self.name_2_code_dict.keys()

        # 创建二级字典
        for item in data:
            k = re.sub(r'[^\u4e00-\u9fa5]', '', item)
            # 处理没有汉字的字段
            if k == '':
                k = "无"
            if k not in group_dict.keys():
                group_dict[k] = [item]
            else:
                group_dict[k].append(item)

        # 保存二级字典到本地
        with open(dict_level2_dict_path, 'w', encoding='utf-8') as f:
            json.dump(group_dict, f, ensure_ascii=False)
        return group_dict

    @staticmethod
    def cut_compair(arr_a: str, arr_b: str, condition='nz') -> str:
        """
        :param condition: 词性
        :param arr_a:
        :param arr_b:
        :return: 第一个相同nz词
        """
        # a: w1,f1  w2,f2  w3, f3
        # b: w1,f1  w2,f2  w3, f3

        cut_arr_a = [list(_) for _ in pseg.lcut(arr_a)]
        cut_arr_b = [list(_) for _ in pseg.lcut(arr_b)]
        for i in range(len(cut_arr_a)):
            for j in range(i, len(cut_arr_b)):
                # 只比较nz词性
                if cut_arr_a[i][1] != condition or cut_arr_b[j][1] != condition:
                    continue
                if cut_arr_a[i][0] == cut_arr_b[j][0] and cut_arr_a[i][1] == cut_arr_b[j][1]:
                    return cut_arr_a[i][0]
        return ''

    def __make_level_one_dictionary(self):
        """创建一级字典"""
        group_dict = {}  # 存放二次分组的结果
        # 尝试从本地加载一级字典
        dict_level_1_path = os.path.join(self.plc_dict_root_dir, f'{self.project_id}_dict_level_1.json')
        if os.path.exists(dict_level_1_path):
            with open(dict_level_1_path, 'r', encoding='utf-8') as f:
                group_dict = json.load(f)
            return group_dict

        if self.dict_level_2.keys() is None:
            raise ValueError(f'dict_lev2 is None', self.dict_level_2)
        # 提取二级字典的所有key
        data = self.dict_level_2.keys()

        # 如果不存在就重新生成一级字典
        # 根据用户词典进行分词，筛选出所有带nz词的字段
        no_nz_list = []  # 没有nz词的字段
        nz_list = []  # 有nz词的字段
        for item in data:
            # 判断是否存在nz名词
            is_exist_n = False
            for w, f in pseg.lcut(item):
                if f == 'nz':  # 查看词性
                    is_exist_n = True
                    break
            if is_exist_n:  # 存在词
                nz_list.append(item)
            else:  # 不存在nz词
                no_nz_list.append(item)

        # 聚合具有相同nz名词的字段
        while len(nz_list) > 0:
            pos = [1 for _ in range(len(nz_list))]  # 0表示不被取，1表示需要被取，默认都要被取，用来更新nz_list给下次判断使用
            pos[0] = 0  # 标记第一个单词为不需要处理
            for i in range(len(nz_list)):
                # 查看是否存在相同的nz词
                same_nz_word = self.cut_compair(nz_list[0], nz_list[i])
                if same_nz_word:
                    # 执行聚合
                    if same_nz_word not in group_dict.keys():
                        # 首次聚合，与自身比较，创建自身类别
                        group_dict[same_nz_word] = [nz_list[i]]
                    else:
                        group_dict[same_nz_word].append(nz_list[i])

                    pos[i] = 0
            # 处理完一趟就要变更nz_list
            nz_list = np.array(nz_list)[np.array(pos, dtype=np.bool)].tolist()

        # 聚合不包含nz的名词, 单独占一个类别
        for item in no_nz_list:
            group_dict[item] = [item]

        with open(dict_level_1_path, 'w', encoding='utf-8') as f:
            json.dump(group_dict, f, ensure_ascii=False)

        return group_dict

    @staticmethod
    def field_align(input_str:str)->str:
        """按照锡山中荷命名规范对齐字段，1#UF替换为UF1，1#RO替换为RO1，保持统一"""
        sources_uf = re.findall(r'\d+#UF', input_str, re.IGNORECASE)  # 匹配1#UF
        sources_ro = re.findall(r'\d+#RO', input_str, re.IGNORECASE)  # 匹配1#RO
        sources = sources_uf + sources_ro
        for sou in sources:
            number_, flag_ = sou.split('#')
            input_str = input_str.replace(sou, flag_.upper() + number_) # 统一转为大写
        return input_str

    @ staticmethod
    def quicksort_up_part(arr:list, start:int, end:int)-> int:
        """升序排序"""
        # 双指针
        low = start
        high = end
        pivot = arr[start][1] # 基准值
        # 大数放在基准值右边，小数放在基准值左边
        while low < high:
            # 先从右向左找比基准值小的
            while low< high and arr[high][1] >= pivot:
                high -= 1
            # 此时high指向值小于基准值，交换
            if low < high:
                arr[low], arr[high] = arr[high], arr[low]
                low +=1
            # 现在开始从左向右找，比基准值大的数
            while low < high and arr[low][1] <= pivot:
                low += 1
            # 此时low指向值大于基准值，交换
            if low < high:
                arr[high], arr[low] = arr[low], arr[high]
                high -= 1
        return low


    def quicksort_up(self, arr:list, start:int, end:int):
        """按照元组第二个元素值大小进行升序排序"""
        if start >= end:
            return
        # 先排一次获得基准值位置
        mid = self.quicksort_up_part(arr, start, end)
        # 排左面
        self.quicksort_up(arr, start, mid - 1)
        # 排右面
        self.quicksort_up(arr, mid + 1, end)

    def words_similarity_score_sorted(self, query:str, candidates:list)->list:
        """计算输入语句与候选词的相似度并按照相似度分值进行排序"""
        # 选择算法（示例使用Levenshtein，归一化到0-1）
        candidates = candidates.copy()
        jarowinkler = textdistance.JaroWinkler()
        key_score_list = [(candidate, jarowinkler.normalized_similarity(query, candidate)) for candidate in candidates]
        self.quicksort_up(key_score_list, 0, len(key_score_list) - 1)  # 升序排序
        key_sorted_list = [tuple_element[0] for tuple_element in key_score_list]  # 取出key
        key_sorted_list = key_sorted_list[::-1]  # 反转，变为降序
        return key_sorted_list

    def words_similarity_score_sorted_v2(self, query:str, candidates:list)->list:
        """通过rerank的方式为候选词进行相似度排序"""
        # 调用远程reranker模型
        n = len(candidates)  # 候选词数量
        group_query = [(query, i) for i in candidates]
        score = self.model.compute_score(group_query)
        key_score_list = [(candidates[i], score[i]) for i in range(n)]
        self.quicksort_up(key_score_list, 0, len(key_score_list) - 1)  # 升序排序
        key_sorted_list = [tuple_element[0] for tuple_element in key_score_list]  # 取出key
        key_sorted_list = key_sorted_list[::-1]  # 反转，变为降序
        return key_sorted_list

    def match_v2_on(self, promt: str,is_agent:bool=False):
        """
        模糊匹配v2
        :param is_agent:
        :param promt:
        :return:
        """
        print("=" * 50)
        # 命名风格转换
        print("原始查询：", promt)
        promt = promt.replace('超滤', 'UF').replace('反渗透', 'RO').replace('号', '#').replace('组', '#')
        promt = self.field_align(promt)
        print("转换查询：", promt)
        # 输入分词
        nz_words = []
        for w, f in pseg.lcut(promt):
            print(f'{w}({f})', end="")
            if f == 'nz':
                nz_words.append(w)
        print('\n备查nz词：', nz_words)

        # 处理专有名词的等价词，为了保证高召回率，我们将备查词的所有等价说法都放入备查序列
        equivalent_words = []
        for nz_idx, nz in enumerate(nz_words):
            # 首先判断nz词是否在等价词汇表中，如果不在根本无法替换
            if nz in self.dict_equivalent_wordmap.keys():
                # 然后把等价的说法都添加进去就好了
                equivalent_words = self.dict_equivalent_wordmap.get(nz, [])
        if equivalent_words:
            nz_words += equivalent_words
            nz_words = list(set(nz_words))
        print('等价备查nz词：', nz_words)
        del equivalent_words

        # 进行一级查询,根据nz词是否包含于词典
        query_level_one = []
        for i in range(len(nz_words)):  # 为第i个nz词进行初次匹配
            result = []
            # 如果nz词包含在一级词典中就算匹配成功
            for dict_level_1_key in self.dict_level_1.keys():
                if nz_words[i] in dict_level_1_key:  # 如果nz词包含在一级词典内
                    result+= self.dict_level_1.get(dict_level_1_key)
            query_level_one.append(result)  # 放入一级查询结果中

        # 进行二级查询
        query_level_two = []
        for idx_nz, i_nz_query_result in enumerate(query_level_one):  # 遍历每个nz词的查询结果
            result = []  # 为第i个nz词进行二次匹配
            # 如果第i个nz词一级查询不为空
            if i_nz_query_result: # 第i个nz词的查询结果list
                for res_word_level_one in i_nz_query_result:
                    if res_word_level_one in self.dict_level_2.keys():
                        result += self.dict_level_2.get(res_word_level_one)  # self.dict_level_2的value本身就是字典，所以用+=拼接
            # 虽然一级查询失败，但是并不意味着映射词典里没有，因为一级词典忽略英文。
            else:  # 如果一级查询失败，就直接在name2code字典中查询
                if nz_words[idx_nz] in self.name_2_code_dict.keys():# 如果第i个nz词在2级词典，就直接添加到结果中
                    result.append(nz_words[idx_nz])
            # 如果第i个nz词的一级查询结果为空,则添加空列表占位
            query_level_two.append(result)

        # 常规精确匹配结束，如果匹配成功，结构为二维列表，否则为空列表
        matched_keys = query_level_two  # 获取已匹配的字段
        # 备查词合并，我们约定所有备查词进行统一的查询，后面怎么用这些结果取决于外部的应用，对于agent模式，将会输出许多结果，对月非agent只会输出概率最高的结果
        tem_matched_keys = []
        for item in matched_keys:
            tem_matched_keys += item
        matched_keys = [list(set(tem_matched_keys))]
        del tem_matched_keys

        # 如果精确匹配失败，没有匹配到任何结果则按照语义进行模糊匹配，返回满足条件的置信度最高的结果
        # if not nz_words or ([] in matched_keys):
        # 比起手动维护词典，我们更相信语义相似度
        top_k = 5
        confi = 0.2 # 置信度阈值
        print(f'进入模糊匹配,召回Top:{top_k} 置信度阈值:{confi}...')
        # 调用远程bge-m3模型进行embedding
        query_embedding = np.array(self.model.encode([promt], normalize=True), dtype=np.float32) # 要求query_embedding是一个二维矩阵，形状为(1, 1024)
        distances, indices = self.knowledge.search(query_embedding, top_k)
        group_query = [(promt, self.plc_database_name_template_list[indices[0][i]]) for i in range(top_k)]
        # 我们更愿意相信bge，因此把词典关键词匹配的结果一并放进去重排序
        group_query_manuel = [(promt, k) for keys in matched_keys for k in keys]
        group_query += group_query_manuel
        del group_query_manuel
        group_query = list(set(group_query))  # 去重
        # 调用远程bge-reranker模型
        score = self.model.compute_score(group_query)
        rerank_result = sorted([(group_query[i][1], score[i]) for i in range(len(group_query))], key=lambda x: x[1], reverse=True)
        print(F'打印前top{top_k}候选词结果：', rerank_result[:top_k])
        print(f'首元素模糊匹配到{rerank_result[0][0]}, 置信度为{rerank_result[0][1]}')
        # matched_keys 为最终结果，保持形状为二维列表
        matched_keys = [[i[0] for i in rerank_result]]
        # 每个匹配结果的置信度
        matched_keys_score = [[i[1] for i in rerank_result]]

        # 为结果创建映射字典
        result_list = []
        for i_nz_keys in matched_keys:
            result_list.append([{key: self.name_2_code_dict.get(key)} for key in i_nz_keys])
        print(f"查询到{len([_ for _ in result_list if _])}个结果：")

        if not is_agent:
            # 非agent模式每个匹配结果只取第一个元素的英文
            tem_list = []
            for res in result_list:
                if res:
                    for k, v in res[0].items():  # 每个nz词的查询结果都是一个list，每个list可能包含多个字典
                        tem_list.append(f'{k}:{v}')
            result_list = tem_list
            print('以非agent模式返回：', result_list)
            return result_list

        print('以agent模式返回：', result_list)
        print('='*50)
        return result_list, matched_keys_score

if __name__ == '__main__':
    pj = 92  # pcl点位
    pcl_helper = PLCMatch(project_id=pj)
    # 用户输入
    my_promt = "我想要查询锡山中荷进水电导率"
    # query_res = pcl_helper.match_v2_on(my_promt, is_agent=True)
    query_res = pcl_helper.match_v2_on(my_promt, is_agent=False)

    pass