摘要：局部敏感哈希，Python，矢量检索，推荐系统

单独记录一下LSH算法的原理，结合代码深入理解一下，因为这个算法的调参对结果影响极大，不懂原理就不会调参，导致最终效果不理想

LSH概述

知识准备

重点概念

算法流程

其他策略

Python代码实现

这个代码参考了https://blog.csdn.net/sgyuanshi/article/details/108132214
我在他的基础上增加了idMap的映射，在模型中把向量的标识也灌了进去

import numpy as np
from typing import List, Union


class EuclideanLSH:
    def __init__(self, num_hash_tables: int, bucket_len: int, embedding_size: int):
        """
        LSH
        :param num_hash_tables:
        :param bucket_len:
        :param embedding_size:
        """
        self.num_hash_tables = num_hash_tables
        self.bucket_len = bucket_len
        # 同一个hash_table采用同一个hash函数，k和hash函数相同，R相同
        self.R = np.random.random([embedding_size, num_hash_tables])
        # 一个hash——table一个随机的b
        self.b = np.random.uniform(0, bucket_len, [1, num_hash_tables])
        # 初始化空的hash_table
        self.hash_tables = [dict() for i in range(num_hash_tables)]
        # ids和vector的对应关系
        self.ids_map = {}

    def _hash(self, inputs: Union[List[List], np.ndarray]):
        """
        将向量映射到对应的hash_table的索引
        :param inputs: 输入的单个或多个向量
        :return: 每一行代表一个向量输出的所有索引，每一列代表位于一个hash_table中的索引
        """
        # H(V) = |V·R + b| / a，R是一个随机向量，a是桶宽，b是一个在[0,a]之间均匀分布的随机变量，这个乘10是为了hash值分布地更开
        hash_val = np.floor(np.abs(np.matmul(inputs, self.R) + self.b) * 10 / self.bucket_len)

        return hash_val

    def insert(self, inputs, ids):  # 增加id和向量的对应关系
        """
        将向量映射到对应的hash_table的索引，并插入到所有hash_table中
        :param inputs:
        :return:
        """
        self.ids_map = dict(zip(ids, inputs))
        # 将inputs转化为二维向量
        inputs = np.array(inputs)
        if len(inputs.shape) == 1:
            inputs = inputs.reshape([1, -1])

        hash_index = self._hash(inputs)
        # 一条输入向量，一条映射到所有hash_table的索引值
        for id_one, inputs_one, indexs in zip(ids, inputs, hash_index):
            # i代表第i个hash_table，key则为当前hash_table的索引位置
            for i, key in enumerate(indexs):
                # 第n个hash_table的第k个桶位置，将这条数据灌进去
                # 还可以这样写, 这个地方用元组是因为后面需要加入set，不可变可以hash
                self.hash_tables[i].setdefault(key, []).append(id_one)

    def query(self, id_one, nums=20):
        """
        查询与id_one的inputs相似的向量，并输出相似度最高的nums个
        :param id_one:
        :param nums:
        :return:
        """
        assert id_one in self.ids_map, "元素不存在在"
        id_vector = self.ids_map[id_one]
        hash_val = self._hash(id_vector).ravel()  # 计算新输入的在所有hash_table的值，然后拉平为一维向量
        candidates = set()

        # 每一张hash table中相同桶位置的向量全部加入候选集，去重
        for i, key in enumerate(hash_val):
            # 后面是一个集合，所以用update，将集合拆碎加入,集合内部元素必须是不可变的，所以提前转了元组
            candidates.update(self.hash_tables[i][key])
        candidates = [x for x in candidates if x != id_one]
        print("LSH之后所有hash_table一个桶下的候选集总计{}".format(len(candidates)))

        # 根据向量距离进行排序
        # 候选集暴力求解
        res = [(x, self.euclidean_dis(self.ids_map[x], id_vector)) for x in candidates]
        return sorted(res, key=lambda x: x[1])[:nums]

    @staticmethod
    def euclidean_dis(x, y):
        """
        计算欧式距离
        :param x:
        :param y:
        :return:
        """
        x = np.array(x)
        y = np.array(y)

        return np.sqrt(np.sum(np.power(x - y, 2)))

现在我们了来调用一发，数据采用离线训练好的170万的实体embedding向量，向量维度为16，大概长这个样子

下一步初始化模型参数，向量维度16固定，设置num_hash_table和bucket_len都是8

LSH近似搜索

暴力搜索

结果是LSH从170万实体中先召回了所有和输入实体共同桶的7.1万个候选，最终在桶下暴力搜索耗时3.6秒，而全表暴力搜索耗时60秒，从准确率来看全表扫描更准备，但是LSH的Top1也是全表扫描的第三名，整体来看LSH准确率表现也不错