Python实现基本搜索引擎
这是一个基于Python的轻量级搜索引擎，包含网页爬取、索引构建和搜索功能：
```python
import re
import os
import json
from collections import defaultdict
from typing import List, Dict, Set, Tuple
import hashlib
import math
class SimpleSearchEngine:
def __init__(self):
"""初始化搜索引擎"""
self.index = defaultdict(set) # 倒排索引: 词 -> 文档ID集合
self.documents = {} # 文档存储: doc_id -> 文档信息
self.doc_freq = defaultdict(int) # 文档频率: 词 -> 包含该词的文档数
self.total_docs = 0 # 总文档数
def tokenize(self, text: str) -> List[str]:
"""文本分词，提取关键词"""
# 转为小写，提取字母数字组合
text = text.lower()
# 使用正则提取单词（支持中文）
tokens = re.findall(r'[\w\u4e00-\u9fff]+', text)
# 过滤停用词（这里简单示例，实际可以更丰富）
stop_words = {'的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '个',
'上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好',
'自己', '这', '那', '它', '他', '她', '们', '与', '或', '且', '等'}
return [token for token in tokens if token not in stop_words and len(token) > 1]
def add_document(self, doc_id: str, content: str, metadata: Dict = None):
"""
添加文档到搜索引擎
:param doc_id: 文档唯一标识
:param content: 文档内容
:param metadata: 文档元数据（标题、URL等）
"""
if doc_id in self.documents:
print(f"文档 {doc_id} 已存在，跳过添加")
return
# 分词
tokens = self.tokenize(content)
# 去重后添加到倒排索引
unique_tokens = set(tokens)
for token in unique_tokens:
self.index[token].add(doc_id)
# 保存文档
self.documents[doc_id] = {
'content': content,
'metadata': metadata or {},
'tokens': tokens
}
self.total_docs += 1
def build_doc_freq(self):
"""构建文档频率（用于TF-IDF）"""
for token, docs in self.index.items():
self.doc_freq[token] = len(docs)
def calculate_tf_idf(self, token: str, doc_id: str) -> float:
"""
计算TF-IDF值
:param token: 词项
:param doc_id: 文档ID
:return: TF-IDF分数
"""
if token not in self.index or doc_id not in self.documents:
return 0.0
# 词频 TF
tf = self.documents[doc_id]['tokens'].count(token)
# 逆文档频率 IDF
idf = math.log(self.total_docs / (1 + self.doc_freq.get(token, 0)))
return tf * idf
def search(self, query: str, top_n: int = 10) -> List[Tuple[str, float]]:
"""
搜索文档
:param query: 查询字符串
:param top_n: 返回前N个结果
:return: 文档ID和相关性分数列表
"""
# 分词
query_tokens = self.tokenize(query)
if not query_tokens:
return []
# 查找包含查询词的文档
doc_scores = defaultdict(float)
for token in query_tokens:
if token in self.index:
# 计算每个文档的TF-IDF分数
for doc_id in self.index[token]:
score = self.calculate_tf_idf(token, doc_id)
doc_scores[doc_id] += score
# 按分数降序排序
sorted_docs = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)
# 返回前N个结果
return sorted_docs[:top_n]
def get_document_info(self, doc_id: str) -> Dict:
"""获取文档信息"""
return self.documents.get(doc_id, {})
def save_index(self, filepath: str):
"""保存索引到文件"""
data = {
'index': {k: list(v) for k, v in self.index.items()},
'documents': self.documents,
'doc_freq': dict(self.doc_freq),
'total_docs': self.total_docs
}
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"索引已保存到 {filepath}")
def load_index(self, filepath: str):
"""从文件加载索引"""
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
self.index = {k: set(v) for k, v in data['index'].items()}
self.documents = data['documents']
self.doc_freq = defaultdict(int, data['doc_freq'])
self.total_docs = data['total_docs']
print(f"索引已从 {filepath} 加载")
# 示例使用
def main():
# 创建搜索引擎实例
se = SimpleSearchEngine()
# 添加示例文档
documents = [
{
'id': 'doc1',
'content': 'Python是一种高级编程语言，具有简洁易读的语法。它广泛应用于数据分析、人工智能和Web开发。',
'metadata': {'title': 'Python简介', 'url': 'https://example.com/python'}
},
{
'id': 'doc2',
'content': '搜索引擎是帮助用户快速查找信息的工具。它通过索引和排名算法提供相关结果。',
'metadata': {'title': '搜索引擎基础', 'url': 'https://example.com/search-engine'}
},
{
'id': 'doc3',
'content': '人工智能（AI）是计算机科学的重要分支，包括机器学习、深度学习和自然语言处理。',
'metadata': {'title': '人工智能概述', 'url': 'https://example.com/ai'}
},
{
'id': 'doc4',
'content': '数据分析涉及收集、处理和分析数据，以发现有价值的信息和模式。Python是最常用的数据分析工具之一。',
'metadata': {'title': '数据分析入门', 'url': 'https://example.com/data-analysis'}
}
]
for doc in documents:
se.add_document(doc['id'], doc['content'], doc['metadata'])
# 构建文档频率
se.build_doc_freq()
# 执行搜索
queries = ['Python', '人工智能', '数据分析', '搜索引擎']
print("=" * 60)
print("搜索引擎测试")
print("=" * 60)
for query in queries:
print(f"\n查询: '{query}'")
results = se.search(query, top_n=3)
if results:
print("搜索结果:")
for idx, (doc_id, score) in enumerate(results, 1):
info = se.get_document_info(doc_id)
title = info.get('metadata', {}).get('title', '无标题')
content_preview = info['content'][:50] + '...' if len(info['content']) > 50 else info['content']
print(f" {idx}. [{doc_id}] {title}")
print(f" 相关性分数: {score:.4f}")
print(f" 内容预览: {content_preview}")
else:
print(" 未找到相关结果")
# 保存索引
se.save_index('search_index.json')
# 演示加载索引
se2 = SimpleSearchEngine()
se2.load_index('search_index.json')
print(f"\n加载的索引包含 {len(se2.documents)} 个文档")
if __name__ == "__main__":
main()
```
这个搜索引擎的实现包含以下核心功能：
核心特性：
1. 倒排索引 - 快速定位包含特定词的文档
2. TF-IDF评分 - 评估文档与查询的相关性
3. 中文分词 - 支持中英文混合文本
4. 停用词过滤 - 去除常见无意义词汇
5. 持久化存储 - 保存和加载索引
使用示例：
```python
# 创建搜索引擎
se = SimpleSearchEngine()
# 添加文档
se.add_document('doc1', '这是文档内容...', {'title': '标题'})
# 构建索引
se.build_doc_freq()
# 搜索
results = se.search('关键词', top_n=10)
# 获取文档详情
doc_info = se.get_document_info('doc1')
```
扩展建议：
· 添加更复杂的评分算法（BM25）
· 实现网页爬虫自动采集内容
· 添加缓存机制提升性能
· 实现布尔查询（AND/OR/NOT）
· 添加同义词扩展
· 实现分布式索引
后续可以根据需求调整分词器、停用词列表和评分算法。这个实现适合了解搜索引擎原理，生产环境建议使用Elasticsearch等成熟方案。