NLP实践：如何从问题列表高效提取常见问题(FAQ)及方案优化

需求说明 #

现有6万余条以字符串形式存储的问题列表，目标是从原始列表中提取得到高频常见问题（FAQ）列表。

原有实现方案与现存问题 #

原有方案核心逻辑 #

• 遍历列表中每个问题，计算其与列表内其余所有问题（排除自身）的余弦相似度
• 构建字典存储匹配结果：字典键为当前处理问题的索引，对应值为与该问题余弦相似度高于设定阈值的所有问题的索引列表
• 字典构建完成后，将对应值列表长度最长的问题判定为高频问题，可按需选取TopN（如Top10）结果作为最终FAQ列表

现存问题 #

• 执行效率极低：6万+数据量按现有方案跑完全流程预计需要14天，完全无法落地
• 方案合理性存疑：不确定当前思路是否是该场景的最优解法
• 需要可复用的高性能实现思路，供遇到同类问题的开发者参考

原有实现代码 #

import nltk
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
nltk.download('punkt')
nltk.download('stopwords')

list_of_questions = ['How does umap know which high dimensional datapoint belongs to which cluster?',...]

score = dict()
threshold = 0.7
#tokenization

#sw contains the list of stopwords
sw = stopwords.words('english')

for index, main_question in enumerate(list_of_questions):
    similarities = []
    temp_list = list_of_questions.copy()
    X_list = word_tokenize(main_question)
    temp_list.pop(index)
    for question_ in temp_list:
        l1 =[];l2 =[]
        Y_list = word_tokenize(question_)
        
        if len(X_list) == 0 or len(Y_list) == 0:
            continue
        #remove stop words from the string
        X_set = {w for w in X_list if not w in sw} 
        Y_set = {w for w in Y_list if not w in sw}

        #form a set containing keywords of both strings 
        rvector = X_set.union(Y_set) 
        for w in rvector:
            if w in X_set: l1.append(1) # create a vector
            else: l1.append(0)
            if w in Y_set: l2.append(1)
            else: l2.append(0)
        c = 0

        #cosine formula 
        try:
            for i in range(len(rvector)):
                    c+= l1[i]*l2[i]
            cosine = c / float((sum(l1)*sum(l2))**0.5)
            if cosine > threshold:
                similarities.append(list_of_questions.index(question_))
                print("Cosine similarity: ", cosine)
        except:
            continue
    score[index] = similarities

性能瓶颈原因分析 #

原代码跑满14天的核心原因是存在多层效率问题，不是简单优化循环就能解决的：

• 算法复杂度高：采用O(n²)的全量两两比对逻辑，6万条数据需要完成近36亿次比对，计算量本身就极大
• 大量重复计算：每一次两两比对时，都会重新对两个问题做分词、去停用词、构建联合词表、生成one-hot向量，相同的预处理操作被重复执行了数万次
• 冗余操作过多：每次外层循环都完整复制6万长度的问题列表、弹出当前元素，额外消耗大量内存和时间
• API使用低效：list.index()方法是线性扫描实现，每次查找索引都要遍历一遍列表，在亿级调用场景下会带来巨量额外耗时
• 计算效率低：纯Python手写循环做向量运算，比numpy、sklearn这类基于C实现的向量化计算效率低100倍以上
• 逻辑存在漏洞：相似关系是双向的，原方案会把同一组相似问题重复计数两次，还会因为list.index()的特性在列表存在重复问题时返回错误索引

优化实现方案 #

整体优化思路是把重复操作提前批量完成，用成熟的近邻检索算法替换全量两两比对，最后通过聚类归并避免重复计数，全流程在普通PC上跑6万条数据只需要几十秒到几分钟。

第一步：批量预处理，一次性生成全量问题的向量表示 #

不要在比对阶段才做文本预处理，先一次性跑完所有文本的特征提取：

• 批量对所有问题做转小写、分词、去停用词操作，每个问题只处理一次
• 根据准确率需求选择向量生成方式：
  • 追求极致速度：用TF-IDF生成词袋稀疏向量，不需要额外模型，速度最快
  • 追求语义匹配准确率：用轻量句向量模型（如all-MiniLM-L6-v2）生成稠密语义向量，能识别不同表述的同义问题

第二步：用近邻检索替换全量两两比对 #

构建向量索引后批量检索每个问题的相似项，把时间复杂度从O(n²)降到O(nlogn)：

• 针对TF-IDF稀疏向量，直接用sklearn的NearestNeighbors组件，指定余弦距离度量，开启多线程并行计算
• 针对语义稠密向量，可用faiss构建向量索引，检索速度更快
• 检索完成后批量过滤掉相似度低于阈值、以及问题自身的匹配项，直接生成每个问题对应的相似问题索引列表

第三步：聚类归并生成最终FAQ列表 #

原方案的计数逻辑会重复统计同一类问题，通过连通域检测把互相相似的问题归为同一个簇：

• 遍历所有相似关系，用广度优先搜索找到所有连通的相似问题，归为一个簇
• 按簇内问题数量倒序排序，簇规模越大代表对应问题的出现频率越高
• 每个簇选代表性问题（比如距离簇中心最近的问题、簇内最早出现的问题）作为标准FAQ，取TopN即可得到最终FAQ列表

优化后参考代码 #

import nltk
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.neighbors import NearestNeighbors
from collections import defaultdict

nltk.download('punkt')
nltk.download('stopwords')
sw = stopwords.words('english')

# 自定义分词预处理函数，所有文本只需要处理一次
def tokenize(text):
    tokens = word_tokenize(text.lower())
    return [t for t in tokens if t not in sw and t.isalpha()]

list_of_questions = ['How does umap know which high dimensional datapoint belongs to which cluster?', ...]
threshold = 0.7

# 1. 一次性生成全量问题的TF-IDF向量
vectorizer = TfidfVectorizer(tokenizer=tokenize)
tfidf_matrix = vectorizer.fit_transform(list_of_questions)

# 2. 构建近邻检索索引，批量查询所有相似项
nn = NearestNeighbors(metric='cosine', n_jobs=-1) # n_jobs=-1开启全核心并行
nn.fit(tfidf_matrix)

# 批量检索所有近邻，返回余弦距离和对应索引，可根据数据分布调整n_neighbors大小
distances, indices = nn.kneighbors(tfidf_matrix, n_neighbors=min(1000, len(list_of_questions)))

score = defaultdict(list)
for idx in range(len(list_of_questions)):
    # 余弦相似度 = 1 - 余弦距离，过滤低于阈值的项，排除自身（第一个匹配项）
    similar_mask = (1 - distances[idx]) > threshold
    similar_indices = indices[idx][similar_mask][1:]
    score[idx] = similar_indices.tolist()

# 3. 连通域聚类归并，统计高频FAQ
visited = set()
clusters = []
for q_idx in range(len(list_of_questions)):
    if q_idx in visited:
        continue
    cluster = []
    stack = [q_idx]
    visited.add(q_idx)
    while stack:
        cur = stack.pop()
        cluster.append(cur)
        for neighbor in score[cur]:
            if neighbor not in visited:
                visited.add(neighbor)
                stack.append(neighbor)
    clusters.append(cluster)

# 按簇规模倒序排序，取Top10作为FAQ结果
clusters.sort(key=lambda x: len(x), reverse=True)
faq_result = []
for cluster in clusters[:10]:
    # 取簇内第一个问题作为FAQ代表，可替换为距离簇中心最近的问题
    faq_result.append({
        "faq": list_of_questions[cluster[0]],
        "count": len(cluster)
    })

如果需要更高的语义匹配准确率，只需要把TF-IDF向量替换为句向量即可，后续检索、聚类逻辑完全通用。

内容的提问来源于stack exchange，提问作者Shodai Thox