百万条用户话术，如何快速找到相似语义占比TopN？（ES向量检索实操全方案）

Yakai Zheng

2024-12-10

搜索引擎

elasticsearch

在我们公司业务里，积累了大量用户话术（比如客服对话、用户反馈、评论留言等），当话术量级达到一定量级时，一个核心需求就会浮现：如何找出这些话术中，相似语义的话术占比最高的那些？

比如客服场景中，我们想知道用户最常咨询的几类问题（如“登录失败”“改绑手机号”“退款申请”），以及每类问题的占比，从而针对性优化服务；再比如用户反馈分析，快速定位核心痛点话题。

很多人初次接触这个需求，会陷入两个误区：一是不知道如何量化“语义相似”，二是误以为生成向量必须调用ChatGPT等在线AI接口，既花钱又麻烦。今天就结合Elasticsearch向量检索、开源本地Embedding模型，给大家一套可落地、高可用、零成本的完整解决方案，全程不依赖在线AI接口，百万条数据也能轻松扛住。

一、核心需求拆解

数据规模：百万条用户话术（文本格式，单条话术长度10-50字为主）；
核心目标：找出语义相似的话术簇，筛选出占比最高的20个簇，输出每个簇的语义主题和占比；
约束条件：不依赖在线AI接口（避免成本、隐私泄露、接口限流），方案要简单可落地，无需复杂组件；
性能要求：百万条数据可在1小时内完成处理，结果准确可解释。

二、关键认知：语义相似怎么量化？（向量基础回顾）

在解决问题前，先理清一个核心概念：语义相似无法通过关键词匹配实现（比如“登录不上”和“登录报错”，关键词不同但语义一致），必须通过“向量Embedding”将文本转化为可计算的数字，再通过特定算法计算向量相似度判断语义相似度。

这里补充两个关键知识点，避免大家踩坑：

1. 向量Embedding不仅仅是只靠在线AI

很多人生成向量必须调用ChatGPT、文心一言等在线Embedding接口（比如我这个需求实际用的还是 openai 的向量生成），在成本有限的情况下我们可以用开源的中文Embedding模型，在本地运行，不联网、不花钱、不限次数，百万条话术随便处理。

推荐中文开源模型（按优先级排序）：

BGE-base-zh-v1.5：效果最优，中文语义识别准确率高，输出维度768，轻量易运行；
text2vec-large-chinese：速度快，轻量，适合对速度要求高的场景；
m3e-base：国产小而强，资源占用低，适合普通电脑运行。

2. 向量的维度?

dim是“向量维度”的缩写，dim=768表示：一条话术被模型转换成包含768个数字的数组（比如“我登录不上去”→ [0.12, 0.55, -0.31, …, 共768个数字]）。

这个维度是模型的固定输出，768维是中文开源模型的黄金选择——语义表达足够强，同时不会占用过多存储空间和计算资源；如果需要更强的语义表达，也可以选择1024维的模型（如bge-large-zh），但会增加存储和计算成本。

三、完整解决方案

整体思路：文本向量化（本地开源模型）→ 向量存储（ES）→ 聚类分组（MiniBatch K-Means）→ 类别分配（ES向量检索）→ 统计占比（Top20），全程本地运行，零成本、高可用。

Step 1：环境准备（简单易搭，新手也能搞定）

需要准备3个核心工具，全部免费开源：

Python 3.10+：用于加载开源模型、生成向量、数据处理；
开源Embedding模型：推荐BGE-base-zh-v1.5（第一次运行会自动下载，之后完全离线）；
Elasticsearch 8.0+：用于存储向量、执行向量检索（聚类后分配类别）。

Python依赖安装（终端执行）：

1	pip install torch transformers sentence-transformers pandas elasticsearch

Step 2：本地生成向量（核心步骤，不调任何在线接口）

用Python加载本地开源模型，将百万条话术批量转换成768维向量，全程离线，速度快、无成本。

示例代码（可直接复制运行，替换自己的话术数据即可）：

from sentence_transformers import SentenceTransformer
import pandas as pd

# 1. 加载本地开源Embedding模型（第一次下载后，后续完全离线）
model = SentenceTransformer("BAAI/bge-base-zh-v1.5")

# 2. 读取百万条用户话术（这里用示例数据，实际替换成自己的CSV/Excel文件）
# 假设话术存在"text"列，数据量100万条
df = pd.read_csv("user_texts.csv")
texts = df["text"].tolist()

# 3. 批量生成向量（normalize_embeddings=True：归一化，提升相似度计算准确性）
# 百万条数据：CPU（普通电脑）约30-60分钟，GPU约3-10分钟
embeddings = model.encode(texts, normalize_embeddings=True)

# 4. 将向量和话术合并，保存为CSV（方便后续存入ES）
df["embedding"] = embeddings.tolist()
df.to_csv("user_texts_with_embedding.csv", index=False)

运行后，你会得到一个包含“话术文本”和“768维向量”的CSV文件，接下来将这些数据存入ES。

Step 3：创建ES向量索引，存入数据

ES 8.0+ 支持dense_vector类型，专门用于存储向量，同时支持HNSW算法，可快速执行向量相似检索。

第一步：创建ES向量索引（用Kibana Dev Tools或Python代码执行）：

PUT /user_texts_vector
{
  "mappings": {
    "properties": {
      "text": { "type": "text", "analyzer": "ik_max_word" },  // 话术文本，支持全文检索
      "embedding": {
        "type": "dense_vector",
        "dims": 768,  // 对应模型输出的768维向量
        "index": true,
        "similarity": "cosine"  // 相似度算法，余弦相似度最适合语义匹配
      }
    }
  }
}

第二步：将带向量的话术批量存入ES（Python代码）：

from elasticsearch import Elasticsearch
import pandas as pd

# 1. 连接ES（替换成自己的ES地址和端口）
es = Elasticsearch("http://localhost:9200", basic_auth=("username", "password"))

# 2. 读取带向量的话术数据
df = pd.read_csv("user_texts_with_embedding.csv")

# 3. 批量存入ES（批量提交，提升效率）
batch_size = 1000
for i in range(0, len(df), batch_size):
    batch = df.iloc[i:i+batch_size]
    actions = []
    for _, row in batch.iterrows():
        action = {
            "index": {
                "_index": "user_texts_vector",
                "_id": str(row.name)  # 用行号作为ES文档ID
            }
        }
        doc = {
            "text": row["text"],
            "embedding": row["embedding"]
        }
        actions.append(action)
        actions.append(doc)
    # 批量提交
    es.bulk(body=actions)

百万条数据批量存入ES，约30分钟内可完成（根据ES性能调整批量大小）。

Step 4：聚类分组，找出相似语义簇

百万条数据无法暴力两两比较相似性，必须用“聚类算法”将语义相似的向量聚成一组（簇）。这里推荐MiniBatch K-Means算法，适合海量数据，速度快、效果稳定。

核心思路：先采样1-5万条向量做聚类，得到若干个“聚类中心”（每个中心代表一个语义主题），再用ES向量检索，将百万条向量分配到最近的聚类中心，实现分组。

示例代码：

from sklearn.cluster import MiniBatchKMeans
import pandas as pd

# 1. 读取带向量的话术数据
df = pd.read_csv("user_texts_with_embedding.csv")
embeddings = df["embedding"].tolist()

# 2. 采样聚类（采样1万条，避免全量聚类过慢）
sample_size = 10000
sample_embeddings = embeddings[:sample_size]

# 3. MiniBatch K-Means聚类（n_clusters=100，可根据实际情况调整，越多越精细）
kmeans = MiniBatchKMeans(n_clusters=100, random_state=42)
cluster_labels = kmeans.fit_predict(sample_embeddings)
cluster_centers = kmeans.cluster_centers_  # 得到100个聚类中心（768维）

# 4. 将聚类中心存入ES（用于后续全量数据分配类别）
from elasticsearch import Elasticsearch
es = Elasticsearch("http://localhost:9200", basic_auth=("username", "password"))

# 创建聚类中心索引
es.indices.create(
    index="cluster_centers",
    body={
        "mappings": {
            "properties": {
                "center_id": {"type": "integer"},
                "embedding": {
                    "type": "dense_vector",
                    "dims": 768,
                    "similarity": "cosine"
                }
            }
        }
    }
)

# 存入聚类中心
for i, center in enumerate(cluster_centers):
    es.index(
        index="cluster_centers",
        id=i,
        document={
            "center_id": i,
            "embedding": center.tolist()
        }
    )

Step 5：全量类别分配 + 统计占比，取Top20

将百万条话术向量，通过ES的KNN向量检索，找到每个向量最近的聚类中心（即所属语义簇），然后统计每个簇的话术数量和占比，最终筛选出占比最高的20个簇。

示例代码：

from elasticsearch import Elasticsearch
import pandas as pd
from collections import Counter

# 1. 连接ES
es = Elasticsearch("http://localhost:9200", basic_auth=("username", "password"))

# 2. 读取百万条话术和向量
df = pd.read_csv("user_texts_with_embedding.csv")
total_count = len(df)  # 总话术数量

# 3. 批量分配类别（每个向量找最近的聚类中心）
batch_size = 1000
cluster_ids = []

for i in range(0, len(df), batch_size):
    batch = df.iloc[i:i+batch_size]
    for _, row in batch.iterrows():
        # KNN查询：找到最近的1个聚类中心
        response = es.search(
            index="cluster_centers",
            body={
                "knn": {
                    "field": "embedding",
                    "query_vector": row["embedding"],
                    "k": 1,  # 只取最近的1个中心
                    "num_candidates": 10
                },
                "_source": ["center_id"]
            }
        )
        # 获取所属聚类ID
        center_id = response["hits"]["hits"][0]["_source"]["center_id"]
        cluster_ids.append(center_id)

# 4. 统计每个聚类的数量和占比
cluster_counter = Counter(cluster_ids)
cluster_stats = []

for center_id, count in cluster_counter.items():
    ratio = (count / total_count) * 100  # 占比（百分比）
    # 取每个聚类的代表性话术（随便选一条）
    representative_text = df.iloc[cluster_ids.index(center_id)]["text"]
    cluster_stats.append({
        "cluster_id": center_id,
        "count": count,
        "ratio": round(ratio, 2),
        "representative_text": representative_text  # 语义主题代表话术
    })

# 5. 按占比降序排序，取Top20
top20_clusters = sorted(cluster_stats, key=lambda x: x["ratio"], reverse=True)[:20]

# 6. 输出结果（可保存为Excel，方便查看）
top20_df = pd.DataFrame(top20_clusters)
top20_df.to_excel("top20_similar_text_clusters.xlsx", index=False)

# 打印Top20结果
print("相似语义话术占比Top20：")
for i, cluster in enumerate(top20_clusters, 1):
    print(f"{i}. 语义主题（代表话术）：{cluster['representative_text']}")
    print(f"   数量：{cluster['count']} 条，占比：{cluster['ratio']}%")
    print("-" * 50)

四、最终结果示例

运行完成后，你会得到一个Excel文件和控制台输出，类似这样：

语义主题（代表话术）：我的账号登不上数量：123000 条，占比：12.30%
语义主题（代表话术）：想改绑手机号数量：98000 条，占比：9.80%
语义主题（代表话术）：申请退款怎么操作数量：73000 条，占比：7.30%
…
\20. 语义主题（代表话术）：查询订单物流数量：12000 条，占比：1.20%

五、避坑

不要全量K-Means聚类：百万条向量全量聚类会非常慢，采样5万条聚类即可，结果足够准确；
ES批量提交大小：批量存入ES时，batch_size设为1000-2000，避免过大导致ES崩溃；
向量归一化：生成向量时一定要加normalize_embeddings=True，否则相似度计算会失真；
聚类数量选择：n_clusters建议设为50-100，太少会导致语义簇太粗，太多会导致占比分散。

六、总结

百万条用户话术找相似语义占比TopN，核心是“向量量化语义 + 聚类分组 + 向量检索分配”，全程不用调用任何在线AI接口，用开源模型+ES就能零成本落地。

整个方案的优势的是：简单易搭、性能可控、隐私安全，既适合新手实操，也能满足工业级场景需求。如果你的话术数据量更大（千万级），可以适当优化ES集群配置、增加GPU加速向量生成，整体流程完全通用。

最后，附上整套方案的核心逻辑总结，方便大家记忆和面试复用：

对百万条用户话术，用本地开源Embedding模型生成768维向量，存入ES向量索引；通过MiniBatch K-Means聚类获取语义中心，再利用ES KNN检索对全量数据做类别分配，统计各类别数量与占比，最终筛选出占比最高的N个相似语义簇。