原始内容
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 个问题:
输入数据路径(用户消息里没明示就问)
文本列名(用户消息里没明示且文件有多列时就问)
聚类目的(必问,无论如何都要问)。给以下选项 + Other:
- 「看用户都在问什么 / 内容主题分布」
- 「发现异常或风险行为」
- 「做推荐分类 / 打标签」
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_transformers走normalize_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,每个簇含 members、samples(≤50 条)、name: null、description: null。
步骤 5:base cluster 起名(agent inline LLM 调用)
state/base.json 是个数组,每个元素一个 base cluster。对每个 cluster:
- 按
prompts/5.base_rename.md模板(替换{goal}和{members})生成 prompt - LLM 回答里解析
<name>和<description>标签 - 写回这个 cluster 的
name和description字段
优化:如果 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.json 含 probe_input.samples(100 个随机 cluster 的 name+description)。
6.2 决定 target_n / should_stop(agent inline)
读 state/level_$LEVEL.json 的 n_cur、probe_input.samples、k_top、goal,按 prompts/6.2.level_probe.md 模板调用 LLM。解析 <answer> JSON,写回 level_$LEVEL.json.probe:
{"target_n": 50, "should_stop": false, "reason": "..."}
6.3 检查停止条件
- 如果
should_stop=true且level >= 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 数组的每一条:
- 按
prompts/6.5.propose.md模板(带{goal}、{internal_clusters}、{external_clusters}、{desired_n}等)调用 LLM - 解析
<answer>里的编号列表,得到候选父级名字数组 - 写回这个 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(每个父级先只填 name,description 留 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 每一条:
- 按
prompts/6.8.assign.md模板调用 LLM - 解析
<answer>里的父级名字 - 写回这条任务的
assigned_parent
整理出 level_$LEVEL.json.assignments:{child_id: parent_id} 字典(用父级 name 找回对应的 cluster_id)。
并行优化:assign 任务多时用 sub-agent。
6.9 rename(agent inline,每个父级一次)
对每个 parent:
- 收集它实际收到的所有子簇(assignments 里指向它的 children)
- 按
prompts/6.9.rename.md模板调用 LLM - 解析
<name><description>写回 parent
把 parents 数组(含完整 name + description)整理写到 level_$LEVEL.json.final,每条带 cluster_id、name、description、size(实际孩子数)。
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.json 的 parents / assignments / final 字段,不要另开 parents_lvl*.json、assign_lvl*.json、naming/ 之类的 scratch 文件。
确认 out/result.xlsx、out/report.html 都存在;缺哪个回去看对应步骤。然后把 state/ 里的 scratch 删掉,只保留 canonical(embeddings.npy、items.json、probe.json、embed_cache_meta.json、base.json、level_*.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 没跑完