【算法】局部敏感哈希（LSH）：高效解决相似性搜索问题的利器

什么是局部敏感哈希？

简单来说，LSH是一种在海量数据中快速找到相似项的技术。与传统的哈希函数不同，LSH的特点是：相似的输入更可能被哈希到相同的"桶"中。

为什么要用LSH？

想象一下，你有一个包含数百万用户信息的数据库，你需要找出与某个特定用户最相似的人。传统方法可能需要遍历整个数据库，这在大数据时代是非常耗时的。而LSH可以大大加快这个过程。

LSH如何工作？一个简单的例子

让我们从一个简单的例子开始：假设我们只考虑用户的年龄。

def simple_lsh(age):
    return age // 10

users = [
    ("Alice", 22), ("Bob", 25), ("Charlie", 31),
    ("David", 23), ("Eve", 38), ("Frank", 41), ("Grace", 29)
]

hash_table = {}
for name, age in users:
    hash_value = simple_lsh(age)
    if hash_value not in hash_table:
        hash_table[hash_value] = []
    hash_table[hash_value].append((name, age))

# 打印结果
for hash_value, group in hash_table.items():
    print(f"年龄在 {hash_value * 10} 到 {(hash_value + 1) * 10 - 1} 之间的人:")
    for name, age in group:
        print(f"  {name}: {age}")
    print()

这个简单的LSH将年龄相近的用户分到同一个桶中。但是，这种方法有一个问题：它无法处理跨越桶边界的情况，比如29岁和31岁的用户会被分到不同的桶中。

改进：多重哈希

为了解决这个问题，我们可以使用多重哈希：

def multi_hash(age):
    return [age // 10, (age + 5) // 10]

hash_table = {}
for name, age in users:
    for hash_value in multi_hash(age):
        if hash_value not in hash_table:
            hash_table[hash_value] = set()
        hash_table[hash_value].add((name, age))

# 查找相似用户
def find_similar(name, age):
    similar = set()
    for hash_value in multi_hash(age):
        similar.update(hash_table[hash_value])
    similar.remove((name, age))  # 移除自己
    return similar

# 测试
test_name, test_age = "Charlie", 31
similar = find_similar(test_name, test_age)
print(f"{test_name} ({test_age}) 的相似用户:")
for name, age in similar:
    print(f"  {name}: {age}")

这种方法增加了找到真正相似项的概率，即使它们跨越了原来的桶边界。

现实场景

在我给出的简化例子中，使用哈希的优势并不明显。让我解释一下为什么在实际应用中LSH仍然非常有用，并给出一个更贴近实际的例子来说明LSH的优势。

在小规模数据集或维度较低的情况下，LSH可能并不比直接比较更有优势。LSH的真正威力体现在处理大规模、高维度数据时。让我们考虑一个更现实的场景：

假设我们有一个包含100万用户的数据库，每个用户有100个特征（比如年龄、收入、兴趣爱好等）。我们想找出与某个特定用户最相似的前10个用户。

暴力搜索方法：
- 需要比较100万个用户。
- 每次比较需要计算100个特征的距离。
- 总计算量：100万 * 100 = 1亿次特征比较。
使用LSH方法：
- 预处理：为每个用户计算LSH值（假设使用10个哈希函数）。
- 查询时：
  1. 计算目标用户的10个哈希值。
  2. 在哈希表中查找这10个哈希值对应的桶。
  3. 只对这些桶中的用户进行详细比较。

假设每个哈希桶平均包含100个用户：

需要比较的用户数：约10 * 100 = 1000个（而不是100万）。
总计算量：1000 * 100 = 10万次特征比较。

这比暴力搜索减少了约1000倍的计算量！

让我们用Python代码来模拟这个更realistic的场景：

import numpy as np
from collections import defaultdict

# 模拟100万用户，每个用户有100个特征
num_users = 1_000_000
num_features = 100
users = np.random.rand(num_users, num_features)

# LSH函数（这里使用一个简化的版本）
def lsh(vector, num_hashes=10):
    return tuple(np.dot(vector, np.random.randn(num_features)) > 0 for _ in range(num_hashes))

# 构建LSH索引
lsh_index = defaultdict(list)
for i, user in enumerate(users):
    for h in lsh(user):
        lsh_index[h].append(i)

# 查询函数
def query(target_user, top_k=10):
    candidate_set = set()
    for h in lsh(target_user):
        candidate_set.update(lsh_index[h])
    
    # 只对候选集中的用户计算实际距离
    distances = [(i, np.linalg.norm(users[i] - target_user)) for i in candidate_set]
    return sorted(distances, key=lambda x: x[1])[:top_k]

# 测试
target_user = np.random.rand(num_features)
result = query(target_user)

print(f"找到的前 {len(result)} 个最相似用户:")
for i, distance in result:
    print(f"用户ID: {i}, 距离: {distance}")

print(f"\n总用户数: {num_users}")
print(f"LSH候选集大小: {len(set().union(*[lsh_index[h] for h in lsh(target_user)]))}")