TA算法高维度场景下cnt_access统计异常问题排查

问题代码 #

from typing import Tuple, defaultdict

def TA(cls, num_dim, top_k) -> Tuple[list, int]:
    uids_result = []
    uids_score={}
    cnt_access = 0
    topknum=0
    finish=0
    uid2dim2value = defaultdict(dict)                    
    for row in range(10000):
        t=0
        for dim in range(num_dim):  
         uid,value= cls.list_sorted_entities[dim][row]
         uid2dim2value[uid][dim] = value
         t+=value
         
         if uid not in uids_score:
             alldim=list(range(num_dim))
             uids_score[uid] =0
             for count in range(len(alldim)):
                    cnt_access+=1
                    uids_score[uid]+= cls.random_access(uid,alldim[count])
         else :
             cnt_access+=1
        uid_record=sorted(uids_score.keys())
        for count2 in range(len(uids_score)):
            RUID=uid_record[count2]
            if uids_score[RUID]>= t:
                if RUID not in uids_result:     
                 uids_result.append(RUID)
                 topknum+=1
                 if topknum==top_k:
                    finish=1
                    break
        if finish==1:
            break            
    return uids_result, cnt_access

场景与问题 #

场景背景 #

现有10000名学生的10门成绩排序列表，需通过TA算法获取top_k结果，算法流程：

1. 对各维度数据列表排序；
2. 按行扫描数据并累加当前行各维度值到变量t；若遇到新学生ID（uid），则随机访问该uid的所有维度数据；
3. 对比t与已扫描uid的总分，若总分≥t则将该uid存入结果列表，当结果数量达到top_k时终止流程。

问题现象 #

• 低维度场景下，cnt_access（随机访问计数）统计值与预期一致；
• 高维度场景下，统计值与预期存在1~7的误差（例：预期25690，实际25659）；
• 怀疑终止逻辑存在问题，但低维度无异常，需排查代码错误。

错误分析与修正 #

核心错误点 #

1. 重复计数逻辑错误：在维度遍历循环中，若uid已存在于uids_score，代码会执行cnt_access +=1，但此时并没有发起实际的随机访问——当前维度的value是从排序列表中直接读取的，不属于算法定义的“随机访问”，且同一行中同一个uid会被多次计数，高维度下重复次数多，误差被放大；
2. 低维度场景误差不明显是因为同一行内重复出现的uid数量少，误统计次数可忽略。

修正后代码 #

from typing import Tuple, defaultdict

def TA(cls, num_dim, top_k) -> Tuple[list, int]:
    uids_result = []
    uids_score = {}
    cnt_access = 0
    topknum = 0
    finish = 0
    uid2dim2value = defaultdict(dict)                    
    
    for row in range(10000):
        t = 0
        processed_uids = set()  # 记录当前行已处理过的uid，避免重复计数
        for dim in range(num_dim):  
            uid, value = cls.list_sorted_entities[dim][row]
            uid2dim2value[uid][dim] = value
            t += value
            
            if uid not in uids_score:
                # 首次遇到该uid，发起全维度随机访问并计数
                uids_score[uid] = 0
                for d in range(num_dim):
                    cnt_access += 1
                    uids_score[uid] += cls.random_access(uid, d)
                processed_uids.add(uid)
            elif uid not in processed_uids:
                # 同一行重复出现的uid，无额外随机访问，不计数
                processed_uids.add(uid)
        
        # 筛选符合条件的uid加入结果
        for uid in sorted(uids_score.keys()):
            if uids_score[uid] >= t:
                if uid not in uids_result:     
                    uids_result.append(uid)
                    topknum += 1
                    if topknum == top_k:
                        finish = 1
                        break
        if finish == 1:
            break            
    
    return uids_result, cnt_access

修正说明 #

• 新增processed_uids集合，避免同一行内重复uid的误计数；
• 仅在uid首次出现时，才发起全维度随机访问并计数，后续同一行内的重复uid不再统计；
• 移除了原代码中无实际随机访问的cnt_access +=1逻辑，确保计数仅对应真实的随机访问操作。

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