推荐系统之局部敏感哈希（LSH）

前言

局部敏感哈希的基本思想：是希望让相邻的点落入同一个“桶”，这样在进行最近邻搜索时，我们仅需要在一个桶内，或相邻几个桶内的元素中进行搜索即可。如果保持每个桶中的元素个数在一个常数附近，我们就可以把最近邻搜索的时间复杂度降低到常数级别。
高维空间映射到低维空间相似度的定性结论：欧式空间中，将高维空间的点映射到低维空间，原本接近的点在低维空间中肯定依然接近，但原本远离的点则有一定概率变成接近的点。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eBUZv8Ix-1622533742078)(./imgs/LSH.png)]
构造原则：

利用低维空间可以保留高维空间相近距离关系的性质，我们就可以构造局部敏感哈希“桶”。
对于 Embedding 向量来说，由于 Embedding 大量使用内积操作计算相似度，因此我们也可以用内积操作来构建局部敏感哈希桶。假设 v 是高维空间中的 k 维 Embedding 向量，x 是随机生成的 k 维映射向量。那我们利用内积操作可以将 v 映射到一维空间，得到数值 h ( v ) = v ⋅ x h(v)=v⋅x h(v)=v⋅x
而且，一维空间也会部分保存高维空间的近似距离信息。因此，我们可以使用哈希函数 h(v) 进行分桶，公式为： h ( v ) = ⌊ ( x ⋅ v + b ) / w ⌋ h(v)=⌊(x⋅v+b)/w⌋ h(v)=⌊(x⋅v+b)/w⌋ 。其中， ⌊⌋ 是向下取整操作， w 是分桶宽度，b 是 0 到 w 间的一个均匀分布随机变量，避免分桶边界固化。
不过，映射操作会损失部分距离信息，如果我们仅采用一个哈希函数进行分桶，必然存在相近点误判的情况，因此，我们可以采用 m 个哈希函数同时进行分桶。如果两个点同时掉进了 m 个桶，那它们是相似点的概率将大大增加。通过分桶找到相邻点的候选集合后，我们就可以在有限的候选集合中通过遍历找到目标点真正的 K 近邻了。

最后，局部敏感哈希能在常数时间得到最近邻的结果吗？
答案是可以的，如果我们能够精确地控制每个桶内的点的规模是 C，假设每个 Embedding 的维度是 N，那么找到最近邻点的时间开销将永远在 O ( C ⋅ N ) O(C⋅N) O(C⋅N)量级。采用多桶策略之后，假设分桶函数数量是 K，那么时间开销也在 O ( K ⋅ C ⋅ N ) O(K⋅C⋅N) O(K⋅C⋅N)量级，这仍然是一个常数。

LSH代码

代码流程：
1、首先，在离线状态下，将所有的向量通过特定的哈希函数映射到对应的索引位置；

2、输入一个向量，用同样的哈希函数计算哈希值，找到对应哈希值位置的所有向量；

3、根据对应的距离度量方法，与第2步查到的所有向量计算距离；

4、筛选出距离最小的n个向量，即为与输入向量最为相似的n个结果。

import numpy as np
from typing import List, Union

class EuclideanLSH:

    def __init__(self, tables_num: int, a: int, depth: int):
        """

        :param tables_num: hash_table的个数
        :param a: a越大，被纳入同个位置的向量就越多，即可以提高原来相似的向量映射到同个位置的概率，
                反之，则可以降低原来不相似的向量映射到同个位置的概率。
        :param depth: 向量的维度数
        """
        self.tables_num = tables_num
        self.a = a
        # 为了方便矩阵运算，调整了shape，每一列代表一个hash_table的随机向量
        self.R = np.random.random([depth, tables_num])
        self.b = np.random.uniform(0, a, [1, tables_num])
        # 初始化空的hash_table
        self.hash_tables = [dict() for i in range(tables_num)]

    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]之间均匀分布的随机变量
        hash_val = np.floor(np.abs(np.matmul(inputs, self.R) + self.b) / self.a)

        return hash_val

    def insert(self, inputs):
        """
        将向量映射到对应的hash_table的索引，并插入到所有hash_table中
        :param inputs:
        :return:
        """
        # 将inputs转化为二维向量
        inputs = np.array(inputs)
        if len(inputs.shape) == 1:
            inputs = inputs.reshape([1, -1])
        # indexs: [21. 23. 24. 26. 25. 22. 25. 25. 22. 26.]
        hash_index = self._hash(inputs)
        for inputs_one, indexs in zip(inputs, hash_index):
            for i, key in enumerate(indexs):
                # i代表第i个hash_table，key则为当前hash_table的索引位置
                # inputs_one代表当前向量, 具有相同hash index的输入放入到同一个桶中
                # [{"hash_key": [inputs_ones]}]
                self.hash_tables[i].setdefault(key, []).append(tuple(inputs_one))

    def query(self, inputs, nums=20, op = "or"):
        """
        查询与inputs相似的向量，并输出相似度最高的nums个
        :param inputs: 输入向量
        :param nums:
        :return:
        """
        hash_val = self._hash(inputs).ravel()
        candidates = set()

        # 将相同索引位置的向量添加到候选集中
        if op == "or":
            for i, key in enumerate(hash_val):
                candidates.update(self.hash_tables[i][key])
                
        elif op == "and":
            for i, key in enumerate(hash_val[1:], start=1):
                if i == 1:
                    candidates = self.hash_tables[i - 1][key]
                l2 = self.hash_tables[i][key]
                candidates = set(candidates)&set(l2)
        # 根据向量距离进行排序
        candidates = sorted(candidates, key=lambda x: self.euclidean_dis(x, inputs))
        
        return candidates[: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)))


if __name__ == '__main__':
    
    data = np.random.random([10000, 100])
    query = np.random.random([100])

    lsh = EuclideanLSH(10, 1, 100)
    lsh.insert(data)
    res = lsh.query(query, 20)
    res = np.array(res)
    print("LSH: \n", np.sum(np.power(res - query, 2), axis=-1))
    
    # 获得排序后的index数值
    # 一一计算欧式距离，进行比较
    sort = np.argsort(np.sum(np.power(data - query, 2), axis=-1))
    print("Original data distance（Top-k）: \n", np.sum(np.power(data[sort[:20]] - query, 2), axis=-1))
    print("Original data distance（Bottom-k）: \n",np.sum(np.power(data[sort[-20:]] - query, 2), axis=-1))

LSH: 
 [10.08980867 10.11797684 10.12562829 10.278338   10.93401227 11.07314636
 11.07426626 11.12684973 11.17300039 11.17381982 11.17470708 11.1872176
 11.18894438 11.33127372 11.34651465 11.36570334 11.37687416 11.39542495
 11.46208599 11.47289804]
Original data distance（Top-k）: 
 [10.08980867 10.11797684 10.12562829 10.13820015 10.278338   10.55935714
 10.68896141 10.78527213 10.93401227 10.994057   11.07314636 11.07426626
 11.12684973 11.17300039 11.17381982 11.17470708 11.1872176  11.18894438
 11.33127372 11.34651465]
Original data distance（Bottom-k）: 
 [21.88635607 21.96086613 21.96727453 22.06513118 22.09067843 22.09252059
 22.10877624 22.13209293 22.25154688 22.33314858 22.39606409 22.40801702
 22.43777882 22.45125568 22.54360093 22.71898604 22.87555687 22.93491961
 22.96644845 23.05905718]

参考

12-Embedding召回中的局部敏感哈希策略
 局部敏感哈希(LSH)：高维数据下的最近邻查找

标签：inputs,hash,self,LSH,敏感,哈希,np,向量
来源： https://blog.csdn.net/weixin_35154281/article/details/117444943

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

推荐系统之局部敏感哈希（LSH）

推荐系统之局部敏感哈希（LSH）

前言

LSH代码

参考