AI自然语言处理折腾手记

import re

def clean_text(text):
    # 移除多余空白
    text = re.sub(r'\s+', ' ', text).strip()

    # 处理表情符号 - 选择性保留或转换为描述
    emoji_pattern = re.compile("["
        u"\U0001F600-\U0001F64F"  # emoticons
        u"\U0001F300-\U0001F5FF"  # symbols & pictographs
        u"\U0001F680-\U0001F6FF"  # transport & map symbols
        u"\U0001F1E0-\U0001F1FF"  # flags (iOS)
        "]+", flags=re.UNICODE)
    text = emoji_pattern.sub('', text)

    # 移除 HTML 标签
    text = re.sub(r'<[^>]+>', '', text)

    return text

def normalize_text(text):
    # 转小写（英文场景）
    text = text.lower()

    # 全半角转换
    text = text.replace('，', ',').replace('。', '.')
    text = text.replace('（', '(').replace('）', ')')

    # 繁简转换需要使用第三方库
    # from opencc import OpenCC
    # cc = OpenCC('t2s')
    # text = cc.convert(text)

    return text

import jieba
import jieba.posseg as pseg

def tokenize_text(text):
    # 基础分词
    words = jieba.lcut(text)

    # 词性标注过滤（保留名词、动词、形容词）
    tagged_words = []
    for word, flag in pseg.cut(text):
        if flag.startswith(('n', 'v', 'a')):
            tagged_words.append(word)

    return tagged_words

from sklearn.feature_extraction.text import TfidfVectorizer

# 训练 TF-IDF 向量化器
tfidf_vectorizer = TfidfVectorizer(
    max_features=5000,        # 限制特征维度
    ngram_range=(1, 2),       # 使用 unigram 和 bigram
    min_df=2,                 # 忽略出现次数少于2的词
    max_df=0.8,               # 忽略出现在超过80%文档中的词
    stop_words=None           # 中文停用词需要自定义
)

# 在训练集上拟合
tfidf_matrix = tfidf_vectorizer.fit_transform(train_texts)

from transformers import AutoTokenizer, AutoModel

# 加载中文预训练模型
model_name = "hfl/chinese-roberta-wwm-ext"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModel.from_pretrained(model_name)

def get_embeddings(text):
    # 分词并转换为模型输入
    inputs = tokenizer(
        text,
        return_tensors="pt",
        padding=True,
        truncation=True,
        max_length=512
    )

    # 获取模型输出
    outputs = model(**inputs)

    # 使用 [CLS] token 的表示作为句子级嵌入
    cls_embedding = outputs.last_hidden_state[:, 0, :]

    return cls_embedding.detach().numpy()

from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC
from transformers import pipeline

# 方案1：传统机器学习 + TF-IDF
clf = LogisticRegression(
    multi_class='multinomial',
    solver='lbfgs',
    max_iter=1000
)
clf.fit(tfidf_matrix, labels)

# 方案2：预训练模型微调
classifier = pipeline(
    "text-classification",
    model="hfl/chinese-roberta-wwm-ext-classification",
    device=0  # 使用 GPU
)

from textblob import TextBlob
from snownlp import SnowNLP

# 轻量级方案：SnowNLP
def sentiment_analysis_snownlp(text):
    s = SnowNLP(text)
    score = s.sentiments  # 0-1之间，越接近1越积极

    if score > 0.6:
        return 'positive'
    elif score < 0.4:
        return 'negative'
    else:
        return 'neutral'

# 深度方案：预训练情感模型
sentiment_analyzer = pipeline(
    "sentiment-analysis",
    model="lxyuan/distilbert-base-multilingual-cased-sentiments-student",
    return_all_scores=True
)

from transformers import pipeline

# 命名实体识别
ner = pipeline(
    "ner",
    model="hfl/chinese-roberta-wwm-ext",
    aggregation_strategy="simple"
)

def extract_entities(text):
    entities = ner(text)
    return {
        'products': [e['word'] for e in entities if e['entity_group'] == 'PRODUCT'],
        'features': [e['word'] for e in entities if e['entity_group'] == 'FEATURE'],
        'issues': [e['word'] for e in entities if e['entity_group'] == 'ISSUE']
    }

def filter_by_confidence(results, threshold=0.7):
    filtered = []
    for result in results:
        if result['score'] >= threshold:
            filtered.append(result)
        else:
            # 低置信度结果标记为需要人工审核
            result['needs_review'] = True
            filtered.append(result)
    return filtered

import pandas as pd

def aggregate_results(results, time_window='1D'):
    df = pd.DataFrame(results)

    # 按时间窗口聚合
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    time_series = df.set_index('timestamp').resample(time_window)

    # 计算各类指标
    metrics = {
        'total_comments': time_series.size(),
        'positive_ratio': time_series.apply(
            lambda x: (x['sentiment'] == 'positive').mean()
        ),
        'top_issues': time_series.apply(
            lambda x: x['issue_type'].value_counts().head(3).to_dict()
        )
    }

    return metrics

def data_quality_check(df):
    # 检查重复评论
    duplicates = df.duplicated(subset=['comment_text'], keep=False)

    # 检查异常长度（可能是垃圾数据）
    length_outliers = (
        (df['comment_text'].str.len() < 5) |
        (df['comment_text'].str.len() > 1000)
    )

    # 检查特殊字符比例
    special_char_ratio = df['comment_text'].apply(
        lambda x: len(re.findall(r'[^\w\s一-鿿]', x)) / len(x)
    )
    text_outliers = special_char_ratio > 0.5

    issues = duplicates | length_outliers | text_outliers
    return issues

def monitor_model_drift(new_data, reference_data, threshold=0.1):
    # 计算新旧数据的分布差异
    from scipy.stats import wasserstein_distance

    drift_scores = {}
    for feature in new_data.columns:
        if new_data[feature].dtype in ['float64', 'int64']:
            score = wasserstein_distance(
                new_data[feature].dropna(),
                reference_data[feature].dropna()
            )
            drift_scores[feature] = score

    # 判断是否超过阈值
    needs_retraining = any(score > threshold for score in drift_scores.values())

    return needs_retraining, drift_scores

class NLPPipeline:
    def __init__(self):
        self.classifier = load_classifier()
        self.sentiment_analyzer = load_sentiment_analyzer()
        self.entity_extractor = load_entity_extractor()

    def process(self, text):
        results = {}

        # 并行执行独立任务
        results['category'] = self.classifier.predict(text)
        results['sentiment'] = self.sentiment_analyzer.predict(text)

        # 依赖前序结果的任务
        if results['category'] == 'feedback':
            results['entities'] = self.entity_extractor.extract(text)

        return results