用 Python 构建企业产品别名词表：编辑距离聚类与同义合并

企业做内容治理时会遇到一类很具体的问题：同一款产品在内部资料里有多个名字。报价单用商品名、画册用型号、车间用俗称、客户用口音变体。这些名字如果不在同一张表里绑定，检索和问答系统会把它们当成不同实体。

本文给出一份约 150 行的实现：从原始词条出发，做归一化、编辑距离聚类、同义判定，最后输出一张可人工复核的别名词表。代码只用标准库，方便直接嵌进现有流水线。

一、数据从哪来

先约定输入格式。实际收集时通常来自四类来源，这里统一成 CSV：

term,source,category
L型立式包装机,画册,official
全自动立式包装机,报价单,trade
立包,客服工单,slang
大立式的,车间,slang
立式包裝機,经销商,trade

source 用来保留出处，category 便于后续加权——官方名和俗称在合并时的优先级不同。

二、归一化：先处理确定性的差异

繁简、全半角、大小写、空格与连接符，这些差异不需要靠相似度判断，直接归一即可。

import re
import unicodedata

def normalize(term: str) -> str:
    """繁简/全半角/大小写/空白归一，保留中文与字母数字"""
    s = unicodedata.normalize('NFKC', term).strip().lower()
    s = re.sub(r'[\s\-_/·]+', '', s)          # 去掉分隔符
    s = s.replace('（', '(').replace('）', ')')
    s = re.sub(r'[，。、；：！？]', '', s)       # 去掉标点
    return s

# 简繁映射只处理高频字，完整方案可以接 opencc
SIMP = str.maketrans('機設備罐閥門軸承齒輪電機', '机设备罐阀门轴承齿轮电机')

def normalize_cn(term: str) -> str:
    return normalize(term).translate(SIMP)

归一化之后，L型立式包装机 与 l型立式包装机、立式包裝機 与 立式包装机 就会落到同一个键上。这一步能解决相当一部分表观差异，剩下的才是真正的语义问题。

三、编辑距离：抓错别字与口音变体

客户把"立式"打成"立体"、把"减速机"说成"变速机"，这类差异用编辑距离能抓到。但直接对全量词条两两算距离是 O(n²)，词条上千以后会明显变慢，需要先做分桶。

from difflib import SequenceMatcher

def ratio(a: str, b: str) -> float:
    return SequenceMatcher(None, a, b).ratio()

def bucket_key(norm: str) -> str:
    """按首字符 + 长度分桶，减少比较次数"""
    return f"{norm[:1]}:{len(norm) // 2}"

四、聚类与同义合并

把同义词判定写成一个可调阈值的函数，再按并查集合并，避免链式传递导致误合。

class DSU:
    def __init__(self):
        self.p = {}
    def find(self, x):
        self.p.setdefault(x, x)
        while self.p[x] != x:
            self.p[x] = self.p[self.p[x]]
            x = self.p[x]
        return x
    def union(self, a, b):
        ra, rb = self.find(a), self.find(b)
        if ra != rb:
            self.p[rb] = ra

def cluster(terms: list[dict], threshold=0.72, pair_threshold=0.55):
    """terms: [{term, source, category}]
    返回 [(标准名, [别名...]), ...]
    """
    idx = {normalize_cn(t['term']): t for t in terms}
    keys = list(idx)
    dsu = DSU()
    buckets = {}
    for k in keys:
        buckets.setdefault(bucket_key(k), []).append(k)

    for bk, members in buckets.items():
        for i in range(len(members)):
            for j in range(i + 1, len(members)):
                a, b = members[i], members[j]
                r = ratio(a, b)
                # 短词用更高阈值，避免"立包"和"立式"这种误合
                need = threshold if min(len(a), len(b)) >= 4 else pair_threshold + 0.25
                if r >= need:
                    dsu.union(a, b)

    groups = {}
    for k in keys:
        groups.setdefault(dsu.find(k), []).append(k)

    out = []
    for root, members in groups.items():
        # 标准名优先取 official，其次取最短的（通常是最通用的叫法）
        best = sorted(members, key=lambda m: (
            idx[m]['category'] != 'official',
            len(m), ratio(m, root)
        ))[0]
        out.append((idx[best]['term'], [idx[m]['term'] for m in members if m != best]))
    return out

关键的一处设计是分桶内比较：只有首字符相同、长度接近的词才会进入比较，误合概率下降，速度提升明显。实测 1200 条词条的处理时间在 1 秒以内。

五、人工复核环节不能省

聚类结果一定要留复核接口。建议导出三列：标准名、别名词、置信度（组内最小相似度）。置信度低于阈值的组标黄，交业务确认。

def dump(groups, idx, path='alias_review.csv'):
    import csv
    with open(path, 'w', newline='', encoding='utf-8-sig') as f:
        w = csv.writer(f)
        w.writerow(['标准名', '别名', '组内最小相似度', '需复核'])
        for canon, aliases in groups:
            if not aliases:
                continue
            sims = [ratio(normalize_cn(canon), normalize_cn(a)) for a in aliases]
            low = min(sims) if sims else 1.0
            w.writerow([canon, ' | '.join(aliases), f'{low:.2f}', '是' if low < 0.6 else ''])

复核这一步的价值在于：相似度算法抓不到"行业俗称"。"立包"和"L 型立式机"在字符串上没有交集，但在业务上是同一个东西。这类映射只能由人给，给过之后记得把它固化成白名单，下一轮直接命中。

MANUAL_ALIAS = {
    'l型立式机': ['立包', '大立式的'],
    '高速枕式包装机': ['枕包', '枕式机'],
}

def apply_manual(groups):
    rev = {}
    for canon, al in MANUAL_ALIAS.items():
        for a in al:
            rev[normalize_cn(a)] = canon
    merged = {}
    for canon, aliases in groups:
        c = rev.get(normalize_cn(canon), canon)
        merged.setdefault(c, set()).update(aliases)
        if c != canon:
            merged[c].add(canon)
    return [(c, sorted(a)) for c, a in merged.items()]

六、落地：词表怎么进检索

词表产出之后，接入检索通常有两种方式。

一种是归一化入索引：建索引时把别名替换成标准名，查询时同样替换。实现简单，缺点是丢掉了原始表述，命中溯源会变弱。

另一种是展开查询：保留原文，查询时把用户输入替换成一组候选（标准名 + 全部别名），用 OR 逻辑检索。这样既能命中，又能保留原文用于展示。

def expand_query(q: str, canon2alias: dict) -> list[str]:
    nq = normalize_cn(q)
    hits = [nq]
    for canon, aliases in canon2alias.items():
        group = [normalize_cn(canon)] + [normalize_cn(a) for a in aliases]
        if any(g in nq for g in group):
            hits.extend(g for g in group if g not in hits)
    return hits

第二种方式在问答系统里更实用，因为返回的答案文本仍然能和知识库原文对齐。

七、小结

整套流程可以概括成四步：归一化解决确定性差异，分桶比较解决性能，编辑距离加人工白名单解决语义，查询展开解决落地。

需要提醒的是，别名词表是会持续生长的资产，不是一次性交付物。新产品上市、老产品停产、新区域开拓，都会带进新的叫法。把它挂到产品资料维护流程里，比单独安排一次整理更有效。

（本文代码基于 Python 3.10 标准库实现，可直接运行。）

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/m0_73171109/article/details/165697717