本文目录导读：
Python 实现文本均值漂移聚类
import numpy as np
from sklearn.cluster import MeanShift
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import StandardScaler
class TextMeanShiftCluster:
def __init__(self, bandwidth=None):
self.vectorizer = TfidfVectorizer(
max_features=1000,
stop_words='english'
)
self.cluster = MeanShift(bandwidth=bandwidth)
self.scaler = StandardScaler()
def read_file(self, filepath):
"""读取文件内容"""
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
return content
def process_lines(self, filepath):
"""按行处理文件内容"""
lines = []
with open(filepath, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line:
lines.append(line)
return lines
def process_documents(self, filepath, delimiter='\n\n'):
"""按段落处理文件内容"""
content = self.read_file(filepath)
documents = content.split(delimiter)
return [doc.strip() for doc in documents if doc.strip()]
def cluster(self, text_data):
"""执行均值漂移聚类"""
# 文本向量化
X = self.vectorizer.fit_transform(text_data).toarray()
# 标准化
X_scaled = self.scaler.fit_transform(X)
# 聚类
labels = self.cluster.fit_predict(X_scaled)
return labels, X_scaled
# 使用示例
def cluster_file_content(filepath):
clusterer = TextMeanShiftCluster()
# 读取文档
documents = clusterer.process_documents(filepath)
if len(documents) < 2:
print("文档数量太少，无法聚类")
return
# 执行聚类
labels, features = clusterer.cluster(documents)
# 输出结果
n_clusters = len(set(labels)) - (1 if -1 in labels else 0)
print(f"聚类数量: {n_clusters}")
for i, (doc, label) in enumerate(zip(documents, labels)):
print(f"文档 {i+1}: 类别 {label}")
# 只显示前50个字符
print(f" 内容预览: {doc[:50]}...")
print()
# 主程序
if __name__ == "__main__":
# 处理单个文件
# cluster_file_content('path/to/your/file.txt')
# 或者完整示例
sample_file = 'sample.txt'
# 创建示例文件
with open(sample_file, 'w', encoding='utf-8') as f:
f.write("Python是一门流行的编程语言\n\n")
f.write("机器学习是人工智能的重要分支\n\n")
f.write("Python的库非常丰富\n\n")
f.write("深度学习是机器学习的一种方法\n\n")
f.write("数据科学需要Python和R语言\n\n")
# 执行聚类
cluster_file_content(sample_file)
基于数值特征的均值漂移聚类
import numpy as np
from sklearn.cluster import MeanShift
import matplotlib.pyplot as plt
class NumericFileCluster:
def __init__(self, bandwidth=None):
self.clusterer = MeanShift(bandwidth=bandwidth)
def read_numeric_data(self, filepath, delimiter=','):
"""读取数值数据"""
data = []
with open(filepath, 'r') as f:
for line in f:
if line.strip() and not line.startswith('#'):
values = [float(x) for x in line.strip().split(delimiter)]
data.append(values)
return np.array(data)
def cluster(self, data):
"""执行聚类"""
labels = self.clusterer.fit_predict(data)
cluster_centers = self.clusterer.cluster_centers_
return labels, cluster_centers
def visualize(self, data, labels):
"""可视化聚类结果（仅适用于2D数据）"""
if data.shape[1] == 2:
plt.scatter(data[:, 0], data[:, 1], c=labels, cmap='viridis')
plt.title('Mean Shift Clustering Result')
plt.colorbar()
plt.show()
else:
print(f"数据维度 {data.shape[1]}，无法可视化")
# 使用示例
def cluster_numeric_file(filepath):
clusterer = NumericFileCluster(bandwidth=0.5)
data = clusterer.read_numeric_data(filepath)
labels, centers = clusterer.cluster(data)
# 输出结果
print(f"数据点数量: {len(data)}")
print(f"聚类中心数量: {len(centers)}")
print(f"聚类标签: {labels}")
return labels, centers
自定义均值漂移算法实现
import numpy as np
from collections import defaultdict
class CustomMeanShift:
def __init__(self, bandwidth=1.0, max_iter=100):
self.bandwidth = bandwidth
self.max_iter = max_iter
def gaussian_kernel(self, distance):
"""高斯核函数"""
return np.exp(-0.5 * (distance / self.bandwidth) ** 2)
def shift_point(self, point, points):
"""计算一个点的均值漂移向量"""
# 计算所有点到当前点的距离
distances = np.linalg.norm(points - point, axis=1)
# 应用核函数
weights = self.gaussian_kernel(distances)
# 计算加权平均
weighted_sum = np.sum(points * weights[:, np.newaxis], axis=0)
weight_sum = np.sum(weights)
return weighted_sum / weight_sum
def fit(self, data):
"""执行均值漂移聚类"""
# 初始化每个点为候选聚类中心
centroids = data.copy()
# 迭代漂移
for iteration in range(self.max_iter):
new_centroids = []
for centroid in centroids:
# 计算新的中心点
new_center = self.shift_point(centroid, data)
new_centroids.append(new_center)
new_centroids = np.array(new_centroids)
# 检查收敛
shift_distance = np.linalg.norm(centroids - new_centroids, axis=1)
if np.max(shift_distance) < 1e-3:
break
centroids = new_centroids
# 合并相近的聚类中心
unique_centroids = self.merge_centroids(centroids)
# 分配标签
labels = self.assign_labels(data, unique_centroids)
self.cluster_centers_ = unique_centroids
self.labels_ = labels
return labels
def merge_centroids(self, centroids, min_distance=0.5):
"""合并相近的聚类中心"""
merged = []
used = set()
for i, c1 in enumerate(centroids):
if i in used:
continue
group = [c1]
used.add(i)
for j, c2 in enumerate(centroids[i+1:], i+1):
if j not in used:
distance = np.linalg.norm(c1 - c2)
if distance < min_distance:
group.append(c2)
used.add(j)
# 计算组内平均作为新的中心
merged.append(np.mean(group, axis=0))
return np.array(merged)
def assign_labels(self, data, centroids):
"""为每个数据点分配标签"""
labels = []
for point in data:
# 计算到所有中心的距离
distances = [np.linalg.norm(point - c) for c in centroids]
# 分配最近的标签
label = np.argmin(distances)
labels.append(label)
return np.array(labels)
# 文本特征提取的自定义实现
class TextFeatureExtractor:
def __init__(self, max_features=1000):
self.max_features = max_features
self.vocabulary = {}
def extract_features(self, texts):
"""提取文本特征（词袋模型）"""
# 构建词汇表
word_freq = defaultdict(int)
for text in texts:
words = text.lower().split()
for word in words:
if len(word) > 2: # 过滤短词
word_freq[word] += 1
# 选择最频繁的词
sorted_words = sorted(word_freq.items(),
key=lambda x: x[1],
reverse=True)[:self.max_features]
self.vocabulary = {word: idx for idx, (word, _) in enumerate(sorted_words)}
# 构建特征矩阵
features = []
for text in texts:
vector = np.zeros(len(self.vocabulary))
words = text.lower().split()
for word in words:
if word in self.vocabulary:
vector[self.vocabulary[word]] += 1
features.append(vector)
return np.array(features)
# 完整的使用示例
def complete_example():
# 创建示例数据
sample_texts = [
"Python编程语言很流行",
"Java也是一种主流语言",
"JavaScript用于网页开发",
"深度学习需要大量数据",
"机器学习是AI的核心",
"数据科学包含统计知识"
]
# 提取特征
extractor = TextFeatureExtractor(max_features=100)
features = extractor.extract_features(sample_texts)
# 自定义均值漂移聚类
custom_ms = CustomMeanShift(bandwidth=0.8)
labels = custom_ms.fit(features)
# 输出结果
for text, label in zip(sample_texts, labels):
print(f"文本: {text} -> 类别: {label}")
# 运行示例
if __name__ == "__main__":
complete_example()
命令行工具版本
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import argparse
import numpy as np
from sklearn.cluster import MeanShift
def main():
parser = argparse.ArgumentParser(description='文件内容均值漂移聚类')
parser.add_argument('file', help='输入文件路径')
parser.add_argument('--bandwidth', type=float, default=None,
help='聚类带宽参数')
parser.add_argument('--delimiter', default='\n\n',
help='文档分隔符')
parser.add_argument('--max-features', type=int, default=100,
help='最大特征数量')
parser.add_argument('--output', help='输出文件路径')
args = parser.parse_args()
# 这里可以添加处理逻辑
print(f"处理文件: {args.file}")
print(f"带宽参数: {args.bandwidth or 'auto'}")
if __name__ == "__main__":
main()
注意事项
- 数据预处理：文本数据需要适当的预处理（分词、去停用词等）
- 参数调整：带宽(bandwidth)是均值漂移的关键参数
- 性能考虑：大规模数据时需要注意内存使用
- 特征选择：合适的特征提取方法对聚类效果很重要
根据你的具体需求选择合适的实现方式,可以配合其他自然语言处理技术提升聚类效果。
