#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2018 Vit Novotny
# Licensed under the GNU LGPL v2.1 - https://www.gnu.org/licenses/old-licenses/lgpl-2.1.en.html
"""
This module provides classes that deal with term similarities.
"""
from array import array
from itertools import chain
import logging
from math import sqrt
import numpy as np
from scipy import sparse
from gensim.matutils import corpus2csc
from gensim.utils import SaveLoad, is_corpus
logger = logging.getLogger(__name__)
NON_NEGATIVE_NORM_ASSERTION_MESSAGE = (
u"sparse documents must not contain any explicit "
u"zero entries and the similarity matrix S must satisfy x^T * S * x >= 0 for any "
u"nonzero bag-of-words vector x."
)
class TermSimilarityIndex(SaveLoad):
"""
Base class = common interface for retrieving the most similar terms for a given term.
See Also
--------
:class:`~gensim.similarities.termsim.SparseTermSimilarityMatrix`
A sparse term similarity matrix built using a term similarity index.
"""
def most_similar(self, term, topn=10):
"""Get most similar terms for a given term.
Return the most similar terms for a given term along with their similarities.
Parameters
----------
term : str
The term for which we are retrieving `topn` most similar terms.
topn : int, optional
The maximum number of most similar terms to `term` that will be retrieved.
Returns
-------
iterable of (str, float)
Most similar terms along with their similarities to `term`. Only terms distinct from
`term` must be returned.
"""
raise NotImplementedError
def __str__(self):
members = ', '.join('%s=%s' % pair for pair in vars(self).items())
return '%s<%s>' % (self.__class__.__name__, members)
class UniformTermSimilarityIndex(TermSimilarityIndex):
"""
Retrieves most similar terms for a given term under the hypothesis that the similarities between
distinct terms are uniform.
Parameters
----------
dictionary : :class:`~gensim.corpora.dictionary.Dictionary`
A dictionary that specifies the considered terms.
term_similarity : float, optional
The uniform similarity between distinct terms.
See Also
--------
:class:`~gensim.similarities.termsim.SparseTermSimilarityMatrix`
A sparse term similarity matrix built using a term similarity index.
Notes
-----
This class is mainly intended for testing SparseTermSimilarityMatrix and other classes that
depend on the TermSimilarityIndex.
"""
def __init__(self, dictionary, term_similarity=0.5):
self.dictionary = sorted(dictionary.items())
self.term_similarity = term_similarity
def most_similar(self, t1, topn=10):
for __, (t2_index, t2) in zip(range(topn), (
(t2_index, t2) for t2_index, t2 in self.dictionary if t2 != t1)):
yield (t2, self.term_similarity)
class WordEmbeddingSimilarityIndex(TermSimilarityIndex):
"""
Computes cosine similarities between word embeddings and retrieves most
similar terms for a given term.
Notes
-----
By fitting the word embeddings to a vocabulary that you will be using, you
can eliminate all out-of-vocabulary (OOV) words that you would otherwise
receive from the `most_similar` method. In subword models such as fastText,
this procedure will also infer word-vectors for words from your vocabulary
that previously had no word-vector.
>>> from gensim.test.utils import common_texts, datapath
>>> from gensim.corpora import Dictionary
>>> from gensim.models import FastText
>>> from gensim.models.word2vec import LineSentence
>>> from gensim.similarities import WordEmbeddingSimilarityIndex
>>>
>>> model = FastText(common_texts, vector_size=20, min_count=1) # train word-vectors on a corpus
>>> different_corpus = LineSentence(datapath('lee_background.cor'))
>>> dictionary = Dictionary(different_corpus) # construct a vocabulary on a different corpus
>>> words = [word for word, count in dictionary.most_common()]
>>> word_vectors = model.wv.vectors_for_all(words) # remove OOV word-vectors and infer word-vectors for new words
>>> assert len(dictionary) == len(word_vectors) # all words from our vocabulary received their word-vectors
>>> termsim_index = WordEmbeddingSimilarityIndex(word_vectors)
Parameters
----------
keyedvectors : :class:`~gensim.models.keyedvectors.KeyedVectors`
The word embeddings.
threshold : float, optional
Only embeddings more similar than `threshold` are considered when retrieving word embeddings
closest to a given word embedding.
exponent : float, optional
Take the word embedding similarities larger than `threshold` to the power of `exponent`.
kwargs : dict or None
A dict with keyword arguments that will be passed to the
:meth:`~gensim.models.keyedvectors.KeyedVectors.most_similar` method
when retrieving the word embeddings closest to a given word embedding.
See Also
--------
:class:`~gensim.similarities.levenshtein.LevenshteinSimilarityIndex`
Retrieve most similar terms for a given term using the Levenshtein distance.
:class:`~gensim.similarities.termsim.SparseTermSimilarityMatrix`
Build a term similarity matrix and compute the Soft Cosine Measure.
"""
def __init__(self, keyedvectors, threshold=0.0, exponent=2.0, kwargs=None):
self.keyedvectors = keyedvectors
self.threshold = threshold
self.exponent = exponent
self.kwargs = kwargs or {}
super(WordEmbeddingSimilarityIndex, self).__init__()
def most_similar(self, t1, topn=10):
if t1 not in self.keyedvectors:
logger.debug('an out-of-dictionary term "%s"', t1)
else:
most_similar = self.keyedvectors.most_similar(positive=[t1], topn=topn, **self.kwargs)
for t2, similarity in most_similar:
if similarity > self.threshold:
yield (t2, similarity**self.exponent)
def _shortest_uint_dtype(max_value):
"""Get the shortest unsingned integer data-type required for representing values up to a given
maximum value.
Returns the shortest unsingned integer data-type required for representing values up to a given
maximum value.
Parameters
----------
max_value : int
The maximum value we wish to represent.
Returns
-------
data-type
The shortest unsigned integer data-type required for representing values up to a given
maximum value.
"""
if max_value < 2**8:
return np.uint8
elif max_value < 2**16:
return np.uint16
elif max_value < 2**32:
return np.uint32
return np.uint64
def _create_source(index, dictionary, tfidf, symmetric, dominant, nonzero_limit, dtype):
"""Build a sparse term similarity matrix using a term similarity index.
Returns
-------
matrix : :class:`scipy.sparse.coo_matrix`
The sparse term similarity matrix.
"""
assert isinstance(index, TermSimilarityIndex)
assert dictionary is not None
matrix_order = len(dictionary)
if matrix_order == 0:
raise ValueError('Dictionary provided to SparseTermSimilarityMatrix must not be empty')
logger.info("constructing a sparse term similarity matrix using %s", index)
if nonzero_limit is None:
nonzero_limit = matrix_order
def tfidf_sort_key(term_index):
if isinstance(term_index, tuple):
term_index, *_ = term_index
term_idf = tfidf.idfs[term_index]
return (-term_idf, term_index)
if tfidf is None:
columns = sorted(dictionary.keys())
logger.info("iterating over %i columns in dictionary order", len(columns))
else:
assert max(tfidf.idfs) == matrix_order - 1
columns = sorted(tfidf.idfs.keys(), key=tfidf_sort_key)
logger.info("iterating over %i columns in tf-idf order", len(columns))
nonzero_counter_dtype = _shortest_uint_dtype(nonzero_limit)
column_nonzero = np.array([0] * matrix_order, dtype=nonzero_counter_dtype)
if dominant:
column_sum = np.zeros(matrix_order, dtype=dtype)
if symmetric:
assigned_cells = set()
row_buffer = array('Q')
column_buffer = array('Q')
if dtype is np.float16 or dtype is np.float32:
data_buffer = array('f')
elif dtype is np.float64:
data_buffer = array('d')
else:
raise ValueError('Dtype %s is unsupported, use numpy.float16, float32, or float64.' % dtype)
def cell_full(t1_index, t2_index, similarity):
if dominant and column_sum[t1_index] + abs(similarity) >= 1.0:
return True # after adding the similarity, the matrix would cease to be strongly diagonally dominant
assert column_nonzero[t1_index] <= nonzero_limit
if column_nonzero[t1_index] == nonzero_limit:
return True # after adding the similarity, the column would contain more than nonzero_limit elements
if symmetric and (t1_index, t2_index) in assigned_cells:
return True # a similarity has already been assigned to this cell
return False
def populate_buffers(t1_index, t2_index, similarity):
column_buffer.append(t1_index)
row_buffer.append(t2_index)
data_buffer.append(similarity)
column_nonzero[t1_index] += 1
if symmetric:
assigned_cells.add((t1_index, t2_index))
if dominant:
column_sum[t1_index] += abs(similarity)
try:
from tqdm import tqdm as progress_bar
except ImportError:
def progress_bar(iterable):
return iterable
for column_number, t1_index in enumerate(progress_bar(columns)):
column_buffer.append(column_number)
row_buffer.append(column_number)
data_buffer.append(1.0)
if nonzero_limit <= 0:
continue
t1 = dictionary[t1_index]
num_nonzero = column_nonzero[t1_index]
num_rows = nonzero_limit - num_nonzero
most_similar = [
(dictionary.token2id[term], similarity)
for term, similarity in index.most_similar(t1, topn=num_rows)
if term in dictionary.token2id
] if num_rows > 0 else []
if tfidf is None:
rows = sorted(most_similar)
else:
rows = sorted(most_similar, key=tfidf_sort_key)
for t2_index, similarity in rows:
if cell_full(t1_index, t2_index, similarity):
continue
if not symmetric:
populate_buffers(t1_index, t2_index, similarity)
elif not cell_full(t2_index, t1_index, similarity):
populate_buffers(t1_index, t2_index, similarity)
populate_buffers(t2_index, t1_index, similarity)
data_buffer = np.frombuffer(data_buffer, dtype=dtype)
row_buffer = np.frombuffer(row_buffer, dtype=np.uint64)
column_buffer = np.frombuffer(column_buffer, dtype=np.uint64)
matrix = sparse.coo_matrix((data_buffer, (row_buffer, column_buffer)), shape=(matrix_order, matrix_order))
logger.info(
"constructed a sparse term similarity matrix with %0.06f%% density",
100.0 * matrix.getnnz() / matrix_order**2,
)
return matrix
def _normalize_dense_vector(vector, matrix, normalization):
"""Normalize a dense vector after a change of basis.
Parameters
----------
vector : 1xN ndarray
A dense vector.
matrix : NxN ndarray
A change-of-basis matrix.
normalization : {True, False, 'maintain'}
Whether the vector will be L2-normalized (True; corresponds to the soft
cosine measure), maintain its L2-norm during the change of basis
('maintain'; corresponds to query expansion with partial membership),
or kept as-is (False; corresponds to query expansion).
Returns
-------
vector : ndarray
The normalized dense vector.
"""
if not normalization:
return vector
vector_norm = vector.T.dot(matrix).dot(vector)[0, 0]
assert vector_norm >= 0.0, NON_NEGATIVE_NORM_ASSERTION_MESSAGE
if normalization == 'maintain' and vector_norm > 0.0:
vector_norm /= vector.T.dot(vector)
vector_norm = sqrt(vector_norm)
normalized_vector = vector
if vector_norm > 0.0:
normalized_vector /= vector_norm
return normalized_vector
def _normalize_dense_corpus(corpus, matrix, normalization):
"""Normalize a dense corpus after a change of basis.
Parameters
----------
corpus : MxN ndarray
A dense corpus.
matrix : NxN ndarray
A change-of-basis matrix.
normalization : {True, False, 'maintain'}
Whether the vector will be L2-normalized (True; corresponds to the soft
cosine measure), maintain its L2-norm during the change of basis
('maintain'; corresponds to query expansion with partial membership),
or kept as-is (False; corresponds to query expansion).
Returns
-------
normalized_corpus : ndarray
The normalized dense corpus.
"""
if not normalization:
return corpus
# use the following equality: np.diag(A.T.dot(B).dot(A)) == A.T.dot(B).multiply(A.T).sum(axis=1).T
corpus_norm = np.multiply(corpus.T.dot(matrix), corpus.T).sum(axis=1).T
assert corpus_norm.min() >= 0.0, NON_NEGATIVE_NORM_ASSERTION_MESSAGE
if normalization == 'maintain':
corpus_norm /= np.multiply(corpus.T, corpus.T).sum(axis=1).T
corpus_norm = np.sqrt(corpus_norm)
normalized_corpus = np.multiply(corpus, 1.0 / corpus_norm)
normalized_corpus = np.nan_to_num(normalized_corpus) # account for division by zero
return normalized_corpus
def _normalize_sparse_corpus(corpus, matrix, normalization):
"""Normalize a sparse corpus after a change of basis.
Parameters
----------
corpus : MxN :class:`scipy.sparse.csc_matrix`
A sparse corpus.
matrix : NxN :class:`scipy.sparse.csc_matrix`
A change-of-basis matrix.
normalization : {True, False, 'maintain'}
Whether the vector will be L2-normalized (True; corresponds to the soft
cosine measure), maintain its L2-norm during the change of basis
('maintain'; corresponds to query expansion with partial membership),
or kept as-is (False; corresponds to query expansion).
Returns
-------
normalized_corpus : :class:`scipy.sparse.csc_matrix`
The normalized sparse corpus.
"""
if not normalization:
return corpus
# use the following equality: np.diag(A.T.dot(B).dot(A)) == A.T.dot(B).multiply(A.T).sum(axis=1).T
corpus_norm = corpus.T.dot(matrix).multiply(corpus.T).sum(axis=1).T
assert corpus_norm.min() >= 0.0, NON_NEGATIVE_NORM_ASSERTION_MESSAGE
if normalization == 'maintain':
corpus_norm /= corpus.T.multiply(corpus.T).sum(axis=1).T
corpus_norm = np.sqrt(corpus_norm)
normalized_corpus = corpus.multiply(sparse.csr_matrix(1.0 / corpus_norm))
normalized_corpus[normalized_corpus == np.inf] = 0 # account for division by zero
return normalized_corpus
class SparseTermSimilarityMatrix(SaveLoad):
"""
Builds a sparse term similarity matrix using a term similarity index.
Examples
--------
>>> from gensim.test.utils import common_texts as corpus, datapath
>>> from gensim.corpora import Dictionary
>>> from gensim.models import Word2Vec
>>> from gensim.similarities import SoftCosineSimilarity, SparseTermSimilarityMatrix, WordEmbeddingSimilarityIndex
>>> from gensim.similarities.index import AnnoyIndexer
>>>
>>> model_corpus_file = datapath('lee_background.cor')
>>> model = Word2Vec(corpus_file=model_corpus_file, vector_size=20, min_count=1) # train word-vectors
>>>
>>> dictionary = Dictionary(corpus)
>>> tfidf = TfidfModel(dictionary=dictionary)
>>> words = [word for word, count in dictionary.most_common()]
>>> word_vectors = model.wv.vectors_for_all(words, allow_inference=False) # produce vectors for words in corpus
>>>
>>> indexer = AnnoyIndexer(word_vectors, num_trees=2) # use Annoy for faster word similarity lookups
>>> termsim_index = WordEmbeddingSimilarityIndex(word_vectors, kwargs={'indexer': indexer})
>>> similarity_matrix = SparseTermSimilarityMatrix(termsim_index, dictionary, tfidf) # compute word similarities
>>>
>>> tfidf_corpus = tfidf[[dictionary.doc2bow(document) for document in common_texts]]
>>> docsim_index = SoftCosineSimilarity(tfidf_corpus, similarity_matrix, num_best=10) # index tfidf_corpus
>>>
>>> query = 'graph trees computer'.split() # make a query
>>> sims = docsim_index[dictionary.doc2bow(query)] # find the ten closest documents from tfidf_corpus
Check out `the Gallery `_
for more examples.
Parameters
----------
source : :class:`~gensim.similarities.termsim.TermSimilarityIndex` or :class:`scipy.sparse.spmatrix`
The source of the term similarity. Either a term similarity index that will be used for
building the term similarity matrix, or an existing sparse term similarity matrix that will
be encapsulated and stored in the matrix attribute. When a matrix is specified as the
source, any other parameters will be ignored.
dictionary : :class:`~gensim.corpora.dictionary.Dictionary` or None, optional
A dictionary that specifies a mapping between terms and the indices of rows and columns
of the resulting term similarity matrix. The dictionary may only be None when source is
a :class:`scipy.sparse.spmatrix`.
tfidf : :class:`gensim.models.tfidfmodel.TfidfModel` or None, optional
A model that specifies the relative importance of the terms in the dictionary. The columns
of the term similarity matrix will be build in a decreasing order of importance of
terms, or in the order of term identifiers if None.
symmetric : bool, optional
Whether the symmetry of the term similarity matrix will be enforced. Symmetry is a necessary
precondition for positive definiteness, which is necessary if you later wish to derive a
unique change-of-basis matrix from the term similarity matrix using Cholesky factorization.
Setting symmetric to False will significantly reduce memory usage during matrix construction.
dominant: bool, optional
Whether the strict column diagonal dominance of the term similarity matrix will be enforced.
Strict diagonal dominance and symmetry are sufficient preconditions for positive
definiteness, which is necessary if you later wish to derive a change-of-basis matrix from
the term similarity matrix using Cholesky factorization.
nonzero_limit : int or None, optional
The maximum number of non-zero elements outside the diagonal in a single column of the
sparse term similarity matrix. If None, then no limit will be imposed.
dtype : numpy.dtype, optional
The data type of the sparse term similarity matrix.
Attributes
----------
matrix : :class:`scipy.sparse.csc_matrix`
The encapsulated sparse term similarity matrix.
Raises
------
ValueError
If `dictionary` is empty.
See Also
--------
:class:`~gensim.similarities.docsim.SoftCosineSimilarity`
A document similarity index using the soft cosine similarity over the term similarity matrix.
:class:`~gensim.similarities.termsim.LevenshteinSimilarityIndex`
A term similarity index that computes Levenshtein similarities between terms.
:class:`~gensim.similarities.termsim.WordEmbeddingSimilarityIndex`
A term similarity index that computes cosine similarities between word embeddings.
"""
def __init__(self, source, dictionary=None, tfidf=None, symmetric=True, dominant=False,
nonzero_limit=100, dtype=np.float32):
if not sparse.issparse(source):
index = source
args = (index, dictionary, tfidf, symmetric, dominant, nonzero_limit, dtype)
source = _create_source(*args)
assert sparse.issparse(source)
self.matrix = source.tocsc()
def inner_product(self, X, Y, normalized=(False, False)):
"""Get the inner product(s) between real vectors / corpora X and Y.
Return the inner product(s) between real vectors / corpora vec1 and vec2 expressed in a
non-orthogonal normalized basis, where the dot product between the basis vectors is given by
the sparse term similarity matrix.
Parameters
----------
vec1 : list of (int, float) or iterable of list of (int, float)
A query vector / corpus in the sparse bag-of-words format.
vec2 : list of (int, float) or iterable of list of (int, float)
A document vector / corpus in the sparse bag-of-words format.
normalized : tuple of {True, False, 'maintain'}, optional
First/second value specifies whether the query/document vectors in the inner product
will be L2-normalized (True; corresponds to the soft cosine measure), maintain their
L2-norm during change of basis ('maintain'; corresponds to query expansion with partial
membership), or kept as-is (False; corresponds to query expansion; default).
Returns
-------
`self.matrix.dtype`, `scipy.sparse.csr_matrix`, or :class:`numpy.matrix`
The inner product(s) between `X` and `Y`.
References
----------
The soft cosine measure was perhaps first described by [sidorovetal14]_.
Further notes on the efficient implementation of the soft cosine measure are described by
[novotny18]_.
.. [sidorovetal14] Grigori Sidorov et al., "Soft Similarity and Soft Cosine Measure: Similarity
of Features in Vector Space Model", 2014, http://www.cys.cic.ipn.mx/ojs/index.php/CyS/article/view/2043/1921.
.. [novotny18] Vít Novotný, "Implementation Notes for the Soft Cosine Measure", 2018,
http://dx.doi.org/10.1145/3269206.3269317.
"""
if not X or not Y:
return self.matrix.dtype.type(0.0)
normalized_X, normalized_Y = normalized
valid_normalized_values = (True, False, 'maintain')
if normalized_X not in valid_normalized_values:
raise ValueError('{} is not a valid value of normalize'.format(normalized_X))
if normalized_Y not in valid_normalized_values:
raise ValueError('{} is not a valid value of normalize'.format(normalized_Y))
is_corpus_X, X = is_corpus(X)
is_corpus_Y, Y = is_corpus(Y)
if not is_corpus_X and not is_corpus_Y:
X = dict(X)
Y = dict(Y)
word_indices = np.array(sorted(set(chain(X, Y))))
dtype = self.matrix.dtype
X = np.array([X[i] if i in X else 0 for i in word_indices], dtype=dtype)
Y = np.array([Y[i] if i in Y else 0 for i in word_indices], dtype=dtype)
matrix = self.matrix[word_indices[:, None], word_indices].todense()
X = _normalize_dense_vector(X, matrix, normalized_X)
Y = _normalize_dense_vector(Y, matrix, normalized_Y)
result = X.T.dot(matrix).dot(Y)
if normalized_X is True and normalized_Y is True:
result = np.clip(result, -1.0, 1.0)
return result[0, 0]
elif not is_corpus_X or not is_corpus_Y:
if is_corpus_X and not is_corpus_Y:
X, Y = Y, X # make Y the corpus
is_corpus_X, is_corpus_Y = is_corpus_Y, is_corpus_X
normalized_X, normalized_Y = normalized_Y, normalized_X
transposed = True
else:
transposed = False
dtype = self.matrix.dtype
expanded_X = corpus2csc([X], num_terms=self.matrix.shape[0], dtype=dtype).T.dot(self.matrix)
word_indices = np.array(sorted(expanded_X.nonzero()[1]))
del expanded_X
X = dict(X)
X = np.array([X[i] if i in X else 0 for i in word_indices], dtype=dtype)
Y = corpus2csc(Y, num_terms=self.matrix.shape[0], dtype=dtype)[word_indices, :].todense()
matrix = self.matrix[word_indices[:, None], word_indices].todense()
X = _normalize_dense_vector(X, matrix, normalized_X)
Y = _normalize_dense_corpus(Y, matrix, normalized_Y)
result = X.dot(matrix).dot(Y)
if normalized_X is True and normalized_Y is True:
result = np.clip(result, -1.0, 1.0)
if transposed:
result = result.T
return result
else: # if is_corpus_X and is_corpus_Y:
dtype = self.matrix.dtype
X = corpus2csc(X if is_corpus_X else [X], num_terms=self.matrix.shape[0], dtype=dtype)
Y = corpus2csc(Y if is_corpus_Y else [Y], num_terms=self.matrix.shape[0], dtype=dtype)
matrix = self.matrix
X = _normalize_sparse_corpus(X, matrix, normalized_X)
Y = _normalize_sparse_corpus(Y, matrix, normalized_Y)
result = X.T.dot(matrix).dot(Y)
if normalized_X is True and normalized_Y is True:
result.data = np.clip(result.data, -1.0, 1.0)
return result