AI 创作日记 | 客服对话中的「黄金矿脉」，DeepSeek非结构化数据挖掘实战

import jieba
from textblob import TextBlob
import re

class DialogPreprocessor:
    def __init__(self):
        self.stopwords = set(open('stopwords.txt').read().splitlines())
        
    def clean_text(self, text):
        """对话文本瑞士军刀"""
        # 去除特殊字符
        text = re.sub(r'[【】★↓←→◆■▼▲]', '', text)  
        # 合并重复标点
        text = re.sub(r'([!?。])\1+', r'\1', text)  
        return text
    
    def analyze_sentiment(self, text):
        """情感雷达扫描"""
        blob = TextBlob(text)
        return {
            'polarity': blob.sentiment.polarity,
            'subjectivity': blob.sentiment.subjectivity
        }
    
    def extract_keywords(self, text, topK=5):
        """语义金矿探测器"""
        words = [word for word in jieba.cut(text) 
                if word not in self.stopwords and len(word) > 1]
        return Counter(words).most_common(topK)

# 使用示例
processor = DialogPreprocessor()
sample_text = "顾客说：这衣服质量太差了！！才洗一次就起球！"
clean_text = processor.clean_text(sample_text)
print(processor.extract_keywords(clean_text))
# 输出：[('衣服', 1), ('质量', 1), ('起球', 1)]

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

class IntentClassifier:
    def __init__(self, model_path="deepseek-ai/deepseek-7b"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_path)
        
    def train(self, dataset):
        """模型调教师"""
        # 此处简化训练流程，实际需配置TrainingArguments
        trainer = Trainer(
            model=self.model,
            train_dataset=dataset,
            data_collator=lambda data: {
                'input_ids': torch.stack([item['input_ids'] for item in data]),
                'labels': torch.tensor([item['label'] for item in data])
            }
        )
        trainer.train()
    
    def predict(self, text):
        """意图雷达"""
        inputs = self.tokenizer(text, return_tensors="pt")
        outputs = self.model(**inputs)
        return torch.argmax(outputs.logits)

# 示例训练数据格式
train_data = [
    {"text": "我要退货", "label": 0},
    {"text": "尺码推荐", "label": 1},
    {"text": "物流查询", "label": 2}
]

用户：你们那个新出的智能咖啡机怎么老是出奶泡啊？  
客服：抱歉给您带来不便，是CF-2025型号吗？  
用户：对！刚买一周就这样，还不如我之前买的便宜款  
客服：我们将安排工程师上门检修...

# 实战代码示例
miner = DialogMiner()
dialog = load_customer_service_log('coffee_machine_case.json')
analysis = miner.analyze_dialog(dialog)

# 生成商业洞察报告
report_template = """
**产品改进建议**：
{product_issues}

**客户画像更新**：
{user_profile}

**市场机会发现**：
{market_insight}
"""
print(report_template.format(**analysis['insights']))

检测到23次关于CF-2025奶泡系统的负面反馈  
发现老客户对新品满意度低于经典款（-35%）  
潜在需求：便携式清洁配件（提及率18%）

# 声纹情感增强模块
class VoiceTextureEnhancer:
    def __init__(self):
        self.audio_net = load_pretrained('voice2vec')
        self.text_net = load_pretrained('bert-base')
    
    def __call__(self, audio, text):
        # 音频特征提取
        voice_feat = self.audio_net(audio)[..., :256]  
        # 文本特征融合
        text_feat = self.text_net(text).last_hidden_state.mean(dim=1)
        # 跨模态注意力
        fused_feat = CrossAttentionLayer()(voice_feat, text_feat)
        return fused_feat

# 使用示例：捕捉哽咽中的真实需求
enhancer = VoiceTextureEnhancer()
true_demand = enhancer(audio_clip, "怎么老是出奶泡...")

# 添加自定义词典
jieba.add_word('孩', freq=100, tag='n')  # 修正方言发音问题

from Levenshtein import distance

def is_sensitive(word):
    variants = ['发票', '发嘌', 'fapiao']
    return any(distance(word, v) <=1 for v in variants)