从清洗走到标注：AI数据处理笔记
最近一个月折腾了一堆 NLP 数据，从 JSON 格式的对话记录到用户评论文本，再到一些半结构化的标注数据。
拿到的第一批数据是这样的：一个 2.3GB 的 JSON 文件，解压后发现是 15 万条用户对话记录，包含文本、时间戳、用户标签和一些元数据。
起手：先看看手里有什么
拿到的第一批数据是这样的：一个 2.3GB 的 JSON 文件，解压后发现是 15 万条用户对话记录，包含文本、时间戳、用户标签和一些元数据。第一反应是用 Pandas 直接读。
import pandas as pd
import json
# 尝试直接读
df = pd.read_json('user_dialogues.json')
报错来了：JSONDecodeError: Expecting value: line 1 column 1 (char 0)
打开文件一看，不是标准 JSON 格式，是每行一条 JSON 的 NDJSON 格式。改用 jsonlines 或者一行行读：
import json
data = []
with open('user_dialogues.json', 'r', encoding='utf-8') as f:
for line in f:
data.append(json.loads(line))
print(f"读取了 {len(data)} 条记录")
跑了一分钟，结果打印出来只有 120,000 多条，比预期少了三万多。回头看原始文件，发现有些行是空的，还有些行是注释符号开头的（# 开头），有些行是重复的。
这时候意识到一个问题：数据到手后不要急着直接处理，先用最笨的方式扫一遍，看看真实情况。
清洗：那些你以为不会出问题的地方
清洗这部分花的时间比预期多很多，主要踩了几个坑。
字符编码问题
第一批数据有些字段是乱码，看起来像编码问题。尝试了几种方案：
# 方案1：指定编码
with open('user_dialogues.json', 'r', encoding='utf-8') as f:
content = f.read()
# 方案2：尝试自动检测编码
import chardet
with open('user_dialogues.json', 'rb') as f:
result = chardet.detect(f.read())
print(f"检测到编码: {result['encoding']}")
最后发现不是编码问题，是有些字段本身就有特殊字符或者不可见字符。处理方式是直接过滤掉那些明显异常的行：
def is_valid_text(text):
if not isinstance(text, str):
return False
# 过滤掉空文本和只包含空白字符的
if not text.strip():
return False
# 过滤掉包含过多控制字符的
control_chars = sum(1 for c in text if ord(c) < 32 and c != '\n')
if control_chars > len(text) * 0.1: # 控制字符超过10%就认为异常
return False
return True
clean_data = [
item for item in data
if is_valid_text(item.get('text', ''))
]
去重逻辑陷阱
去重看起来简单，但实际场景里有很多细节。比如用户可能会发送完全相同的消息，在时间上相近但不算重复；而有些恶意刷屏的重复内容又确实需要过滤。
def deduplicate_by_hash(data, time_window=60, max_len=256):
"""
基于内容和时间窗口的去重
time_window: 秒数，在这个时间内的相同内容算重复
max_len: 超过这个长度的文本不进行去重（误伤风险高）
"""
seen = {} # {hash: last_time}
result = []
for item in data:
text = item.get('text', '')
timestamp = item.get('timestamp', 0)
# 只对短文本去重
if len(text) > max_len:
result.append(item)
continue
# 简单hash（生产环境可以用更好的hash函数）
text_hash = hash(text[:100]) # 只hash前100个字符
if text_hash in seen:
if timestamp - seen[text_hash] < time_window:
continue # 重复
seen[text_hash] = timestamp
result.append(item)
return result
异常字段处理
有些记录字段缺失，有些字段类型不对，有些是明显错误的数据（比如时间戳在 1970 年之前或者 2050 年之后）：
import datetime
def validate_timestamp(timestamp):
"""验证时间戳是否合理"""
if not isinstance(timestamp, (int, float)):
return False
try:
dt = datetime.datetime.fromtimestamp(timestamp)
# 设定一个合理的时间范围
min_date = datetime.datetime(2020, 1, 1)
max_date = datetime.datetime.now() + datetime.timedelta(days=30)
return min_date <= dt <= max_date
except:
return False
valid_data = [
item for item in clean_data
if validate_timestamp(item.get('timestamp'))
and isinstance(item.get('user_id'), str) # 确保user_id是字符串
]
经过这些清洗步骤，15 万条原始数据最后剩下了 8.2 万条可用记录。看起来损失很大，但总比用脏数据训练模型要好。
标注工具选择：从 fancy 到实际
清洗完数据接下来是标注，这部分最头疼的是工具选择。
Label Studio
试了 Label Studio，界面很漂亮，功能也全，支持多种标注任务类型。安装配置也很简单：
docker run -it -p 8080:8080 -v $(pwd)/data:/root/data label-studio/label-studio:latest
但遇到的问题是：它的学习曲线比预期的陡，配置自定义标注模板需要改 JSON 配置文件，而且对于我们这种简单文本分类任务来说有点过度设计了。最致命的是它占用内存比较高，在 8GB 内存的机器上跑起来就比较吃力。
Prodigy
Prodigy 是 spaCy 团队做的，主打效率和自定义。优点是它有一个很赞的主动学习机制，能智能选择对模型最有价值的数据进行标注。
但问题来了：它是商业软件，免费版功能有限，而且价格不算便宜。对于我们这种小团队来说性价比不高。
Doccano
最后选了 Doccano，一个开源的轻量级文本标注工具。
docker run -d --name doccano -p 8000:8000 ghcr.io/doccano/doccano:latest
启动后访问 http://localhost:8000 即可使用。界面相对简单，但对于文本分类、序列标注这类任务够用了。
创建项目、导入数据、分配标注人员、导出标注结果，整个流程还算顺滑。唯一的问题是它的批量操作功能不够强，比如批量修改标注类型、批量导出特定条件的数据，需要自己写 SQL 或者脚本来处理。
标注质量控制：比你想象的复杂
标注质量这部分，比预期复杂很多。不只是标注人员的问题，数据本身的歧义、标注指南的清晰度、任务设计都会影响质量。
标注指南的重要性
一开始我们就直接把数据和任务丢给标注人员，结果回来一看，质量参差不齐。同样的一句话，有人标成 A 类，有人标成 B 类。
后来意识到必须先写一个清晰的标注指南，明确每种分类的定义、边界情况、举例说明。
比如我们的分类任务是判断用户意图，标注指南里就写了：
- 意图 A（产品咨询）：用户在询问产品功能、价格、使用方法等
- 明确包含：“这个产品多少钱”、“怎么用”、“支持哪些功能”
- 不包含：“你们公司怎么样”、“产品经理是谁”
- 意图 B（技术支持）：用户遇到了具体问题需要帮助解决
- 明确包含：“出错了”、“不行”、“不能正常使用”
- 不包含：“我想了解技术细节”、“技术架构是什么”
写完指南后，标注一致度明显提升了很多。
双盲标注和 adjudication
为了进一步保证质量，我们采用了双盲标注+仲裁的方案：
import numpy as np
from sklearn.metrics import cohen_kappa_score
def calculate_agreement(annotations_1, annotations_2):
"""计算两个标注人员的一致性"""
return cohen_kappa_score(annotations_1, annotations_2)
# 示例数据
ann1 = [0, 1, 2, 0, 1, 2] # 标注人员A的结果
ann2 = [0, 1, 1, 0, 1, 2] # 标注人员B的结果
kappa = calculate_agreement(ann1, ann2)
print(f"Kappa系数: {kappa}")
Kappa 系数大于 0.8 说明一致性很好，0.6-0.8 可以接受，低于 0.6 就需要重新培训标注人员或者修改标注指南。
标注抽样审查
除了双盲标注，还需要定期抽检已完成标注的数据。我们按照 5% 的比例随机抽取，由资深人员审查。
import random
def sample_annotations(all_annotations, sample_rate=0.05):
"""随机抽取一定比例的标注数据进行审查"""
sample_count = int(len(all_annotations) * sample_rate)
return random.sample(all_annotations, sample_count)
sample_data = sample_annotations(valid_data)
审查过程中发现的错误要反馈给标注人员，并统计错误类型，看是哪些类型的样本容易出错，针对性地改进指南或者调整任务设计。
自动化标注：能省点时间就省点
完全靠人工标注成本太高，尤其是数据量大的时候。我们尝试了一些自动化方案。
基于规则的预标注
对于一些规则比较明确的任务，可以用规则引擎先进行预标注，然后人工只审查有疑问的样本。
import re
def rule_based_intent_detection(text):
"""基于规则的意图识别"""
text_lower = text.lower()
# 价格相关
if any(keyword in text_lower for keyword in ['多少钱', '价格', '费用', '成本']):
return 'product_inquiry'
# 故障相关
if any(keyword in text_lower for keyword in ['出错', '错误', '不行', '不能用']):
return 'technical_support'
# 账号相关
if any(keyword in text_lower for keyword in ['登录', '注册', '密码', '账号']):
return 'account'
return None # 无法确定
# 预标注
auto_labeled = []
unlabeled = []
for item in valid_data:
text = item.get('text', '')
intent = rule_based_intent_detection(text)
if intent:
item['auto_label'] = intent
auto_labeled.append(item)
else:
unlabeled.append(item)
print(f"自动标注: {len(auto_labeled)} 条")
print(f"需人工标注: {len(unlabeled)} 条")
这样能把一些明显的样本先筛掉，人工只需要标注那些规则覆盖不到的样本。
预训练模型辅助
对于更复杂的任务，可以用预训练模型（如 BERT、GPT）先进行预测，人工只修改那些置信度低的样本。
from transformers import pipeline
classifier = pipeline("text-classification", model="bert-base-uncased")
def predict_with_confidence(text, threshold=0.8):
"""预测并返回置信度"""
result = classifier(text)
label = result[0]['label']
confidence = result[0]['score']
if confidence >= threshold:
return label, confidence
else:
return None, confidence
high_confidence = []
low_confidence = []
for item in unlabeled:
text = item.get('text', '')
label, confidence = predict_with_confidence(text)
if label:
item['auto_label'] = label
item['confidence'] = confidence
high_confidence.append(item)
else:
low_confidence.append(item)
需要注意的是，自动标注的质量取决于规则或者模型的质量，而且会有放大错误的风险。如果预标注质量本身就很差，反而会增加人工审核的工作量。
数据版本管理：最容易被忽略的一环
数据处理过程中最容易被忽略的就是版本管理。我们一开始就直接在原文件上修改，后来发现标注错了想回退，却找不到原始数据了。
后来学了乖，每次处理都保存一个新文件：
# 原始数据
raw_data/user_dialogues.json
# 清洗后
processed_data/user_dialogues_cleaned_v1.json
processed_data/user_dialogues_cleaned_v2.json
# 标注后
annotated_data/user_dialogues_annotated_v1.json
更好的方式是用 Git LFS 来管理大型数据文件，或者使用专门的版本控制工具如 DVC。
# 使用 DVC 初始化
dvc init
# 追踪数据文件
dvc add data/user_dialogues.json
git add data/.gitignore
git add data/user_dialogues.json.dvc
# 提交到 Git
git commit -m "add dataset tracking"
一些被浪费的下午
回想起来，有几个下午的时间是完全可以避免浪费的：
- 第一下午：直接在内存里处理 2.3GB 的 JSON 文件，导致 OOM。后来才知道应该分批处理或者使用数据库。
- 第二下午：盲目相信数据提供方的"数据质量保证"，结果清洗时发现 40% 的数据都有问题。
- 第三下午：用 Excel 打开标注结果文件进行编辑，结果把文件格式搞乱了。
- 第四下午：没有先做小规模试运行，直接把所有数据都丢给标注平台，结果任务设计有缺陷，只能重新来。
经验就是：先在小规模数据上把整个流程跑通，再放大到全量数据。
最后
数据处理不是最光鲜的工作，但绝对是最关键的环节。模型好坏先不说，至少要保证数据本身是干净的、标注是准确的、流程是可复现的。
如果你正在做类似的任务，建议先想清楚这几件事：
- 数据到底从哪里来，源头质量如何
- 预期的数据规模和结构，有没有现成的工具可以直接用
- 标注任务的复杂度，需要投入多少人力
- 有哪些地方可以自动化，哪些地方必须人工参与
- 整个流程如何保证可追溯和可复现
很多时候花在数据处理上的时间比模型训练要多，但这很正常。毕竟垃圾进，垃圾出，这点永远不会变。
版权声明: 本文首发于 指尖魔法屋-从清洗走到标注：AI数据处理笔记（https://blog.thinkmoon.cn/post/150-ai-data-processing-cleaning-labeling-practice/） 转载或引用必须申明原指尖魔法屋来源及源地址！
评论
使用 GitHub 账号登录后即可留言，支持 Markdown。