HJQ/tianxuan
2
1
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

fix: raw row clustering — remove entity concept, use row index, all numeric features

Browse Source
This commit is contained in:
PM-pinou
2026-07-24 11:35:18 +08:00
parent fa13ebcf72
commit e2c64fd068
3 changed files with 23 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
analysis/views/auto.py
+1 -1
View File
@@ -274,7 +274,7 @@ def run_llm_analysis_view(request):
_run_clustering_pipeline( _run_clustering_pipeline(
run=run, run=run,
store=store, store=store,
entity_ds_id=entity_ds_id, ds_id=entity_ds_id,
feature_columns=None, feature_columns=None,
algorithm='agglomerative', algorithm='agglomerative',
min_cluster_size=5, min_cluster_size=5,
analysis/views/clustering.py
+18 -19
View File
@@ -196,16 +196,19 @@ def cluster_detail(request, display_id, cluster_label):
@profile @profile
def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorithm, def _run_clustering_pipeline(run, store, ds_id, feature_columns, algorithm,
min_cluster_size, run_umap=True, head=None): min_cluster_size, run_umap=True, head=None):
"""Unified clustering pipeline: clustering → feature extraction → UMAP → ORM save. """Clustering pipeline on raw rows: clustering → feature extraction → UMAP → ORM save.
Operates directly on raw data rows (no entity aggregation).
Per-row results stored in EntityProfile with row index as identifier.
Args: Args:
run: AnalysisRun ORM object. run: AnalysisRun ORM object.
store: SessionStore instance. store: SessionStore instance.
entity_ds_id: dataset ID in SessionStore pointing to entity-aggregated data. ds_id: dataset ID in SessionStore pointing to raw data.
feature_columns: list of feature column names (None = auto-detect). feature_columns: list of feature column names (None = auto-detect all numeric).
algorithm: 'hdbscan' or 'kmeans'. algorithm: clustering algorithm.
min_cluster_size: int for HDBSCAN. min_cluster_size: int for HDBSCAN.
run_umap: bool, whether to compute and save UMAP-2D embeddings. run_umap: bool, whether to compute and save UMAP-2D embeddings.
head: optional int, downsample to at most this many rows (min 100). head: optional int, downsample to at most this many rows (min 100).
@@ -217,10 +220,10 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
from analysis.tool_registry import _handle_run_clustering, _handle_extract_features from analysis.tool_registry import _handle_run_clustering, _handle_extract_features
import polars as pl import polars as pl
entry = store.get_dataset(entity_ds_id) entry = store.get_dataset(ds_id)
if entry is None: if entry is None:
run.status = 'failed' run.status = 'failed'
run.error_message = 'Entity dataset not found in session store' run.error_message = '数据集在会话中不存在'
run.save(update_fields=['status', 'error_message']) run.save(update_fields=['status', 'error_message'])
return return
@@ -241,11 +244,11 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
run.save(update_fields=['progress_msg']) run.save(update_fields=['progress_msg'])
lf = lf.head(head) lf = lf.head(head)
store.store_dataset( store.store_dataset(
entity_ds_id, lf, ds_id, lf,
schema=schema, schema=schema,
metadata=entry.get('metadata', {}), metadata=entry.get('metadata', {}),
) )
entry = store.get_dataset(entity_ds_id) # refresh entry = store.get_dataset(ds_id)
row_count = head row_count = head
# Resolve feature columns — use actual LazyFrame dtypes (not stored schema dict # Resolve feature columns — use actual LazyFrame dtypes (not stored schema dict
@@ -268,7 +271,7 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
feature_cols = [ feature_cols = [
live_names[i] for i, dt in enumerate(live_dtypes) live_names[i] for i, dt in enumerate(live_dtypes)
if dt in numeric_types and not live_names[i].startswith('_') if dt in numeric_types and not live_names[i].startswith('_')
][:10] ]
# ── Step 1: Clustering ──────────────────────────────────────────────── # ── Step 1: Clustering ────────────────────────────────────────────────
# If no numeric columns detected, pass empty list to let # If no numeric columns detected, pass empty list to let
@@ -279,7 +282,7 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
run.save(update_fields=['status', 'progress_pct', 'progress_msg']) run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
result = asyncio.run(_handle_run_clustering( result = asyncio.run(_handle_run_clustering(
dataset_id=entity_ds_id, dataset_id=ds_id,
cluster_columns=feature_cols, cluster_columns=feature_cols,
algorithm=algorithm, algorithm=algorithm,
params={'min_cluster_size': min_cluster_size}, params={'min_cluster_size': min_cluster_size},
@@ -307,7 +310,7 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
run.save(update_fields=['status', 'progress_pct', 'progress_msg']) run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
feat_result = asyncio.run(_handle_extract_features( feat_result = asyncio.run(_handle_extract_features(
dataset_id=entity_ds_id, dataset_id=ds_id,
cluster_result_id=cluster_id, cluster_result_id=cluster_id,
top_k=10, method='zscore', save_to_db=True, top_k=10, method='zscore', save_to_db=True,
run_id=run.id, run_id=run.id,
@@ -431,10 +434,6 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
if coords is None: if coords is None:
raise Exception('UMAP produced no output') raise Exception('UMAP produced no output')
non_num = [c for c in df_umap.columns
if c not in num_cols and not c.startswith('_')]
ent_col = non_num[0] if non_num else df_umap.columns[0]
cluster_entry = store.get_cluster_result(cluster_id) cluster_entry = store.get_cluster_result(cluster_id)
labels_list = cluster_entry.get('labels', []) if cluster_entry else [] labels_list = cluster_entry.get('labels', []) if cluster_entry else []
@@ -443,9 +442,9 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
profiles = [] profiles = []
cr_cache = {} cr_cache = {}
for i, row in enumerate(df_umap.iter_rows(named=True)): for i in range(len(df_umap)):
ev = str(row.get(ent_col, '')) or 'entity' # Use row index as identifier — no entity column
ev = f'{ev}_{i}' ev = f'row_{i}'
lbl = labels_list[i] if i < len(labels_list) else -1 lbl = labels_list[i] if i < len(labels_list) else -1
cache_key = f'{run.id}_{lbl}' cache_key = f'{run.id}_{lbl}'
if cache_key not in cr_cache: if cache_key not in cr_cache:
analysis/views/manual.py
+4 -4
View File
@@ -23,9 +23,9 @@ def manual_page(request):
# Get all tools metadata # Get all tools metadata
tools = get_tools_meta() tools = get_tools_meta()
core_names = {'profile_data', 'build_entity_profiles', 'compute_scores', core_names = {'profile_data', 'run_clustering', 'extract_features',
'run_clustering', 'extract_features', 'detect_anomalies', 'filter_data', 'preprocess_data',
'visualize_anomalies'} 'detect_anomalies', 'visualize_anomalies'}
diag_names = {'validate_data', 'explore_distributions', 'find_outliers', diag_names = {'validate_data', 'explore_distributions', 'find_outliers',
'diagnose_clustering', 'compare_datasets', 'export_debug_sample', 'diagnose_clustering', 'compare_datasets', 'export_debug_sample',
'repair_schema'} 'repair_schema'}
@@ -165,7 +165,7 @@ def manual_run_analysis(request):
# Use the unified clustering pipeline (clustering → extraction → UMAP) # Use the unified clustering pipeline (clustering → extraction → UMAP)
_run_clustering_pipeline( _run_clustering_pipeline(
run=run, store=store, entity_ds_id=ds_id, run=run, store=store, ds_id=ds_id,
feature_columns=feature_columns, feature_columns=feature_columns,
algorithm=algorithm, min_cluster_size=min_cluster_size, algorithm=algorithm, min_cluster_size=min_cluster_size,
run_umap=True, head=head, run_umap=True, head=head,