更快地计算余弦相似度

Question

我想在我的 IR 项目中使用余弦相似度，但由于向量的大小很大并且必须多次乘以浮点数，所以需要很长时间。

有什么方法可以更快地计算余弦相似度？

这是我的代码：

private double diffrence(HashMap<Integer, Float> hashMap,
 HashMap<Integer, Float> hashMap2 ) {
    Integer[] keys = new Integer[hashMap.size()];
    hashMap.keySet().toArray(keys);

     float ans = 0;

    for (int i = 0; i < keys.length; i++) {
        if (hashMap2.containsKey(keys[i])) {
             ans += hashMap.get(keys[i]) * hashMap2.get(keys[i]);

        }
    }

     float hashLength = 0;
    for (int i = 0; i < keys.length; i++) {
         hashLength += (hashMap.get(keys[i]) * hashMap.get(keys[i]));
    }
     hashLength = (float) Math.sqrt(hashLength);

    Integer[] keys2 = new Integer[hashMap2.size()];
    hashMap2.keySet().toArray(keys2);

     float hash2Length = 0;
    for (int i = 0; i < keys2.length; i++) {

         hash2Length += hashMap2.get(keys2[i]) * hashMap2.get(keys2[i]);

    }
     hash2Length = (float) Math.sqrt(hash2Length);

    return (float) (ans /(hash2Length*hashLength));
}

Answer 1

通常在 IR 中，一个向量的非零元素比另一个向量少得多（通常查询向量是更稀疏的向量，但即使对于文档向量也是如此）。您可以通过循环遍历 sparser 向量的键来节省时间，即较小的哈希映射，在较大的哈希映射中查找它们。

至于 pkacprzak 对查找表的建议和您的内存不足：请意识到可以在余弦相似度计算之前进行标准化。对于每个向量，在存储它之前，计算它的范数并将每个元素除以它。然后，您可以计算一个点积并得到一个余弦相似度。

即余弦相似度通常定义为

x·y / (||x|| × ||y||)

但这等于

(x / ||x||) · (y / ||y||)

其中/ 是逐元素除法。如果你每个人都用x / ||x||替换x，那么你只需要计算x·y。

如果您将这两个建议结合起来，您将得到一个余弦相似度算法，该算法只对两个输入中较小的一个进行一个循环。

可以通过使用更智能的sparse vector 结构进行进一步改进；哈希表在查找和迭代中都有很多的开销。

Answer 2

通常有太多向量无法预先计算每对的余弦相似度，但您可以预先计算每个向量的长度并使用查找表存储它。这减少了计算两个向量的余弦相似度的常数因子 - 实际上它节省了大量时间，因为有很多浮点运算。

我假设您不会通过在向量中存储零来浪费内存。

Answer 3

除了按照其他人的建议对向量进行预规范化并假设向量列表没有改变之外，将它们转换为数组对一次（在相似度函数之外）并按关键索引，例如：

Integer[] keys = new Integer[hashMap.size()];
Float values[] = new Float[keys.size()];
int i = 0;
float norm = ...;    
for (Map.Entry<Integer, Float> entry : new TreeMap<Integer, Float>(hashMap).entrySet())
{
   keys[i] = entry.getKey();
   values[i++] = entry.getValue() / norm;
}

然后进行实际的相似度计算（假设您随后通过keys1、values、keys2、values2 而不是两个HashMaps），您的最内层循环减少为：

float ans = 0;
int i,j = 0;
while (i < keys1.length && j < keys2.length)
{
  if (keys1[i] < keys2[j])
    ++i;
  else if (keys1[i] > keys2[j])
    ++j;
  else
    // we have the same key in 1 and 2
    ans += values1[i] * values2[j];
}

您甚至可以考虑将所有向量的所有keys 和values 连续存储在int 和float 的大数组中，并在第一个位置保留另一个具有索引的数组：

int sumOfAllVectorLengths = ...;
int allKeys[] = new int[sumOfAllVectorLengths];
float allValues[] = new float[sumOfAllVectorLengths];
int firstPos = new int[numberOfVectors + 1]; 
firstPos[numberOfVectors] = sumOfAllVectorLengths;

int nextFirstPos = 0;
int index = 0;

for (HashMap<Integer, Float> vector : allVectors)
{
   firstPos[index] = nextFirstPos;

   float norm = ...;    
   for (Map.Entry<Integer, Float> entry : new TreeMap<Integer, Float>(hashMap).entrySet())
   {
      keys[nextFirstPos] = entry.getKey();
      values[nextFirstPos++] = entry.getValue() / norm;
   }

   ++index; 
}

然后将数组和向量的索引传递给比较函数。

Answer 4

可以查看项目simbasehttps://github.com/guokr/simbase，它是一个向量相似度nosql数据库。

Simbase 使用以下概念：

• 向量集：一组向量
• 基：向量的基，一个向量集中的向量具有相同的基
• 推荐：两个具有相同基的向量集之间的单向二元关系

写操作在每个基中在单个线程中处理，并且需要在任意两个向量之间进行比较，因此写操作的缩放时间为 O(n)。

我们在 i7-cpu Macbook 上对密集向量进行了非最终性能测试，它可以在 0.14 秒内轻松处理 100k 1k 维向量，每次写入操作；如果线性比例可以保持，这意味着 Simbase 可以在 1 秒内处理 70 万个密集向量，每次写入操作。

Answer 5

我至少可以清楚地看到一个地方，你只是在浪费 CPU 周期：

for (int i = 0; i < keys.length; i++) {
    if (hashMap2.containsKey(keys[i])) {
         ans += hashMap.get(keys[i]) * hashMap2.get(keys[i]);
    }
}

float hashLength = 0;
for (int i = 0; i < keys.length; i++) {
     hashLength += (hashMap.get(keys[i]) * hashMap.get(keys[i]));
}

在这里，您在相同的 2 个 hashMap 上有 2 个相同边界的循环。为什么不在一个周期内完成：

float hashLength = 0;
int hm = 0;
for (int i = 0; i < keys.length; i++) {
    hm = hashMap.get(keys[i])*hashMap2.get(keys[i]);
    hashLength += hm;
    if (hashMap2.containsKey(keys[i])) {
         ans += hm;
    }
}

顺便问一下，使用hashMap有什么特别的原因吗？或者你可以用一些更简单的数组来做？