基于置信度提取字典非停用词值的高效规模化实现问询

需求背景 #

输入数据 #

inp = [{'father_husband_mother_name': [['Father s Name', 0.8603670001029968],
   ['Shripati', 0.8603670001029968],
   ['Father s Name', 0.8903670001029969],
   ['Shpppati', 0.8903670001029969]],
  'doc_id': [['GGX2176', 0.8435981869697571],
   ['GGC2176', 0.8835981869697571]],
  'name': [['Elector s Name', 0.8301510810852051],
   ['Shibshankar Ghosh', 0.8301510810852051],
   ['Elector s Name', 0.8501510810852051],
   ['Shibshankar Ghosh', 0.8501510810852051]],
  'date_of_birth': [['Age as on 1.1.2000', 0.8067844915390014],
   ['15', 0.8067844915390014],
   ['Age as on 1.1.2000', 0.8267844915390015],
   ['15', 0.8267844915390015]],
  'gender_sex': [['Sex', 0.7784658074378967],
   ['M', 0.7784658074378967],
   ['Sex', 0.8784658074378967],
   ['M', 0.8784658074378967]]}]

STOPWORDS = ['Sex', 'Father s Name', 'Elector s Name', 'Address', 'Name', 'Gender', 'Mother s Name', 
             'Husband s Name']

预期输出 #

{'father_husband_mother_name': 'Shpppati',
 'doc_id': 'GGC2176',
 'name': 'Shibshankar Ghosh',
 'date_of_birth': 'Age as on 1.1.2000,15',
 'gender_sex': 'M'}

处理规则 #

• 为每个键筛选出最高置信度（列表中的float值）的条目集合
• 从该集合中排除属于STOPWORDS的文本，将剩余文本用逗号拼接（若无剩余则留空）
• 仅保留置信度超过阈值（默认0.7）的结果

当前实现问题 #

现有实现采用多层嵌套循环，处理大规模数据时性能瓶颈明显，核心问题包括：

1. 多次遍历列表计算置信度、筛选最大值，时间复杂度高
2. 中间数据结构设计冗余，增加内存开销和遍历成本
3. 停用词匹配效率低（注：可忽略编辑距离实现，重点优化循环结构）

现有代码：

def process_kie_dict(voter_raw_labels, threshold=0.7):
    cleaned_dict = {}
    intermediate_dict = {}
    for entity_dict in voter_raw_labels:
        for entity, val in entity_dict.items():
            conf_val = [item[1] for item in val]
            unique_val = list(set(conf_val))
            max_conf = max(unique_val)
            if max_conf > threshold:
                if len(unique_val)==1:
                    add_val = [item[0] for item in val]
                else:
                    max_conf_index = conf_val.index(max_conf)
                    add_val = [item[0] for item in val[max_conf_index:]]

                if entity not in intermediate_dict.keys():
                    intermediate_dict[entity] = [add_val,max_conf]
                else:
                    if intermediate_dict[entity][1] < max_conf:
                        intermediate_dict[entity] = [add_val,max_conf]
    
#     print(intermediate_dict)
    for key, val in intermediate_dict.items():
        final_value = ''
        for value in val[0]:
            m = len(str.strip(value))
            edit_dist_list = []
            for word in STOPWORDS:
                n = len(word)
                edit_dist = editDistDP(value, word, m, n)
                edit_dist_list.append(edit_dist)

            if min(edit_dist_list) < 2:
                value=''
            
            final_value = final_value + value + ','
            clean_value = final_value.strip(",")
        
        cleaned_dict[key]=clean_value
    return cleaned_dict

def editDistDP(str1, str2, m, n):
    # Create a table to store results of subproblems
    dp = [[0 for x in range(n + 1)] for x in range(m + 1)]
 
    # Fill d[][] in bottom up manner
    for i in range(m + 1):
        for j in range(n + 1):
 
            # If first string is empty, only option is to
            # insert all characters of second string
            if i == 0:
                dp[i][j] = j    # Min. operations = j
 
            # If second string is empty, only option is to
            # remove all characters of second string
            elif j == 0:
                dp[i][j] = i    # Min. operations = i
 
            # If last characters are same, ignore last char
            # and recur for remaining string
            elif str1[i-1] == str2[j-1]:
                dp[i][j] = dp[i-1][j-1]
 
            # If last character are different, consider all
            # possibilities and find minimum
            else:
                dp[i][j] = 1 + min(dp[i][j-1],        # Insert
                                   dp[i-1][j],        # Remove
                                   dp[i-1][j-1])    # Replace
 
    return dp[m][n]

优化方案 #

核心思路 #

• 用集合加速停用词查找（O(1)时间复杂度）
• 单次遍历完成有效条目筛选、最高置信度计算和文本收集，避免重复操作
• 简化中间数据结构，减少内存占用和遍历次数

优化代码 #

def process_kie_dict_optimized(voter_raw_labels, threshold=0.7):
    # 转为集合加速停用词查找
    stop_set = set(STOPWORDS)
    cleaned_dict = {}
    
    for entity_dict in voter_raw_labels:
        for entity, items in entity_dict.items():
            # 筛选置信度达标条目
            valid_items = [item for item in items if item[1] > threshold]
            if not valid_items:
                continue
            
            # 找到最高置信度并提取对应文本
            max_conf = max(item[1] for item in valid_items)
            max_conf_texts = [text for text, conf in valid_items if conf == max_conf]
            
            # 过滤停用词后拼接
            filtered_texts = [text for text in max_conf_texts if text not in stop_set]
            cleaned_value = ','.join(filtered_texts)
            
            # 处理多实体字典，保留最高置信度结果
            if entity not in cleaned_dict:
                cleaned_dict[entity] = (cleaned_value, max_conf)
            else:
                _, current_conf = cleaned_dict[entity]
                if max_conf > current_conf:
                    cleaned_dict[entity] = (cleaned_value, max_conf)
    
    # 提取最终结果值
    return {k: v[0] for k, v in cleaned_dict.items()}

优化点说明 #

1. 停用词查找优化：集合查找替代线性遍历，时间复杂度从O(n)降至O(1)
2. 单次遍历处理：一次遍历完成有效条目筛选、最高置信度计算和文本收集，避免多次列表操作
3. 简化中间结构：直接生成清理后的值，减少中间字典的内存开销和二次遍历
4. 多实体处理逻辑：简洁处理同一实体的多批次数据，保留最高置信度结果

测试验证 #

用给定输入测试优化后的函数：

result = process_kie_dict_optimized(inp)
print(result)

输出与预期完全一致：

{'father_husband_mother_name': 'Shpppati',
 'doc_id': 'GGC2176',
 'name': 'Shibshankar Ghosh',
 'date_of_birth': 'Age as on 1.1.2000,15',
 'gender_sex': 'M'}

内容的提问来源于stack exchange，提问作者Vivek Kalyanarangan