每次搜索都扫描全部文章，逻辑简单，但同一份文本会被反复处理。倒排索引先把文本拆成词项，保存“这个词项出现在哪些文档”，查询时再合并候选集合。本练习用标准库建立一个本地原型，重点是索引不变量和正确性，并不替代博客现有的搜索实现。
明确检索语义
输入语料是“文档ID到正文字符串”的JSON对象。基础汉字逐字作为词项，连续的其他字母数字构成单词；先做NFKC规范化与大小写折叠。查询词项用AND连接，文档含有全部词项才命中，重复查询词不重复加权。空查询返回空列表。
这不是中文词法分词，也不保证查询字符在文中相邻出现。查询“算法”会检查“算”和“法”两个字是否都存在，可能出现隔得很远的匹配。要支持短语、词义或相关性，需要保存词位、引入分词或改变评分模型；先把当前语义说清楚，才能判断结果是否正确。
from collections import Counter, defaultdict
import heapq
import unicodedata
def tokenize(text):
text = unicodedata.normalize('NFKC', text).casefold()
tokens, word = [], []
for char in text:
if '\u4e00' <= char <= '\u9fff':
if word:
tokens.append(''.join(word))
word.clear()
tokens.append(char)
elif char.isalnum():
word.append(char)
elif word:
tokens.append(''.join(word))
word.clear()
if word:
tokens.append(''.join(word))
return tokens
class SearchIndex:
def __init__(self):
self.documents = {}
self.postings = defaultdict(set)
def remove(self, doc_id):
old = self.documents.pop(doc_id, None)
if old is None:
return False
for token in old:
self.postings[token].remove(doc_id)
if not self.postings[token]:
del self.postings[token]
return True
def upsert(self, doc_id, text):
if not isinstance(doc_id, str) or not isinstance(text, str):
raise ValueError('document ID and text must be strings')
self.remove(doc_id)
counts = Counter(tokenize(text))
self.documents[doc_id] = counts
for token in counts:
self.postings[token].add(doc_id)
def search(self, query, limit=5):
if limit < 0:
raise ValueError('limit must be non-negative')
terms = set(tokenize(query))
if not terms or not limit or any(term not in self.postings for term in terms):
return []
groups = sorted((self.postings[term] for term in terms), key=len)
candidates = groups[0].copy()
for group in groups[1:]:
candidates.intersection_update(group)
scored = ((-sum(self.documents[doc_id][term] for term in terms), doc_id)
for doc_id in candidates)
return [(doc_id, -negative) for negative, doc_id in heapq.nsmallest(limit, scored)]
if __name__ == '__main__':
import argparse
import json
from pathlib import Path
parser = argparse.ArgumentParser(description='Local AND search')
parser.add_argument('query', nargs='?', default='图 算法')
parser.add_argument('corpus', nargs='?')
args = parser.parse_args()
corpus = {'a': '图算法与最短路径', 'b': 'Python 算法 算法', 'c': '图算法 图算法'}
if args.corpus:
corpus = json.loads(Path(args.corpus).read_text(encoding='utf-8'))
if not isinstance(corpus, dict):
parser.error('corpus must be a JSON object mapping IDs to text')
index = SearchIndex()
for doc_id, text in corpus.items():
index.upsert(doc_id, text)
print(index.search(args.query))
documents[id] 保存文档的词频表，postings[token] 保存包含该词项的ID集合。二者必须互相一致：某文档词频为正，当且仅当对应倒排表里存在这个ID。替换文档时先删除旧词项的倒排记录，再加入新词项，避免修改后仍能搜到已经消失的词。
查询与排序
从最短倒排表开始求交，能尽早缩小候选。任一词项没有倒排记录便可直接返回空。每个候选的得分是各查询词在文中的频次之和，堆选择前K项；同分按文档ID排序，让输出稳定。这个简单评分偏向重复次数高的文本，没有做长度归一化，也没有TF-IDF或BM25的文档稀有性权重。
构建时间与文本总长度以及词项操作有关；索引空间取决于不同词项和词项—文档关系数量。求交的实际成本受倒排表长度影响，不只取决于查询字符数。候选数R、查询不同词项数Q时，打分需要约 O(RQ)，堆选前K项通常 O(R log K)。极端高频词可能仍命中大部分语料。
使用一份小语料
下载 corpus.json 与两个Python文件，执行 python example.py "图 算法" corpus.json。也可以不传参数，直接运行内置三文档演示。JSON键必须是字符串，值必须是正文字符串。先用小语料手算候选与分数，再换成自己的文本。
这个原型只在进程内存保存索引，退出后重新构建；不执行同步保存、鉴权或远程更新。若要持久化，应给分词规则和索引格式加版本，先离线构建新索引再原子替换，防止读到半更新状态。多线程修改时也需要统一同步，不应以为单个字典操作可用就能保证整次替换一致。
完成条件与扩展
测试大小写、全角字符、组合音标、重复词、空查询、不存在词、相同分数、删除与替换。用全量扫描同一份词频表作为基线，逐个比较查询结果。规范化可能改变下标，若加结果高亮，需要另外保存原文到规范化文本的位置映射。
扩展练习先加入词位，实现必须相邻的短语查询；再比较文档频率加权与当前词频排序。面试自查：倒排表和正向词频表为何都需要，删除旧索引为什么不能省略，从最短集合开始求交改善的是哪部分成本，以及为什么检索规则变化后旧索引必须重建。
本地运行
示例仅使用 Python 标准库，适用于 Python 3.10 及以上。把 example.py 与 test_example.py 保存到同一个目录，在该目录运行：
python example.py
python -m unittest -v test_example.py
测试检查本文列出的边界；新增练习仍需自己补充断言。