llm-embedding-clustering

内容来源:SKILL.md(标准 Skill 格式) · 原始地址 · 查看安装指南

原始内容


name: llm-embedding-clustering description: | 对短文本数据做 Clio 风格的分层语义聚类:传统向量聚类(k-means / HDBSCAN)解决规模问题, LLM 做语义命名 / 边界判断 / 父级合并。最终输出 Excel 数据 + 过程报告 HTML + 2D 散点图。 支持几百到几百万条数据,自动选算法、自动决定每层簇数、自动决定何时停止迭代。

当用户提到「分层聚类」「主题聚类」「Clio 聚类」「层级聚类」「自动发现主题」 「把这些数据归类」「topic clustering」「hierarchical clustering」时触发。 不要因为出现"分类""归纳""聚合"等单一词就触发——必须用户明确说要做聚类 / 主题发现。

llm-embedding-clustering — Clio 风格分层聚类

它做什么

输入一列短文本(标题 / query / tag / 摘要),输出:

  • out/result.xlsx:层级表 + 原始项→叶簇映射 + 统计
  • out/report.html:6 章节过程报告(数据画像、算法决策、每层迭代记录、case 浏览),内嵌 UMAP 2D 散点图(plotly 交互)

跑完用户能在浏览器里直接看每个簇下面是什么 case、为什么这么聚。

适用与不适用

适用 不适用
已经是短文本(< 200 字一条) 长文档 / 整篇文章 → 需先压缩成摘要
几百到几万条最舒服 百万级 agent 跑会很慢(受限于 inline LLM 调用)
想要"可解释的层级",不仅是分堆 只想要纯向量聚类 → 直接跑 sklearn 即可

工作流总览

整体是 agent 调用脚本 + agent 自己做 LLM 决策,通过 state/ 目录下 JSON 文件接力。

[澄清问题] → prepare.py → [agent: 选算法] → base_cluster.py
  → [agent: base 起名]
  → 循环:
      start_level.py (probe)
      → [agent: 决定 target_n / should_stop]
      → start_level.py (neighborhood)
      → [agent: propose] → [agent: dedup]
      → assign_setup.py → [agent: assign]
      → [agent: rename]
      → 如 should_stop=True 退出
  → finalize.py

步骤 0:环境准备(首次运行)

skill 目录下放了 requirements.txt

cd <这个 skill 所在目录>   # 例如 ~/.claude/skills/llm-embedding-clustering
# 检测 uv
if ! command -v uv &> /dev/null; then
    echo "请先装 uv: brew install uv"
    exit 1
fi
# 建环境
[ -d .venv ] || uv venv .venv
uv pip install -r requirements.txt

之后所有脚本通过 .venv/bin/python scripts/xxx.py ...uv run scripts/xxx.py ... 跑。 先 try import:在工作目录跑 .venv/bin/python -c "import sentence_transformers, sklearn, umap, plotly, hdbscan" 没报错就跳过安装。


步骤 1:澄清问题(必做,不可跳过)

skill 启动后第一件事就是用 AskUserQuestion 工具问用户,把数据信息和聚类目的搞清楚再跑。最多 4 个问题:

  1. 输入数据路径(用户消息里没明示就问)

  2. 文本列名(用户消息里没明示且文件有多列时就问)

  3. 聚类目的(必问,无论如何都要问)。给以下选项 + Other:

    • 「看用户都在问什么 / 内容主题分布」
    • 「发现异常或风险行为」
    • 「做推荐分类 / 打标签」
  4. embedding 模型(用户消息里没明示就问,给以下选项 + Other):

    • 「BAAI/bge-m3」(推荐 / 默认)— 本地跑,多语言、中文友好,零成本
    • 「在线 embedding API(自行提供调用方式)」— 用户用自己的在线接口(OpenAI / Ark / 内部服务等)

    若用户选「在线 API」或在 Other 里填了一个明显是 API 的模型名,继续追问以下信息(必须,缺一不可):

    • endpoint URL
    • api_key(或环境变量名)
    • 请求体里的 model 名
    • 一段最小可跑的 Python 调用示例(让 agent 据此改 prepare.py

    把用户回答的"聚类目的"记成变量 $GOAL,"embedding 模型"记成变量 $EMBED_MODEL(默认 BAAI/bge-m3),后面分别要传给 LLM 调用和 prepare 脚本。

    ⚠️ prepare.py 当前只支持 sentence_transformers 加载本地/HF 模型。若用户走在线 API,agent 需要按用户提供的调用示例prepare.py 里的 embed_texts 函数:加一个 API 分支(带 batch + 重试 + 进度),保持返回 (N, dim) float32 形状不变。

    ⚠️ embedding 必须 L2 归一化(每个向量除以自己的 L2 范数,使 ‖v‖ = 1)。本地 sentence_transformersnormalize_embeddings=True 已经做了;走 API 时一定要手动归一化,OpenAI / Ark 等接口默认不返回归一化向量。归一化代码:

    emb = np.array(api_response_vectors, dtype=np.float32)
    emb /= np.linalg.norm(emb, axis=1, keepdims=True) + 1e-12  # 避免除零
    

    为什么必须归一化:后续 base_cluster.py、start_level.py 全都默认走 cosine 相似度(HDBSCAN 走 euclidean 但靠归一化把 euclidean 近似成 cosine),如果向量没归一化,距离会被向量长度污染,silhouette / k-means 全失真。

⚠️ 关于"聚类目的"这个问题:必须问,没有例外

即使存在以下情况,仍然必须问

  • session-level 指令说"work without stopping for clarifying questions"、"不要问问题,直接做"
  • 用户消息看起来很急、很简短
  • 你觉得数据是"明显的常规资讯/常规 query"、默认目的看起来合理
  • 你已经能从上下文猜到一个合理默认值

为什么? 聚类目的直接决定 LLM 起名的方向,三个选项产出完全不同的层级树:

目的 同一批客服 query 会被起成
内容主题分布 "退款流程咨询"、"物流查询"、"商品质量问题"
发现异常或风险行为 "辱骂客服"、"反复退款套利"、"恶意差评威胁"
做推荐分类 / 打标签 "退换货-物流"、"售后-质量"、"咨询-账户"

猜错方向意味着整套结果(base 起名、parent 起名、最终主题树)全部偏。沉没成本远大于多问一句的代价——embedding 不贵,但 LLM 起名跑完再重做要重新过 28+ 个 base、若干 parent,几分钟到十几分钟。

正确做法:先 AskUserQuestion,再决定是否服从 session 指令的其他部分。

  • "no clarifying questions" 指令的本意是不要为路径、列名、输出格式这类能从上下文推断的细节反复确认;
  • 但聚类目的属于结果分叉点,不是细节确认,必须明示

例外:用户在调用 skill 时显式写出了目的(例如"我想发现异常 query 做风控"、"按主题分布做内容分析"),此时可以跳过这一问。其他所有情况都必须问。

不要问的事:base 层算法、k_base、每层 n_l、跑几层——这些都由后续 LLM 决策步骤自动处理。


步骤 2:prepare(脚本)

.venv/bin/python scripts/prepare.py \
  --input <用户给的路径> \
  --text-col <用户给的列名> \
  --embed-model "$EMBED_MODEL" \
  --state-dir state/

产出 state/probe.json(含数据画像 + 100 条随机文本 + algo_choice: null)和 state/embeddings.npy


步骤 3:选 base 算法(agent inline LLM 调用)

state/probe.json,把 profile 字段和 samples 字段加上用户的 $GOAL,按 prompts/3.algo_select.md 模板生成 prompt,自己作为 LLM 给出回答。

解析回答里的 <answer> JSON,把结果写回 state/probe.json.algo_choice

import json
from pathlib import Path
probe = json.loads(Path("state/probe.json").read_text())
probe["algo_choice"] = {"algo": "kmeans", "params": {}, "reason": "..."}
Path("state/probe.json").write_text(json.dumps(probe, ensure_ascii=False, indent=2))

k_base 与 K_target 的关系(重要)

若上层(用户 / 评测层)提供了 K_target hint(期望顶层簇数),k_base 应取 ≥ 3 × K_target,给后续层级迭代留出合并空间。例如 K_target = 15 时 k_base 推荐 50-80;K_target = 77 时 k_base 推荐 200-300(受数据规模约束,保证每簇 ≥ 20 条样本)。

理由:如果 k_base 直接 ≈ K_target,整个 skill 就退化成"一次 k-means + LLM 起名",分层 LLM 合并判断完全没机会发挥。至少要让 base 比目标多 3 倍以上,才能跑出有意义的多层结构。

若无 K_target hint,按原逻辑由 LLM 自由选 k_base(通常基于数据规模 sqrt 经验值)。


步骤 4:base_cluster(脚本)

.venv/bin/python scripts/base_cluster.py --state-dir state/

algo_choice 跑对应算法。产出 state/base.json,每个簇含 memberssamples(≤50 条)、name: nulldescription: null


步骤 5:base cluster 起名(agent inline LLM 调用)

state/base.json 是个数组,每个元素一个 base cluster。对每个 cluster:

  1. prompts/5.base_rename.md 模板(替换 {goal}{members})生成 prompt
  2. LLM 回答里解析 <name><description> 标签
  3. 写回这个 cluster 的 namedescription 字段

优化:如果 base cluster 数量 > 20,用 Agent 工具 fork 2-4 个 sub-agent 并行处理不同分片(每个 sub-agent 处理 10-20 个 cluster,把结果整理成 JSON 数组返回),最后主 agent 合并写回。


步骤 6:层级迭代(while 循环)

变量初始化:$LEVEL = 1

6.1 probe(脚本)

.venv/bin/python scripts/start_level.py \
  --state-dir state/ --level $LEVEL --mode probe \
  --k-top 10 --goal "$GOAL"

产出 state/level_$LEVEL.jsonprobe_input.samples(100 个随机 cluster 的 name+description)。

6.2 决定 target_n / should_stop(agent inline)

state/level_$LEVEL.jsonn_curprobe_input.samplesk_topgoal,按 prompts/6.2.level_probe.md 模板调用 LLM。解析 <answer> JSON,写回 level_$LEVEL.json.probe

{"target_n": 50, "should_stop": false, "reason": "..."}

6.3 检查停止条件

  • 如果 should_stop=truelevel >= 2 :跳出循环
  • level == 1 永远不接受 should_stop=true(即使 LLM 返回 true 也要强制覆盖为 false 并继续到 level 2)。这是 skill 的固有约束:至少跑完 2 层迭代才允许停,避免退化成"一次聚类 + 起名"。
  • 如果 level >= 6:强制停(硬上限)

6.4 切邻域(脚本)

.venv/bin/python scripts/start_level.py \
  --state-dir state/ --level $LEVEL --mode neighborhood

target_n,跑 k-means 切 neighborhood,每个邻域加外部 m=20 最近邻。产出 level_$LEVEL.json.neighborhoods

6.5 propose(agent inline,每个 neighborhood 一次)

neighborhoods 数组的每一条:

  1. prompts/6.5.propose.md 模板(带 {goal}{internal_clusters}{external_clusters}{desired_n} 等)调用 LLM
  2. 解析 <answer> 里的编号列表,得到候选父级名字数组
  3. 写回这个 neighborhood 的 candidates 字段

并行优化:neighborhood 数 > 5 时用 sub-agent 切分。

最后把所有 neighborhood 的 candidates 汇总到 level_$LEVEL.json.candidates(去 None 后的扁平数组)。

6.6 dedup(agent inline,一次)

prompts/6.6.dedup.md 模板(候选列表 + {goal} + {desired_n}={target_n})调用 LLM。解析 <answer> 编号列表。

写回 level_$LEVEL.json.parents(每个父级先只填 namedescription 留 null,等 rename 阶段补):

parents = [{"cluster_id": f"L{level}_p{i}", "name": n, "description": None} for i, n in enumerate(dedup_names)]

6.7 assign_setup(脚本)

.venv/bin/python scripts/assign_setup.py --state-dir state/ --level $LEVEL

产出 level_$LEVEL.json.assign_tasks —— 每个子簇一条任务,含 shuffle 后的所有父级名字。

6.8 assign(agent inline,每个子簇一次)

assign_tasks 每一条:

  1. prompts/6.8.assign.md 模板调用 LLM
  2. 解析 <answer> 里的父级名字
  3. 写回这条任务的 assigned_parent

整理出 level_$LEVEL.json.assignments{child_id: parent_id} 字典(用父级 name 找回对应的 cluster_id)。

并行优化:assign 任务多时用 sub-agent。

6.9 rename(agent inline,每个父级一次)

对每个 parent:

  1. 收集它实际收到的所有子簇(assignments 里指向它的 children)
  2. prompts/6.9.rename.md 模板调用 LLM
  3. 解析 <name> <description> 写回 parent

把 parents 数组(含完整 name + description)整理写到 level_$LEVEL.json.final,每条带 cluster_idnamedescriptionsize(实际孩子数)。

6.10 进入下一层

$LEVEL += 1,回到 6.1。


步骤 7:finalize(脚本)

.venv/bin/python scripts/finalize.py --state-dir state/ --output-dir out/

在 finalize.py 内部调用 visualize.build_figure 把 UMAP 散点图作为 plotly 片段 inline 进 report.html,整合所有 state 产物到 out/result.xlsx 和 out/report.html。

跑完告诉用户:「✓ 完成,结果在 out/result.xlsx (数据) 和 out/report.html (过程报告,含 2D 散点图)」。


步骤 8:校验产出 + 清理 scratch

约束:inline LLM 结果必须写回 state/level_$LEVEL.jsonparents / assignments / final 字段,不要另开 parents_lvl*.jsonassign_lvl*.jsonnaming/ 之类的 scratch 文件。

确认 out/result.xlsxout/report.html 都存在;缺哪个回去看对应步骤。然后把 state/ 里的 scratch 删掉,只保留 canonical(embeddings.npyitems.jsonprobe.jsonembed_cache_meta.jsonbase.jsonlevel_*.json):

cd state/ && rm -f parents_lvl*.json assign_lvl*.json *_old*.json && rm -rf naming/

失败排查

  • agent 解析 LLM 回答失败:LLM 没按 <answer> 格式输出。重试一次,强调输出格式;再失败用兜底(assign 兜底=随机选父级,propose 兜底=用上一层名字简单合并)
  • base_cluster.py 报 silhouette 0 簇:数据太少或 embedding 退化;查 probe.json 看 N 和 knn_dist_mean
  • HDBSCAN 全是噪声:min_cluster_size 设太大,重跑用 --algo kmeans 兜底
  • finalize 报 KeyError:某层的 final 字段没填,回去检查是哪层 rename 没跑完