Python实现基本搜索引擎

📅 2026/7/18 20:24:09 👁️ 阅读次数 📝 编程学习
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等成熟方案。