文本相似度计算算法与应用实践-平芜编程栈

1. CSP第二题相似度计算概述

CSP（Content Security Policy）第二题的相似度计算是一个典型的文本处理与算法设计问题。这类题目通常要求参赛者设计算法来量化两个文本片段之间的相似程度，在信息安全、内容过滤和文本分析等领域有广泛应用。

相似度计算的核心在于建立有效的文本特征表示和选择合适的距离度量方法。常见的应用场景包括：

网页内容相似性检测
抄袭识别系统
搜索引擎结果去重
文本聚类分析

2. 相似度计算的核心算法

2.1 基于词频的余弦相似度

最基础的相似度计算方法是将文本视为词频向量，然后计算它们的余弦夹角：

from collections import Counter import math def cosine_similarity(text1, text2): # 构建词频向量 vec1 = Counter(text1.split()) vec2 = Counter(text2.split()) # 获取所有唯一词 words = set(vec1.keys()).union(set(vec2.keys())) # 计算点积 dot_product = sum(vec1.get(word,0) * vec2.get(word,0) for word in words) # 计算模长 magnitude1 = math.sqrt(sum(vec1.get(word,0)**2 for word in words)) magnitude2 = math.sqrt(sum(vec2.get(word,0)**2 for word in words)) return dot_product / (magnitude1 * magnitude2)

注意：这种方法对词序不敏感，适合处理内容相似但表述方式不同的文本。

2.2 基于编辑距离的方法

对于需要考虑词序的场景，可以使用编辑距离算法：

def levenshtein_distance(s1, s2): if len(s1) < len(s2): return levenshtein_distance(s2, s1) if len(s2) == 0: return len(s1) previous_row = range(len(s2) + 1) for i, c1 in enumerate(s1): current_row = [i + 1] for j, c2 in enumerate(s2): insertions = previous_row[j + 1] + 1 deletions = current_row[j] + 1 substitutions = previous_row[j] + (c1 != c2) current_row.append(min(insertions, deletions, substitutions)) previous_row = current_row return previous_row[-1] def normalized_edit_similarity(text1, text2): distance = levenshtein_distance(text1, text2) max_len = max(len(text1), len(text2)) return 1 - distance / max_len

3. 高级相似度计算技术

3.1 基于N-gram的相似度

N-gram方法可以捕捉局部文本特征：

def ngram_similarity(text1, text2, n=2): def get_ngrams(text): return [text[i:i+n] for i in range(len(text)-n+1)] ngrams1 = set(get_ngrams(text1)) ngrams2 = set(get_ngrams(text2)) intersection = ngrams1 & ngrams2 union = ngrams1 | ngrams2 return len(intersection) / len(union)

3.2 基于词向量的语义相似度

使用预训练的词向量可以捕捉语义信息：

import numpy as np from gensim.models import KeyedVectors # 加载预训练词向量 word_vectors = KeyedVectors.load_word2vec_format('GoogleNews-vectors-negative300.bin', binary=True) def sentence_vector(sentence): words = sentence.split() vectors = [word_vectors[word] for word in words if word in word_vectors] if not vectors: return np.zeros(300) return np.mean(vectors, axis=0) def semantic_similarity(text1, text2): vec1 = sentence_vector(text1) vec2 = sentence_vector(text2) return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

4. 性能优化与工程实践

4.1 大规模文本处理的优化策略

当处理大量文本时，需要考虑以下优化方法：

索引预处理：

from sklearn.feature_extraction.text import TfidfVectorizer import numpy as np corpus = ["text1 content", "text2 content", ...] # 文本集合 vectorizer = TfidfVectorizer() tfidf_matrix = vectorizer.fit_transform(corpus)

近似最近邻搜索：

from annoy import AnnoyIndex # 构建索引 dim = tfidf_matrix.shape[1] annoy_index = AnnoyIndex(dim, 'angular') for i in range(tfidf_matrix.shape[0]): annoy_index.add_item(i, tfidf_matrix[i].toarray()[0]) annoy_index.build(10) # 10 trees

4.2 分布式计算方案

对于超大规模文本集合，可以使用Spark等分布式框架：

from pyspark.ml.feature import HashingTF, IDF from pyspark.ml.linalg import Vectors from pyspark.sql import SparkSession spark = SparkSession.builder.appName("TextSimilarity").getOrCreate() # 创建示例数据 data = spark.createDataFrame([ (0, "hello world"), (1, "hello spark"), (2, "goodbye world") ], ["id", "text"]) # 特征提取 hashingTF = HashingTF(inputCol="text", outputCol="rawFeatures") featurizedData = hashingTF.transform(data) idf = IDF(inputCol="rawFeatures", outputCol="features") idfModel = idf.fit(featurizedData) rescaledData = idfModel.transform(featurizedData)

5. 实际应用中的挑战与解决方案

5.1 处理不同长度的文本

长短文本的相似度计算需要特殊处理：

def asymmetric_similarity(text1, text2): """处理长短文本的相似度""" len1, len2 = len(text1), len(text2) if len1 > len2: text1, text2 = text2, text1 # 使用滑动窗口比较 max_sim = 0 for i in range(len(text2) - len(text1) + 1): window = text2[i:i+len(text1)] current_sim = cosine_similarity(text1, window) if current_sim > max_sim: max_sim = current_sim return max_sim

5.2 多语言支持

处理多语言文本需要考虑字符编码和语言特性：

import langid def detect_language(text): return langid.classify(text)[0] def multilingual_similarity(text1, text2): lang1 = detect_language(text1) lang2 = detect_language(text2) if lang1 != lang2: return 0 # 不同语言直接判定不相似 # 根据语言选择适当的处理方法 if lang1 == 'zh': # 中文需要分词 import jieba text1 = ' '.join(jieba.cut(text1)) text2 = ' '.join(jieba.cut(text2)) return cosine_similarity(text1, text2)

6. 评估与调优

6.1 相似度算法的评估指标

建立评估体系对算法性能进行量化：

from sklearn.metrics import precision_recall_curve, auc def evaluate_model(true_labels, pred_scores): precision, recall, thresholds = precision_recall_curve(true_labels, pred_scores) pr_auc = auc(recall, precision) # 找到最佳阈值 f1_scores = 2 * (precision * recall) / (precision + recall + 1e-10) best_idx = np.argmax(f1_scores) best_threshold = thresholds[best_idx] return { 'pr_auc': pr_auc, 'best_threshold': best_threshold, 'best_f1': f1_scores[best_idx] }

6.2 算法融合与集成

结合多种算法提升效果：

class HybridSimilarity: def __init__(self): self.weights = { 'cosine': 0.4, 'edit': 0.3, 'ngram': 0.3 } def similarity(self, text1, text2): scores = { 'cosine': cosine_similarity(text1, text2), 'edit': normalized_edit_similarity(text1, text2), 'ngram': ngram_similarity(text1, text2) } return sum(w * scores[k] for k, w in self.weights.items())

7. 实际案例分析

7.1 网页内容相似度检测

检测两个网页内容的相似度：

import requests from bs4 import BeautifulSoup def get_webpage_text(url): try: response = requests.get(url, timeout=5) soup = BeautifulSoup(response.text, 'html.parser') # 移除脚本和样式 for script in soup(["script", "style"]): script.decompose() return ' '.join(soup.stripped_strings) except: return "" def webpage_similarity(url1, url2): text1 = get_webpage_text(url1) text2 = get_webpage_text(url2) if not text1 or not text2: return 0 # 使用混合相似度 hybrid = HybridSimilarity() return hybrid.similarity(text1, text2)

7.2 代码相似度检测

对于编程题目，需要特殊的处理方法：

import ast from io import StringIO import tokenize def normalize_code(code): """标准化代码：移除注释、标准化变量名等""" try: tree = ast.parse(code) # 标准化变量名 for node in ast.walk(tree): if isinstance(node, ast.Name) and not isinstance(node.ctx, ast.Load): node.id = 'var' # 重新生成代码 normalized = ast.unparse(tree) # 移除注释 tokens = [] for tok in tokenize.generate_tokens(StringIO(normalized).readline): if tok.type != tokenize.COMMENT: tokens.append(tok) return tokenize.untokenize(tokens) except: return code def code_similarity(code1, code2): norm1 = normalize_code(code1) norm2 = normalize_code(code2) # 使用编辑距离更适合代码比较 return normalized_edit_similarity(norm1, norm2)

8. 性能优化实战技巧

缓存预处理结果：

from functools import lru_cache @lru_cache(maxsize=1000) def cached_similarity(text1, text2): return cosine_similarity(text1, text2)

使用更高效的数据结构：

import numpy as np from datasketch import MinHash, MinHashLSH # 创建MinHash对象 def create_minhash(text, num_perm=128): mh = MinHash(num_perm=num_perm) for word in text.split(): mh.update(word.encode('utf8')) return mh # 建立LSH索引 lsh = MinHashLSH(threshold=0.5, num_perm=128) for idx, text in enumerate(text_collection): mh = create_minhash(text) lsh.insert(idx, mh)

并行计算：

from multiprocessing import Pool def parallel_similarity(args): i, j, text1, text2 = args return (i, j, cosine_similarity(text1, text2)) def batch_compute(texts, workers=4): pool = Pool(workers) tasks = [(i, j, texts[i], texts[j]) for i in range(len(texts)) for j in range(i+1, len(texts))] return pool.map(parallel_similarity, tasks)

9. 常见问题与解决方案

9.1 内存不足问题

当处理大规模文本时：

使用生成器而非列表
采用流式处理
使用更紧凑的数据表示

def stream_similarity(text_stream, query): query_vec = sentence_vector(query) for text in text_stream: text_vec = sentence_vector(text) yield np.dot(query_vec, text_vec) / (np.linalg.norm(query_vec) * np.linalg.norm(text_vec))

9.2 精度与效率的权衡

根据场景需求调整算法：

def adaptive_similarity(text1, text2, fast_threshold=100): len1, len2 = len(text1), len(text2) if len1 < fast_threshold and len2 < fast_threshold: # 短文本使用精确算法 return normalized_edit_similarity(text1, text2) else: # 长文本使用近似算法 mh1 = create_minhash(text1) mh2 = create_minhash(text2) return mh1.jaccard(mh2)

9.3 处理特殊字符和格式

清洗文本数据：

import re def clean_text(text): # 移除HTML标签 text = re.sub(r'<[^>]+>', '', text) # 移除特殊字符 text = re.sub(r'[^\w\s]', '', text) # 标准化空白字符 text = re.sub(r'\s+', ' ', text).strip() return text.lower()

10. 前沿技术与未来方向

基于Transformer的相似度计算：

from sentence_transformers import SentenceTransformer model = SentenceTransformer('all-MiniLM-L6-v2') def transformer_similarity(text1, text2): emb1 = model.encode(text1) emb2 = model.encode(text2) return np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2))