在数据可视化和新闻分析领域将新闻周期动态映射到空间布局是一个极具挑战性的任务。传统新闻聚合器通常按时间或主题分类但很少能直观展示新闻事件之间的语义关联和集群效应。一个能够每小时从新闻标题中重建语义地图的系统需要结合现代自然语言处理、向量数据库和降维可视化技术栈。这类系统通常面向数据分析师、新闻研究者、产品经理或任何需要宏观把握新闻趋势的开发者。通过本文你将理解如何构建一个从新闻标题抓取、文本向量化、语义聚类到交互式地图渲染的完整流程。最终产出是一个类似“新闻星系图”的可视化界面每个新闻簇代表一个热点事件簇的大小反映事件热度距离反映语义相关性。1. 理解新闻语义地图的技术栈与核心机制新闻语义地图的核心思想是将非结构化的文本数据转化为结构化的空间坐标。整个过程涉及四个关键阶段文本嵌入、向量索引、聚类分析和可视化映射。1.1 文本嵌入从标题文字到高维向量文本嵌入模型如 OpenAI 的 text-embedding-3-large将每个新闻标题转换为一个高维向量例如 1536 维。这个向量捕获了标题的语义信息语义相近的标题在向量空间中距离更近。选择嵌入模型时需要考虑嵌入维度、推理速度和语义质量。高维向量虽然包含更多信息但直接处理会面临“维度灾难”需要后续降维。1.2 向量索引快速近似最近邻搜索当新闻标题数量达到数千或数万时精确计算向量间距离变得非常耗时。HNSWHierarchical Navigable Small World算法通过构建分层图结构实现高效近似最近邻搜索。HNSWLib 是一个轻量级库特别适合在内存中处理大规模向量集合支持实时插入和查询。1.3 聚类分析从向量到新闻簇DBSCANDensity-Based Spatial Clustering of Applications with Noise是一种基于密度的聚类算法适合处理不规则形状的簇并能识别噪声点即孤立新闻。与 K-means 需要预设簇数量不同DBSCAN 通过eps邻域半径和min_samples最小样本数参数自动确定簇的数量更适应新闻数据动态变化的特性。1.4 降维可视化从高维到二维坐标UMAPUniform Manifold Approximation and Projection是一种降维算法能够更好地保留高维数据的拓扑结构。相比 t-SNEUMAP 计算速度更快尤其适合大规模数据集。降维后的二维坐标可以直接用于地图渲染。1.5 前端渲染交互式新闻地图Deck.gl 是一个基于 WebGL 的大规模数据可视化库适合渲染成千上万的新闻点及其聚类效果。结合 React 或纯 JavaScript可以构建交互式地图支持缩放、拖拽、点击查看详情等操作。2. 环境准备与依赖配置构建新闻语义地图需要 Python 后端处理数据和 JavaScript 前端渲染可视化。以下环境基于常见项目设置实际版本需根据项目需求调整。2.1 Python 后端环境建议使用 Python 3.9 环境。创建并激活虚拟环境后安装核心依赖python -m venv news_map_env source news_map_env/bin/activate # Linux/Mac # news_map_env\Scripts\activate # Windows pip install requests beautifulsoup4 openai hnswlib umap-learn scikit-learn pandas numpy flask flask-cors关键依赖说明requests和beautifulsoup4用于抓取新闻网站标题openai调用 OpenAI Embedding APIhnswlib构建向量索引库umap-learn执行 UMAP 降维scikit-learn提供 DBSCAN 聚类算法pandas和numpy数据处理flask和flask-cors提供数据 API 并处理跨域2.2 前端环境配置前端项目可以使用现代构建工具如 Vite 初始化npm create vitelatest news-map-frontend -- --template react cd news-map-frontend npm install deck.gl deck.gl/core deck.gl/layers luma.gl/core如果需要更多交互功能可以额外安装地图库npm install react-map-gl maplibre-gl2.3 API 密钥与配置管理使用 OpenAI Embedding API 需要配置 API 密钥。建议通过环境变量管理敏感信息export OPENAI_API_KEYyour-api-key-here # Linux/Mac # set OPENAI_API_KEYyour-api-key-here # Windows在 Python 代码中通过os.environ读取import os openai_api_key os.environ.get(OPENAI_API_KEY)3. 构建新闻抓取与向量化管道新闻语义地图的数据流程始于新闻标题的采集和向量化。这个阶段需要处理多个新闻源确保数据质量和代表性。3.1 新闻源选择与标题提取选择具有代表性的新闻源包括主流媒体、行业媒体和区域媒体。使用 Requests 和 BeautifulSoup 实现简单的爬虫import requests from bs4 import BeautifulSoup import re def fetch_news_titles(source_url, title_selector): 从指定新闻源提取标题 try: response requests.get(source_url, timeout10) response.raise_for_status() soup BeautifulSoup(response.content, html.parser) titles [] # 根据网站结构提取标题元素 title_elements soup.select(title_selector) for element in title_elements: title_text element.get_text().strip() # 清理标题中的多余空格和特殊字符 title_text re.sub(r\s, , title_text) if len(title_text) 10: # 过滤过短标题 titles.append(title_text) return titles except Exception as e: print(f抓取 {source_url} 失败: {e}) return [] # 示例配置多个新闻源 news_sources [ { url: https://example-news.com/latest, selector: .headline a }, { url: https://another-news.site/recent, selector: h3.title } ] all_titles [] for source in news_sources: titles fetch_news_titles(source[url], source[selector]) all_titles.extend(titles) print(f从 {source[url]} 获取 {len(titles)} 个标题)3.2 标题去重与清洗新闻标题经常存在重复或轻微变体需要去重处理def deduplicate_titles(titles, similarity_threshold0.9): 基于简单规则去重标题 unique_titles [] seen_patterns set() for title in titles: # 标准化标题小写、移除标点 normalized re.sub(r[^\w\s], , title.lower()) words normalized.split() # 生成标题指纹取前5个单词 if len(words) 3: fingerprint .join(sorted(words[:5])) else: fingerprint .join(sorted(words)) if fingerprint not in seen_patterns: seen_patterns.add(fingerprint) unique_titles.append(title) return unique_titles cleaned_titles deduplicate_titles(all_titles) print(f去重后剩余 {len(cleaned_titles)} 个唯一标题)3.3 调用 OpenAI Embedding API使用 OpenAI 的 text-embedding-3-large 模型生成标题向量import openai import numpy as np import time def get_embeddings(titles, modeltext-embedding-3-large, batch_size100): 批量获取标题的嵌入向量 embeddings [] for i in range(0, len(titles), batch_size): batch titles[i:i batch_size] try: response openai.embeddings.create( modelmodel, inputbatch ) batch_embeddings [item.embedding for item in response.data] embeddings.extend(batch_embeddings) # 避免速率限制 time.sleep(0.1) except Exception as e: print(f第 {i//batch_size 1} 批处理失败: {e}) # 失败时填充零向量 embeddings.extend([np.zeros(1536)] * len(batch)) return np.array(embeddings) # 获取所有标题的嵌入向量 title_embeddings get_embeddings(cleaned_titles) print(f生成 {title_embeddings.shape[0]} 个标题的嵌入向量维度: {title_embeddings.shape[1]})3.4 构建 HNSW 向量索引使用 HNSWLib 建立高效的向量索引支持快速相似度搜索import hnswlib def build_hnsw_index(embeddings, spacecosine, ef_construction200, M16): 构建 HNSW 向量索引 dim embeddings.shape[1] num_elements embeddings.shape[0] # 初始化索引 index hnswlib.Index(spacespace, dimdim) index.init_index(max_elementsnum_elements, ef_constructionef_construction, MM) # 添加向量 index.add_items(embeddings) # 设置查询参数 index.set_ef(50) return index # 构建索引 vector_index build_hnsw_index(title_embeddings) print(HNSW 向量索引构建完成)4. 实现聚类分析与降维映射获得标题向量后需要将其聚类成有意义的新闻簇并降维到二维空间进行可视化。4.1 DBSCAN 聚类参数调优DBSCAN 对参数敏感需要根据数据特性调整from sklearn.cluster import DBSCAN from sklearn.neighbors import NearestNeighbors import matplotlib.pyplot as plt def find_optimal_eps(embeddings, k5): 使用 k-距离图寻找最优 eps 参数 neighbors NearestNeighbors(n_neighborsk) neighbors_fit neighbors.fit(embeddings) distances, indices neighbors_fit.kneighbors(embeddings) k_distances np.sort(distances[:, k-1], axis0) plt.plot(k_distances) plt.xlabel(数据点排序) plt.ylabel(f{k}-最近邻距离) plt.title(k-距离图) plt.show() # 寻找拐点 gradients np.gradient(k_distances) inflection_point np.argmax(gradients) suggested_eps k_distances[inflection_point] return suggested_eps # 寻找合适的 eps 值 suggested_eps find_optimal_eps(title_embeddings) print(f建议的 eps 参数: {suggested_eps:.4f}) # 执行 DBSCAN 聚类 dbscan DBSCAN(epssuggested_eps, min_samples3, metriccosine) cluster_labels dbscan.fit_predict(title_embeddings) # 分析聚类结果 unique_labels set(cluster_labels) n_clusters len(unique_labels) - (1 if -1 in unique_labels else 0) n_noise list(cluster_labels).count(-1) print(f发现 {n_clusters} 个聚类{n_noise} 个噪声点)4.2 UMAP 降维配置UMAP 降维需要平衡计算效率和结构保持import umap def reduce_dimensionality(embeddings, n_neighbors15, min_dist0.1, n_components2): 使用 UMAP 进行降维 reducer umap.UMAP( n_neighborsn_neighbors, min_distmin_dist, n_componentsn_components, metriccosine, random_state42 ) embedding_2d reducer.fit_transform(embeddings) return embedding_2d # 执行降维 coordinates_2d reduce_dimensionality(title_embeddings) print(f降维后坐标形状: {coordinates_2d.shape})4.3 聚类结果分析与标签生成为每个聚类生成有意义的标签帮助理解新闻主题from collections import Counter import re def generate_cluster_labels(titles, cluster_labels, top_n3): 为每个聚类生成代表性标签 cluster_info {} for cluster_id in set(cluster_labels): if cluster_id -1: continue # 跳过噪声点 cluster_titles [title for i, title in enumerate(titles) if cluster_labels[i] cluster_id] # 提取常见关键词 all_words [] for title in cluster_titles: words re.findall(r\b[a-zA-Z]{4,}\b, title.lower()) # 提取长度≥4的单词 all_words.extend(words) word_freq Counter(all_words) common_words [word for word, count in word_freq.most_common(10) if count 1] # 至少出现2次 cluster_info[cluster_id] { size: len(cluster_titles), titles: cluster_titles[:5], # 前5个标题作为样本 keywords: common_words[:top_n], representative_title: cluster_titles[0] if cluster_titles else } return cluster_info cluster_details generate_cluster_labels(cleaned_titles, cluster_labels) # 输出聚类分析结果 for cluster_id, info in cluster_details.items(): print(f聚类 {cluster_id}: {info[size]} 条新闻, 关键词: {, .join(info[keywords])})5. 构建交互式新闻地图前端使用 Deck.gl 和 React 构建前端可视化界面展示新闻聚类结果。5.1 准备前端数据结构首先将后端处理结果转换为前端可用的格式import json from datetime import datetime def prepare_frontend_data(titles, coordinates_2d, cluster_labels, cluster_details): 准备前端可视化数据 features [] for i, (title, coord, cluster_id) in enumerate(zip(titles, coordinates_2d, cluster_labels)): feature { type: Feature, geometry: { type: Point, coordinates: [float(coord[0]), float(coord[1])] }, properties: { id: i, title: title, clusterId: int(cluster_id), clusterSize: cluster_details.get(cluster_id, {}).get(size, 1), isNoise: cluster_id -1 } } features.append(feature) # 聚类元数据 clusters_meta {} for cluster_id, info in cluster_details.items(): clusters_meta[cluster_id] { keywords: info[keywords], size: info[size], representativeTitle: info[representative_title] } result { type: FeatureCollection, features: features, metadata: { generatedAt: datetime.now().isoformat(), totalPoints: len(features), clustersMeta: clusters_meta } } return result # 生成前端数据 frontend_data prepare_frontend_data(cleaned_titles, coordinates_2d, cluster_labels, cluster_details) # 保存为 JSON 文件 with open(news_map_data.json, w, encodingutf-8) as f: json.dump(frontend_data, f, ensure_asciiFalse, indent2)5.2 React 组件实现创建主要的新闻地图组件import React, { useState, useMemo } from react; import DeckGL from deck.gl/react; import { ScatterplotLayer } from deck.gl/layers; import { StaticMap } from react-map-gl; const MAPBOX_TOKEN your-mapbox-token-here; // 需要注册 Mapbox function NewsMap({ data }) { const [viewport, setViewport] useState({ longitude: 0, latitude: 0, zoom: 1, pitch: 0, bearing: 0 }); const [selectedCluster, setSelectedCluster] useState(null); // 处理点击事件 const onClick (info) { if (info.object) { setSelectedCluster(info.object.properties.clusterId); } }; // 创建散点图层 const scatterplotLayer new ScatterplotLayer({ id: news-points, data: data.features, getPosition: d d.geometry.coordinates, getRadius: d Math.min(10 Math.log(d.properties.clusterSize) * 5, 30), getFillColor: d { if (d.properties.isNoise) return [200, 200, 200, 100]; // 灰色噪声点 const clusterId d.properties.clusterId; // 为不同聚类生成不同颜色 const hue (clusterId * 137.5) % 360; // 黄金角分布 return [hue, 70, 50, 180]; // HSL 颜色 }, getLineColor: [0, 0, 0, 200], getLineWidth: 1, pickable: true, autoHighlight: true, radiusScale: 1, radiusMinPixels: 2, radiusMaxPixels: 30, }); // 筛选选中聚类的新闻 const filteredTitles useMemo(() { if (!selectedCluster) return []; return data.features .filter(f f.properties.clusterId selectedCluster) .slice(0, 10); // 显示前10条 }, [selectedCluster, data]); return ( div style{{ position: relative, height: 100vh }} DeckGL initialViewState{viewport} controller{true} layers{[scatterplotLayer]} onClick{onClick} StaticMap mapboxApiAccessToken{MAPBOX_TOKEN} mapStylemapbox://styles/mapbox/light-v10 / /DeckGL {/* 侧边信息面板 */} div style{{ position: absolute, top: 10, right: 10, width: 300, background: white, padding: 15, borderRadius: 5, boxShadow: 0 2px 10px rgba(0,0,0,0.1), maxHeight: 90vh, overflow: auto }} h3新闻语义地图/h3 p总新闻数: {data.metadata.totalPoints}/p {selectedCluster ! null ( div h4聚类 #{selectedCluster}/h4 p关键词: {data.metadata.clustersMeta[selectedCluster]?.keywords.join(, )}/p div {filteredTitles.map((feature, i) ( div key{i} style{{ padding: 5px 0, borderBottom: 1px solid #eee, fontSize: 12px }} {feature.properties.title} /div ))} /div /div )} /div /div ); } export default NewsMap;5.3 主应用入口创建应用主文件import React from react; import NewsMap from ./components/NewsMap; import newsData from ./data/news_map_data.json; function App() { return ( div classNameApp NewsMap data{newsData} / /div ); } export default App;6. 构建数据更新管道与 API 服务新闻数据需要定期更新以反映最新的新闻周期变化。构建一个完整的更新管道和 API 服务。6.1 自动化更新脚本创建定时任务脚本每小时执行一次数据更新# update_pipeline.py import schedule import time import logging from datetime import datetime # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(news_map_updates.log), logging.StreamHandler() ] ) def full_update_pipeline(): 完整的新闻地图更新管道 start_time datetime.now() logging.info(开始新闻地图更新流程) try: # 1. 抓取新闻标题 titles fetch_news_titles_from_sources() logging.info(f抓取到 {len(titles)} 个原始标题) # 2. 清洗和去重 cleaned_titles deduplicate_titles(titles) logging.info(f去重后剩余 {len(cleaned_titles)} 个标题) # 3. 生成嵌入向量 embeddings get_embeddings(cleaned_titles) logging.info(嵌入向量生成完成) # 4. 聚类分析 cluster_labels perform_clustering(embeddings) logging.info(聚类分析完成) # 5. 降维映射 coordinates_2d reduce_dimensionality(embeddings) logging.info(降维映射完成) # 6. 生成前端数据 frontend_data prepare_frontend_data(cleaned_titles, coordinates_2d, cluster_labels) # 7. 保存结果 with open(data/news_map_latest.json, w, encodingutf-8) as f: json.dump(frontend_data, f, ensure_asciiFalse, indent2) # 8. 备份历史数据 timestamp datetime.now().strftime(%Y%m%d_%H%M) with open(fdata/backups/news_map_{timestamp}.json, w, encodingutf-8) as f: json.dump(frontend_data, f, ensure_asciiFalse, indent2) duration (datetime.now() - start_time).total_seconds() logging.info(f更新流程完成耗时 {duration:.2f} 秒) except Exception as e: logging.error(f更新流程失败: {e}) # 设置定时任务 schedule.every().hour.do(full_update_pipeline) if __name__ __main__: # 立即执行一次 full_update_pipeline() # 启动定时任务 while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次6.2 Flask API 服务提供数据接口供前端调用# app.py from flask import Flask, jsonify, send_file from flask_cors import CORS import os import json from datetime import datetime app Flask(__name__) CORS(app) app.route(/api/news-map) def get_news_map(): 获取最新的新闻地图数据 try: with open(data/news_map_latest.json, r, encodingutf-8) as f: data json.load(f) return jsonify(data) except FileNotFoundError: return jsonify({error: 数据文件不存在}), 404 app.route(/api/update-status) def get_update_status(): 获取数据更新状态 try: stat os.stat(data/news_map_latest.json) last_modified datetime.fromtimestamp(stat.st_mtime) return jsonify({ lastUpdated: last_modified.isoformat(), status: ok }) except FileNotFoundError: return jsonify({error: 数据文件不存在}), 404 app.route(/api/cluster/int:cluster_id) def get_cluster_details(cluster_id): 获取特定聚类的详细信息 try: with open(data/news_map_latest.json, r, encodingutf-8) as f: data json.load(f) cluster_titles [ feature[properties][title] for feature in data[features] if feature[properties][clusterId] cluster_id ] cluster_meta data[metadata][clustersMeta].get(str(cluster_id), {}) return jsonify({ clusterId: cluster_id, titles: cluster_titles, metadata: cluster_meta }) except FileNotFoundError: return jsonify({error: 数据文件不存在}), 404 if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)6.3 前端数据获取优化修改前端组件以支持动态数据加载// hooks/useNewsData.js import { useState, useEffect } from react; export function useNewsData() { const [data, setData] useState(null); const [loading, setLoading] useState(true); const [error, setError] useState(null); useEffect(() { const fetchData async () { try { setLoading(true); const response await fetch(http://localhost:5000/api/news-map); if (!response.ok) throw new Error(数据获取失败); const newsData await response.json(); setData(newsData); } catch (err) { setError(err.message); } finally { setLoading(false); } }; fetchData(); // 每5分钟自动更新数据 const interval setInterval(fetchData, 5 * 60 * 1000); return () clearInterval(interval); }, []); return { data, loading, error }; }7. 性能优化与生产环境部署将系统部署到生产环境需要考虑性能、可靠性和可维护性。7.1 向量处理优化处理大规模新闻数据时需要优化内存使用和计算效率# optimization.py import numpy as np from sklearn.utils import shuffle import gc def optimized_embedding_generation(titles, modeltext-embedding-3-large, max_tokens8191, batch_size50): 优化嵌入向量生成处理长文本和内存使用 embeddings [] for i in range(0, len(titles), batch_size): batch titles[i:i batch_size] # 预处理截断过长的标题 processed_batch [] for title in batch: # 简单截断实际项目可用更智能的截断 if len(title) max_tokens * 4: # 粗略估计 title title[:max_tokens * 4] processed_batch.append(title) try: response openai.embeddings.create( modelmodel, inputprocessed_batch, truncationTrue # 允许API端截断 ) batch_embeddings [item.embedding for item in response.data] embeddings.extend(batch_embeddings) # 手动垃圾回收 if i % 200 0: gc.collect() except Exception as e: print(f批处理失败: {e}) # 使用零向量填充保持数据完整性 embeddings.extend([np.zeros(1536)] * len(processed_batch)) return np.array(embeddings) def incremental_clustering(embeddings, existing_labelsNone, eps0.3, min_samples3): 增量聚类支持在已有聚类基础上添加新数据 if existing_labels is None: # 全新聚类 from sklearn.cluster import DBSCAN clustering DBSCAN(epseps, min_samplesmin_samples, metriccosine) labels clustering.fit_predict(embeddings) else: # 增量处理简化版本实际需要更复杂逻辑 labels np.full(len(embeddings), -1) # 初始化为噪声 # 这里可以实现更复杂的增量聚类逻辑 return labels7.2 数据库持久化对于生产环境建议使用数据库存储新闻数据和向量# database.py import sqlite3 import json import numpy as np from datetime import datetime class NewsVectorDB: def __init__(self, db_pathnews_vectors.db): self.db_path db_path self.init_db() def init_db(self): 初始化数据库表结构 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS news_articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, source_url TEXT, published_at DATETIME, embedding BLOB, cluster_id INTEGER, coordinates_json TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ) cursor.execute( CREATE TABLE IF NOT EXISTS update_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, update_type TEXT, items_processed INTEGER, success BOOLEAN, error_message TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def store_news_batch(self, news_items): 批量存储新闻数据 conn sqlite3.connect(self.db_path) cursor conn.cursor() for item in news_items: embedding_blob item[embedding].tobytes() if embedding in item else None coordinates_json json.dumps(item.get(coordinates, [])) cursor.execute( INSERT INTO news_articles (title, source_url, published_at, embedding, cluster_id, coordinates_json) VALUES (?, ?, ?, ?, ?, ?) , ( item[title], item.get(source_url), item.get(published_at), embedding_blob, item.get(cluster_id, -1), coordinates_json )) conn.commit() conn.close()7.3 Docker 容器化部署创建 Dockerfile 实现一键部署# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装 Python 依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建数据目录 RUN mkdir -p data/backups # 暴露端口 EXPOSE 5000 # 启动命令 CMD [python, app.py]对应的 docker-compose.ymlversion: 3.8 services: news-map-api: build: . ports: - 5000:5000 environment: - OPENAI_API_KEY${OPENAI_API_KEY} volumes: - ./data:/app/data restart: unless-stopped news-updater: build: . command: python update_pipeline.py environment: - OPENAI_API_KEY${OPENAI_API_KEY} volumes: - ./data:/app/data restart: unless-stopped8. 常见问题排查与优化建议在实际部署和运行过程中可能会遇到各种问题以下是典型问题及解决方案。8.1 API 限制与错误处理OpenAI API 有速率限制需要实现重试机制# error_handling.py import time from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def robust_embedding_call(titles, modeltext-embedding-3-large): 带重试机制的嵌入调用 try: response openai.embeddings.create( modelmodel, inputtitles, timeout30 # 设置超时 ) return response except openai.RateLimitError: print(速率限制等待重试...) time.sleep(60) # 等待1分钟 raise # 重新抛出异常触发重试 except openai.APITimeoutError: print(API 超时重试...) raise8.2 聚类质量评估与调优DBSCAN 聚类结果需要评估质量并调整参数# clustering_quality.py from sklearn.metrics import silhouette_score import numpy as np def evaluate_clustering_quality(embeddings, labels): 评估聚类质量 if len(set(labels)) 1: return {silhouette_score: -1, message: 聚类数量不足} # 排除噪声点计算轮廓系数 valid_mask labels ! -1 if np.sum(valid_mask) 2: return {silhouette_score: -1, message: 有效数据点不足} valid_embeddings embeddings[valid_mask] valid_labels labels[valid_mask] try: score silhouette_score(valid_embeddings, valid_labels, metriccosine) return { silhouette_score: score, n_clusters: len(set(valid_labels)), n_noise: np.sum(labels -1), quality: good if score 0.5 else fair if score 0.3 else poor } except: return {silhouette_score: -1, message: 计算失败} def auto_tune_dbscan(embeddings, eps_range[0.1, 0.5], min_samples_range[2, 5]): 自动调优 DBSCAN 参数 best_score -1 best_params None best_labels None for eps in np.linspace(eps_range[0], eps_range[1], 10): for min_samples in range(min_samples_range[0], min_samples_range[1] 1): dbscan DBSCAN(epseps, min_samplesmin_samples, metriccosine) labels dbscan.fit_predict(embeddings) evaluation evaluate_clustering_quality(embeddings, labels) score evaluation[silhouette_score] if score best_score and evaluation[n_clusters] 1: best_score score best_params {eps: eps, min_samples: min_samples} best_labels labels return best_params, best_labels, best_score8.3 前端性能优化处理大量数据点时优化前端渲染性能