句子语义相似度计算：核心方法与实战应用指南

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import jieba
 
def tfidf_similarity(sentence1, sentence2):
    # 中文分词
    words1 = ' '.join(jieba.cut(sentence1))
    words2 = ' '.join(jieba.cut(sentence2))
    
    # 构建TF-IDF向量
    vectorizer = TfidfVectorizer()
    tfidf_matrix = vectorizer.fit_transform([words1, words2])
    
    # 计算余弦相似度
    similarity = cosine_similarity(tfidf_matrix[0:1], tfidf_matrix[1:2])[0][0]
    return similarity
 
# 示例
s1 = "机器学习是人工智能的核心技术"
s2 = "深度学习推动了AI技术的快速发展"
print(f"TF-IDF相似度: {tfidf_similarity(s1, s2):.4f}")

import numpy as np
from gensim.models import Word2Vec
import jieba
 
class WordEmbeddingSimilarity:
    def __init__(self, model_path=None):
        if model_path:
            self.model = Word2Vec.load(model_path)
        else:
            # 使用简单的随机向量作为示例
            self.model = None
            self.vector_size = 100
    
    def sentence_vector(self, sentence):
        words = jieba.cut(sentence)
        vectors = []
        
        for word in words:
            if self.model and word in self.model.wv:
                vectors.append(self.model.wv[word])
            else:
                # 随机向量作为fallback
                vectors.append(np.random.randn(self.vector_size))
        
        if not vectors:
            return np.zeros(self.vector_size)
        
        # 加权平均，使用TF-IDF权重
        return np.mean(vectors, axis=0)
    
    def similarity(self, sent1, sent2):
        vec1 = self.sentence_vector(sent1)
        vec2 = self.sentence_vector(sent2)
        
        # 余弦相似度
        dot_product = np.dot(vec1, vec2)
        norm1 = np.linalg.norm(vec1)
        norm2 = np.linalg.norm(vec2)
        
        if norm1 == 0 or norm2 == 0:
            return 0.0
        
        return dot_product / (norm1 * norm2)
 
# 使用示例
embedding_sim = WordEmbeddingSimilarity()
s1 = "自然语言处理很有趣"
s2 = "NLP技术非常吸引人"
print(f"词嵌入相似度: {embedding_sim.similarity(s1, s2):.4f}")

from sentence_transformers import SentenceTransformer
import numpy as np
 
class SBERTSimilarity:
    def __init__(self, model_name='paraphrase-multilingual-MiniLM-L12-v2'):
        try:
            self.model = SentenceTransformer(model_name)
        except:
            print("模型加载失败，使用模拟数据")
            self.model = None
    
    def encode_sentences(self, sentences):
        if self.model:
            return self.model.encode(sentences)
        else:
            # 模拟向量生成
            return np.random.randn(len(sentences), 384)
    
    def similarity_matrix(self, sentences):
        embeddings = self.encode_sentences(sentences)
        
        # 计算相似度矩阵
        similarity_matrix = np.zeros((len(sentences), len(sentences)))
        for i in range(len(sentences)):
            for j in range(len(sentences)):
                similarity_matrix[i][j] = np.dot(embeddings[i], embeddings[j]) / (
                    np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[j])
                )
        
        return similarity_matrix
 
# 实际应用示例
sbert_sim = SBERTSimilarity()
sentences = [
    "今天天气很好",
    "今天阳光明媚",
    "明天下雨的可能性很大",
    "天气预报说明天有雨"
]
 
similarity_matrix = sbert_sim.similarity_matrix(sentences)
print("句子相似度矩阵:")
for i, sent1 in enumerate(sentences):
    for j, sent2 in enumerate(sentences):
        print(f"'{sent1}' vs '{sent2}': {similarity_matrix[i][j]:.4f}")

from transformers import BertTokenizer, BertModel
import torch
import torch.nn.functional as F
 
class BERTInteractionSimilarity:
    def __init__(self, model_name='bert-base-chinese'):
        try:
            self.tokenizer = BertTokenizer.from_pretrained(model_name)
            self.model = BertModel.from_pretrained(model_name)
        except:
            print("BERT模型加载失败")
            self.tokenizer = None
            self.model = None
    
    def calculate_similarity(self, sentence1, sentence2):
        if not self.model or not self.tokenizer:
            # 返回模拟相似度
            return 0.75
        
        # 编码句子对
        inputs = self.tokenizer(
            sentence1, sentence2,
            return_tensors='pt',
            padding=True,
            truncation=True,
            max_length=128
        )
        
        # 获取BERT输出
        with torch.no_grad():
            outputs = self.model(**inputs)
            # 使用[CLS]token的表示
            cls_embedding = outputs.last_hidden_state[:, 0, :]
            
            # 计算余弦相似度
            similarity = F.cosine_similarity(cls_embedding[0:1], cls_embedding[1:2])
            return similarity.item()
 
# 使用示例
bert_sim = BERTInteractionSimilarity()
s1 = "人工智能改变世界"
s2 = "AI技术正在改变我们的生活"
print(f"BERT交互式相似度: {bert_sim.calculate_similarity(s1, s2):.4f}")

class FAQMatcher:
    def __init__(self, faq_data, similarity_model):
        self.faq_data = faq_data
        self.similarity_model = similarity_model
    
    def find_best_match(self, user_query, threshold=0.7):
        best_match = None
        best_score = 0
        
        for faq in self.faq_data:
            score = self.similarity_model.similarity(user_query, faq['question'])
            if score > best_score and score >= threshold:
                best_score = score
                best_match = faq
        
        return best_match, best_score
 
# 示例FAQ数据
faq_data = [
    {"question": "如何重置密码？", "answer": "点击登录页面的'忘记密码'链接..."},
    {"question": "订单什么时候发货？", "answer": "订单通常在24小时内发货..."},
    {"question": "如何申请退款？", "answer": "在订单页面点击申请退款按钮..."}
]
 
faq_matcher = FAQMatcher(faq_data, SBERTSimilarity())
user_query = "我忘记了登录密码怎么办"
match, score = faq_matcher.find_best_match(user_query)
print(f"最佳匹配: {match['question']} (相似度: {score:.4f})")

class ContentRecommender:
    def __init__(self, content_data, similarity_model):
        self.content_data = content_data
        self.similarity_model = similarity_model
        # 预计算内容向量
        self.content_vectors = self._precompute_vectors()
    
    def _precompute_vectors(self):
        contents = [item['content'] for item in self.content_data]
        if hasattr(self.similarity_model, 'encode_sentences'):
            return self.similarity_model.encode_sentences(contents)
        else:
            return None
    
    def recommend(self, user_content, top_k=5):
        if self.content_vectors is not None:
            # 使用预计算向量加速
            user_vector = self.similarity_model.encode_sentences([user_content])[0]
            similarities = []
            
            for i, content_vector in enumerate(self.content_vectors):
                sim = np.dot(user_vector, content_vector) / (
                    np.linalg.norm(user_vector) * np.linalg.norm(content_vector)
                )
                similarities.append((i, sim))
        else:
            # 实时计算
            similarities = []
            for i, item in enumerate(self.content_data):
                sim = self.similarity_model.similarity(user_content, item['content'])
                similarities.append((i, sim))
        
        # 排序并返回top-k
        similarities.sort(key=lambda x: x[1], reverse=True)
        recommendations = []
        
        for i, (idx, score) in enumerate(similarities[:top_k]):
            recommendations.append({
                'rank': i + 1,
                'content': self.content_data[idx]['title'],
                'similarity_score': score
            })
        
        return recommendations

class TextDeduplicator:
    def __init__(self, similarity_model, threshold=0.85):
        self.similarity_model = similarity_model
        self.threshold = threshold
    
    def find_duplicates(self, documents):
        duplicates = []
        n = len(documents)
        
        for i in range(n):
            for j in range(i + 1, n):
                similarity = self.similarity_model.similarity(
                    documents[i]['content'], 
                    documents[j]['content']
                )
                
                if similarity >= self.threshold:
                    duplicates.append({
                        'doc1_id': documents[i]['id'],
                        'doc2_id': documents[j]['id'],
                        'similarity': similarity
                    })
        
        return duplicates
    
    def cluster_similar_documents(self, documents, clustering_threshold=0.8):
        clusters = []
        processed = set()
        
        for i, doc in enumerate(documents):
            if doc['id'] in processed:
                continue
            
            cluster = [doc]
            processed.add(doc['id'])
            
            for j, other_doc in enumerate(documents[i+1:], i+1):
                if other_doc['id'] in processed:
                    continue
                
                similarity = self.similarity_model.similarity(
                    doc['content'], other_doc['content']
                )
                
                if similarity >= clustering_threshold:
                    cluster.append(other_doc)
                    processed.add(other_doc['id'])
            
            clusters.append(cluster)
        
        return clusters

import faiss
import numpy as np
 
class VectorIndex:
    def __init__(self, vector_dim, index_type='IVF'):
        self.vector_dim = vector_dim
        self.index_type = index_type
        
        if index_type == 'IVF':
            # IVF索引适合中等规模数据
            quantizer = faiss.IndexFlatIP(vector_dim)  # 内积相似度
            self.index = faiss.IndexIVFFlat(quantizer, vector_dim, 100)
        else:
            # 平面索引适合小规模数据
            self.index = faiss.IndexFlatIP(vector_dim)
    
    def add_vectors(self, vectors):
        if not self.index.is_trained and hasattr(self.index, 'train'):
            self.index.train(vectors.astype(np.float32))
        self.index.add(vectors.astype(np.float32))
    
    def search(self, query_vectors, k=10):
        scores, indices = self.index.search(query_vectors.astype(np.float32), k)
        return scores, indices
 
# 使用示例
vector_dim = 384  # SBERT向量维度
index = VectorIndex(vector_dim, 'IVF')
 
# 假设我们有大量文档向量
doc_vectors = np.random.randn(10000, vector_dim).astype(np.float32)
index.add_vectors(doc_vectors)
 
# 查询相似文档
query_vector = np.random.randn(1, vector_dim).astype(np.float32)
scores, indices = index.search(query_vector, k=5)
print(f"Top-5相似文档索引: {indices[0]}")
print(f"相似度分数: {scores[0]}")

import hashlib
import json
import time
from functools import lru_cache
 
class SimilarityCache:
    def __init__(self, redis_client=None, cache_ttl=3600):
        self.redis_client = redis_client
        self.cache_ttl = cache_ttl
        self.local_cache = {}
    
    def _generate_key(self, text1, text2):
        # 生成缓存键，确保顺序无关
        texts = sorted([text1, text2])
        content = f"{texts[0]}|||{texts[1]}"
        return hashlib.md5(content.encode()).hexdigest()
    
    def get(self, text1, text2):
        key = self._generate_key(text1, text2)
        
        # 本地缓存检查
        if key in self.local_cache:
            timestamp, value = self.local_cache[key]
            if time.time() - timestamp < self.cache_ttl:
                return value
        
        # Redis缓存检查
        if self.redis_client:
            cached_value = self.redis_client.get(key)
            if cached_value:
                value = float(cached_value)
                self.local_cache[key] = (time.time(), value)
                return value
        
        return None
    
    def set(self, text1, text2, similarity_score):
        key = self._generate_key(text1, text2)
        
        # 更新本地缓存
        self.local_cache[key] = (time.time(), similarity_score)
        
        # 更新Redis缓存
        if self.redis_client:
            self.redis_client.setex(key, self.cache_ttl, similarity_score)
    
    @lru_cache(maxsize=1000)
    def cached_similarity(self, similarity_func, text1, text2):
        # 使用函数和文本作为缓存键
        return similarity_func(text1, text2)

import torch
from torch.utils.data import DataLoader, Dataset
 
class SentenceDataset(Dataset):
    def __init__(self, sentences):
        self.sentences = sentences
    
    def __len__(self):
        return len(self.sentences)
    
    def __getitem__(self, idx):
        return self.sentences[idx]
 
class BatchSimilarityProcessor:
    def __init__(self, model, batch_size=32, device='cuda'):
        self.model = model
        self.batch_size = batch_size
        self.device = device if torch.cuda.is_available() else 'cpu'
        
        if hasattr(model, 'to'):
            self.model = model.to(self.device)
    
    def encode_sentences_batch(self, sentences):
        dataset = SentenceDataset(sentences)
        dataloader = DataLoader(dataset, batch_size=self.batch_size, shuffle=False)
        
        all_embeddings = []
        
        for batch_sentences in dataloader:
            # 批量编码
            if hasattr(self.model, 'encode'):
                batch_embeddings = self.model.encode(batch_sentences)
            else:
                # 模拟批量编码
                batch_embeddings = np.random.randn(len(batch_sentences), 384)
            
            all_embeddings.append(batch_embeddings)
        
        return np.vstack(all_embeddings)
    
    def compute_similarity_matrix_batch(self, sentences1, sentences2):
        # 批量编码所有句子
        embeddings1 = self.encode_sentences_batch(sentences1)
        embeddings2 = self.encode_sentences_batch(sentences2)
        
        # 计算相似度矩阵
        similarity_matrix = np.dot(embeddings1, embeddings2.T)
        
        # 归一化
        norm1 = np.linalg.norm(embeddings1, axis=1, keepdims=True)
        norm2 = np.linalg.norm(embeddings2, axis=1, keepdims=True)
        similarity_matrix /= np.dot(norm1, norm2.T)
        
        return similarity_matrix

import re
import string
 
def preprocess_text(text):
    """文本预处理函数"""
    # 转换为小写
    text = text.lower()
    
    # 移除标点符号
    text = re.sub(f'[{re.escape(string.punctuation)}]', '', text)
    
    # 移除多余空白
    text = re.sub(r'\s+', ' ', text).strip()
    
    # 移除停用词（示例）
    stopwords = {'的', '了', '在', '是', '我', '你', '他', '她', '它'}
    words = text.split()
    words = [w for w in words if w not in stopwords]
    
    return ' '.join(words)

class ThresholdOptimizer:
    def __init__(self, similarity_model):
        self.similarity_model = similarity_model
    
    def find_optimal_threshold(self, validation_data, metric='f1'):
        """基于验证数据寻找最优阈值"""
        thresholds = np.arange(0.1, 1.0, 0.05)
        best_threshold = 0.5
        best_score = 0
        
        for threshold in thresholds:
            predictions = []
            true_labels = []
            
            for item in validation_data:
                similarity = self.similarity_model.similarity(
                    item['text1'], item['text2']
                )
                prediction = 1 if similarity >= threshold else 0
                predictions.append(prediction)
                true_labels.append(item['label'])
            
            # 计算指标
            if metric == 'f1':
                score = self._calculate_f1(true_labels, predictions)
            elif metric == 'accuracy':
                score = self._calculate_accuracy(true_labels, predictions)
            
            if score > best_score:
                best_score = score
                best_threshold = threshold
        
        return best_threshold, best_score
    
    def _calculate_f1(self, true_labels, predictions):
        # 简化的F1计算
        tp = sum(1 for t, p in zip(true_labels, predictions) if t == 1 and p == 1)
        fp = sum(1 for t, p in zip(true_labels, predictions) if t == 0 and p == 1)
        fn = sum(1 for t, p in zip(true_labels, predictions) if t == 1 and p == 0)
        
        if tp + fp == 0 or tp + fn == 0:
            return 0
        
        precision = tp / (tp + fp)
        recall = tp / (tp + fn)
        
        if precision + recall == 0:
            return 0
        
        return 2 * precision * recall / (precision + recall)

class SimilarityMonitor:
    def __init__(self):
        self.metrics = {
            'total_requests': 0,
            'cache_hits': 0,
            'avg_similarity': 0.0,
            'response_times': []
        }
    
    def record_request(self, similarity_score, response_time, cache_hit=False):
        self.metrics['total_requests'] += 1
        if cache_hit:
            self.metrics['cache_hits'] += 1
        
        # 更新平均相似度
        n = self.metrics['total_requests']
        self.metrics['avg_similarity'] = (
            (self.metrics['avg_similarity'] * (n-1) + similarity_score) / n
        )
        
        self.metrics['response_times'].append(response_time)
        
        # 保持最近100个响应时间
        if len(self.metrics['response_times']) > 100:
            self.metrics['response_times'] = self.metrics['response_times'][-100:]
    
    def get_stats(self):
        cache_hit_rate = self.metrics['cache_hits'] / max(self.metrics['total_requests'], 1)
        avg_response_time = np.mean(self.metrics['response_times']) if self.metrics['response_times'] else 0
        
        return {
            'total_requests': self.metrics['total_requests'],
            'cache_hit_rate': cache_hit_rate,
            'avg_similarity': self.metrics['avg_similarity'],
            'avg_response_time': avg_response_time,
            'p95_response_time': np.percentile(self.metrics['response_times'], 95) if self.metrics['response_times'] else 0
        }