大模型应用：语料库治理实战：基于 text2vec+BERT 的由浅入深解析.41

from modelscope.hub.snapshot_download import snapshot_download

# 配置本地缓存目录（建议选择空间充足的磁盘）
cache_dir = "D:\\modelscope\\hub"
# 模型名称（ModelScope上的官方名称）
model_name = "Jerry0/text2vec-base-chinese"

# 下载模型到本地
local_text2vec_path = snapshot_download(
    model_name,
    cache_dir=cache_dir,
    revision="master"  # 下载主分支版本
)

print(f"text2vec模型本地路径：{local_text2vec_path}")

from transformers import AutoTokenizer, AutoModel

# 加载本地text2vec模型
tokenizer = AutoTokenizer.from_pretrained("D:\\modelscope\\hub\\Jerry0\\text2vec-base-chinese")
model = AutoModel.from_pretrained("D:\\modelscope\\hub\\Jerry0\\text2vec-base-chinese")

# 测试文本向量化
text = "亲，这个商品啥时候发货？"
inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
outputs = model(**inputs)

# 输出向量维度（验证加载成功）
print(f"文本向量维度：{outputs.last_hidden_state.shape}")

from modelscope.hub.snapshot_download import snapshot_download

cache_dir = "D:\\modelscope\\hub"
model_name = "google-bert/bert-base-chinese"

# 下载模型到本地
local_bert_path = snapshot_download(
    model_name,
    cache_dir=cache_dir,
    revision="master"
)

print(f"BERT模型本地路径：{local_bert_path}")

from transformers import BertTokenizer, BertModel

# 加载本地BERT模型
tokenizer = BertTokenizer.from_pretrained("D:\\modelscope\\hub\\google-bert\\bert-base-chinese")
model = BertModel.from_pretrained("D:\\modelscope\\hub\\google-bert\\bert-base-chinese")

# 测试文本编码
text = "苹果手机的电池容量为3000毫安时"
inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
outputs = model(**inputs)

# 输出隐藏层维度（验证加载成功）
print(f"BERT隐藏层维度：{outputs.last_hidden_state.shape}")

import pandas as pd
import random

# ===================== 1. 模板定义（贴合真实语境） =====================
ecommerce_templates = [
    "亲，{商品}啥时候发货？{语气} {问题}？",
    "客服在吗？想咨询下{问题}，{附加信息}？",
    "咋地{商品}还不发货？是不是{原因}？",
    "发货后能改{信息}吗？{补充说明}",
    "{商品}的{属性}有问题，能{解决方案}吗？！"
]
home_templates = [
    "【爆款】{家具}限时特价！{促销信息}",
    "<p>{家具}用着{用户评价}，{具体描述}</p>",
    "{家具}也太好看了吧，{具体描述}！",
    "这个{家具}质量{评价_key}，{评价_value}",
    "我吃的{食物}口感{口感}，{推荐理由}！",
    "{家具}的{材质}怎么样？用着{使用体验}吗？"
]
digital_templates = [
    "{品牌}{数码产品}的{参数}是{数值}，{效果}{附加功能}",
    "请问{数码产品}的{参数}适合{使用场景}吗？",
    "【秒杀】{数码产品}限时{折扣}！{促销时间}",
    "{数码产品}的{参数}比上一代{对比信息}，{效果}",
    "{品牌}新款{数码产品}主打{亮点功能}，价格{价格}"
]

# ===================== 2. 填充词表（全英文变量名，避免中文拼写错误） =====================
# 基础词表
products = ["手机", "衣服", "鞋子", "沙发", "床垫", "电脑", "耳机", "充电宝", "手表", "键盘"]
brands = ["苹果", "华为", "小米", "vivo", "OPPO", "三星", "荣耀"]
params = ["电池容量", "续航时间", "摄像头像素", "屏幕尺寸", "处理器型号", "内存大小"]
values = ["3000mAh", "4000mAh", "5000mAh", "20000mAh", "6.5英寸", "8GB", "12GB"]
effects = ["拍照超清晰", "能充3次手机", "用一整天不关机", "游戏流畅不卡顿", "音质很赞"]

# 电商场景词表
tones = ["急急急！", "麻烦快点~", "谢谢啦～", "求回复！", ""]  # 语气
additional_infos = ["地址写错了想改", "订单号123456", "快递单号能发我吗", ""]  # 附加信息
reasons = ["仓库没货", "物流延迟", "系统出问题了", "我地址填错了", ""]  # 原因（修复原reason拼写错误）
supplementary_notes = ["地址是北京市朝阳区", "收货人是张三", "联系电话13800138000", ""]  # 补充说明
attributes = ["颜色", "尺寸", "功能", "包装"]  # 属性
solutions = ["换货", "退款", "补发", "维修"]  # 解决方案
questions = ["发货时间", "改地址", "退款", "退货", "换货"]  # 问题
infos = ["地址", "收货人", "联系方式", "订单号"]  # 信息

# 家居场景词表
promotion_infos = ["买一送一！", "满1000减200", "前100名送赠品", ""]  # 促销信息
user_evaluations = ["超舒服", "性价比超高", "颜值绝了", "一般般"]  # 用户评价
detailed_descs = ["设计简约大气", "做工特别精细", "颜色很正", "尺寸很合适"]  # 具体描述
evaluation_dict = {  # 评价字典
    "差": ["千万别买！", "建议换一家"],
    "一般": ["凑合用", "可以入手试试"],
    "好": ["特别推荐！", "必买！"],
    "很好": ["强烈推荐！", "闭眼入！"]
}
tastes = ["非常好吃", "味道一般", "超级辣", "清淡可口"]  # 口感
recommend_reasons = ["适合家庭聚餐", "孩子特别爱吃", "营养又健康"]  # 推荐理由
materials = ["实木", "金属", "布艺", "皮质", "板材"]  # 材质
use_experiences = ["很舒服", "有点硬", "容易清洁", "耐用"]  # 使用体验
furnitures = ["沙发", "床垫", "桌子", "椅子", "衣柜", "书架"]  # 家具
foods = ["饭", "面条", "火锅", "烤肉", "披萨", "寿司"]  # 食物

# 数码场景词表
additional_funcs = ["支持快充", "防水防尘", "无线充电", ""]  # 附加功能
use_scenes = ["出差用", "打游戏", "日常办公", ""]  # 使用场景
discounts = ["5折", "7折", "8折", "9折"]  # 折扣
promotion_times = ["今晚24点截止", "限时3天", "仅限今日"]  # 促销时间
highlight_funcs = ["AI拍照", "超长续航", "折叠屏", "快充"]  # 亮点功能
prices = ["2999元", "3999元", "性价比超高", "高端旗舰款"]  # 价格
compare_infos = ["提升很多", "没区别", "略差一点", "", ""]  # 对比信息
digital_products = ["手机", "电脑", "耳机", "充电宝", "平板", "智能手表"]  # 数码产品
extended_params = params + ["重量", "厚度", "分辨率"]  # 扩展参数
extended_values = values + ["1.5kg", "8mm", "4K"]  # 扩展数值
extended_brands = brands + ["联想", "戴尔", "索尼"]  # 扩展品牌

# ===================== 3. 生成语料 =====================
corpus = []

# 1. 电商客服：5000条
for _ in range(5000):
    template = random.choice(ecommerce_templates)
    text = template.format(
        商品=random.choice(products),
        问题=random.choice(questions),
        信息=random.choice(infos),
        语气=random.choice(tones),
        附加信息=random.choice(additional_infos),
        原因=random.choice(reasons),  # 修复：使用正确的变量名reasons
        补充说明=random.choice(supplementary_notes),
        属性=random.choice(attributes),
        解决方案=random.choice(solutions)
    )
    if len(text.strip()) >= 10:  # 过滤短文本
        corpus.append({"text": text, "domain": "电商客服"})

# 2. 家居生活：4000条
for _ in range(4000):
    template = random.choice(home_templates)
    eval_key = random.choice(list(evaluation_dict.keys()))
    eval_value = random.choice(evaluation_dict[eval_key])
    text = template.format(
        家具=random.choice(furnitures),
        食物=random.choice(foods),
        促销信息=random.choice(promotion_infos),
        用户评价=random.choice(user_evaluations),
        具体描述=random.choice(detailed_descs),
        评价_key=eval_key,
        评价_value=eval_value,
        口感=random.choice(tastes),
        推荐理由=random.choice(recommend_reasons),
        材质=random.choice(materials),
        使用体验=random.choice(use_experiences)
    )
    if len(text.strip()) >= 10:
        corpus.append({"text": text, "domain": "家居生活"})

# 3. 数码产品：4000条
for _ in range(4000):
    template = random.choice(digital_templates)
    text = template.format(
        品牌=random.choice(extended_brands),
        数码产品=random.choice(digital_products),
        参数=random.choice(extended_params),
        数值=random.choice(extended_values),
        效果=random.choice(effects + ["", ""]),
        附加功能=random.choice(additional_funcs),
        使用场景=random.choice(use_scenes),
        折扣=random.choice(discounts),
        促销时间=random.choice(promotion_times),
        亮点功能=random.choice(highlight_funcs),
        价格=random.choice(prices),
        对比信息=random.choice(compare_infos)
    )
    if len(text.strip()) >= 10:
        corpus.append({"text": text, "domain": "数码产品"})

# ===================== 4. 处理重复+保存 =====================
# 转为DataFrame
df = pd.DataFrame(corpus)

# 随机添加短文本（长度小于10字符）
short_texts = [
    {"text": "发货", "domain": "电商客服"},
    {"text": "好", "domain": "家居生活"},
    {"text": "快", "domain": "数码产品"},
    {"text": "改", "domain": "电商客服"},
    {"text": "买", "domain": "家居生活"},
    {"text": "赞", "domain": "数码产品"}
]
df_short = pd.DataFrame(short_texts)

# 控制重复文本比例（总条数的5%左右）
duplicate_num = int(len(df) * 0.05)
df_duplicate = df.sample(n=duplicate_num, random_state=42)
df_final = pd.concat([df, df_duplicate, df_short], ignore_index=True)

# 打乱顺序
df_final = df_final.sample(frac=1, random_state=42).reset_index(drop=True)

# 保存CSV
df_final.to_csv("raw_corpus.csv", index=False, encoding="utf-8")

# 输出统计信息
total_size = df_final.memory_usage().sum() / 1024 / 1024
print(f"生成raw_corpus.csv完成！")
print(f"总语料条数：{len(df_final)}条")
print(f"文件大小：{total_size:.2f}MB")
print(f"重复文本占比：{duplicate_num / len(df_final) * 100:.2f}%")

# 输出示例语料
print("\n=== 示例语料（前5条）===")
for idx, row in df_final.head(5).iterrows():
    print(f"领域：{row['domain']} | 文本：{row['text']}")

import pandas as pd
import hashlib
import re

# 1. 加载本地原始语料
df = pd.read_csv("raw_corpus.csv", encoding="utf-8")
print(f"初始语料条数：{len(df)}，初始大小：{df.memory_usage().sum()/1024/1024:.2f}MB")

# 2. 完全去重（基于MD5哈希，剔除一模一样的文本）
def get_text_hash(text):
    return hashlib.md5(str(text).encode("utf-8")).hexdigest()

df["text_hash"] = df["text"].apply(get_text_hash)
df = df.drop_duplicates(subset=["text_hash"])
print(f"完全去重后条数：{len(df)}")

# 3. 格式规整（去HTML标签、统一标点、去乱码）
def clean_format(text):
    text = str(text).strip()
    # 去HTML标签（比如<p>、<div>）
    text = re.sub(r"<.*?>", "", text)
    # 统一中英文标点（把.换成。，,换成，）
    punc_map = {".": "。", ",": "，", "?": "？", "!": "！", ":": "："}
    for en_punc, cn_punc in punc_map.items():
        text = text.replace(en_punc, cn_punc)
    # 去乱码（只保留中文、数字、常用标点）
    text = re.sub(r"[^\u4e00-\u9fa50-9，。？！：（）、]", "", text)
    return text

df["clean_text"] = df["text"].apply(clean_format)

# 4. 过滤短文本（至少10个字符，剔除“啊啊啊”这种垃圾）
df = df[df["clean_text"].apply(lambda x: len(x) >= 10)]
print(f"格式清洗+短文本过滤后条数：{len(df)}")

# 5. 保存初步清洗后的语料
df.to_csv("clean_corpus_v1.csv", index=False, encoding="utf-8")
print("初步清洗完成，保存为clean_corpus_v1.csv")

import pandas as pd
import numpy as np
from transformers import AutoTokenizer, AutoModel
import torch
from sklearn.metrics.pairwise import cosine_similarity
from tqdm import tqdm
import matplotlib.pyplot as plt
import os

# ===================== 基础配置 =====================
# 设置中文字体（避免可视化乱码）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 本地模型路径（ModelScope下载的text2vec路径）
LOCAL_MODEL_PATH = "D:\\modelscope\\hub\\Jerry0\\text2vec-base-chinese"
# 输入/输出文件路径
INPUT_CORPUS_PATH = "clean_corpus_v1.csv"
OUTPUT_CORPUS_PATH = "clean_corpus_v2.csv"
VISUALIZATION_PATH = "语义相似度分布.png"

# ===================== 核心修复：加载text2vec模型（HuggingFace原生接口） =====================
def load_text2vec_model(model_path, device="cpu"):
    """
    加载本地text2vec模型（兼容ModelScope下载的文件结构）
    :param model_path: 本地模型路径
    :param device: 运行设备（cpu/cuda）
    :return: tokenizer, model
    """
    print(f"===== 加载本地text2vec模型 =====")
    print(f"模型路径：{model_path}")
    print(f"运行设备：{device}")
    
    # 加载tokenizer和model（HuggingFace原生接口，避免ModelScope registry报错）
    tokenizer = AutoTokenizer.from_pretrained(model_path)
    model = AutoModel.from_pretrained(model_path).to(device)
    model.eval()  # 预测模式
    print("模型加载成功！")
    return tokenizer, model

def get_sentence_embedding(texts, tokenizer, model, device="cpu", batch_size=100):
    """
    批量计算文本语义向量（替代ModelScope pipeline）
    :param texts: 文本列表
    :param tokenizer: 分词器
    :param model: text2vec模型
    :param device: 运行设备
    :param batch_size: 批次大小
    :return: 语义向量列表
    """
    embeddings = []
    # 分批处理
    for i in tqdm(range(0, len(texts), batch_size), desc="计算语义向量"):
        batch_texts = texts[i:i+batch_size]
        # 分词
        inputs = tokenizer(
            batch_texts,
            padding=True,
            truncation=True,
            max_length=128,
            return_tensors="pt"
        ).to(device)
        
        # 计算向量（text2vec采用[CLS] token的均值作为句子向量）
        with torch.no_grad():
            outputs = model(**inputs)
            # 取最后一层隐藏层的均值作为句子向量
            last_hidden = outputs.last_hidden_state
            mask = inputs['attention_mask'].unsqueeze(-1).expand(last_hidden.size())
            sum_embeddings = torch.sum(last_hidden * mask, dim=1)
            sum_mask = torch.clamp(mask.sum(1), min=1e-9)
            batch_emb = (sum_embeddings / sum_mask).cpu().numpy()
        
        embeddings.extend(batch_emb)
    return embeddings

# ===================== 主流程：语义去重 =====================
if __name__ == "__main__":
    # 1. 加载初步清洗后的语料
    print("===== 加载语料 =====")
    try:
        df = pd.read_csv(INPUT_CORPUS_PATH, encoding="utf-8")
        # 检查列名（兼容不同清洗版本的列名）
        text_col = "clean_text" if "clean_text" in df.columns else "text"
        texts = df[text_col].tolist()
        # 过滤空值
        texts = [str(t).strip() for t in texts if pd.notna(t) and len(str(t).strip()) > 0]
        df = df[df[text_col].apply(lambda x: pd.notna(x) and len(str(x).strip()) > 0)].reset_index(drop=True)
        
        print(f"待语义去重的文本条数：{len(texts)}")
    except FileNotFoundError:
        print(f"错误：未找到文件 {INPUT_CORPUS_PATH}")
        exit()
    except Exception as e:
        print(f"加载语料失败：{e}")
        exit()
    
    # 2. 加载本地text2vec模型（核心修复：改用HuggingFace接口）
    device = "cuda" if torch.cuda.is_available() else "cpu"
    tokenizer, model = load_text2vec_model(LOCAL_MODEL_PATH, device=device)
    
    # 3. 批量计算语义向量
    embeddings = get_sentence_embedding(texts, tokenizer, model, device=device, batch_size=100)
    emb_array = np.array(embeddings)
    print(f"语义向量计算完成！向量维度：{emb_array.shape}")
    
    # 4. 语义去重（余弦相似度≥0.95判定为近重复）
    print("\n===== 开始语义去重 =====")
    keep_indices = []
    for i in tqdm(range(len(emb_array)), desc="语义去重"):
        # 第一条直接保留
        if i == 0:
            keep_indices.append(i)
            continue
        # 计算当前文本与已保留文本的相似度
        sim_scores = cosine_similarity([emb_array[i]], emb_array[keep_indices])[0]
        # 相似度<0.95才保留（避免近重复）
        if np.max(sim_scores) < 0.95:
            keep_indices.append(i)
    
    # 5. 筛选去重后的语料
    df_dedup = df.iloc[keep_indices].reset_index(drop=True)
    print(f"原始文本条数：{len(df)}")
    print(f"语义去重后条数：{len(df_dedup)}")
    print(f"过滤掉的近重复文本条数：{len(df) - len(df_dedup)}")
    print(f"语义去重保留率：{len(df_dedup)/len(df)*100:.2f}%")
    
    # 6. 保存语义去重后的语料
    df_dedup.to_csv(OUTPUT_CORPUS_PATH, index=False, encoding="utf-8")
    print(f"\n语义去重后语料已保存：{OUTPUT_CORPUS_PATH}")
    
    # 7. 可视化：语义相似度分布（直观看重复程度）
    print("\n===== 生成可视化图表 =====")
    # 随机选1000条计算相似度（避免计算量过大）
    sample_size = min(1000, len(emb_array))
    sample_idx = np.random.choice(len(emb_array), sample_size, replace=False)
    sample_emb = emb_array[sample_idx]
    
    # 计算相似度矩阵
    sim_matrix = cosine_similarity(sample_emb)
    # 取上三角矩阵（排除自身相似度）
    sim_vals = sim_matrix[np.triu_indices_from(sim_matrix, k=1)]
    
    # 绘制直方图
    plt.figure(figsize=(10, 5))
    plt.hist(sim_vals, bins=50, color="#1f77b4", alpha=0.7)
    plt.axvline(x=0.95, color="red", linestyle="--", linewidth=2, label="相似度阈值0.95")
    plt.title("文本语义相似度分布", fontsize=14)
    plt.xlabel("余弦相似度", fontsize=12)
    plt.ylabel("文本对数量", fontsize=12)
    plt.legend(fontsize=10)
    plt.grid(alpha=0.3)
    plt.tight_layout()
    plt.savefig(VISUALIZATION_PATH, dpi=300)
    print(f"可视化图表已保存：{VISUALIZATION_PATH}")
    # plt.show()  # 如需弹窗显示，取消注释
    
    print("\n===== 语义去重流程完成！=====")

import pandas as pd
import torch
from transformers import BertTokenizer, BertForSequenceClassification
from tqdm import tqdm
import matplotlib.pyplot as plt

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 1. 加载语义去重后的语料
df = pd.read_csv("clean_corpus_v2.csv", encoding="utf-8")
texts = df["text"].tolist()
texts = [str(t).strip() for t in texts if pd.notna(t)]
print(f"待质量评分的文本条数：{len(texts)}")

# 2. 加载本地BERT模型（适配质量评分任务）
local_bert_path = "D:\\modelscope\\hub\\google-bert\\bert-base-chinese"
tokenizer = BertTokenizer.from_pretrained(local_bert_path)
# 构建质量评分模型（回归任务，输出0~1的分数）
model = BertForSequenceClassification.from_pretrained(
    local_bert_path,
    num_labels=1  # 回归任务，单标签
).to("cpu")
model.eval()

# 3. 批量预测质量分
quality_scores = []
batch_size = 32
for i in tqdm(range(0, len(texts), batch_size), desc="计算质量分"):
    batch_texts = texts[i:i+batch_size]
    inputs = tokenizer(
        batch_texts,
        truncation=True,
        padding="max_length",
        max_length=128,
        return_tensors="pt"
    )
    with torch.no_grad():
        outputs = model(**inputs)
    # 转换为0~1的分数
    scores = torch.sigmoid(outputs.logits).cpu().numpy().flatten()
    quality_scores.extend(scores)

# 4. 筛选高质量文本（≥0.8分）
df["quality_score"] = quality_scores
df_high_quality = df[df["quality_score"] >= 0.5].reset_index(drop=True)
print(f"质量分≥0.5的文本条数：{len(df_high_quality)}")
print(f"过滤低质量文本条数：{len(df) - len(df_high_quality)}")

# 5. 保存高质量语料
df_high_quality.to_csv("clean_corpus_v3.csv", index=False, encoding="utf-8")

# 6. 可视化质量分分布
plt.figure(figsize=(10, 5))
plt.hist(quality_scores, bins=50, color="#ff7f0e", alpha=0.7)
plt.axvline(x=0.8, color="red", linestyle="--", label="质量分阈值0.8")
plt.title("文本质量分分布", fontsize=14)
plt.xlabel("质量分（0-1）", fontsize=12)
plt.ylabel("文本数量", fontsize=12)
plt.legend()
plt.grid(alpha=0.3)
plt.savefig("文本质量分分布.png", dpi=300)
plt.show()

print(f"质量评分完成！高质量语料保存为clean_corpus_v3.csv")

import pandas as pd
import re

# 1. 加载高质量语料
df = pd.read_csv("clean_corpus_v3.csv", encoding="utf-8")

# 2. 规则细筛
def clean_text(text):
    # 修正错别字
    typo_map = {"mah": "mAh", "啥时候": "什么时候", "咋地": "怎么样"}
    for typo, correct in typo_map.items():
        text = text.replace(typo, correct)
    # 修正病句（如“我吃了饭非常好吃”→“我吃的饭非常好吃”）
    text = re.sub(r"吃了(.*?)非常好吃", r"吃的\1非常好吃", text)
    # 去除HTML标签
    text = re.sub(r"<.*?>", "", text)
    # 统一标点（英文→中文）
    punc_map = {".": "。", ",": "，", "?": "？", "!": "！"}
    for en_punc, cn_punc in punc_map.items():
        text = text.replace(en_punc, cn_punc)
    # 去除偏见词汇
    bias_map = {"垃圾": "劣质", "没用": "效果不佳", "脑残": "不合适"}
    for bias, neutral in bias_map.items():
        text = text.replace(bias, neutral)
    return text.strip()

df["final_text"] = df["text"].apply(clean_text)

# 3. 质量评估
# 3.1 重复率（完全+近重复）
total = len(df)
duplicate_rate = (1 - len(df.drop_duplicates(subset=["final_text"]))/total) * 100

# 3.2 语法正确率（根据数据量动态抽样）
sample_size = min(100, total)  # 如果数据不足100条，则取全部数据
sample_texts = df["final_text"].sample(sample_size, random_state=42).tolist()
# 模拟人工审核（实际需人工标注）
correct_grammar = int(sample_size * 0.96)  # 假设96%语法正确
grammar_correct_rate = (correct_grammar/sample_size) * 100

# 3.3 噪声率（1 - 最终语料/原始语料）
raw_total = len(pd.read_csv("raw_corpus.csv"))
noise_rate = (1 - len(df)/raw_total) * 100

# 4. 保存最终语料
df_final = df[["final_text", "domain", "quality_score"]]
df_final.to_csv("final_high_quality_corpus.csv", index=False, encoding="utf-8")

# 5. 输出评估结果
print("===== 质量评估结果 =====")
print(f"重复率：{duplicate_rate:.2f}%")
print(f"语法正确率：{grammar_correct_rate:.2f}%")
print(f"噪声率：{noise_rate:.2f}%")
print(f"\n最终语料保存为：final_high_quality_corpus.csv")
print(f"最终语料条数：{len(df_final)}条")