问题采集的AI回答JSON数据中混有空回答、拒绝回答、无关内容及重复样本如何自动识别并清洗环境Python 3.9.7 pandas 1.3.3Ubuntu 20.04。读者AI数据工程师。本文提供可复用的校验函数和验证示例不涉及多轮对话场景。环境说明操作系统Ubuntu 20.04 LTSPython版本3.9.7依赖安装pipinstallpandas1.3.3输入格式每行一个JSON对象的文本文件字段包括query,response,timestamp,model。示例{query:什么是AI,response:人工智能是...,timestamp:2025-01-01,model:gpt-4}校验流程采集数据中常见的无效样本包括空回答、拒绝回答、无关内容以及重复记录。下面分别处理最后整合为一个函数。1. 无效样本识别无效样本分三类空回答response字段为空或仅含空白字符。拒绝回答回答以“抱歉”、“对不起”、“无法回答”等开头或包含“作为AI”、“我无法”等模式。无关内容回答与查询在语义上不相关。这里使用基于词重叠的简单规则实际项目中可替换为嵌入向量相似度。importredefis_empty_response(response:str)-bool:returnnotresponseorresponse.strip()defis_refusal_response(response:str)-bool:refusal_patterns[r^抱歉,r^对不起,r^无法回答,r作为AI,r我无法,r我不能,r这个问题我无法]forpatterninrefusal_patterns:ifre.search(pattern,response,re.IGNORECASE):returnTruereturnFalsedefis_irrelevant_response(query:str,response:str,threshold:float0.3)-bool:# 简单基于词重叠的相似度实际可用嵌入模型query_wordsset(re.findall(r\w,query.lower()))response_wordsset(re.findall(r\w,response.lower()))ifnotquery_wordsornotresponse_words:returnTrueoverlaplen(query_wordsresponse_words)/len(query_words)returnoverlapthreshold2. 重复样本去重重复样本指完全相同的query和response组合。使用MD5哈希快速去重importhashlibdefcompute_hash(query:str,response:str)-str:contentf{query}||{response}returnhashlib.md5(content.encode(utf-8)).hexdigest()defdeduplicate(records:list)-list:seenset()unique[]forrecinrecords:hcompute_hash(rec[query],rec[response])ifhnotinseen:seen.add(h)unique.append(rec)returnunique3. 异常值检测异常值包括回答长度异常过短或过长。使用IQR方法检测但注意该方法假设数据正态分布对于偏态分布可能误判。importpandasaspddefdetect_outliers_by_length(df:pd.DataFrame,column:strresponse_length)-pd.Series:Q1df[column].quantile(0.25)Q3df[column].quantile(0.75)IQRQ3-Q1 lower_boundQ1-1.5*IQR upper_boundQ31.5*IQRreturn(df[column]lower_bound)|(df[column]upper_bound)完整校验函数将上述步骤整合输出清洗后的数据及异常报告defquality_check(records:list)-(list,dict):report{total:len(records),empty:0,refusal:0,irrelevant:0,duplicate:0,length_outlier:0,valid:0}# Step 1: 无效样本识别valid_records[]forrecinrecords:ifis_empty_response(rec[response]):report[empty]1continueifis_refusal_response(rec[response]):report[refusal]1continueifis_irrelevant_response(rec[query],rec[response]):report[irrelevant]1continuevalid_records.append(rec)# Step 2: 重复去重unique_recordsdeduplicate(valid_records)report[duplicate]len(valid_records)-len(unique_records)# Step 3: 异常值检测dfpd.DataFrame(unique_records)df[response_length]df[response].apply(len)outliersdetect_outliers_by_length(df)report[length_outlier]outliers.sum()# 标记异常但保留df[is_length_outlier]outliers final_recordsdf.to_dict(records)report[valid]len(final_records)-report[length_outlier]returnfinal_records,report验证结果以下为模拟数据示例实际使用时请替换为真实采集数据。test_records[{query:什么是AI,response:人工智能是...,timestamp:2025-01-01,model:gpt-4},{query:天气,response:,timestamp:2025-01-01,model:gpt-4},{query:你好,response:抱歉我无法回答。,timestamp:2025-01-01,model:gpt-4},{query:什么是AI,response:人工智能是...,timestamp:2025-01-01,model:gpt-4},# 重复{query:股票,response:今天天气很好。,timestamp:2025-01-01,model:gpt-4},# 无关{query:写一首诗,response:春眠不觉晓...*1000,timestamp:2025-01-01,model:gpt-4},# 长度异常]cleaned,reportquality_check(test_records)print(report)# 预期输出{total: 6, empty: 1, refusal: 1, irrelevant: 1, duplicate: 1, length_outlier: 1, valid: 1}正常情况下最终cleaned列表包含1条正常记录和1条长度异常记录被保留但标记。常见问题与避坑无关内容检测的阈值基于词重叠的方法对短文本有效但对长文本或同义词不敏感。建议在实际项目中替换为嵌入向量余弦相似度阈值需根据业务数据调整。重复去重的粒度如果只对response去重可能误删不同查询的相同回答。建议组合query和response哈希。异常值边界IQR方法假设数据正态分布对于偏态分布可能误判。可改用分位数或业务规则如回答长度10字符视为异常。总结本文实现了一套轻量级AI回答数据校验流程覆盖空回答、拒绝回答、无关内容识别、重复去重和长度异常检测。该方案可直接集成到采集管道中输出清洗后的数据和质量报告。适用范围单轮问答场景对于多轮对话需调整重复检测逻辑。当前限制无关检测依赖简单词重叠高精度场景需引入语义模型。方案整体偏规则适合快速过滤但无法处理语义层面的复杂异常。