AI数据处理实战指南：从清洗到增强

import re

def standardize_text(text):
    # 统一换行符
    text = text.replace('\r\n', '\n').replace('\r', '\n')
    # 去除多余空格
    text = re.sub(r'\s+', ' ', text).strip()
    # 统一引号风格
    text = text.replace('"', '"').replace('"', '"')
    text = text.replace(''', "'").replace(''', "'")
    return text

import json

data = []
with open('user_dialogues.json', 'r', encoding='utf-8') as f:
    for line in f:
        line = line.strip()
        if not line or line.startswith('#'):
            continue
        try:
            data.append(json.loads(line))
        except json.JSONDecodeError:
            continue  # 坏行隔离

def filter_noise_samples(df, text_col='text', min_length=10):
    """过滤明显噪声样本"""
    df = df.copy()

    # 过滤过短文本
    df = df[df[text_col].str.len() >= min_length]

    # 过滤重复文本（保留第一个）
    df = df.drop_duplicates(subset=[text_col], keep='first')

    # 过滤疑似乱码文本
    def is_garbage(text):
        special_chars = re.findall(r'[^\w\s一-鿿]', text)
        ratio = len(special_chars) / len(text) if len(text) > 0 else 0
        return ratio > 0.5

    df = df[~df[text_col].apply(is_garbage)]

    return df.reset_index(drop=True)

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

        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

import hashlib
import re

def mask_pii(text):
    """脱敏隐私信息"""
    text = str(text)

    # 手机号脱敏：前3后4
    text = re.sub(
        r'1[3-9]\d{9}',
        lambda m: m.group(0)[:3] + '****' + m.group(0)[-4:],
        text
    )

    # 身份证号脱敏：前6后4
    text = re.sub(
        r'\d{17}[\dXx]',
        lambda m: m.group(0)[:6] + '********' + m.group(0)[-4:],
        text
    )

    # 邮箱脱敏
    text = re.sub(
        r'([a-zA-Z0-9._%+-]+)@([a-zA-Z0-9.-]+\.[a-zA-Z]{2,})',
        lambda m: m.group(1)[:2] + '***@' + m.group(2),
        text
    )

    # 订单号用哈希替换（保留可追溯性）
    text = re.sub(
        r'\b\d{18}\b',
        lambda m: 'ORDER_' + hashlib.md5(m.group(0).encode()).hexdigest()[:8],
        text
    )

    return text

# 朴素版本：慢
from difflib import SequenceMatcher

def check_label_consistency_naive(df):
    inconsistent_pairs = []
    for i in range(len(df)):
        for j in range(i+1, len(df)):
            sim = SequenceMatcher(None, df.iloc[i]['text'], df.iloc[j]['text']).ratio()
            if sim >= 0.85 and df.iloc[i]['label'] != df.iloc[j]['label']:
                inconsistent_pairs.append((i, j, sim))
    return inconsistent_pairs

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def fast_label_consistency_check(df, text_col='text', label_col='label',
                                  similarity_threshold=0.85, top_k=20):
    """快速标签一致性检查"""
    # 计算 TF-IDF 矩阵
    vectorizer = TfidfVectorizer(max_features=5000)
    tfidf_matrix = vectorizer.fit_transform(df[text_col])

    # 计算相似度矩阵
    similarity_matrix = cosine_similarity(tfidf_matrix)

    # 只检查每行最相似的 top_k 个
    inconsistent_pairs = []

    for i in range(len(df)):
        similar_indices = np.argsort(similarity_matrix[i])[-(top_k+1):-1][::-1]

        for j in similar_indices:
            similarity = similarity_matrix[i][j]
            if similarity >= similarity_threshold and df.iloc[i][label_col] != df.iloc[j][label_col]:
                inconsistent_pairs.append({
                    'index_1': i,
                    'index_2': j,
                    'similarity': similarity,
                    'label_1': df.iloc[i][label_col],
                    'label_2': df.iloc[j][label_col]
                })

    return inconsistent_pairs

def analyze_label_distribution(df, label_col='label'):
    label_counts = df[label_col].value_counts()

    print(f"总样本数: {len(df)}")
    print(f"标签类别数: {len(label_counts)}")
    print(label_counts)

    imbalance_ratio = label_counts.max() / label_counts.min()
    if imbalance_ratio > 10:
        print(f"警告: 存在严重类别不平衡 (最大/最小 = {imbalance_ratio:.1f})")

    return label_counts

class DataQualityMonitor:
    def __init__(self, baseline_stats=None):
        self.baseline_stats = baseline_stats or {}

    def check_drift(self, current_stats, threshold=0.1):
        """检查数据漂移"""
        drift_report = []

        for key, current_value in current_stats.items():
            if key in self.baseline_stats:
                baseline_value = self.baseline_stats[key]

                if isinstance(current_value, (int, float)):
                    if baseline_value != 0:
                        relative_change = abs(current_value - baseline_value) / baseline_value
                        if relative_change > threshold:
                            drift_report.append({
                                'metric': key,
                                'baseline': baseline_value,
                                'current': current_value,
                                'change': f"{relative_change*100:.1f}%"
                            })

        return drift_report

from sklearn.metrics import cohen_kappa_score

def calculate_agreement(annotations_1, annotations_2):
    """计算两个标注人员的一致性"""
    return cohen_kappa_score(annotations_1, annotations_2)

docker run -d --name doccano -p 8000:8000 ghcr.io/doccano/doccano:latest

def rule_based_intent_detection(text):
    """基于规则的意图识别"""
    text_lower = text.lower()

    if any(kw in text_lower for kw in ['多少钱', '价格', '费用']):
        return 'product_inquiry'
    if any(kw in text_lower for kw in ['出错', '错误', '不行', '不能用']):
        return 'technical_support'
    if any(kw in text_lower for kw in ['登录', '注册', '密码']):
        return 'account'

    return None  # 无法确定

from transformers import pipeline

classifier = pipeline("text-classification", model="bert-base-uncased")

def predict_with_confidence(text, threshold=0.8):
    result = classifier(text)
    confidence = result[0]['score']

    if confidence >= threshold:
        return result[0]['label'], confidence
    else:
        return None, confidence

import torchvision.transforms as transforms

transform = transforms.Compose([
    transforms.RandomHorizontalFlip(p=0.5),
    transforms.RandomRotation(degrees=15, fill=(0, 0, 0)),  # 明确填充
    transforms.ColorJitter(brightness=0.1, contrast=0.1, saturation=0.1, hue=0.05),
])

def mixup_data(x, y, alpha=1.0):
    if alpha > 0:
        lam = np.random.beta(alpha, alpha)
    else:
        lam = 1

    batch_size = x.size()[0]
    index = torch.randperm(batch_size)

    mixed_x = lam * x + (1 - lam) * x[index]
    y_a, y_b = y, y[index]
    return mixed_x, y_a, y_b, lam

def mixup_criterion(criterion, pred, y_a, y_b, lam):
    return lam * criterion(pred, y_a) + (1 - lam) * criterion(pred, y_b)

from nltk.corpus import wordnet

def synonym_replacement(text, n=2):
    words = text.split()
    new_words = words.copy()

    random_word_list = [w for w in words if wordnet.synsets(w)]
    random.shuffle(random_word_list)

    num_replaced = 0
    for random_word in random_word_list:
        synonyms = wordnet.synsets(random_word)
        if synonyms:
            synonym = synonyms[0].lemmas()[0].name()
            new_words = [synonym if w == random_word else w for w in new_words]
            num_replaced += 1
        if num_replaced >= n:
            break

    return ' '.join(new_words)

from transformers import pipeline

generator = pipeline("text2text-generation", model="google/flan-t5-small")

def contextual_replacement(text):
    prompt = f"Rewrite this sentence with different words but keep the meaning: {text}"
    result = generator(prompt, num_return_sequences=3)
    return [item['generated_text'] for item in result]

def back_translate(text, src='zh-cn', temp_lang='en'):
    temp_translation = translator.translate(text, src=src, dest=temp_lang).text
    back_translation = translator.translate(temp_translation, src=temp_lang, dest=src).text
    return back_translation

PROTECTED_TERMS = {
    '心肌梗死', '心力衰竭', '高血压', '糖尿病',
    '冠状动脉', '心律失常', '脑卒中', '肺炎'
}

def safe_augmentation(text):
    tokens = tokenize(text)
    protected_positions = [i for i, t in enumerate(tokens) if t in PROTECTED_TERMS]

    augmented_tokens = tokens.copy()
    for i in range(len(augmented_tokens)):
        if i not in protected_positions:
            augmented_tokens[i] = augment_word(augmented_tokens[i])

    return detokenize(augmented_tokens)

def generate_synthetic_samples(original_samples, n_per_sample=3):
    prompt = """
    你是一个医疗数据增强助手。请根据以下文本，生成语义相似但表述不同的新文本。
    要求：
    1. 保持语义准确
    2. 改变句式结构
    3. 替换同义词
    4. 保持专业性和准确性

    原始文本：{original_text}

    请生成 {n} 个不同的版本：
    """

    synthetic_samples = []
    for sample in original_samples:
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{
                "role": "user",
                "content": prompt.format(original_text=sample, n=n_per_sample)
            }],
            temperature=0.7
        )
        synthetic_samples.extend(response.choices[0].message.content.split('\n'))

    return synthetic_samples

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list:
    """把文本切分成固定大小的块，保留重叠部分"""
    paragraphs = re.split(r'\n\s*\n', text)
    chunks = []
    current_chunk = ""

    for para in paragraphs:
        para = para.strip()
        if not para:
            continue

        if len(current_chunk) + len(para) > chunk_size:
            if current_chunk:
                chunks.append(current_chunk)
            current_chunk = current_chunk[-overlap:] + " " + para
        else:
            current_chunk = current_chunk + " " + para if current_chunk else para

    if current_chunk:
        chunks.append(current_chunk)

    return chunks

from openai import OpenAI
import numpy as np
import time

client = OpenAI()

def batch_embeddings(texts, batch_size=10):
    """批量生成向量，处理速率限制"""
    embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        try:
            response = client.embeddings.create(
                input=batch,
                model="text-embedding-3-small"  # 1536 维，性价比高
            )
            embeddings.extend([np.array(d.embedding) for d in response.data])
        except Exception as e:
            print(f"Batch {i} failed: {e}")
            time.sleep(5)
        time.sleep(0.1)  # 避免触发速率限制

    return embeddings

import faiss

class VectorSearch:
    def __init__(self, dimension=1536):
        self.dimension = dimension
        self.index = faiss.IndexFlatL2(dimension)  # L2 距离
        self.documents = []

    def add(self, vectors, docs):
        vectors_array = np.array(vectors, dtype=np.float32)  # faiss 要 float32
        self.index.add(vectors_array)
        self.documents.extend(docs)

    def search(self, query_vector, top_k=5):
        query_vector = np.array([query_vector], dtype=np.float32)
        distances, indices = self.index.search(query_vector, top_k)

        results = []
        for dist, idx in zip(distances[0], indices[0]):
            if idx < len(self.documents):
                similarity = 1 / (1 + dist)  # L2 距离转相似度
                results.append((similarity, self.documents[idx]))

        return sorted(results, reverse=True, key=lambda x: x[0])

# text-embedding-3-small 输出 1536 维
dimension = 1536
index = faiss.IndexFlatL2(dimension)

# 维度不匹配会报错
index.add(vector_768d)  # Error

def efficient_validation(df, sample_size=10000):
    # 第一层：快速检查（全量数据）
    quick_checks = {
        'missing_values': check_missing_values(df),
        'data_size': check_data_size(df)
    }

    # 第二层：详细检查（抽样数据）
    sample_df = df.sample(n=min(sample_size, len(df)))

    detailed_checks = {
        'data_types': check_data_types(sample_df, expected_types),
        'value_ranges': check_value_ranges(sample_df, ranges)
    }

    return {**quick_checks, **detailed_checks}

def order_amount_positive_rule(df):
    """订单金额必须为正数（退货订单除外）"""
    violations = df[(df['order_amount'] <= 0) & (df['order_type'] != 'return')]
    return violations.index.tolist()

def check_anomalies_with_whitelist(df, column, whitelist=None):
    """带白名单的异常值检查"""
    if whitelist is None:
        whitelist = []

    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR

    anomalies = df[(df[column] < lower_bound) | (df[column] > upper_bound)]

    # 过滤掉白名单中的值
    filtered_anomalies = anomalies[~df[column].isin(whitelist)]

    return {
        'total_anomalies': len(anomalies),
        'whitelisted_anomalies': len(anomalies) - len(filtered_anomalies),
        'real_anomalies': len(filtered_anomalies)
    }

# 初始化
dvc init

# 追踪数据文件
dvc add data/user_dialogues.json
git add data/.gitignore data/user_dialogues.json.dvc

# 提交
git commit -m "add dataset tracking"

# 检查增强分布
def check_augmentation_bias(dataset, transform, num_samples=1000):
    transformed_labels = []
    for i in range(min(num_samples, len(dataset))):
        image, label = dataset[i]
        transformed = transform(image)
        transformed_labels.append(label)

    label_counts = Counter(transformed_labels)
    original_counts = Counter([dataset[i][1] for i in range(min(num_samples, len(dataset)))])

    for label in set(list(label_counts.keys()) + list(original_counts.keys())):
        print(f"Label {label}: original={original_counts[label]}, augmented={label_counts[label]}")

def sanitize_urls(text):
    return re.sub(r'https?://[^\s]+', '[URL]', text)

def smart_augmentation(original_data, augment_ratio=0.3):
    """只对部分数据增强"""
    augment_count = int(len(original_data) * augment_ratio)
    to_augment = random.sample(original_data, augment_count)
    # 每条生成 2-3 条
    ...