本文目录导读：
的K-means聚类,以Python为例。
基本实现步骤
安装必要的库
pip install scikit-learn pandas numpy jieba # 中文分词
完整实现脚本
import os
import jieba
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt
class FileContentCluster:
def __init__(self, file_paths, n_clusters=3):
self.file_paths = file_paths
self.n_clusters = n_clusters
self.documents = []
self.filenames = []
def load_files(self):
"""加载文件内容"""
for file_path in self.file_paths:
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
self.documents.append(content)
self.filenames.append(os.path.basename(file_path))
except Exception as e:
print(f"Error loading {file_path}: {e}")
def preprocess_text(self, text):
"""文本预处理（中文分词）"""
# 使用jieba分词
words = jieba.cut(text)
# 过滤停用词（可选）
stop_words = set(['的', '了', '是', '在', '和', '有', '就', '不', '人', '都'])
filtered_words = [word for word in words if word not in stop_words and len(word) > 1]
return ' '.join(filtered_words)
def vectorize_texts(self):
"""将文本向量化"""
# 预处理所有文档
processed_docs = [self.preprocess_text(doc) for doc in self.documents]
# TF-IDF向量化
self.vectorizer = TfidfVectorizer(max_features=1000)
self.X = self.vectorizer.fit_transform(processed_docs)
return self.X
def perform_clustering(self):
"""执行K-means聚类"""
self.kmeans = KMeans(
n_clusters=self.n_clusters,
random_state=42,
n_init=10
)
self.labels = self.kmeans.fit_predict(self.X)
return self.labels
def visualize_clusters(self):
"""可视化聚类结果（使用PCA降维）"""
# PCA降维到2维
pca = PCA(n_components=2)
X_pca = pca.fit_transform(self.X.toarray())
# 绘制散点图
plt.figure(figsize=(10, 6))
colors = ['red', 'green', 'blue', 'yellow', 'purple']
for i in range(self.n_clusters):
cluster_points = X_pca[self.labels == i]
plt.scatter(
cluster_points[:, 0],
cluster_points[:, 1],
c=colors[i % len(colors)],
label=f'Cluster {i+1}',
alpha=0.6
)
plt.title('File Content Clustering (PCA)')
plt.xlabel('First Principal Component')
plt.ylabel('Second Principal Component')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()
def print_cluster_results(self):
"""打印聚类结果"""
clusters = {}
for i, label in enumerate(self.labels):
if label not in clusters:
clusters[label] = []
clusters[label].append(self.filenames[i])
print("\n=== 聚类结果 ===")
for cluster_id, files in clusters.items():
print(f"\n簇 {cluster_id + 1} ({len(files)}个文件):")
for file_name in files:
print(f" - {file_name}")
def analyze_cluster_centers(self):
"""分析聚类中心关键词"""
print("\n=== 各簇关键词 ===")
order_centroids = self.kmeans.cluster_centers_.argsort()[:, ::-1]
terms = self.vectorizer.get_feature_names_out()
for i in range(self.n_clusters):
print(f"\n簇 {i+1} 的关键词:")
top_keywords = [terms[ind] for ind in order_centroids[i, :10]]
print(f" {', '.join(top_keywords)}")
# 使用示例
def main():
# 示例文件列表（替换为你的实际文件路径）
file_paths = [
"document1.txt",
"document2.txt",
"document3.txt",
# ... 更多文件
]
# 创建聚类器
cluster = FileContentCluster(file_paths, n_clusters=3)
# 执行聚类
cluster.load_files()
cluster.vectorize_texts()
cluster.perform_clustering()
# 输出结果
cluster.print_cluster_results()
cluster.analyze_cluster_centers()
# 可视化（可选）
cluster.visualize_clusters()
if __name__ == "__main__":
# 创建测试文件（演示用）
sample_files = {
"tech_1.txt": "计算机科学 人工智能 机器学习 深度学习 神经网络",
"tech_2.txt": "Python编程 数据结构 算法设计 软件开发",
"sports_1.txt": "足球 篮球 体育比赛 运动健康 健身",
"sports_2.txt": "奥林匹克 世界杯 运动员 训练方法",
"food_1.txt": "烹饪 菜谱 美食 营养 健康饮食",
"food_2.txt": "中餐 西餐 食材 调料 做法"
}
file_paths = []
for name, content in sample_files.items():
with open(name, 'w', encoding='utf-8') as f:
f.write(content)
file_paths.append(name)
# 执行聚类
cluster = FileContentCluster(file_paths, n_clusters=3)
cluster.load_files()
cluster.vectorize_texts()
cluster.perform_clustering()
cluster.print_cluster_results()
cluster.analyze_cluster_centers()
# 清理测试文件
for file_path in file_paths:
os.remove(file_path)
简化版本（函数式）
import os
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
def cluster_files(file_paths, n_clusters=3):
"""简化版文件聚类函数"""
# 读取文件内容
documents = []
filenames = []
for file_path in file_paths:
with open(file_path, 'r', encoding='utf-8') as f:
documents.append(f.read())
filenames.append(os.path.basename(file_path))
# 向量化
vectorizer = TfidfVectorizer(max_features=1000)
X = vectorizer.fit_transform(documents)
# 聚类
kmeans = KMeans(n_clusters=n_clusters, random_state=42)
labels = kmeans.fit_predict(X)
# 输出结果
for i, label in enumerate(labels):
print(f"{filenames[i]} -> 簇 {label + 1}")
return labels
# 使用
file_list = ["file1.txt", "file2.txt", "file3.txt"]
cluster_files(file_list, n_clusters=2)
高级功能扩展
class AdvancedFileCluster(FileContentCluster):
def __init__(self, file_paths, n_clusters=3):
super().__init__(file_paths, n_clusters)
def find_optimal_k(self, max_k=10):
"""使用肘部法则找到最佳K值"""
from sklearn.metrics import silhouette_score
inertias = []
silhouette_scores = []
for k in range(2, max_k + 1):
kmeans = KMeans(n_clusters=k, random_state=42)
kmeans.fit(self.X)
inertias.append(kmeans.inertia_)
if k > 1:
sil_score = silhouette_score(self.X.toarray(), kmeans.labels_)
silhouette_scores.append(sil_score)
# 绘制肘部图
plt.figure(figsize=(12, 4))
plt.subplot(1, 2, 1)
plt.plot(range(2, max_k + 1), inertias, 'bo-')
plt.xlabel('K')
plt.ylabel('Inertia')
plt.title('Elbow Method')
plt.subplot(1, 2, 2)
plt.plot(range(2, max_k + 1), silhouette_scores, 'ro-')
plt.xlabel('K')
plt.ylabel('Silhouette Score')
plt.title('Silhouette Score')
plt.tight_layout()
plt.show()
使用建议
-
文件格式支持：可以扩展支持PDF、Word等格式（使用
pypdf2、python-docx） -
预处理优化：
- 添加更完善的停用词表
- 支持词性标注过滤
- 使用词干提取或词形还原
-
性能优化：
- 使用
MiniBatchKMeans处理大量文件 - 使用稀疏矩阵存储
- 使用
-
结果保存：
import joblib joblib.dump(kmeans, 'cluster_model.pkl')
这个脚本可以根据你的具体需求进行调整,支持中文和英文文本的聚类分析。
