CSC 4538 – Introduction à la science des données

CI8 : Recherche d'informations

Implémenter un moteur de recherche simple.

Recherche d'informations

Dans cet exercice, nous allons créer un moteur de recherche. Pour cela, nous allons continuer le code commencé pendant le TP sur l'introduction au traitement du texte.

Nous allons commencer par créer notre index inversé. Pour cela, créez une classe InvertedIndex. Dans cette classe, ajoutez un constructeur qui initialise trois dictionnaires : un premier transformant un terme en une liste ordonnée de documents et le deuxième qui contient pour chaque terme le nombre de documents dans lequel il apparaît, et le troisième qui contient l'identifiant unique de chaque terme. Ajoutez également une liste vide qui contiendra tous les documents. L'identifiant d'un document sera l'index du document dans cette liste.
Ajoutez également un champs nlp contenant les opérations en anglais de Spacy.

class InvertedIndex:

    def __init__(self):
        self.inverted_index = {}
        self.document_frequency = {}
        self.documents = []
        self.term_to_id = dict()
        self.nlp = spacy.load('en_core_web_sm')
        

Reprenez la fonction normalize(text) du TP sur l'introduction au traitement du langage naturel et ajoutez-la à la classe InvertedIndex. Si ce n'est pas déjà le cas, ajoutez le nettoyage par regex qui permettait d'enlever les videos et les sources des images.

    def normalize(self, text):
        text = RE_VIDEO.sub("", text)
        text = RE_SOURCES.sub("", text)
        text = text.replace("\n", " ").replace("  ", " ")
        doc = self.nlp(text)
        return [token.lemma_.lower() for token in doc if not token.is_stop and not token.is_punct]
        

Écrivez une fonction add_document(document) qui ajoute un document à l'index. Pour cela, il faut :

Ajouter le document à la liste de documents et récupérer son index
Normaliser le document
Prendre le set des termes
Pour chaque terme, ajouter le document aux deux dictionnaires (dans une liste pour l'un, en incrémentant le compteur pour l'autre)
Pour chaque terme, s'il n'a pas été rencontré avant, lui attribuer un identifiant unique (en commençant à 0)

    def add_document(self, document):
        self.documents.append(document)
        index = len(self.documents) - 1
        terms = set(self.normalize(document, self.nlp))
        for term in terms:
            if term not in self.inverted_index:
                self.inverted_index[term] = []
                self.document_frequency[term] = 0
                self.term_to_id[term] = len(self.inverted_index) - 1
            self.inverted_index[term].append(index)
            self.document_frequency[term] += 1
        

Écrivez une méthode save(self, filename) et une fonction load(filename) qui respectivement enregistrent et charge l'index dans un fichier. Pour cela, on pourra enregistrer les attributs sous la forme d'un fichier JSON.

    def save(self, filename):
        to_save = {
            "inverted_index": self.inverted_index,
            "document_frequency": self.document_frequency,
            "documents": self.documents,
            "term_to_id": self.term_to_id,
        }
        with open(filename, "w") as f:
            json.dump(to_save, f)

    @staticmethod
    def load(filename):
        index = InvertedIndex()
        with open(filename, "r") as f:
            to_load = json.load(f)
        index.inverted_index = to_load["inverted_index"]
        index.document_frequency = to_load["document_frequency"]
        index.documents = to_load["documents"]
        index.term_to_id = to_load["term_to_id"]
        return index
        

Écrivez une fonction search_conjunction(words) qui retourne tous les documents contenants tous les mots dans words.

On pourra écrire une fonction annexe merge_conjunction(l1, l2) qui fusionne les deux listes de manière adéquate.

    def merge_conjunction(l1, l2):
        idx1 = 0
        idx2 = 0
        res = []
        while idx1  l2[idx2]:
                idx2 += 1
            else:
                res.append(l1[idx1])
                idx1 += 1
                idx2 += 1
        return res

    def search_conjunction(self, words):
        first_word = words[0]
        if first_word not in self.inverted_index:
            return []
        res = self.inverted_index[first_word]
        if len(words) == 1:
            return [self.documents[i] for i in res]
        for word in words[1:]:
            if word not in self.inverted_index:
                return []
            res = self.merge_conjunction(res, self.inverted_index[word])
        return [self.documents[i] for i in res]
        

Écrivez une fonction search_disjunction(words) qui retourne tous les documents contenants au moins un mot de words.

On pourra écrire une fonction annexe merge_disjunction(l1, l2) qui fusionne les deux listes de manière adéquate.

    def merge_disjunction(l1, l2):
        idx1 = 0
        idx2 = 0
        res = []
        while idx1 = len(l2) or (idx1 = len(l1) or l1[idx1] > l2[idx2]:
                res.append(l2[idx2])
                idx2 += 1
            else:
                res.append(l1[idx1])
                idx1 += 1
                idx2 += 1
        return res

    def search_disjunction(self, words):
        res = []
        for word in words:
            if word not in self.inverted_index:
                continue
            res = self.merge_disjunction(res, self.inverted_index[word])
        return [self.documents[i] for i in res]
        

(Optionnel) Adaptez votre code pour faire de l'optimisation de requête comme vu en cours.

(Optionnel) Vous pouvez rajouter la possibilité de recherche d'autres opérations booléennes.

Écrivez une fonction get_cosine_similarity(u, v) calculant la similarité cosinus entre les vecteurs u et v supposés de même taille.

    @staticmethod
    def get_cosine_similarity(u, v):
        norm_u = 0
        norm_v = 0
        dot_product = 0
        for x, y in zip(u, v):
            norm_u += x * x
            norm_v += y * y
            dot_product += x * y
        return dot_product / math.sqrt(norm_u * norm_v)
        

Importez et adaptez les fonctions get_bow(text) et get_tfidf(text) du dernier TP sur le TAL dans la classe InvertedIndex.

    def get_bow(self, text):
        res = [0] * len(self.inverted_index)
        for word in text:
            res[self.term_to_id[word]] += 1
        return res

    def get_tfidf(self, text):
        res = [0] * len(self.inverted_index)
        text = self.normalize(text)
        bow = self.get_bow(text)
        for term, i in self.term_to_id.items():
            res[i] = bow[i] * math.log(len(self.documents) / self.document_frequency[term])
        return res
        

Écrivez une fonction get_top_k(query, k, vocab, to_idx, n_words_in_documents, n_documents) qui retourne les top k meilleurs documents pour la requête query. Pour cela :

Calculez le vecteur tfidf de la requête
Normalisez la requête et récupérer le set des termes
Récupérez tous les documents contenant au moins un terme de la requête
Pour chaque document trouvé, calculez sa similarité cosinus avec le vecteur de la requête
Retournez les top k documents avec la similarité la plus faible

Les top k meilleurs résultats peuvent être obtenu en utilisant la fonction de tri de Python sorted avec le bon paramètre de key (allez voir la documentation)

    def get_top_k(self, query, k):
        query_vector = self.get_tfidf(query)
        terms_query = list(set(self.normalize(query)))
        search_results = self.search_disjunction(terms_query)
        similarities = []
        for document in search_results:
            vector = self.get_tfidf(document)
            similarities.append(self.get_cosine_similarity(query_vector, vector))
        ranked_results = sorted(zip(similarities, search_results), key=lambda x: -x[0])
        return ranked_results[:k]


if __name__ == '__main__':
    filename = "index.json"
    save = False
    if save:
        inverted_index = InvertedIndex()
        articles = json.load(open('../intro_nlp/all_articles.json'))
        for article in articles:
            inverted_index.add_document(article)
        inverted_index.save(filename)
    else:
        inverted_index = InvertedIndex.load(filename)
    search_results = inverted_index.search_conjunction(["elon", "musk", "twitter"])
    print("Number of results:", len(search_results))
    if search_results:
        print(search_results[0])
    search_results = inverted_index.search_conjunction(["microsoft", "gates"])
    print("Number of results:", len(search_results))
    if search_results:
        print("Article", search_results[0])
    search_results = inverted_index.search_disjunction(["elon", "musk", "twitter"])
    print("Number of results:", len(search_results))
    print(inverted_index.get_top_k("elon musk and twitter", 1))
        