全文搜索无处不在。在 Scribd(一个文档分享平台)上搜索一本书,在 Netflix 上搜索一部电影,在亚马逊上搜索卫生纸商品,或者通过谷歌搜索东西,你都在搜索大量的非结构化数据。更令人感到惊奇地是,即使你搜索的是数百万(或数十亿)条记录,也能够获得毫秒级的响应体验。在这篇文章中,我们将探索全文搜索引擎的基本组件,并用它们来构建一个可以搜索数百万个文档、根据相关性对文档进行排名的搜索引擎。我们将用不到 150 行的 Python 代码来开发这个搜索引擎。
这篇文章中所有的代码都可以在 Github 上找到(https://github.com/bartdegoede/python-searchengine/)。我将在文章中提供代码片段和链接,你可以尝试自己运行它们。你可以安装运行示例所需的组件(pip install -r requirements.txt),然后运行 python run.py(https://github.com/bartdegoede/python-searchengine/blob/master/run.py)。它会下载所有的数据,并运行带排名和不带排名的搜索示例。
在开始构建搜索引擎之前,我们需要一些非结构化的数据。我们将搜索英文维基百科中的文章摘要。维基百科被打包成一个约 785MB 的压缩 XML 文件包,其中包含了约 627 万篇摘要。我写了一个简单的函数(https://github.com/bartdegoede/python-searchengine/blob/master/download.py)用来下载 XML 压缩包,当然你也可以手动下载这个文件。
这个文件是一个包含所有摘要的大型 XML 文件。每一个摘要内容都包含在<doc>标签中,看起来大致如下所示(我省略了我们不感兴趣的标签):
<doc>
<title>Wikipedia: London Beer Flood</title>
<url>https://en.wikipedia.org/wiki/London_Beer_Flood</url>
<abstract>The London Beer Flood was an accident at Meux & Co's Horse Shoe Brewery, London, on 17 October 1814. It took place when one of the wooden vats of fermenting porter burst.</abstract>
...
</doc>
我们感兴趣的是 title、url 和 abstract 这几个标签。为了方便访问数据,我们将用 Python 数据类(https://realpython.com/python-data-classes/)来表示文档。我们将添加一个属性来连接标题和摘要内容,代码可以在这里找到(https://github.com/bartdegoede/python-searchengine/blob/master/search/documents.py)。
from dataclasses import dataclass
@dataclass
class Abstract:
"""Wikipedia abstract"""
ID: int
title: str
abstract: str
url: str
@property
def fulltext(self):
return ' '.join([self.title, self.abstract])
然后,我们从 XML 中提取摘要数据,对其进行解析,并创建 Abstract 实例。我们将通过流的方式来读取 XML,不会将整个文件加载到内存中。我们将按照加载顺序为每个文档分配一个 ID(即第一个文档 ID=1,第二个文档 ID=2,以此类推)。相关代码可以在这里找到(https://github.com/bartdegoede/python-searchengine/blob/master/load.py)。
import gzip
from lxml import etree
from search.documents import Abstract
def load_documents():
# open a filehandle to the gzipped Wikipedia dump
with gzip.open('data/enwiki.latest-abstract.xml.gz', 'rb') as f:
doc_id = 1
# iterparse will yield the entire `doc` element once it finds the
# closing `</doc>` tag
for _, element in etree.iterparse(f, events=('end',), tag='doc'):
title = element.findtext('./title')
url = element.findtext('./url')
abstract = element.findtext('./abstract')
yield Abstract(ID=doc_id, title=title, url=url, abstract=abstract)
doc_id += 1
# the `element.clear()` call will explicitly free up the memory
# used to store the element
element.clear()
我们将把这些数据保存成“倒排索引”。我们可以把它想象成一本书后面的索引,它有一个按字母顺序排列的单词和概念的清单,读者可以在相应的页码上找到它们。
我们需要创建一个字典,将语料库所有的单词与它们所在文档 ID 映射起来,看起来是这样的:
{
...
"london": [5245250, 2623812, 133455, 3672401, ...],
"beer": [1921376, 4411744, 684389, 2019685, ...],
"flood": [3772355, 2895814, 3461065, 5132238, ...],
...
}
注意,在上面的例子中,字典中的单词都是小写的。在构建索引之前,我们需要把原始文本分解为单词。我们首先将文本分解为单词,然后对每个单词应用零个或多个过滤器(如小写或词干筛选),以提高查询与文本匹配的几率。
我们将进行非常简单的解析,只根据空格来拆分文本。然后,我们对每个单词进行过滤:将单词转成小写,移除标点符号,移除英语中最常见的 25 个单词(包括“维基百科”这个单词,因为它出现在每一个标题和摘要中),并提取词干(确保不同形式的同一个单词映射到相同的词干,如 brewery 和 breweries)。
分解和小写过滤器非常简单:
import Stemmer
STEMMER = Stemmer.Stemmer('english')
def tokenize(text):
return text.split()
def lowercase_filter(text):
return [token.lower() for token in tokens]
def stem_filter(tokens):
return STEMMER.stemWords(tokens)
使用正则表达式来移除标点符号:
import re
import string
PUNCTUATION = re.compile('[%s]' % re.escape(string.punctuation))
def punctuation_filter(tokens):
return [PUNCTUATION.sub('', token) for token in tokens]
停顿词是非常常见的单词,(几乎)会出现在语料库的每一篇文档中。因此,当我们在搜索它们时,它们不会对搜索有多大贡献(因为几乎每个文档都会匹配到),它们只会占用更多的空间,所以我们会在进行索引时过滤掉它们。维基百科摘要语料库的每个标题中都包含“Wikipedia”一词,因此我们将把这个词也添加到停顿词清单中。我们还去掉了英语中最常见的 25 个单词。
# top 25 most common words in English and "wikipedia":
# https://en.wikipedia.org/wiki/Most_common_words_in_English
STOPWORDS = set(['the', 'be', 'to', 'of', 'and', 'a', 'in', 'that', 'have',
'I', 'it', 'for', 'not', 'on', 'with', 'he', 'as', 'you',
'do', 'at', 'this', 'but', 'his', 'by', 'from', 'wikipedia'])
def stopword_filter(tokens):
return [token for token in tokens if token not in STOPWORDS]
将所有这些过滤器组合在一起变成 analyze 函数,它将处理每个摘要中的文本,将文本拆分为单词(更确切地说是节点),然后对每一个单词进行过滤。顺序很重要,因为我们使用了一个没有经过提取词干的停顿词清单,所以应该在 stem_filter 之前应用 stopword_filter。
def analyze(text):
tokens = tokenize(text)
tokens = lowercase_filter(tokens)
tokens = punctuation_filter(tokens)
tokens = stopword_filter(tokens)
tokens = stem_filter(tokens)
return [token for token in tokens if token]
我们将创建一个 Index 类来存储 index 和 documents。documents 字典按照 ID 来存储数据类,index 的键就是单词,值就是单词所在的文档的 ID:
class Index:
def __init__(self):
self.index = {}
self.documents = {}
def index_document(self, document):
if document.ID not in self.documents:
self.documents[document.ID] = document
for token in analyze(document.fulltext):
if token not in self.index:
self.index[token] = set()
self.index[token].add(document.ID)
现在我们已经为所有单词建立了索引,接下来的搜索就要用到分析文档时所使用的分析器,这样就可以得到与索引中的单词相匹配的单词。对于每个单词,我们将在字典中进行查找,查找单词所在的文档 ID。每一个单词都要这么做,然后找出在所有这些集合中都存在的文档 ID(也就是说,目标文档需要包含所有的查询单词)。然后,我们将获取文档 ID 结果列表,并从 documents 中获取实际的数据。
def _results(self, analyzed_query):
return [self.index.get(token, set()) for token in analyzed_query]
def search(self, query):
"""
Boolean search; this will return documents that contain all words from the
query, but not rank them (sets are fast, but unordered).
"""
analyzed_query = analyze(query)
results = self._results(analyzed_query)
documents = [self.documents[doc_id] for doc_id in set.intersection(*results)]
return documents
In [1]: index.search('London Beer Flood')
search took 0.16307830810546875 milliseconds
Out[1]:
[Abstract(ID=1501027, title='Wikipedia: Horse Shoe Brewery', abstract='The Horse Shoe Brewery was an English brewery in the City of Westminster that was established in 1764 and became a major producer of porter, from 1809 as Henry Meux & Co. It was the site of the London Beer Flood in 1814, which killed eight people after a porter vat burst.', url='https://en.wikipedia.org/wiki/Horse_Shoe_Brewery'),
Abstract(ID=1828015, title='Wikipedia: London Beer Flood', abstract="The London Beer Flood was an accident at Meux & Co's Horse Shoe Brewery, London, on 17 October 1814. It took place when one of the wooden vats of fermenting porter burst.", url='https://en.wikipedia.org/wiki/London_Beer_Flood')]
这样会让搜索非常精确,特别是在使用较长的字符串进行搜索时(搜索包含的单词越多,文档中包含所有这些单词的可能性就越小)。我们可以优化搜索函数,允许用户指定当有一个单词匹配时就算匹配整个搜索,以提高召回率(recall)而不是精确度:
def search(self, query, search_type='AND'):
"""
Still boolean search; this will return documents that contain either all words
from the query or just one of them, depending on the search_type specified.
We are still not ranking the results (sets are fast, but unordered).
"""
if search_type not in ('AND', 'OR'):
return []
analyzed_query = analyze(query)
results = self._results(analyzed_query)
if search_type == 'AND':
# all tokens must be in the document
documents = [self.documents[doc_id] for doc_id in set.intersection(*results)]
if search_type == 'OR':
# only one token has to be in the document
documents = [self.documents[doc_id] for doc_id in set.union(*results)]
return documents
In [2]: index.search('London Beer Flood', search_type='OR')
search took 0.02816295623779297 seconds
Out[2]:
[Abstract(ID=5505026, title='Wikipedia: Addie Pryor', abstract='| birth_place = London, England', url='https://en.wikipedia.org/wiki/Addie_Pryor'),
Abstract(ID=1572868, title='Wikipedia: Tim Steward', abstract='|birth_place = London, United Kingdom', url='https://en.wikipedia.org/wiki/Tim_Steward'),
Abstract(ID=5111814, title='Wikipedia: 1877 Birthday Honours', abstract='The 1877 Birthday Honours were appointments by Queen Victoria to various orders and honours to reward and highlight good works by citizens of the British Empire. The appointments were made to celebrate the official birthday of the Queen, and were published in The London Gazette on 30 May and 2 June 1877.', url='https://en.wikipedia.org/wiki/1877_Birthday_Honours'),
...
In [3]: len(index.search('London Beer Flood', search_type='OR'))
search took 0.029065370559692383 seconds
Out[3]: 49627
我们已经用 Python 实现了一个非常快的搜索引擎,但还少了个东西,那就是相关度。现在我们只返回一个无序的文档列表,并由用户来确定他们真正感兴趣的是哪些文档。如果返回的是一个大型的结果集,那将是一件很痛苦的事情,或者根本不可能确定哪些是用户真正感兴趣的(在我们的 OR 示例中,返回将近 50000 个结果)。
相关度的概念是这样来的:我们可以给每个文档分配一个分数,表示它与查询的匹配度,并根据这个分数进行排序。给文档分配分数的一种简单的方法是计算文档出现检索词的频率。毕竟,文档出现某个检索词的次数越多,它就越有可能与我们的搜索相关!
我们将 Abstract 数据类扩展一下,在建立索引时计算并存储它的检索词频率。这样,当我们想对无序列表中的文档进行排序时,就可以很容易地使用这些数字:
# in documents.py
from collections import Counter
from .analysis import analyze
@dataclass
class Abstract:
# snip
def analyze(self):
# Counter will create a dictionary counting the unique values in an array:
# {'london': 12, 'beer': 3, ...}
self.term_frequencies = Counter(analyze(self.fulltext))
def term_frequency(self, term):
return self.term_frequencies.get(term, 0)
我们要确保在建立索引时生成这些频率计数:
# in index.py we add `document.analyze()
def index_document(self, document):
if document.ID not in self.documents:
self.documents[document.ID] = document
document.analyze()
接下来我们要修改搜索函数,以便对结果集中的文档进行排名。我们先从索引和文档存储中获取文档,对于结果集中的每个文档,我们简单地将每个检索词在该文档中出现的频率相加起来。
def search(self, query, search_type='AND', rank=True):
# snip
if rank:
return self.rank(analyzed_query, documents)
return documents
def rank(self, analyzed_query, documents):
results = []
if not documents:
return results
for document in documents:
score = sum([document.term_frequency(token) for token in analyzed_query])
results.append((document, score))
return sorted(results, key=lambda doc: doc[1], reverse=True)
这样已经好多了,但仍然有一些明显的不足。在评估搜索相关度时,我们认为所有的搜索条件都是等价的。但实际上,某些检索词可能只有很小的识别度,甚至没有。例如,如果一个文档集合大都包含了“啤酒”这个单词,“啤酒”这个单词会经常出现在几乎每个文档中(我们已经试图通过从索引中移除 25 个最常见的英语单词来解决这个问题)。对于这种情况,搜索“啤酒”实质上是进行了另一种随机的排序。
为了解决这个问题,我们将在评分算法中添加另一个组件,以减少索引中出现频率较高的检索词对最终分数的影响。我们可以使用检索词集合频率(即这个检索词在所有文档中出现的频率),但实际上使用的是逆文本频率(即索引中有多少个文档包含这个检索词)。因为我们要对文档进行排序,所以需要文档级别的统计信息。
我们用索引中的文档数量(N)除以包含检索词的文档数量,并对其取对数,得出检索词的逆文本频率。
然后,在进行排名时,我们将检索词频率与逆文本频率相乘,这样语料库中出现较少的检索词将对相关度得分有更大的影响。我们很容易就能根据索引中可用的数据计算出逆文本频率:
# index.py
import math
def document_frequency(self, token):
return len(self.index.get(token, set()))
def inverse_document_frequency(self, token):
# Manning, Hinrich and Schütze use log10, so we do too, even though it
# doesn't really matter which log we use anyway
# https://nlp.stanford.edu/IR-book/html/htmledition/inverse-document-frequency-1.html
return math.log10(len(self.documents) / self.document_frequency(token))
def rank(self, analyzed_query, documents):
results = []
if not documents:
return results
for document in documents:
score = 0.0
for token in analyzed_query:
tf = document.term_frequency(token)
idf = self.inverse_document_frequency(token)
score += tf * idf
results.append((document, score))
return sorted(results, key=lambda doc: doc[1], reverse=True)
这是一个很基础的搜索引擎,只需要几行 Python 代码!你可以在 Github 上找到所有的代码(https://github.com/bartdegoede/python-searchengine),我还提供了一个实用函数,可以下载维基百科摘要并构建索引。安装好必要的组件,在 Python 控制台中运行它,并享受数据结构和搜索给你带来的乐趣吧。
当然,这个项目是为了解释搜索的概念,以及搜索为何会如此之快(我可以用 Python 这样的“慢”语言在我的笔记本电脑上搜索 627 万个文档,并进行排名)。一些开源项目(如 Lucene)利用了非常高效的数据结构,甚至优化了磁盘搜索,还有一些项目(如 Elasticsearch 和 Solr)将 Lucene 扩展到可以运行在数百台甚至数千台机器上。
我们也可以考虑对这个只具有基本功能的搜索引擎做一些扩展。例如,我们假设文档中的每个字段对相关度都有相同的贡献,而标题中的检索词匹配应该比摘要中的检索词匹配具有更大的权重。另外,我们也可以对解析器进行扩展,既可以匹配所有检索词,有可以匹配单个检索词。我们也可以忽略某些检索词,或者支持检索词之间的 AND 或 OR 关系。我们也可以将索引持久化到磁盘上,打破单台笔记本电脑内存的限制。
原文链接:
https://bart.degoe.de/building-a-full-text-search-engine-150-lines-of-code/
领取专属 10元无门槛券
私享最新 技术干货