1. 从评分矩阵到用户分群NMF 聚类到底在解决什么问题Non-negative Matrix Factorization非负矩阵分解NMF做 Clustering 的核心思路是把一个非负的大矩阵 V 拆成两个非负小矩阵 W 和 H 的乘积即 V ≈ WH。放到推荐或用户行为场景里V 是 m 个用户对 n 个物品的评分或行为矩阵W 是 n×r 的基向量矩阵H 是 r×m 的软分配权重矩阵。r 就是你想要的聚类个数H 的每一列代表某个用户属于各个簇的权重分布。它和 K-means 最大的区别在于K-means 给每个样本一个硬标签一个用户只能属于一个簇NMF 给的是软标签一个用户可以 60% 属于价格敏感型、30% 属于高频复购型。这种软分配在用户画像、文本主题聚类里非常实用因为现实中的用户本来就不是非黑即白的。这篇要交付的是一套可复现的工程落地流程用 TaoToken 统一管理实验调用的 Key 和 API 通道把 NMF 聚类脚本的参数配置、结果一致性验证动作全部串起来。适合已经会写 Python、但想把聚类实验从「跑一次看看」变成「每次都能复现」的工程师。下面从环境准备开始一步步跑通。2. TaoToken 前置统一 Key 与 API 通道管理做聚类实验时经常遇到一个麻烦脚本里散落着各种模型的调用地址和 Key换一台机器就要重新配一遍实验记录也没法追溯到底用的是哪个通道。TaoToken 在这里的作用是提供一个统一的 API 入口把模型对话、编码辅助、Key 管理都收敛到一套配置里。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。你需要先去控制台创建一个 API Key然后把它写进本地配置文件而不是硬编码在脚本里。具体操作路径打开控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsole_nmf_clusteringutm_campaignrewrite 登录后进入 API Keys 页面。新建一个 Key命名建议带上用途比如nmf-clustering-exp方便后面区分实验。复制 Key 之后不要直接贴进.py文件而是写进config.toml或settings.json用环境变量或配置文件读取。如果你后面要做长期编码或 Agent 类的实验可以了解 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_plan_nmfutm_campaignrewrite 。模型对话调试入口在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodels_nmf_clusteringutm_campaignrewrite 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdoc_nmf_clusteringutm_campaignrewrite 。注意Key 只存在本地配置文件或环境变量里不要提交到 Git 仓库。建议在.gitignore里加上config.toml和settings.json。3. 可复制配置config.toml 与 settings.json 骨架先建一个实验目录结构建议如下nmf-clustering/ ├── config.toml ├── settings.json ├── nmf_cluster.py ├── requirements.txt └── data/ └── ratings.csvconfig.toml负责放 TaoToken 的通道配置和实验级参数[taotoken] api_base https://taotoken.net/api api_key sk-your-key-here default_model gpt-4o-mini timeout_seconds 60 [experiment] name nmf-user-clustering-v1 random_seed 42 n_clusters 5 max_iter 500 tol 1e-4 init_method nndsvd [data] input_path data/ratings.csv user_col user_id item_col item_id value_col ratingsettings.json负责放运行时开关和输出路径方便不改 TOML 就能切换行为{ taotoken: { api_base: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, model: gpt-4o-mini }, nmf: { n_clusters: 5, max_iter: 500, tol: 0.0001, init: nndsvd, beta_loss: frobenius, solver: cd }, output: { model_dir: artifacts/model, report_dir: artifacts/report, save_top_terms: 20 }, verify: { repeat_runs: 3, consistency_threshold: 0.85 } }安装依赖pip install scikit-learn pandas numpy tomli如果你用的是 Python 3.11 以上tomllib已经内置不需要额外装tomli。读取配置的代码片段import json import os import tomllib from pathlib import Path def load_config(config_pathconfig.toml, settings_pathsettings.json): with open(config_path, rb) as f: config tomllib.load(f) with open(settings_path, r, encodingutf-8) as f: settings json.load(f) api_key os.environ.get(settings[taotoken][api_key_env]) if not api_key: api_key config[taotoken][api_key] settings[taotoken][api_key] api_key return config, settings这样 Key 优先从环境变量读读不到再回退到 TOML本地调试和 CI 环境都能兼容。4. NMF 聚类脚本参数配置与完整跑通核心脚本nmf_cluster.py分四步加载数据、构建矩阵、跑 NMF、输出软标签和一致性验证。先看数据加载和矩阵构建import numpy as np import pandas as pd from sklearn.decomposition import NMF from sklearn.preprocessing import normalize def build_matrix(df, user_col, item_col, value_col): pivot df.pivot_table( indexuser_col, columnsitem_col, valuesvalue_col, fill_value0 ) return pivot def run_nmf(matrix, n_clusters, max_iter, tol, init, seed): model NMF( n_componentsn_clusters, initinit, solvercd, beta_lossfrobenius, max_itermax_iter, toltol, random_stateseed ) W model.fit_transform(matrix.values) H model.components_ return model, W, H这里几个参数值得说清楚参数作用建议值n_components聚类个数 r先用 3–8 试配合一致性验证调init初始化方式nndsvd 比 random 稳定推荐默认solver求解器cd 适合中小矩阵mu 适合大数据beta_loss损失函数frobenius 对应标准 NMFmax_iter最大迭代500 起步不收敛再往上加tol收敛阈值1e-4 是常用平衡点跑完之后H 的每一列就是某个用户到各簇的权重。取 argmax 得到硬标签取整列得到软标签def get_labels(H): soft normalize(H.T, norml1, axis1) hard np.argmax(soft, axis1) return soft, hard把结果写回 DataFrame 并保存def save_result(pivot, soft, hard, report_dir): Path(report_dir).mkdir(parentsTrue, exist_okTrue) result pd.DataFrame(soft, indexpivot.index) result.columns [fcluster_{i} for i in range(soft.shape[1])] result[hard_label] hard result.to_csv(f{report_dir}/user_cluster_labels.csv) return result主流程串起来if __name__ __main__: config, settings load_config() df pd.read_csv(config[data][input_path]) pivot build_matrix( df, config[data][user_col], config[data][item_col], config[data][value_col] ) model, W, H run_nmf( pivot, settings[nmf][n_clusters], settings[nmf][max_iter], settings[nmf][tol], settings[nmf][init], config[experiment][random_seed] ) soft, hard get_labels(H) result save_result(pivot, soft, hard, settings[output][report_dir]) print(freconstruction error: {model.reconstruction_err_:.4f}) print(result.head())跑一次看输出reconstruction_err_是重构误差越小说明 WH 越接近原始 V。但这个值不能单独用来判断聚类好坏还要看下面的一致性验证。5. 验证请求与成功结果一致性验证动作NMF 有个坑不同随机种子跑出来的簇编号可能不一样甚至簇的划分本身会漂移。所以「跑通」不等于「可复现」必须加一致性验证。做法是固定数据、固定参数只改随机种子跑多次然后比较两次硬标签的匹配程度。用调整兰德指数Adjusted Rand Index衡量两次聚类结果的一致性from sklearn.metrics import adjusted_rand_score def consistency_check(pivot, settings, config): labels_list [] for i in range(settings[verify][repeat_runs]): seed config[experiment][random_seed] i _, _, H run_nmf( pivot, settings[nmf][n_clusters], settings[nmf][max_iter], settings[nmf][tol], settings[nmf][init], seed ) _, hard get_labels(H) labels_list.append(hard) scores [] for i in range(len(labels_list)): for j in range(i 1, len(labels_list)): score adjusted_rand_score(labels_list[i], labels_list[j]) scores.append(score) print(frun{i} vs run{j}: ARI {score:.4f}) mean_score np.mean(scores) threshold settings[verify][consistency_threshold] print(fmean ARI {mean_score:.4f}, threshold {threshold}) if mean_score threshold: print(WARNING: consistency below threshold, consider nndsvd init or more iterations) else: print(PASS: clustering is reproducible) return mean_score实测下来用nndsvd初始化时ARI 通常能到 0.9 以上用纯随机初始化时可能只有 0.6 左右。如果 ARI 低于阈值优先换初始化方式其次加max_iter再不行就说明数据本身簇结构不明显需要重新考虑 r 的取值。成功跑通的标志有三个重构误差稳定在合理范围、多次运行 ARI 高于阈值、输出的user_cluster_labels.csv里软标签每行和为 1。三个都满足这次 NMF 聚类实验就算可复现了。6. 本篇常见错排查报错一ValueError: Input contains negative valuesNMF 要求输入矩阵非负。检查你的评分数据里有没有负数或者 pivot 之后有没有 NaN 被当成负值处理。用df[value_col].min()确认一下有负数就做平移或截断。报错二ConvergenceWarning: Maximum number of iterations reached迭代没收敛。先把max_iter从 500 加到 1000同时把tol放宽到 1e-3 试试。如果还不收敛检查矩阵是不是太稀疏稀疏度太高时 NMF 本身就不容易稳定。报错三每次跑出来的簇编号对不上这是正常现象NMF 的簇编号没有固定顺序。不要直接比较两次的hard_label数值要用 ARI 这类指标比较划分结构。如果你需要固定编号可以在跑完后按簇中心排序重新映射。报错四tomllib导入失败Python 3.11 以下没有内置tomllib装tomli然后改成import tomli as tomllib。或者干脆把配置全放settings.json少一个依赖。报错五API Key 读取为空检查环境变量名和settings.json里的api_key_env是否一致。用echo $TAOTOKEN_API_KEY确认环境变量真的注入了。如果是在 IDE 里跑注意 IDE 可能不继承 shell 的环境变量需要在运行配置里手动加。7. 把实验通道固定下来NMF 聚类本身不复杂难的是让每次实验都能复现、每个 Key 都能追溯。把 TaoToken 的 API 通道写进config.toml把实验参数写进settings.json再用 ARI 做一致性验证这套流程跑顺之后换数据集、换 r 值、换初始化方式都只是改配置的事。如果你在接入过程中遇到 Key 或通道问题直接看接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdoc_nmf_clusteringutm_campaignrewrite 。需要管理多个实验的 Key去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keys_nmf_clusteringutm_campaignrewrite 。想先验证模型对话通道是否通用模型对话入口 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodels_nmf_clusteringutm_campaignrewrite 。长期做编码和 Agent 实验的话Coding Plan 在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_plan_nmfutm_campaignrewrite 。最后留一个实用技巧把consistency_check的 mean ARI 写进实验报告文件和reconstruction_err_一起存到artifacts/report/metrics.json。下次换参数时直接对比这两个数比凭感觉判断靠谱得多。