218 lines
8.9 KiB
Python
218 lines
8.9 KiB
Python
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Optional, Tuple
|
|
|
|
import numpy as np
|
|
import matplotlib.pyplot as plt
|
|
|
|
# 尝试可选依赖
|
|
_umap_available = False
|
|
try:
|
|
import umap # type: ignore
|
|
_umap_available = True
|
|
except Exception:
|
|
_umap_available = False
|
|
|
|
from sklearn.decomposition import PCA
|
|
from sklearn.manifold import TSNE
|
|
from sklearn.metrics import silhouette_score, pairwise_distances
|
|
from sklearn.preprocessing import StandardScaler
|
|
|
|
|
|
def load_embeddings_json(path: str) -> np.ndarray:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
emb = np.asarray(data, dtype=np.float32)
|
|
if emb.ndim != 2:
|
|
raise ValueError(f"embeddings 不是二维数组: shape={emb.shape}")
|
|
return emb
|
|
|
|
|
|
def load_labels_json(path: str) -> np.ndarray:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
labels = np.asarray(data, dtype=np.int64)
|
|
if labels.ndim != 1:
|
|
raise ValueError(f"labels 不是一维数组: shape={labels.shape}")
|
|
return labels
|
|
|
|
|
|
def subsample(X: np.ndarray, y: np.ndarray, max_points: Optional[int], seed: int = 42) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
|
|
n = X.shape[0]
|
|
idx = np.arange(n)
|
|
if max_points is not None and n > max_points:
|
|
rng = np.random.default_rng(seed)
|
|
sel = rng.choice(n, size=max_points, replace=False)
|
|
sel = np.sort(sel)
|
|
return X[sel], y[sel], sel
|
|
return X, y, idx
|
|
|
|
|
|
def diagnostics(X: np.ndarray, y: np.ndarray, reduced2d: Optional[np.ndarray] = None, method: str = "pca") -> str:
|
|
# 基本统计
|
|
N, D = X.shape
|
|
means = X.mean(axis=0)
|
|
stds = X.std(axis=0)
|
|
mean_norm = float(np.linalg.norm(means))
|
|
mean_std = float(stds.mean())
|
|
zero_var_ratio = float((stds == 0).mean())
|
|
|
|
# 余弦相似度（抽样，避免 O(N^2)）
|
|
max_pairs = 20000
|
|
rng = np.random.default_rng(0)
|
|
if N > 1:
|
|
sample_size = min(N, int(np.sqrt(max_pairs)) + 1)
|
|
idx = rng.choice(N, size=sample_size, replace=False)
|
|
Xs = X[idx]
|
|
# 归一化以计算余弦
|
|
Xn = Xs / (np.linalg.norm(Xs, axis=1, keepdims=True) + 1e-12)
|
|
cos = Xn @ Xn.T
|
|
iu = np.triu_indices_from(cos, k=1)
|
|
cos_vals = cos[iu]
|
|
cos_mean = float(cos_vals.mean()) if cos_vals.size > 0 else float("nan")
|
|
cos_std = float(cos_vals.std()) if cos_vals.size > 0 else float("nan")
|
|
else:
|
|
cos_mean = float("nan")
|
|
cos_std = float("nan")
|
|
|
|
# silhouette（类内/类间可分性），需要至少2类且每类>=2个点
|
|
sil = float("nan")
|
|
unique, counts = np.unique(y, return_counts=True)
|
|
if unique.size >= 2 and (counts >= 2).all():
|
|
# 为避免过大计算，对子集做
|
|
subN = min(N, 5000)
|
|
if N > subN:
|
|
idx = rng.choice(N, size=subN, replace=False)
|
|
Xs, ys = X[idx], y[idx]
|
|
else:
|
|
Xs, ys = X, y
|
|
# 用欧氏距离
|
|
try:
|
|
sil = float(silhouette_score(Xs, ys, metric="euclidean"))
|
|
except Exception:
|
|
sil = float("nan")
|
|
|
|
# PCA前两维解释率（若提供reduced2d且method=pca，则更有意义）
|
|
pca_explained_2 = float("nan")
|
|
try:
|
|
pca = PCA(n_components=2, random_state=0)
|
|
pca.fit(X)
|
|
pca_explained_2 = float(pca.explained_variance_ratio_.sum())
|
|
except Exception:
|
|
pass
|
|
|
|
lines = [
|
|
f"N={N}, D={D}, 类别数={unique.size}, 每类样本数范围=[{counts.min() if counts.size>0 else 0}, {counts.max() if counts.size>0 else 0}]",
|
|
f"均值向量范数={mean_norm:.6f}, 维度平均标准差={mean_std:.6f}, 零方差维度占比={zero_var_ratio*100:.2f}%",
|
|
f"余弦相似度: 均值={cos_mean:.6f}, 标准差={cos_std:.6f} (越接近1且方差小，越可能坍塌)",
|
|
f"Silhouette 分数={sil:.4f} (接近0或为负，类间不可分)",
|
|
f"PCA前两维累计解释率={pca_explained_2*100:.2f}%",
|
|
f"降维方法={method}",
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def reduce_dim(X: np.ndarray, method: str, seed: int, tsne_perplexity: int, umap_n_neighbors: int, umap_min_dist: float) -> np.ndarray:
|
|
if method == "pca":
|
|
# 标准化后做PCA，通常更稳定
|
|
Xs = StandardScaler(with_mean=True, with_std=True).fit_transform(X)
|
|
comp = PCA(n_components=2, random_state=seed)
|
|
Z = comp.fit_transform(Xs)
|
|
return Z
|
|
elif method == "tsne":
|
|
# t-SNE 对规模敏感，建议先PCA到50维再t-SNE
|
|
d = min(50, X.shape[1])
|
|
Xp = PCA(n_components=d, random_state=seed).fit_transform(X)
|
|
ts = TSNE(n_components=2, perplexity=tsne_perplexity, learning_rate="auto", init="pca", random_state=seed, n_iter=1000)
|
|
Z = ts.fit_transform(Xp)
|
|
return Z
|
|
elif method == "umap":
|
|
if not _umap_available:
|
|
raise RuntimeError("未安装 umap-learn，请先: pip install umap-learn")
|
|
reducer = umap.UMAP(n_components=2, n_neighbors=umap_n_neighbors, min_dist=umap_min_dist, random_state=seed, metric="euclidean")
|
|
Z = reducer.fit_transform(X)
|
|
return Z
|
|
else:
|
|
raise ValueError(f"未知降维方法: {method}")
|
|
|
|
|
|
def plot_2d(Z: np.ndarray, y: np.ndarray, title: str, out_path: Optional[str] = None) -> None:
|
|
plt.figure(figsize=(8, 7), dpi=120)
|
|
classes = np.unique(y)
|
|
# 构建调色板
|
|
cmap = plt.get_cmap("tab20")
|
|
colors = [cmap(i % 20) for i in range(len(classes))]
|
|
|
|
for i, c in enumerate(classes):
|
|
mask = (y == c)
|
|
plt.scatter(Z[mask, 0], Z[mask, 1], s=8, c=[colors[i]], label=str(c), alpha=0.7, edgecolors='none')
|
|
|
|
plt.title(title)
|
|
plt.xlabel("Dim 1")
|
|
plt.ylabel("Dim 2")
|
|
if len(classes) <= 20:
|
|
plt.legend(markerscale=2, frameon=True, fontsize=8)
|
|
plt.tight_layout()
|
|
if out_path:
|
|
plt.savefig(out_path, bbox_inches="tight")
|
|
plt.show()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="可视化高维 embedding 并进行坍塌诊断")
|
|
parser.add_argument("--embeddings", type=str, default=os.path.join("DishClassification/faiss_index", "embeddings.json"), help="embeddings.json 路径")
|
|
# parser.add_argument("--embeddings", type=str, default=os.path.join("WholeIngredientRecognition/faiss_index", "embeddings.json"), help="embeddings.json 路径")
|
|
# parser.add_argument("--embeddings", type=str, default=os.path.join("ProcessedIngredientRecognition/faiss_index", "embeddings.json"), help="embeddings.json 路径")
|
|
parser.add_argument("--labels", type=str, default=os.path.join("DishClassification/faiss_index", "labels.json"), help="labels.json 路径")
|
|
# parser.add_argument("--labels", type=str, default=os.path.join("WholeIngredientRecognition/faiss_index", "labels.json"), help="labels.json 路径")
|
|
# parser.add_argument("--labels", type=str, default=os.path.join("ProcessedIngredientRecognition/faiss_index", "labels.json"), help="labels.json 路径")
|
|
parser.add_argument("--method", type=str, default="pca", choices=["pca", "tsne", "umap"], help="降维方法")
|
|
parser.add_argument("--seed", type=int, default=42)
|
|
parser.add_argument("--max_points", type=int, default=None, help="抽样上限，避免t-SNE/UMAP过慢；None为全量")
|
|
parser.add_argument("--tsne_perplexity", type=int, default=30, help="t-SNE perplexity")
|
|
parser.add_argument("--umap_n_neighbors", type=int, default=15, help="UMAP n_neighbors")
|
|
parser.add_argument("--umap_min_dist", type=float, default=0.1, help="UMAP min_dist")
|
|
parser.add_argument("--no_center_scale", action="store_true", help="禁用标准化（默认PCA会标准化）")
|
|
args = parser.parse_args()
|
|
|
|
if not os.path.exists(args.embeddings):
|
|
print(f"找不到 embeddings: {args.embeddings}", file=sys.stderr)
|
|
sys.exit(1)
|
|
if not os.path.exists(args.labels):
|
|
print(f"找不到 labels: {args.labels}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print("加载数据...")
|
|
X = load_embeddings_json(args.embeddings)
|
|
y = load_labels_json(args.labels)
|
|
|
|
if X.shape[0] != y.shape[0]:
|
|
n = min(X.shape[0], y.shape[0])
|
|
print(f"警告: embeddings与labels数量不一致, 截断到最小长度 {n}。", file=sys.stderr)
|
|
X = X[:n]
|
|
y = y[:n]
|
|
|
|
# 抽样（t-SNE/UMAP 推荐抽样）
|
|
Xs, ys, sel_idx = subsample(X, y, args.max_points, seed=args.seed)
|
|
if Xs.shape[0] < X.shape[0]:
|
|
print(f"已抽样: {Xs.shape[0]}/{X.shape[0]}")
|
|
|
|
# 降维
|
|
print(f"降维方法: {args.method}")
|
|
Z = reduce_dim(Xs, args.method, args.seed, args.tsne_perplexity, args.umap_n_neighbors, args.umap_min_dist)
|
|
|
|
# 诊断
|
|
print("\n==== 诊断信息 ====")
|
|
print(diagnostics(Xs, ys, reduced2d=Z, method=args.method))
|
|
|
|
# 绘图
|
|
out_png = os.path.join(os.path.dirname(args.embeddings), f"embedding_{args.method}_2d.png")
|
|
title = f"Embedding {args.method.upper()} 2D (N={Xs.shape[0]})"
|
|
plot_2d(Z, ys, title, out_path=out_png)
|
|
print(f"已保存图像: {out_png}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |