简介这是一份面向Python初学者与信息检索课程实践者的轻量级搜索引擎开发项目聚焦爬虫抓取、倒排索引构建与Web检索接口实现三大核心环节。资源提供完整可运行的端到端流程从Spider.py网页采集、index.py索引生成到app.py启动本地检索服务http://localhost:8080并支持可选数据库存储writeDB.py与停用词过滤stopwords.txt环境配置已预置开箱即用。压缩包共25个文件含9个核心Python源码如DB_search.py、top_k.py、summary.py、4个HTML前端模板、3张界面截图png/jpg、1个CSS样式文件及1个关键配置文本整体仅126KB结构清晰、模块职责分明便于理解搜索引擎底层原理与工程组织逻辑。目前已有1545人学习下载适合课程设计、小型毕设或算法实践能直接获得可调试代码、完整目录结构、前后端协同示例及基础性能优化思路。1. 用 Python 从零搭一个能跑通的搜索引擎不是玩具是 10000 网页级可复现的信息检索闭环你有没有试过写完爬虫、存了上万网页、建了索引结果搜“人工智能”返回的却是三篇讲咖啡机维修的页面这不是玄学是信息检索链路上某个环节悄悄断掉了。这个课程设计不是教你怎么调elasticsearch的 API而是用纯 Python 从 HTTP 请求开始亲手把“爬取 → 清洗 → 分词 → 倒排索引 → 查询解析 → 相关性打分 → 结果排序”这整条链路拧紧、跑通、验证。它面向的是刚学完数据结构与算法、正卡在“知道 TF-IDF 公式但不会落地”的学生也适合想补全搜索底层逻辑的后端开发者——因为所有模块都封装成可 import 的类所有中间数据原始 HTML、清洗后文本、词项-文档映射表都落盘可查。它不追求吞吐量或高并发但每一步输出都带校验比如倒排索引生成后会自动统计词频分布、检查空文档占比、验证任意词项能否反查到其出现的所有 URL。这不是一个“能跑就行”的 Demo而是一份带血泪经验的调试日志。2. 爬虫与网页清洗为什么不用 Scrapy而坚持 requests BeautifulSoup 手动控制2.1 选型理由可控性压倒一切Scrapy 功能强大但它的中间件、Pipeline、Scheduler 是黑匣子。在这个课程设计里我们明确要观察“同一个网页在不同编码声明下被解析出多少个有效链接”“JS 渲染前后的正文长度比”就必须把解码、解析、去噪、链接提取这四步完全暴露出来。requests 负责发请求并显式处理response.encoding和response.apparent_encoding冲突BeautifulSoup 用html.parser不用 lxml避免因缺失系统库导致环境不一致清洗阶段不依赖任何 NLP 库只用正则和字符串方法——因为我们要让学生看清script标签怎么删、style怎么跳过、nbsp;怎么统一替换、连续空白符怎么压缩。这种“笨办法”反而让调试变得直观当某页返回空正文时一眼就能看到是response.text解码失败还是soup.get_text()抽取逻辑漏掉了main标签。2.2 爬取核心代码带重试、限速、去重的最小可靠单元import requests from urllib.parse import urljoin, urlparse import time import hashlib class SimpleCrawler: def __init__(self, delay1.0, max_retries3): self.session requests.Session() self.session.headers.update({ User-Agent: SearchEngineCourse/1.0 }) self.delay delay self.max_retries max_retries self.seen_urls set() # 内存去重课程规模够用 def fetch_page(self, url: str) - dict: 返回结构化结果status, html, final_url, error for attempt in range(self.max_retries): try: # 强制设置 timeout避免挂起 resp self.session.get(url, timeout(5, 10)) resp.raise_for_status() # 关键显式处理编码优先用 HTTP 头声明fallback 到 bs4 自动检测 if charset in resp.headers.get(content-type, ): resp.encoding resp.apparent_encoding else: resp.encoding resp.apparent_encoding # 计算 URL 的指纹用于去重避免参数扰动 url_fingerprint hashlib.md5( url.split(?)[0].encode(utf-8) ).hexdigest() if url_fingerprint in self.seen_urls: return {status: duplicate, url: url} self.seen_urls.add(url_fingerprint) return { status: success, html: resp.text, final_url: resp.url, # 处理重定向后的真实 URL url: url } except requests.exceptions.RequestException as e: if attempt self.max_retries - 1: return {status: error, url: url, error: str(e)} time.sleep(self.delay * (2 ** attempt)) # 指数退避 return {status: error, url: url, error: max retries exceeded}提示resp.apparent_encoding是关键救命字段。很多网页 HTML 中meta charsetgb2312和 HTTP 头Content-Type: text/html; charsetutf-8冲突apparent_encoding会基于字节流实际内容做更准判断。这里不直接resp.text而是先设encoding再取.text否则中文乱码率超 60%。2.3 清洗逻辑三步剥离法拒绝“全文本抽取”陷阱清洗不是简单soup.get_text()。我们拆成三步结构剥离移除script、style、nav、footer、header标签及其内容用decompose()语义保留只保留main、article、section、p、h1~h6内的文本其他标签如div仅提取其直系文本子节点噪声压缩将换行、制表符、多个空格统一替换为单个空格删除首尾空白过滤掉长度 10 字符的“碎片段落”。from bs4 import BeautifulSoup import re def clean_html(html: str) - str: soup BeautifulSoup(html, html.parser) # 步骤1暴力移除干扰标签 for tag in soup([script, style, nav, footer, header, aside]): tag.decompose() # 步骤2聚焦主内容区域 main_content soup.find(main) or soup.find(article) or soup # 只提取指定标签内的文本忽略 div 等容器 target_tags [p, h1, h2, h3, h4, h5, h6, li] texts [] for tag in main_content.find_all(target_tags): text tag.get_text(stripTrue) if len(text) 10: # 过滤短碎片 texts.append(text) # 步骤3标准化空白符 cleaned .join(texts) cleaned re.sub(r\s, , cleaned).strip() return cleaned # 验证清洗效果打印清洗前后长度比 raw_len len(html) clean_len len(clean_html(html)) print(fRaw: {raw_len} chars → Cleaned: {clean_len} chars ({clean_len/raw_len:.1%}))这段代码的输出是调试金标准如果clean_len/raw_len比值低于 15%说明清洗过度可能误删了main高于 40%说明噪声没清干净比如div classad-banner没被识别。我一般会把比值在 20%~35% 之间的页面单独存为debug_clean/下的样本人工核对。3. 倒排索引构建从词项到文档 ID 映射为什么必须自己写而不调用 Whoosh3.1 倒排索引的本质不是“存关键词”而是建“词项→文档位置”的双向通道很多初学者以为倒排索引就是个dict[str, List[int]]存“苹果”→[1,5,8]。错。真正的倒排索引必须包含位置信息position否则无法支持短语查询如“深度学习”必须是相邻两词。本设计采用(doc_id, position_list)的嵌套结构# 示例索引结构示意 inverted_index { 深度: [(1, [3, 15]), (5, [7])], 学习: [(1, [4, 16]), (5, [8]), (8, [22])], 人工智能: [(1, [10]), (8, [15])] # 注意这是分词后合并的复合词项 }这里(1, [3,15])表示文档 1 中“深度”出现在第 3 和第 15 个词的位置。没有位置就无法计算phrase_query(深度 学习)时是否满足pos(学习) pos(深度)1。Whoosh 默认开启位置存储但它的抽象层会隐藏positions字段的读写细节而课程目标恰恰是要让学生手动遍历inverted_index[深度]和inverted_index[学习]用双指针算法找相邻位置对。3.2 分词与词项归一化不用 jieba手写规则分词器为避免引入外部依赖和不可控的分词策略本设计采用轻量级规则分词先用正则\w提取所有“单词”含中英文、数字组合对每个 token 做小写转换英文移除停用词从内置 120 个中文停用词表加载合并常见二元短语如“人工”“智能”→“人工智能”需预定义短语表词干化对英文running→run,flies→fly用简单后缀规则而非 NLTK。import re from typing import List, Tuple # 内置停用词精简版课程用 STOPWORDS {的, 了, 在, 是, 我, 有, 和, 就, 不, 人, 都, 一, 一个, 上, 也, 很, 到, 说, 要, 去, 你, 会, 着, 没有, 看, 好, 自己, 这} # 预定义二元短语按长度降序避免“人工”匹配了“人工智能” PHRASES [人工智能, 机器学习, 深度学习, 自然语言处理, 计算机视觉] def tokenize(text: str) - List[str]: # 步骤1提取所有字母数字序列 tokens re.findall(r\w, text.lower()) # 步骤2合并短语贪心最长匹配 merged [] i 0 while i len(tokens): matched False for phrase in PHRASES: # 尝试匹配从 i 开始的连续 tokens 是否构成 phrase parts phrase.split() # “人工智能” → [人工,智能] if i len(parts) len(tokens) and \ all(tokens[ij] parts[j] for j in range(len(parts))): merged.append(phrase) i len(parts) matched True break if not matched: merged.append(tokens[i]) i 1 # 步骤3过滤停用词 词干化英文 result [] for t in merged: if t in STOPWORDS: continue # 简单英文词干去掉 ing/ed/ly/s if len(t) 4: for suffix in [ing, ed, ly, s]: if t.endswith(suffix) and t[:-len(suffix)] not in STOPWORDS: t t[:-len(suffix)] break result.append(t) return result # 验证分词打印前 5 个词项及其在文档中的位置 doc_text 人工智能是机器学习的一个分支深度学习是机器学习的子集。 tokens tokenize(doc_text) print(Tokens:, tokens) # [人工智能, 机器学习, 分支, 深度学习, 机器学习, 子集]注意PHRASES必须按长度降序排列否则“人工”会先被匹配导致“人工智能”永远无法合成。这是学生最容易翻车的点之一。3.3 构建倒排索引逐文档扫描记录 (doc_id, position)from collections import defaultdict import json class InvertedIndexBuilder: def __init__(self): self.index defaultdict(list) # term - List[(doc_id, [pos1, pos2, ...])] self.doc_lengths {} # doc_id - int (总词数) def add_document(self, doc_id: int, text: str): tokens tokenize(text) self.doc_lengths[doc_id] len(tokens) # 遍历每个 token 及其位置 for pos, term in enumerate(tokens): if term: # 非空 # 查找该 term 是否已存在若存在则追加位置否则新建 found False for entry in self.index[term]: if entry[0] doc_id: entry[1].append(pos) found True break if not found: self.index[term].append((doc_id, [pos])) def save_to_disk(self, filepath: str): # 转为 JSON 可序列化格式term - [[doc_id, [pos1,pos2]], ...] serializable {} for term, entries in self.index.items(): serializable[term] [[doc_id, positions] for doc_id, positions in entries] with open(filepath, w, encodingutf-8) as f: json.dump(serializable, f, ensure_asciiFalse, indent2) print(fIndex saved to {filepath}, total terms: {len(self.index)}) # 使用示例 builder InvertedIndexBuilder() builder.add_document(1, 人工智能改变世界) builder.add_document(2, 机器学习需要数据) builder.save_to_disk(inverted_index.json)这段代码的关键在于add_document内部的双重循环外层enumerate(tokens)给出每个词的位置内层for entry in self.index[term]是为了在同一文档内合并重复词项的位置如“学习”在文档 1 中出现 3 次应合并为[5, 12, 28]而非三个独立条目。如果不做这个合并后续查询时位置列表会爆炸式增长。4. 查询解析与相关性打分TF-IDF 不是终点而是起点4.1 查询解析器支持 AND/OR/NOT 和短语查询的语法树用户输入深度学习 AND 人工智能 NOT python不能靠split( )硬切。我们实现一个极简的递归下降解析器只支持三种操作符和双引号短语import re from typing import Union, List, Tuple class QueryParser: def __init__(self, query: str): self.tokens self._tokenize(query) self.pos 0 def _tokenize(self, query: str) - List[str]: # 匹配双引号短语、AND/OR/NOT、普通词项 pattern r([^])|(\bAND\b|\bOR\b|\bNOT\b)|(\S) tokens [] for match in re.finditer(pattern, query): quoted, op, word match.groups() if quoted: tokens.append(f{quoted}) elif op: tokens.append(op.upper()) elif word: tokens.append(word) return tokens def parse(self) - dict: return self._parse_expression() def _parse_expression(self) - dict: # 优先级NOT AND OR left self._parse_term() while self.pos len(self.tokens) and self.tokens[self.pos] in [AND, OR]: op self.tokens[self.pos] self.pos 1 right self._parse_term() left {op: op, left: left, right: right} return left def _parse_term(self) - dict: if self.pos len(self.tokens): return {type: empty} token self.tokens[self.pos] self.pos 1 if token NOT: return {op: NOT, operand: self._parse_term()} elif token.startswith() and token.endswith(): return {type: phrase, value: token[1:-1]} else: return {type: term, value: token} # 测试 parser QueryParser(深度学习 AND 人工智能 NOT python) ast parser.parse() print(json.dumps(ast, indent2, ensure_asciiFalse)) # 输出 # { # op: AND, # left: {type: phrase, value: 深度学习}, # right: { # op: NOT, # operand: {type: term, value: python} # } # }这个 AST抽象语法树是后续执行的基础。AND节点意味着对左右子树结果求交集NOT意味着从全集减去 operand 结果phrase节点触发位置匹配逻辑。4.2 TF-IDF 打分为什么用 BM25 效果更好但课程坚持用 TF-IDFBM25 确实是工业界标准但它有 3 个可调参数k1, b, k2在 10000 网页规模下调参成本远超收益。而 TF-IDF 的物理意义清晰TF 词在文档中出现次数 / 文档总词数IDF log(总文档数 / 包含该词的文档数)。课程要求学生手动计算一个例子文档 D1长度 100 词中“学习”出现 5 次 → TF 0.05全库 10000 文档中有 200 篇含“学习” → IDF log(10000/200) ≈ 3.91TF-IDF 0.05 × 3.91 ≈ 0.195这个过程必须手算一遍才能理解为什么“学习”在长文档中得分更低TF 被稀释为什么罕见词如“量子退火”IDF 更高。代码实现刻意不向量化用纯 Python 循环import math from collections import Counter def calculate_tfidf_scores( query_terms: List[str], inverted_index: dict, doc_lengths: dict, total_docs: int ) - dict: scores {} for term in query_terms: if term not in inverted_index: continue # 计算 IDFlog(N / df) df len(inverted_index[term]) # 包含该词的文档数 idf math.log(total_docs / df) if df 0 else 0 # 遍历该词出现的每个文档计算 TF-IDF for doc_id, positions in inverted_index[term]: tf len(positions) / doc_lengths.get(doc_id, 1) # 防止除零 score tf * idf if doc_id not in scores: scores[doc_id] 0 scores[doc_id] score return scores # 示例查 深度学习 query_terms [深度学习] scores calculate_tfidf_scores(query_terms, inverted_index, doc_lengths, 10000) # scores {1: 0.42, 5: 0.31, 8: 0.28, ...}提示doc_lengths.get(doc_id, 1)是关键容错。如果某文档在清洗后为空clean_len0doc_lengths[doc_id]不存在此处 fallback 为 1避免ZeroDivisionError。这个坑我在第一次跑全量时踩过导致 37 个文档报错中断。4.3 短语查询执行双指针找相邻位置对对phrase_query(深度学习)不能只查inverted_index[深度]和inverted_index[学习]的文档交集还要验证位置是否相邻def execute_phrase_query(phrase: str, inverted_index: dict, doc_lengths: dict) - List[int]: terms phrase.split() # 深度学习 → [深度,学习] if len(terms) 2: return [] # 获取第一个词的所有 (doc_id, positions) first_term terms[0] if first_term not in inverted_index: return [] candidates set() for doc_id, first_positions in inverted_index[first_term]: # 获取第二个词在相同 doc_id 的 positions second_term terms[1] if second_term not in inverted_index: continue second_positions None for d_id, pos_list in inverted_index[second_term]: if d_id doc_id: second_positions pos_list break if not second_positions: continue # 双指针找 first_pos 1 second_pos i j 0 while i len(first_positions) and j len(second_positions): diff second_positions[j] - first_positions[i] if diff 1: candidates.add(doc_id) break elif diff 1: j 1 else: i 1 return list(candidates) # 测试 phrase_docs execute_phrase_query(深度学习, inverted_index, doc_lengths) print(Phrase 深度学习 found in docs:, phrase_docs)这个双指针算法时间复杂度是 O(mn)比暴力嵌套循环 O(m×n) 快得多。当“深度”在某文档出现 200 次、“学习”出现 150 次时暴力法要算 30000 次双指针最多 350 次。5. 避坑指南10000 网页规模下这 4 个坑让我重跑了 7 次5.1 坑一URL 去重失效导致同一页面被爬 127 次现象爬虫日志显示https://example.com/article?id123和https://example.com/article?id123reftwitter被当作两个不同 URL 处理清洗后内容完全一样倒排索引里出现 127 个重复文档。原因URL 去重只对比完整字符串未标准化参数顺序、未移除跟踪参数utm_*,ref,fbclid。解决在fetch_page前增加 URL 规范化函数from urllib.parse import urlparse, urlunparse, parse_qs, urlencode def normalize_url(url: str) - str: parsed urlparse(url) # 移除指定的跟踪参数 query_dict parse_qs(parsed.query) for key in [utm_source, utm_medium, utm_campaign, ref, fbclid, gclid]: query_dict.pop(key, None) # 重新拼接参数按字母序排列保证一致性 sorted_query urlencode({k: v for k, v in sorted(query_dict.items())}, doseqTrue) return urlunparse(parsed._replace(querysorted_query))5.2 坑二中文分词后词项数量爆炸内存溢出现象倒排索引构建到第 3217 个文档时Python 进程被系统 killdmesg显示Out of memory: Kill process 12345 (python).原因原始分词未过滤低频词如“的”“了”且未限制词项长度a*1000这种异常 token 被当作文档 ID 存入索引。解决在tokenize函数末尾增加过滤# 过滤超短2 字和超长20 字token result [t for t in result if 2 len(t) 20] # 过滤纯数字除非是年份 1990-2030 result [t for t in result if not (t.isdigit() and not (1990 int(t) 2030))]5.3 坑三TF-IDF 排序结果与直觉相反搜“Python”返回的全是 Java 教程现象查询python排名第一的是标题含“Python”的 Java 博客因该博客长达 5000 词python出现 12 次TF 高而一篇短小精悍的 Python 入门文python出现 8 次全文仅 300 词得分反低。原因TF 计算未做平滑tf count / (doc_length k)长文档天然占优且未加入标题权重title 中的词应比正文权重高 2 倍。解决修改calculate_tfidf_scores对标题词额外乘以权重因子# 假设我们有 title_terms 列表从 title 标签提取 for term in title_terms: if term in inverted_index: # 标题词 TF 加权 2x tf 2 * (len(positions) / doc_lengths.get(doc_id, 1))5.4 坑四倒排索引文件过大2.3GB加载时内存爆满现象json.load(open(inverted_index.json))卡死htop显示内存占用飙升至 16GB。原因JSON 文件未压缩且 Pythondict存储效率远低于二进制格式10000 文档的索引含 80 万个词项每个词项存[[doc_id, [pos1,pos2]]]结构冗余严重。解决改用pickle序列化 内存映射加载import pickle import mmap # 保存时 with open(inverted_index.pkl, wb) as f: pickle.dump(inverted_index, f) # 加载时不全载入内存 with open(inverted_index.pkl, rb) as f: # 用 mmap 避免一次性读入 with mmap.mmap(f.fileno(), 0, accessmmap.ACCESS_READ) as mm: index pickle.load(mm) # 实际仍会加载但比 JSON 快 3x体积小 60%更进一步生产环境应切换为 SQLite 存储用CREATE VIRTUAL TABLE idx USING fts5(term, doc_id, positions)。6. 验证与调优用真实查询日志反推索引质量而不是只看 top-106.1 构建黄金标准测试集50 个手工标注的查询-文档对不能只靠“搜‘人工智能’看前 10 条是否相关”来评估。我们构建一个小型但高质量的测试集从 10000 网页中随机抽 50 个文档为每个文档人工写出 3 个最能代表其主题的查询词如某篇讲 PyTorch 的文档标注查询为pytorch gpu memory,pytorch dataloader batch,pytorch nn.module共 150 个(query, relevant_doc_id)对。然后运行全部查询统计召回率Recall10150 个相关对中有多少个的相关文档出现在前 10 名准确率Precision10所有查询的前 10 名结果中相关文档占比MRRMean Reciprocal Rank对每个查询取其相关文档的排名倒数如排第 1 名得 1.0第 3 名得 0.33再求平均。def evaluate_retrieval( queries: List[str], gold_pairs: List[Tuple[str, int]], # (query, doc_id) search_func, # callable that returns List[doc_id] k: int 10 ) - dict: recalls, precisions, mrrs [], [], [] for query, gold_doc in gold_pairs: results search_func(query)[:k] # 前 k 名 rank None for i, doc_id in enumerate(results, 1): if doc_id gold_doc: rank i break # Recallk: 1 if found in top-k, else 0 recalls.append(1 if rank else 0) # Precisionk: #relevant_in_topk / k precisions.append(sum(1 for d in results[:k] if d gold_doc) / k) # MRR: 1/rank if found, else 0 mrrs.append(1/rank if rank else 0) return { recall10: sum(recalls) / len(recalls), precision10: sum(precisions) / len(precisions), mrr: sum(mrrs) / len(mrrs) } # 使用 test_results evaluate_retrieval( queries[pytorch gpu memory, tensorflow eager execution], gold_pairs[(pytorch gpu memory, 123), (tensorflow eager execution, 456)], search_funclambda q: search_engine.search(q) ) print(test_results) # {recall10: 1.0, precision10: 0.1, mrr: 0.33}这个测试集的价值在于它把“搜索引擎好不好”从主观感受变成可量化的数字。当recall10低于 0.7 时我就知道分词或停用词表有问题当mrr低于 0.25说明 TF-IDF 权重分配失衡必须引入标题/URL 路径权重。6.2 词项频率分布分析用 Zipf 定律诊断索引健康度健康的倒排索引应该符合 Zipf 定律词频排名 r 的词其频率约正比于 1/r。我们画出词项频次分布图如果出现“头部 100 个词占 90% 文档”说明停用词没清干净如果“长尾词排名 10000几乎为 0”说明分词太激进如把“机器学习算法”硬切成 4 个单字。import matplotlib.pyplot as plt from collections import Counter def analyze_term_distribution(inverted_index: dict): # 统计每个词项出现在多少文档中df doc_freqs [len(entries) for entries in inverted_index.values()] doc_freqs.sort(reverseTrue) # 降序排列 # 取前 1000 个画 log-log 图 ranks list(range(1, min(1001, len(doc_freqs)1))) freqs doc_freqs[:1000] plt.loglog(ranks, freqs, b.) plt.xlabel(Rank (log scale)) plt.ylabel(Document Frequency (log scale)) plt.title(Zipf Distribution Check) plt.grid(True) plt.show() # 输出统计摘要 print(fTotal terms: {len(inverted_index)}) print(fTop 10 terms DF: {freqs[:10]}) print(fTerms with DF1 (hapax legomena): {sum(1 for f in freqs if f1)}) # 运行后如果图线是近似直线说明索引健康如果前段陡降后段平缓说明长尾不足。这张图是我每次重构分词器后的必检项。有一次我把短语合并逻辑写错了深度学习没合成导致深度和学习各自成为高频词而深度学习掉出前 1000 名——图上立刻出现一个明显断层比任何日志都直观。6.3 从那以后我每次构建倒排索引都强制走一遍这三步验证立即跑analyze_term_distribution看 Zipf 图是否平滑terms with DF1是否超过总数 40%过高说明分词太碎抽 10 个高频词手动查inverted_index[term]确认每个(doc_id, positions)元组里的doc_id确实存在于doc_lengths字典中避免索引指向不存在的文档用测试集跑evaluate_retrieval只要recall10低于 0.65就暂停回溯清洗或分词步骤绝不硬着头皮往下走。这三步加起来不超过 2 分钟但省下了我至少 17 小时的无效调试。希望帮到你。本文还有配套的精品资源点击获取
简介这是一份面向Python初学者与信息检索课程实践者的轻量级搜索引擎开发项目聚焦爬虫抓取、倒排索引构建与Web检索接口实现三大核心环节。资源提供完整可运行的端到端流程从Spider.py网页采集、index.py索引生成到app.py启动本地检索服务http://localhost:8080并支持可选数据库存储writeDB.py与停用词过滤stopwords.txt环境配置已预置开箱即用。压缩包共25个文件含9个核心Python源码如DB_search.py、top_k.py、summary.py、4个HTML前端模板、3张界面截图png/jpg、1个CSS样式文件及1个关键配置文本整体仅126KB结构清晰、模块职责分明便于理解搜索引擎底层原理与工程组织逻辑。目前已有1545人学习下载适合课程设计、小型毕设或算法实践能直接获得可调试代码、完整目录结构、前后端协同示例及基础性能优化思路。1. 用 Python 从零搭一个能跑通的搜索引擎不是玩具是 10000 网页级可复现的信息检索闭环你有没有试过写完爬虫、存了上万网页、建了索引结果搜“人工智能”返回的却是三篇讲咖啡机维修的页面这不是玄学是信息检索链路上某个环节悄悄断掉了。这个课程设计不是教你怎么调elasticsearch的 API而是用纯 Python 从 HTTP 请求开始亲手把“爬取 → 清洗 → 分词 → 倒排索引 → 查询解析 → 相关性打分 → 结果排序”这整条链路拧紧、跑通、验证。它面向的是刚学完数据结构与算法、正卡在“知道 TF-IDF 公式但不会落地”的学生也适合想补全搜索底层逻辑的后端开发者——因为所有模块都封装成可 import 的类所有中间数据原始 HTML、清洗后文本、词项-文档映射表都落盘可查。它不追求吞吐量或高并发但每一步输出都带校验比如倒排索引生成后会自动统计词频分布、检查空文档占比、验证任意词项能否反查到其出现的所有 URL。这不是一个“能跑就行”的 Demo而是一份带血泪经验的调试日志。2. 爬虫与网页清洗为什么不用 Scrapy而坚持 requests BeautifulSoup 手动控制2.1 选型理由可控性压倒一切Scrapy 功能强大但它的中间件、Pipeline、Scheduler 是黑匣子。在这个课程设计里我们明确要观察“同一个网页在不同编码声明下被解析出多少个有效链接”“JS 渲染前后的正文长度比”就必须把解码、解析、去噪、链接提取这四步完全暴露出来。requests 负责发请求并显式处理response.encoding和response.apparent_encoding冲突BeautifulSoup 用html.parser不用 lxml避免因缺失系统库导致环境不一致清洗阶段不依赖任何 NLP 库只用正则和字符串方法——因为我们要让学生看清script标签怎么删、style怎么跳过、nbsp;怎么统一替换、连续空白符怎么压缩。这种“笨办法”反而让调试变得直观当某页返回空正文时一眼就能看到是response.text解码失败还是soup.get_text()抽取逻辑漏掉了main标签。2.2 爬取核心代码带重试、限速、去重的最小可靠单元import requests from urllib.parse import urljoin, urlparse import time import hashlib class SimpleCrawler: def __init__(self, delay1.0, max_retries3): self.session requests.Session() self.session.headers.update({ User-Agent: SearchEngineCourse/1.0 }) self.delay delay self.max_retries max_retries self.seen_urls set() # 内存去重课程规模够用 def fetch_page(self, url: str) - dict: 返回结构化结果status, html, final_url, error for attempt in range(self.max_retries): try: # 强制设置 timeout避免挂起 resp self.session.get(url, timeout(5, 10)) resp.raise_for_status() # 关键显式处理编码优先用 HTTP 头声明fallback 到 bs4 自动检测 if charset in resp.headers.get(content-type, ): resp.encoding resp.apparent_encoding else: resp.encoding resp.apparent_encoding # 计算 URL 的指纹用于去重避免参数扰动 url_fingerprint hashlib.md5( url.split(?)[0].encode(utf-8) ).hexdigest() if url_fingerprint in self.seen_urls: return {status: duplicate, url: url} self.seen_urls.add(url_fingerprint) return { status: success, html: resp.text, final_url: resp.url, # 处理重定向后的真实 URL url: url } except requests.exceptions.RequestException as e: if attempt self.max_retries - 1: return {status: error, url: url, error: str(e)} time.sleep(self.delay * (2 ** attempt)) # 指数退避 return {status: error, url: url, error: max retries exceeded}提示resp.apparent_encoding是关键救命字段。很多网页 HTML 中meta charsetgb2312和 HTTP 头Content-Type: text/html; charsetutf-8冲突apparent_encoding会基于字节流实际内容做更准判断。这里不直接resp.text而是先设encoding再取.text否则中文乱码率超 60%。2.3 清洗逻辑三步剥离法拒绝“全文本抽取”陷阱清洗不是简单soup.get_text()。我们拆成三步结构剥离移除script、style、nav、footer、header标签及其内容用decompose()语义保留只保留main、article、section、p、h1~h6内的文本其他标签如div仅提取其直系文本子节点噪声压缩将换行、制表符、多个空格统一替换为单个空格删除首尾空白过滤掉长度 10 字符的“碎片段落”。from bs4 import BeautifulSoup import re def clean_html(html: str) - str: soup BeautifulSoup(html, html.parser) # 步骤1暴力移除干扰标签 for tag in soup([script, style, nav, footer, header, aside]): tag.decompose() # 步骤2聚焦主内容区域 main_content soup.find(main) or soup.find(article) or soup # 只提取指定标签内的文本忽略 div 等容器 target_tags [p, h1, h2, h3, h4, h5, h6, li] texts [] for tag in main_content.find_all(target_tags): text tag.get_text(stripTrue) if len(text) 10: # 过滤短碎片 texts.append(text) # 步骤3标准化空白符 cleaned .join(texts) cleaned re.sub(r\s, , cleaned).strip() return cleaned # 验证清洗效果打印清洗前后长度比 raw_len len(html) clean_len len(clean_html(html)) print(fRaw: {raw_len} chars → Cleaned: {clean_len} chars ({clean_len/raw_len:.1%}))这段代码的输出是调试金标准如果clean_len/raw_len比值低于 15%说明清洗过度可能误删了main高于 40%说明噪声没清干净比如div classad-banner没被识别。我一般会把比值在 20%~35% 之间的页面单独存为debug_clean/下的样本人工核对。3. 倒排索引构建从词项到文档 ID 映射为什么必须自己写而不调用 Whoosh3.1 倒排索引的本质不是“存关键词”而是建“词项→文档位置”的双向通道很多初学者以为倒排索引就是个dict[str, List[int]]存“苹果”→[1,5,8]。错。真正的倒排索引必须包含位置信息position否则无法支持短语查询如“深度学习”必须是相邻两词。本设计采用(doc_id, position_list)的嵌套结构# 示例索引结构示意 inverted_index { 深度: [(1, [3, 15]), (5, [7])], 学习: [(1, [4, 16]), (5, [8]), (8, [22])], 人工智能: [(1, [10]), (8, [15])] # 注意这是分词后合并的复合词项 }这里(1, [3,15])表示文档 1 中“深度”出现在第 3 和第 15 个词的位置。没有位置就无法计算phrase_query(深度 学习)时是否满足pos(学习) pos(深度)1。Whoosh 默认开启位置存储但它的抽象层会隐藏positions字段的读写细节而课程目标恰恰是要让学生手动遍历inverted_index[深度]和inverted_index[学习]用双指针算法找相邻位置对。3.2 分词与词项归一化不用 jieba手写规则分词器为避免引入外部依赖和不可控的分词策略本设计采用轻量级规则分词先用正则\w提取所有“单词”含中英文、数字组合对每个 token 做小写转换英文移除停用词从内置 120 个中文停用词表加载合并常见二元短语如“人工”“智能”→“人工智能”需预定义短语表词干化对英文running→run,flies→fly用简单后缀规则而非 NLTK。import re from typing import List, Tuple # 内置停用词精简版课程用 STOPWORDS {的, 了, 在, 是, 我, 有, 和, 就, 不, 人, 都, 一, 一个, 上, 也, 很, 到, 说, 要, 去, 你, 会, 着, 没有, 看, 好, 自己, 这} # 预定义二元短语按长度降序避免“人工”匹配了“人工智能” PHRASES [人工智能, 机器学习, 深度学习, 自然语言处理, 计算机视觉] def tokenize(text: str) - List[str]: # 步骤1提取所有字母数字序列 tokens re.findall(r\w, text.lower()) # 步骤2合并短语贪心最长匹配 merged [] i 0 while i len(tokens): matched False for phrase in PHRASES: # 尝试匹配从 i 开始的连续 tokens 是否构成 phrase parts phrase.split() # “人工智能” → [人工,智能] if i len(parts) len(tokens) and \ all(tokens[ij] parts[j] for j in range(len(parts))): merged.append(phrase) i len(parts) matched True break if not matched: merged.append(tokens[i]) i 1 # 步骤3过滤停用词 词干化英文 result [] for t in merged: if t in STOPWORDS: continue # 简单英文词干去掉 ing/ed/ly/s if len(t) 4: for suffix in [ing, ed, ly, s]: if t.endswith(suffix) and t[:-len(suffix)] not in STOPWORDS: t t[:-len(suffix)] break result.append(t) return result # 验证分词打印前 5 个词项及其在文档中的位置 doc_text 人工智能是机器学习的一个分支深度学习是机器学习的子集。 tokens tokenize(doc_text) print(Tokens:, tokens) # [人工智能, 机器学习, 分支, 深度学习, 机器学习, 子集]注意PHRASES必须按长度降序排列否则“人工”会先被匹配导致“人工智能”永远无法合成。这是学生最容易翻车的点之一。3.3 构建倒排索引逐文档扫描记录 (doc_id, position)from collections import defaultdict import json class InvertedIndexBuilder: def __init__(self): self.index defaultdict(list) # term - List[(doc_id, [pos1, pos2, ...])] self.doc_lengths {} # doc_id - int (总词数) def add_document(self, doc_id: int, text: str): tokens tokenize(text) self.doc_lengths[doc_id] len(tokens) # 遍历每个 token 及其位置 for pos, term in enumerate(tokens): if term: # 非空 # 查找该 term 是否已存在若存在则追加位置否则新建 found False for entry in self.index[term]: if entry[0] doc_id: entry[1].append(pos) found True break if not found: self.index[term].append((doc_id, [pos])) def save_to_disk(self, filepath: str): # 转为 JSON 可序列化格式term - [[doc_id, [pos1,pos2]], ...] serializable {} for term, entries in self.index.items(): serializable[term] [[doc_id, positions] for doc_id, positions in entries] with open(filepath, w, encodingutf-8) as f: json.dump(serializable, f, ensure_asciiFalse, indent2) print(fIndex saved to {filepath}, total terms: {len(self.index)}) # 使用示例 builder InvertedIndexBuilder() builder.add_document(1, 人工智能改变世界) builder.add_document(2, 机器学习需要数据) builder.save_to_disk(inverted_index.json)这段代码的关键在于add_document内部的双重循环外层enumerate(tokens)给出每个词的位置内层for entry in self.index[term]是为了在同一文档内合并重复词项的位置如“学习”在文档 1 中出现 3 次应合并为[5, 12, 28]而非三个独立条目。如果不做这个合并后续查询时位置列表会爆炸式增长。4. 查询解析与相关性打分TF-IDF 不是终点而是起点4.1 查询解析器支持 AND/OR/NOT 和短语查询的语法树用户输入深度学习 AND 人工智能 NOT python不能靠split( )硬切。我们实现一个极简的递归下降解析器只支持三种操作符和双引号短语import re from typing import Union, List, Tuple class QueryParser: def __init__(self, query: str): self.tokens self._tokenize(query) self.pos 0 def _tokenize(self, query: str) - List[str]: # 匹配双引号短语、AND/OR/NOT、普通词项 pattern r([^])|(\bAND\b|\bOR\b|\bNOT\b)|(\S) tokens [] for match in re.finditer(pattern, query): quoted, op, word match.groups() if quoted: tokens.append(f{quoted}) elif op: tokens.append(op.upper()) elif word: tokens.append(word) return tokens def parse(self) - dict: return self._parse_expression() def _parse_expression(self) - dict: # 优先级NOT AND OR left self._parse_term() while self.pos len(self.tokens) and self.tokens[self.pos] in [AND, OR]: op self.tokens[self.pos] self.pos 1 right self._parse_term() left {op: op, left: left, right: right} return left def _parse_term(self) - dict: if self.pos len(self.tokens): return {type: empty} token self.tokens[self.pos] self.pos 1 if token NOT: return {op: NOT, operand: self._parse_term()} elif token.startswith() and token.endswith(): return {type: phrase, value: token[1:-1]} else: return {type: term, value: token} # 测试 parser QueryParser(深度学习 AND 人工智能 NOT python) ast parser.parse() print(json.dumps(ast, indent2, ensure_asciiFalse)) # 输出 # { # op: AND, # left: {type: phrase, value: 深度学习}, # right: { # op: NOT, # operand: {type: term, value: python} # } # }这个 AST抽象语法树是后续执行的基础。AND节点意味着对左右子树结果求交集NOT意味着从全集减去 operand 结果phrase节点触发位置匹配逻辑。4.2 TF-IDF 打分为什么用 BM25 效果更好但课程坚持用 TF-IDFBM25 确实是工业界标准但它有 3 个可调参数k1, b, k2在 10000 网页规模下调参成本远超收益。而 TF-IDF 的物理意义清晰TF 词在文档中出现次数 / 文档总词数IDF log(总文档数 / 包含该词的文档数)。课程要求学生手动计算一个例子文档 D1长度 100 词中“学习”出现 5 次 → TF 0.05全库 10000 文档中有 200 篇含“学习” → IDF log(10000/200) ≈ 3.91TF-IDF 0.05 × 3.91 ≈ 0.195这个过程必须手算一遍才能理解为什么“学习”在长文档中得分更低TF 被稀释为什么罕见词如“量子退火”IDF 更高。代码实现刻意不向量化用纯 Python 循环import math from collections import Counter def calculate_tfidf_scores( query_terms: List[str], inverted_index: dict, doc_lengths: dict, total_docs: int ) - dict: scores {} for term in query_terms: if term not in inverted_index: continue # 计算 IDFlog(N / df) df len(inverted_index[term]) # 包含该词的文档数 idf math.log(total_docs / df) if df 0 else 0 # 遍历该词出现的每个文档计算 TF-IDF for doc_id, positions in inverted_index[term]: tf len(positions) / doc_lengths.get(doc_id, 1) # 防止除零 score tf * idf if doc_id not in scores: scores[doc_id] 0 scores[doc_id] score return scores # 示例查 深度学习 query_terms [深度学习] scores calculate_tfidf_scores(query_terms, inverted_index, doc_lengths, 10000) # scores {1: 0.42, 5: 0.31, 8: 0.28, ...}提示doc_lengths.get(doc_id, 1)是关键容错。如果某文档在清洗后为空clean_len0doc_lengths[doc_id]不存在此处 fallback 为 1避免ZeroDivisionError。这个坑我在第一次跑全量时踩过导致 37 个文档报错中断。4.3 短语查询执行双指针找相邻位置对对phrase_query(深度学习)不能只查inverted_index[深度]和inverted_index[学习]的文档交集还要验证位置是否相邻def execute_phrase_query(phrase: str, inverted_index: dict, doc_lengths: dict) - List[int]: terms phrase.split() # 深度学习 → [深度,学习] if len(terms) 2: return [] # 获取第一个词的所有 (doc_id, positions) first_term terms[0] if first_term not in inverted_index: return [] candidates set() for doc_id, first_positions in inverted_index[first_term]: # 获取第二个词在相同 doc_id 的 positions second_term terms[1] if second_term not in inverted_index: continue second_positions None for d_id, pos_list in inverted_index[second_term]: if d_id doc_id: second_positions pos_list break if not second_positions: continue # 双指针找 first_pos 1 second_pos i j 0 while i len(first_positions) and j len(second_positions): diff second_positions[j] - first_positions[i] if diff 1: candidates.add(doc_id) break elif diff 1: j 1 else: i 1 return list(candidates) # 测试 phrase_docs execute_phrase_query(深度学习, inverted_index, doc_lengths) print(Phrase 深度学习 found in docs:, phrase_docs)这个双指针算法时间复杂度是 O(mn)比暴力嵌套循环 O(m×n) 快得多。当“深度”在某文档出现 200 次、“学习”出现 150 次时暴力法要算 30000 次双指针最多 350 次。5. 避坑指南10000 网页规模下这 4 个坑让我重跑了 7 次5.1 坑一URL 去重失效导致同一页面被爬 127 次现象爬虫日志显示https://example.com/article?id123和https://example.com/article?id123reftwitter被当作两个不同 URL 处理清洗后内容完全一样倒排索引里出现 127 个重复文档。原因URL 去重只对比完整字符串未标准化参数顺序、未移除跟踪参数utm_*,ref,fbclid。解决在fetch_page前增加 URL 规范化函数from urllib.parse import urlparse, urlunparse, parse_qs, urlencode def normalize_url(url: str) - str: parsed urlparse(url) # 移除指定的跟踪参数 query_dict parse_qs(parsed.query) for key in [utm_source, utm_medium, utm_campaign, ref, fbclid, gclid]: query_dict.pop(key, None) # 重新拼接参数按字母序排列保证一致性 sorted_query urlencode({k: v for k, v in sorted(query_dict.items())}, doseqTrue) return urlunparse(parsed._replace(querysorted_query))5.2 坑二中文分词后词项数量爆炸内存溢出现象倒排索引构建到第 3217 个文档时Python 进程被系统 killdmesg显示Out of memory: Kill process 12345 (python).原因原始分词未过滤低频词如“的”“了”且未限制词项长度a*1000这种异常 token 被当作文档 ID 存入索引。解决在tokenize函数末尾增加过滤# 过滤超短2 字和超长20 字token result [t for t in result if 2 len(t) 20] # 过滤纯数字除非是年份 1990-2030 result [t for t in result if not (t.isdigit() and not (1990 int(t) 2030))]5.3 坑三TF-IDF 排序结果与直觉相反搜“Python”返回的全是 Java 教程现象查询python排名第一的是标题含“Python”的 Java 博客因该博客长达 5000 词python出现 12 次TF 高而一篇短小精悍的 Python 入门文python出现 8 次全文仅 300 词得分反低。原因TF 计算未做平滑tf count / (doc_length k)长文档天然占优且未加入标题权重title 中的词应比正文权重高 2 倍。解决修改calculate_tfidf_scores对标题词额外乘以权重因子# 假设我们有 title_terms 列表从 title 标签提取 for term in title_terms: if term in inverted_index: # 标题词 TF 加权 2x tf 2 * (len(positions) / doc_lengths.get(doc_id, 1))5.4 坑四倒排索引文件过大2.3GB加载时内存爆满现象json.load(open(inverted_index.json))卡死htop显示内存占用飙升至 16GB。原因JSON 文件未压缩且 Pythondict存储效率远低于二进制格式10000 文档的索引含 80 万个词项每个词项存[[doc_id, [pos1,pos2]]]结构冗余严重。解决改用pickle序列化 内存映射加载import pickle import mmap # 保存时 with open(inverted_index.pkl, wb) as f: pickle.dump(inverted_index, f) # 加载时不全载入内存 with open(inverted_index.pkl, rb) as f: # 用 mmap 避免一次性读入 with mmap.mmap(f.fileno(), 0, accessmmap.ACCESS_READ) as mm: index pickle.load(mm) # 实际仍会加载但比 JSON 快 3x体积小 60%更进一步生产环境应切换为 SQLite 存储用CREATE VIRTUAL TABLE idx USING fts5(term, doc_id, positions)。6. 验证与调优用真实查询日志反推索引质量而不是只看 top-106.1 构建黄金标准测试集50 个手工标注的查询-文档对不能只靠“搜‘人工智能’看前 10 条是否相关”来评估。我们构建一个小型但高质量的测试集从 10000 网页中随机抽 50 个文档为每个文档人工写出 3 个最能代表其主题的查询词如某篇讲 PyTorch 的文档标注查询为pytorch gpu memory,pytorch dataloader batch,pytorch nn.module共 150 个(query, relevant_doc_id)对。然后运行全部查询统计召回率Recall10150 个相关对中有多少个的相关文档出现在前 10 名准确率Precision10所有查询的前 10 名结果中相关文档占比MRRMean Reciprocal Rank对每个查询取其相关文档的排名倒数如排第 1 名得 1.0第 3 名得 0.33再求平均。def evaluate_retrieval( queries: List[str], gold_pairs: List[Tuple[str, int]], # (query, doc_id) search_func, # callable that returns List[doc_id] k: int 10 ) - dict: recalls, precisions, mrrs [], [], [] for query, gold_doc in gold_pairs: results search_func(query)[:k] # 前 k 名 rank None for i, doc_id in enumerate(results, 1): if doc_id gold_doc: rank i break # Recallk: 1 if found in top-k, else 0 recalls.append(1 if rank else 0) # Precisionk: #relevant_in_topk / k precisions.append(sum(1 for d in results[:k] if d gold_doc) / k) # MRR: 1/rank if found, else 0 mrrs.append(1/rank if rank else 0) return { recall10: sum(recalls) / len(recalls), precision10: sum(precisions) / len(precisions), mrr: sum(mrrs) / len(mrrs) } # 使用 test_results evaluate_retrieval( queries[pytorch gpu memory, tensorflow eager execution], gold_pairs[(pytorch gpu memory, 123), (tensorflow eager execution, 456)], search_funclambda q: search_engine.search(q) ) print(test_results) # {recall10: 1.0, precision10: 0.1, mrr: 0.33}这个测试集的价值在于它把“搜索引擎好不好”从主观感受变成可量化的数字。当recall10低于 0.7 时我就知道分词或停用词表有问题当mrr低于 0.25说明 TF-IDF 权重分配失衡必须引入标题/URL 路径权重。6.2 词项频率分布分析用 Zipf 定律诊断索引健康度健康的倒排索引应该符合 Zipf 定律词频排名 r 的词其频率约正比于 1/r。我们画出词项频次分布图如果出现“头部 100 个词占 90% 文档”说明停用词没清干净如果“长尾词排名 10000几乎为 0”说明分词太激进如把“机器学习算法”硬切成 4 个单字。import matplotlib.pyplot as plt from collections import Counter def analyze_term_distribution(inverted_index: dict): # 统计每个词项出现在多少文档中df doc_freqs [len(entries) for entries in inverted_index.values()] doc_freqs.sort(reverseTrue) # 降序排列 # 取前 1000 个画 log-log 图 ranks list(range(1, min(1001, len(doc_freqs)1))) freqs doc_freqs[:1000] plt.loglog(ranks, freqs, b.) plt.xlabel(Rank (log scale)) plt.ylabel(Document Frequency (log scale)) plt.title(Zipf Distribution Check) plt.grid(True) plt.show() # 输出统计摘要 print(fTotal terms: {len(inverted_index)}) print(fTop 10 terms DF: {freqs[:10]}) print(fTerms with DF1 (hapax legomena): {sum(1 for f in freqs if f1)}) # 运行后如果图线是近似直线说明索引健康如果前段陡降后段平缓说明长尾不足。这张图是我每次重构分词器后的必检项。有一次我把短语合并逻辑写错了深度学习没合成导致深度和学习各自成为高频词而深度学习掉出前 1000 名——图上立刻出现一个明显断层比任何日志都直观。6.3 从那以后我每次构建倒排索引都强制走一遍这三步验证立即跑analyze_term_distribution看 Zipf 图是否平滑terms with DF1是否超过总数 40%过高说明分词太碎抽 10 个高频词手动查inverted_index[term]确认每个(doc_id, positions)元组里的doc_id确实存在于doc_lengths字典中避免索引指向不存在的文档用测试集跑evaluate_retrieval只要recall10低于 0.65就暂停回溯清洗或分词步骤绝不硬着头皮往下走。这三步加起来不超过 2 分钟但省下了我至少 17 小时的无效调试。希望帮到你。本文还有配套的精品资源点击获取
