简介:检索相关算法的学习


1.TF-IDF

1.1原理

TF:term frequency(词频)

IDF:inverse document frequency(逆文档频率)

字词的重要性随着它在文件中出现的次数成正比增加,但同时会随着它在语料库中出现的频率成反比下降。

TF-IDF的主要思想是:如果某个单词在一篇文章中出现的频率高(TF高),并且在其他文章中很少出现(IDF高),则认为此词或者短语具有很好的类别区分能力,适合用来分类。


$TF=\frac{某个词在文章中出现的次数}{文章的总词数}$

考虑到文章长短不同,除以总词数进行标准化


$IDF=log(\frac{语料库的文章总数}{包含该词的文章数量+1})$

加1防止不存在包含该词的文档时分母为0


$TF-IDF=TF \times IDF$


优点:

  1. 简单快速、易理解

缺点:

  1. 没考虑词语的语义
  2. 仅用词频考虑词语的重要性不够全面。按照传统TF-IDF,往往一些生僻词的IDF(反文档频率)会比较高、因此这些生僻词常会被误认为是文档关键词。
  3. 没有考虑特征词的位置因素对文本的区分度,词条出现在文档的不同位置时,对区分度的贡献大小是不一样的。

1.2自己实现

import math


# 计算每句话的词频
def counter(word_list):
    wordcount = []
    for doc in word_list:
        count = {}
        for word in doc:
            count[word] = count.get(word, 0) + 1
        wordcount.append(count)
    return wordcount


# 计算tf=某个词在文章中出现的总次数/文章的总词数
def tf(word, word_list):
    return word_list.get(word) / sum(word_list.values())


# 统计含有该单词的句子数
def count_sentence(word, wordcount):
    return sum(1 for i in wordcount if i.get(word))


# 计算idf=log(语料库中的文档总数/(包含该词的文档数+1))
def idf(word, wordcount):
    return math.log(len(wordcount) + 1 / count_sentence(word, wordcount) + 1) + 1


# tf-idf=tf*idf
def tfidf(word, word_list, wordcount):
    return tf(word, word_list) * idf(word, wordcount)


if __name__ == "__main__":
    docs = [
        "what is the weather like today",
        "what is for dinner tonight",
        "this is a question worth pondering",
        "it is a beautiful day today"
    ]

    word_list = []  # 记录每个文档分词后的结果
    for doc in docs:
        word_list.append(doc.split(" "))

    # 使用停用词
    # stopwords = ["is", "the"]
    # for i in docs:
    #     all_words = i.split()
    #     new_words = []
    #     for j in all_words:
    #         if j not in stopwords:
    #             new_words.append(j)
    #     word_list.append(new_words)

    wordcount = counter(word_list)  # 统计每个文档词的次数

    for cnt, doc in enumerate(wordcount):
        print("doc{}".format(cnt))
        for word, _ in doc.items():
            print("word:{} --- TF-IDF:{}".format(word, tfidf(word, doc, wordcount)))

1.3使用sklearn库

from sklearn.feature_extraction.text import TfidfVectorizer


if __name__ == "__main__":
    docs = [
        "what is the weather like today",
        "what is for dinner tonight",
        "this is a question worth pondering",
        "it is a beautiful day today"
    ]

    tfidf_vec = TfidfVectorizer()

    # 利用fit_transform得到TFIDF矩阵
    tfidf_matrix = tfidf_vec.fit_transform(docs)

    # 利用get_feature_names_out得到不重复的单词
    print(tfidf_vec.get_feature_names_out())

    # 利用vocabulary_得到各单次的编号
    print(tfidf_vec.vocabulary_)

    # 输出TFIDF矩阵,即每个文档中每个词的tfidf值
    print(tfidf_matrix)

2.BM25

2.1原理

BM25是一种基于概率检索框架的排序函数,用于计算查询(Query)与文档(Document)的相关性得分。

对于TF-IDF算法,如果文档长,词语多,TF值就会很大。BM25通过b参数对文档长度进行打压,随着TF的增加,BM25score趋于稳定,如下图所示。

P1


符号说明:

  • $ Q $: 查询,由一组词项组成 $ Q = (q_1, q_2, ..., q_n) $
  • $ D $: 文档
  • $ f_i $: 词项 $ q_i $ 在文档 $ D $ 中的出现频率(词频,TF)
  • $ dl $: 文档长度(文档中的词项总数)
  • $ avgdl $: 语料库中所有文档的平均长度
  • $ N $: 语料库中文档总数
  • $ n_i $: 包含词项 $ q_i $ 的文档数量
  • $ k $: 词频调节参数(通常取1.2~2.0)
  • $ b $: 文档长度调节参数(通常取0.75)
    参数典型值作用
    $ k $1.2~2.0控制词频饱和度,值越大饱和度变化越慢
    $ b $0.75控制文档长度归一化程度,0=不归一化,1=完全归一化

b 参数默认 0.75,主要是对长文档做惩罚。如果设置为 0,分数将与文档的长度无关。从下图可以看到,b=0时,L与分数无关,b=1时,L越大,分数打压越厉害。

P2


完整公式:

$$ \text{BM25}(D, Q) = \sum_{i=1}^{n} \frac{{(k+1) \cdot f_i}}{{f_i + k \cdot (1 - b + b \cdot \frac{{\lvert D \rvert}}{{\text{avgdl}}})}} \cdot \log\left(\frac{{N - n_i + 0.5}}{{n_i + 0.5}}\right) $$

单个词项得分

对于查询中的单个词项 $ q_i $,其BM25得分为:

$\text{score}(q_i, D) = \text{IDF}(q_i) \cdot \frac{f_i \cdot (k_1 + 1)}{f_i + k_1 \cdot (1 - b + b \cdot \frac{dl}{avgdl})}$

IDF计算

逆文档频率(IDF)的计算公式为:

$\text{IDF}(q_i) = \log \left( \frac{N - n_i + 0.5}{n_i + 0.5} + 1 \right)$

整体查询得分

对于完整查询 $ Q $,BM25得分为所有词项得分的总和: $\text{BM25}(Q, D) = \sum_{i=1}^{n} \text{score}(q_i, D)$


2.2自己实现

import math
from collections import Counter


class BM25:
    def __init__(self, docs, k1=1.5, b=0.75):
        """
        BM25算法的构造器
        :param docs: 分词后的文档列表,每个文档是一个包含词汇的列表
        :param k1: BM25算法中的调节参数k1
        :param b: BM25算法中的调节参数b
        """
        self.docs = docs
        self.k1 = k1
        self.b = b
        self.doc_len = [len(doc) for doc in docs]  # 计算每个文档的长度
        self.avgdl = sum(self.doc_len) / len(docs)  # 计算所有文档的平均长度
        self.doc_freqs = []  # 存储每个文档的词频
        self.idf = {}  # 存储每个词的逆文档频率
        self.initialize()

    def initialize(self):
        """
        初始化方法,计算所有词的逆文档频率
        """
        df = {}  # 用于存储每个词在多少不同文档中出现
        for doc in self.docs:
            # 为每个文档创建一个词频统计
            self.doc_freqs.append(Counter(doc))
            # 更新df值
            for word in set(doc):
                df[word] = df.get(word, 0) + 1
        # 计算每个词的IDF值(出现的文档数越少,得分越高)
        for word, freq in df.items():
            self.idf[word] = math.log((len(self.docs) - freq + 0.5) / (freq + 0.5) + 1)

    def score(self, doc, query):
        """
        计算文档与查询的BM25得分
        :param doc: 文档的索引
        :param query: 查询词列表
        :return: 该文档与查询的相关性得分
        """
        score = 0.0
        for word in query:
            if word in self.doc_freqs[doc]:
                freq = self.doc_freqs[doc][word]  # 词在文档中的频率
                # 应用BM25计算公式
                score += (self.idf[word] * freq * (self.k1 + 1)) / (
                            freq + self.k1 * (1 - self.b + self.b * self.doc_len[doc] / self.avgdl))
        return score


if __name__ == "__main__":
    # 示例文档集和查询
    docs = [["the", "quick", "brown", "fox"],
            ["the", "lazy", "dog"],
            ["the", "quick", "dog"],
            ["the", "quick", "brown", "brown", "fox"]]
    query = ["quick", "brown"]

    # 初始化BM25模型并计算得分
    bm25 = BM25(docs)
    scores = [bm25.score(i, query) for i in range(len(docs))]
    print(scores)

2.3rank_bm25库

from rank_bm25 import BM25Okapi
import jieba

corpus = [
    "BM25是一种常用的信息检索算法",
    "这个Python库实现了BM25算法",
    "信息检索是搜索引擎的核心技术",
    "BM25比传统的TF-IDF效果更好",
    "中文信息检索需要先进行分词处理",
    "自然语言处理是人工智能的重要领域",
    "Python是最受欢迎的编程语言之一",
]

tokenized_corpus = [list(jieba.cut(doc)) for doc in corpus]

bm25 = BM25Okapi(tokenized_corpus)


def search(query, top_n=3):
    tokenized_query = list(jieba.cut(query))
    doc_scores = bm25.get_scores(tokenized_query)
    top_docs = bm25.get_top_n(tokenized_query, corpus, n=top_n)
    return doc_scores, top_docs


query = "Python信息检索"
print(f"查询: '{query}'")

scores, results = search(query)

print("\n相关文档:")
for i, (score, doc) in enumerate(zip(scores, corpus)):
    print(f"{i}. [Score: {score:.2f}] {doc}")

print("\n最相关的3个文档:")
for i, doc in enumerate(results):
    print(f"{i+1}. {doc}")