Jon Sco1. Google搜索核心架构解析
在互联网信息爆炸的时代，搜索引擎已经成为我们获取信息的主要入口。作为全球最大的搜索引擎，Google每天要处理超过85亿次搜索请求，能够在平均不到0.2秒的时间内返回最相关的结果。这背后是一套极其复杂而又精妙的算法系统在支撑。
Google搜索的核心架构可以概括为CIRS体系，即爬虫(Crawl)、索引(Index)、排序(Rank)和服务(Serve)四个关键环节的协同工作。这个架构设计体现了搜索引擎领域最前沿的技术理念，也是Google能够长期保持搜索质量领先的关键所在。
1.1 CIRS架构工作流程
爬虫环节(Crawl) 是整个搜索流程的起点。GoogleBot作为Google的网页爬虫程序，会持续不断地遍历互联网上的网页。但不同于简单的全量抓取，GoogleBot采用了智能调度策略：
- 根据网站的权威性和更新频率动态调整抓取频率
- 通过robots.txt协议尊重网站的抓取意愿
- 采用分布式架构，全球部署数千台服务器协同工作
一个典型的例子是新闻网站和博客的区别。CNN这样的新闻网站可能每几分钟就会被抓取一次，而个人博客可能几天才会被抓取一次。这种差异化的抓取策略确保了有限的计算资源被用在最需要的地方。
索引环节(Index) 将抓取到的网页内容转化为可快速查询的数据结构。Google采用的是倒排索引(Inverted Index)技术，这是一种将"文档-词"的正向关系转化为"词-文档"反向映射的数据结构。简单来说，就是建立一个巨大的词表，记录每个词出现在哪些网页中。
这种结构的优势在于查询速度。当用户搜索"机器学习"时，系统不需要扫描所有网页，而是直接查找"机器学习"这个词对应的文档列表即可。Google的索引系统还采用了先进的压缩算法，将庞大的网页数据压缩到可管理的规模。
排序环节(Rank) 是Google的核心竞争力所在。早期的搜索引擎主要依赖关键词匹配，而Google开创性地引入了PageRank算法，通过分析网页之间的链接关系来评估网页的权威性。如今，Google的排序算法已经发展成为一个包含数百个因素的复杂系统：
- 传统算法：PageRank、TF-IDF、BM25等
- 机器学习模型：RankBrain、BERT等深度学习模型
- 用户行为信号：点击率、停留时间等
这些因素被综合计算，最终生成每个网页的排序分数。值得注意的是，不同查询类型会侧重不同的排序因素。例如，对于"最新新闻"这类查询，时效性的权重会更高；而对于"Python教程"这类查询，内容的深度和质量会更重要。
服务环节(Serve) 负责将排序后的结果快速呈现给用户。这一环节看似简单，实则包含多项优化技术：
- 结果缓存：热门查询的结果会被缓存，响应时间可缩短到50毫秒以内
- 拼写纠正：自动修正"Gooogle"这类拼写错误
- 个性化调整：根据用户位置、搜索历史等因素微调结果
所有这些环节协同工作，确保用户能够在输入查询后几乎瞬间获得最相关的结果。整个流程的耗时通常控制在200毫秒以内，这种极致的速度体验是Google搜索深受用户信赖的重要原因。
1.2 技术演进历程
Google搜索算法并非一成不变，而是经历了持续的演进和革新。我们可以将其发展历程划分为三个阶段：
第一阶段(1998-2004)：链接分析时代
这个阶段的核心是PageRank算法，它革命性地利用了网页之间的链接关系作为质量信号。同时，TF-IDF(词频-逆文档频率)算法帮助评估关键词与文档的相关性。这一时期的Google迅速超越了当时依赖关键词匹配的竞争对手。
第二阶段(2004-2015)：机器学习引入
随着网页数量爆炸式增长，单纯的链接分析已不足以应对复杂的排序需求。Google开始引入机器学习技术：
- 2003年引入Orion系统，使用用户点击数据优化排序
- 2013年推出Hummingbird算法，更好地理解查询意图
- BM25算法取代TF-IDF，提供更精确的相关性评分
第三阶段(2015至今)：深度学习主导
深度学习技术的突破彻底改变了搜索排序的方式：
- 2015年推出RankBrain，首次将深度神经网络应用于搜索排序
- 2018年引入BERT模型，大幅提升对自然语言的理解能力
- 2020年推出MUM模型，支持跨语言、多模态的搜索体验
这种技术演进使得Google搜索从最初的关键词匹配，发展到如今能够理解复杂查询意图、处理多模态内容的智能系统。值得注意的是，新技术的引入并非完全取代旧技术，而是形成了一种"传统算法+深度学习"的混合架构，各自发挥优势。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 爬虫系统深度解析
作为搜索引擎的数据采集端，爬虫系统的设计直接决定了搜索引擎能够覆盖多少网页内容，以及这些内容的时效性如何。Google的爬虫系统GoogleBot是互联网上最复杂的网络爬虫之一，它需要解决海量网页抓取、重复内容识别、网站负载控制等一系列挑战。
2.1 智能调度策略
GoogleBot的核心设计理念是"智能调度"，即根据网站的重要性、更新频率等因素动态调整抓取策略。这种设计主要考虑两个维度：抓取效率和网站友好性。
爬虫预算(Crawl Budget) 是GoogleBot调度策略的核心概念。它决定了GoogleBot会为一个网站分配多少抓取资源，包括：
- 抓取频率：多长时间访问一次该网站
- 抓取深度：每次访问会抓取多少页面
- 抓取速度：每秒发送多少个请求
影响爬虫预算的主要因素包括：
- 网站权威性：通过PageRank等指标评估。权威性高的网站(如Wikipedia)会获得更多抓取资源。
- 内容更新频率：新闻类网站比静态的企业官网抓取更频繁。
- 服务器性能：GoogleBot会监测网站响应时间，自动调整请求速率避免过载。
- 历史数据质量：如果之前抓取的内容大多质量较低，可能会减少抓取。
实际应用中，SEO专家可以通过Google Search Console查看自己网站的爬虫统计信息，了解GoogleBot的抓取情况。如果发现重要页面未被抓取，可以通过提交Sitemap等方式引导爬虫。
2.2 内容去重技术
互联网上存在大量重复或近似内容，如果不加以处理，会严重影响搜索结果的质量和多样性。Google采用多层次的内容去重技术来解决这个问题。
SimHash算法 是Google用于快速识别相似内容的核心技术。它的工作原理如下：
- 对网页内容进行分词处理，提取关键词
- 为每个关键词生成64位的哈希值
- 根据词频加权计算所有哈希值的综合向量
- 最终生成一个代表整个网页内容的64位指纹
判断两个网页是否相似时，只需比较它们的SimHash指纹的汉明距离(即不同位的数量)。如果距离小于3(对于64位哈希)，则认为内容高度相似。
这种算法有几个显著优势：
- 计算效率高，适合海量网页处理
- 对内容的微小修改不敏感，避免过度去重
- 可以配置不同的相似度阈值，灵活控制去重强度
除了内容相似度，Google还会考虑URL规范化(如统一大小写、去除会话ID等)来识别重复页面。一个典型的例子是电商网站，同一商品可能有多个URL(如通过不同分类进入)，Google会识别这些URL指向相同内容。
2.3 Python实现简易爬虫
理解理论后，我们可以用Python实现一个具备基本功能的简易爬虫。这个爬虫将包含GoogleBot的核心特性：智能调度和内容去重。
python复制import requests
from bs4 import BeautifulSoup
from collections import deque
import hashlib
from urllib.parse import urlparse
class AdvancedWebCrawler:
def __init__(self):
self.visited = set() # 已访问URL集合
self.queue = deque() # 待访问URL队列
self.domain_limits = {} # 域名级别的请求限速
self.headers = {
'User-Agent': 'Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)'
}
def get_domain(self, url):
"""提取URL的域名部分"""
parsed = urlparse(url)
return parsed.netloc
def can_fetch(self, url):
"""检查是否允许抓取该URL"""
domain = self.get_domain(url)
# 简单模拟robots.txt规则
if 'disallow' in domain:
return False
# 检查域名级别的请求限制
if domain in self.domain_limits:
if self.domain_limits[domain] <= 0:
return False
return True
def generate_simhash(self, content):
"""生成内容的SimHash指纹(简化版)"""
words = content.split()
hash_bits = [0] * 64
for word in words:
# 计算单词的哈希值
word_hash = int(hashlib.md5(word.encode()).hexdigest(), 16)
# 将哈希值的每一位作为特征
for i in range(64):
bit = (word_hash >> i) & 1
if bit == 1:
hash_bits[i] += 1
else:
hash_bits[i] -= 1
# 生成最终的指纹
fingerprint = 0
for i in range(64):
if hash_bits[i] > 0:
fingerprint |= 1 << i
return fingerprint
def is_duplicate(self, content):
"""检查内容是否重复"""
fingerprint = self.generate_simhash(content)
# 简化版去重：检查指纹是否已存在
if fingerprint in self.visited:
return True
self.visited.add(fingerprint)
return False
def crawl_page(self, url):
"""抓取单个页面"""
if not self.can_fetch(url):
return None
try:
# 控制抓取速度
domain = self.get_domain(url)
if domain in self.domain_limits:
self.domain_limits[domain] -= 1
response = requests.get(url, headers=self.headers, timeout=10)
response.raise_for_status()
response.encoding = response.apparent_encoding or 'utf-8'
# 解析页面内容
soup = BeautifulSoup(response.text, 'html.parser')
title = soup.title.string if soup.title else 'No Title'
content = soup.get_text(strip=True)
# 内容去重检查
if self.is_duplicate(content):
print(f"发现重复内容：{url}")
return None
# 提取页面中的链接
links = []
for a_tag in soup.find_all('a', href=True):
link = a_tag['href']
if link.startswith('http'):
links.append(link)
return {
'url': url,
'title': title,
'content': content,
'links': links
}
except Exception as e:
print(f"抓取失败：{url}，错误：{str(e)}")
return None
def start_crawl(self, start_url, max_pages=50, max_per_domain=10):
"""启动爬虫"""
self.queue.append(start_url)
crawled_pages = 0
# 初始化域名限制
domain = self.get_domain(start_url)
self.domain_limits[domain] = max_per_domain
while self.queue and crawled_pages < max_pages:
url = self.queue.popleft()
page_data = self.crawl_page(url)
if page_data:
crawled_pages += 1
print(f"成功抓取第{crawled_pages}页：{page_data['title']}")
# 将新链接加入队列
for link in page_data['links']:
link_domain = self.get_domain(link)
# 初始化新域名的限制
if link_domain not in self.domain_limits:
self.domain_limits[link_domain] = max_per_domain
# 控制队列大小
if link not in self.queue and len(self.queue) < 1000:
self.queue.append(link)
print(f"抓取完成，共抓取{crawled_pages}页")
# 使用示例
if __name__ == "__main__":
crawler = AdvancedWebCrawler()
crawler.start_crawl(start_url="https://example.com", max_pages=20)
这个爬虫实现了几个关键功能：
- 域名级别的请求限制：防止对单一网站发送过多请求
- 基本的robots.txt模拟：尊重网站的抓取意愿
- SimHash去重：识别内容相似的页面
- 广度优先搜索：系统地遍历网页链接
在实际应用中，这个爬虫还可以进一步优化：
- 增加分布式支持，多机协同抓取
- 实现更精确的robots.txt解析
- 添加动态页面渲染能力(使用Selenium等工具)
- 引入更复杂的调度算法，如基于PageRank的优先级调度
3. 倒排索引技术详解
倒排索引是搜索引擎能够实现毫秒级响应速度的核心技术。与传统的正排索引(文档→词项)不同，倒排索引(词项→文档)的结构特别适合快速查找包含特定关键词的文档。Google的索引系统每天要处理数十亿网页，构建的倒排索引规模极其庞大，需要精妙的设计才能高效运作。
3.1 倒排索引结构解析
一个完整的倒排索引由两部分组成：词典(Term Dictionary)和倒排列表(Posting List)。
词典 包含所有唯一的词项(term)，通常按照字典序排列以便快速查找。在实际实现中，词典可能会被进一步优化：
- 前缀压缩：存储共同前缀一次，如"compute"和"computer"可以共享"comput"前缀
- 块存储：将词典分成多个块，减少内存占用
- 跳跃表：加速词典内的查找速度
倒排列表 是倒排索引的核心部分，每个词项对应一个倒排列表，记录包含该词项的所有文档信息。一个典型的倒排列表条目包含：
- 文档ID：唯一标识一个文档
- 词频(TF)：该词项在文档中出现的次数
- 位置信息：词项在文档中出现的位置(用于短语查询和高亮)
- 其他元数据：如词项在文档中的重要性标记
在Google的实际实现中，倒排列表还会存储更多丰富的信息，如词项在标题、锚文本等特殊位置的出现情况，这些信息都会影响最终的排序评分。
3.2 索引压缩技术
倒排索引通常非常庞大，Google的索引数据据说达到PB级别。为了减少存储空间和提高IO效率，索引压缩技术至关重要。
变长编码(Variable-length Encoding) 是常用的压缩技术之一。其核心思想是用更少的字节表示较小的数字。常见的变长编码包括：
- Varint编码：使用字节的最高位作为延续位，低7位存储数据
- Elias编码：使用前缀表示数字的位数，然后存储实际数值
- Simple9编码：将多个整数打包到一个32位字中
对于位置信息这类通常较小的数字，变长编码可以显著减少存储空间。例如，数字1只需要1个字节，而不用固定的4字节存储。
块压缩(Block Compression) 是另一种重要技术。它将倒排列表分成多个块，每个块单独压缩。这样在查询时，可以只解压需要的块，减少内存占用。常用的块压缩算法包括：
- PForDelta：适用于密集数值序列
- LZ4：快速压缩解压算法
- Zstandard：提供更好的压缩比
Google还采用了分层索引技术，将索引分为内存部分和磁盘部分。热门的词项索引常驻内存，而冷门词项存储在磁盘，通过缓存策略平衡速度和存储成本。
3.3 Python实现倒排索引
下面我们用Python实现一个功能完整的倒排索引系统，包含词典、倒排列表和基本的查询功能。
python复制import math
from collections import defaultdict
import struct
import zlib
class InvertedIndex:
def __init__(self, compression=True):
self.lexicon = {} # 词典：{词项: (偏移量, 倒排列表长度)}
self.postings = bytearray() # 倒排列表存储
self.doc_map = {} # 文档ID到URL的映射
self.next_doc_id = 1 # 下一个文档ID
self.compression = compression # 是否启用压缩
def add_document(self, url, title, content):
"""添加文档到索引"""
doc_id = self.next_doc_id
self.next_doc_id += 1
self.doc_map[doc_id] = {'url': url, 'title': title}
# 简单分词（实际应用应使用专业分词器）
words = self.tokenize(content)
# 记录词项位置
term_positions = defaultdict(list)
for pos, word in enumerate(words):
term_positions[word].append(pos)
# 更新倒排列表
for term, positions in term_positions.items():
frequency = len(positions)
# 准备倒排列表条目
entry = {
'doc_id': doc_id,
'frequency': frequency,
'positions': positions
}
# 序列化条目
serialized = self.serialize_entry(entry)
# 如果启用压缩，压缩数据
if self.compression:
compressed = zlib.compress(serialized)
# 存储压缩标记(1字节)和压缩后长度(4字节)
self.postings.extend(b'\x01')
self.postings.extend(struct.pack('>I', len(compressed)))
self.postings.extend(compressed)
else:
# 存储压缩标记(0)和原始长度
self.postings.extend(b'\x00')
self.postings.extend(struct.pack('>I', len(serialized)))
self.postings.extend(serialized)
# 更新词典
if term not in self.lexicon:
self.lexicon[term] = {
'offset': len(self.postings) - len(serialized) - 5,
'length': 1
}
else:
self.lexicon[term]['length'] += 1
def serialize_entry(self, entry):
"""序列化倒排列表条目"""
# 使用简单二进制格式：doc_id(4B) + frequency(4B) + positions_count(4B) + positions...
data = bytearray()
data.extend(struct.pack('>I', entry['doc_id']))
data.extend(struct.pack('>I', entry['frequency']))
# 使用差值编码存储位置，然后使用变长编码压缩
positions = entry['positions']
prev_pos = 0
encoded_positions = bytearray()
for pos in positions:
delta = pos - prev_pos
# 使用Varint编码
while delta >= 0x80:
encoded_positions.append((delta & 0x7F) | 0x80)
delta >>= 7
encoded_positions.append(delta)
prev_pos = pos
data.extend(struct.pack('>I', len(encoded_positions)))
data.extend(encoded_positions)
return bytes(data)
def deserialize_entry(self, data):
"""反序列化倒排列表条目"""
if data[0] == 1: # 压缩数据
decompressed = zlib.decompress(data[5:])
return self.deserialize_entry(decompressed)
doc_id = struct.unpack('>I', data[:4])[0]
frequency = struct.unpack('>I', data[4:8])[0]
pos_length = struct.unpack('>I', data[8:12])[0]
# 解码位置信息
positions = []
pos_data = data[12:12+pos_length]
delta = 0
shift = 0
current_pos = 0
for byte in pos_data:
delta |= (byte & 0x7F) << shift
if not (byte & 0x80):
current_pos += delta
positions.append(current_pos)
delta = 0
shift = 0
else:
shift += 7
return {
'doc_id': doc_id,
'frequency': frequency,
'positions': positions
}
def tokenize(self, text):
"""简单分词函数"""
# 转换为小写，移除标点
text = text.lower()
for ch in ',.!?;:"\'()[]{}':
text = text.replace(ch, ' ')
return text.split()
def search_term(self, term):
"""查询单个词项"""
if term not in self.lexicon:
return []
entry_info = self.lexicon[term]
offset = entry_info['offset']
length = entry_info['length']
results = []
current_offset = offset
for _ in range(length):
# 读取压缩标记
compressed_flag = self.postings[current_offset]
# 读取数据长度
data_length = struct.unpack('>I', self.postings[current_offset+1:current_offset+5])[0]
# 读取数据
entry_data = self.postings[current_offset+5:current_offset+5+data_length]
# 反序列化条目
entry = self.deserialize_entry(bytes([compressed_flag]) + struct.pack('>I', data_length) + entry_data)
results.append({
'url': self.doc_map[entry['doc_id']]['url'],
'title': self.doc_map[entry['doc_id']]['title'],
'frequency': entry['frequency'],
'positions': entry['positions']
})
# 移动到下一个条目
current_offset += 5 + data_length
return results
def search_phrase(self, phrase):
"""短语查询"""
terms = self.tokenize(phrase)
if not terms:
return []
# 获取每个词项的倒排列表
term_results = []
for term in terms:
postings = self.search_term(term)
if not postings:
return [] # 如果有词项不存在，短语肯定不存在
term_results.append(postings)
# 找出所有文档都包含的文档ID
common_docs = set(doc['url'] for doc in term_results[0])
for postings in term_results[1:]:
current_docs = set(doc['url'] for doc in postings)
common_docs.intersection_update(current_docs)
if not common_docs:
return []
# 检查短语顺序
phrase_results = []
for doc_url in common_docs:
# 收集每个词项在该文档中的位置
term_positions = []
for i, term in enumerate(terms):
for posting in term_results[i]:
if posting['url'] == doc_url:
term_positions.append(posting['positions'])
break
# 检查是否存在连续的位置序列
# 简单实现：检查第一个词的位置+1是否在第二个词的位置中，依此类推
for pos in term_positions[0]:
found = True
for i in range(1, len(terms)):
if (pos + i) not in term_positions[i]:
found = False
break
if found:
phrase_results.append({
'url': doc_url,
'title': next(doc['title'] for doc in term_results[0] if doc['url'] == doc_url),
'positions': list(range(pos, pos + len(terms)))
})
break
return phrase_results
# 使用示例
if __name__ == "__main__":
index = InvertedIndex(compression=True)
# 添加一些文档到索引
documents = [
{
"url": "https://example.com/doc1",
"title": "人工智能概述",
"content": "人工智能是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。"
},
{
"url": "https://example.com/doc2",
"title": "机器学习基础",
"content": "机器学习是人工智能的核心，是使计算机具有智能的根本途径。机器学习主要研究计算机怎样模拟或实现人类的学习行为。"
},
{
"url": "https://example.com/doc3",
"title": "深度学习与应用",
"content": "深度学习是机器学习的分支，是一种以人工神经网络为架构，对数据进行表征学习的算法。深度学习在计算机视觉和自然语言处理等领域有广泛应用。"
}
]
for doc in documents:
index.add_document(doc["url"], doc["title"], doc["content"])
# 执行查询
print("单关键词查询结果(人工智能):")
for result in index.search_term("人工智能"):
print(f"文档: {result['title']}, URL: {result['url']}, 词频: {result['frequency']}")
print("\n短语查询结果(机器学习是):")
for result in index.search_phrase("机器学习是"):
print(f"文档: {result['title']}, URL: {result['url']}, 位置: {result['positions']}")
这个实现包含了几个关键特性：
- 压缩存储：支持zlib压缩倒排列表，减少内存/磁盘占用
- 差值编码：对位置信息使用差值编码，提高压缩率
- Varint编码：对小整数使用变长编码
- 短语查询：支持精确的短语匹配查询
- 高效序列化：使用二进制格式而非JSON等文本格式，提高IO效率
在实际应用中，这个索引系统还可以进一步优化：
- 实现内存映射文件，支持超大规模索引
- 添加并发控制，支持多线程索引构建
- 实现索引合并策略，支持增量索引更新
- 添加更复杂的分词器，支持中文等语言
4. 排序算法深度剖析
排序算法是搜索引擎的核心机密，也是决定搜索结果质量的关键因素。Google的排序算法经历了从简单的PageRank到复杂的深度学习模型的演进过程。理解这些算法的原理和实现，对于构建高质量的搜索系统至关重要。
4.1 PageRank算法详解
PageRank是Google创始人Larry Page和Sergey Brin在斯坦福大学开发的核心算法，它革命性地利用了网页之间的链接关系作为质量信号。PageRank的基本思想是：一个网页的重要性取决于链接到它的其他网页的数量和质量。
算法原理：
PageRank将互联网建模为一个有向图，其中网页是节点，链接是边。每个节点的PageRank值通过以下公式计算：
PR(A) = (1 - d)/N + d × (PR(B)/L(B) + PR(C)/L(C) + ... + PR(N)/L(N))
其中：
- PR(A)：网页A的PageRank值
- d：阻尼系数，通常设为0.85，表示用户继续点击链接的概率
- N：网页总数
- L(B)：网页B的出链数量
- PR(B)/L(B)：网页B将其PageRank值均分给它指向的所有网页
收敛计算：
PageRank值通过迭代计算直到收敛。初始时，所有网页的PageRank值都设为1/N。然后反复应用上述公式更新每个网页的PageRank值，直到变化小于某个阈值。
Python实现：
python复制import numpy as np
def pagerank(links, d=0.85, max_iter=100, tol=1e-6):
"""
计算PageRank值
参数:
links -- 字典，表示链接关系 {page: [linked_pages]}
d -- 阻尼系数，默认0.85
max_iter -- 最大迭代次数，默认100
tol -- 收敛阈值，默认1e-6
返回:
排序后的PageRank值列表 [(page, rank)]
"""
# 收集所有唯一页面
pages = set()
for src, dests in links.items():
pages.add(src)
pages.update(dests)
pages = list(pages)
N = len(pages)
# 创建页面到索引的映射
page_index = {page: i for i, page in enumerate(pages)}
# 初始化转移矩阵M
M = np.zeros((N, N))
for src, dests in links.items():
if not dests: # 处理没有出链的情况
continue
src_idx = page_index[src]
# 均分权重
weight = 1.0 / len(dests)
for dest in dests:
dest_idx = page_index[dest]
M[dest_idx][src_idx] = weight
# 处理悬挂节点(没有出链的页面)
dangling_nodes = np.where(M.sum(axis=0) == 0)[0]
for node in dangling_nodes:
M[:, node] = 1.0 / N # 随机跳转到任何页面
# 添加阻尼因子
M = d * M + (1 - d) / N
# 初始化PageRank向量
pr = np.ones(N) / N
# 迭代计算
for _ in range(max_iter):
new_pr = M @ pr
diff = np.abs(new_pr - pr).sum()
if diff < tol:
break
pr = new_pr
# 返回排序结果
ranked_pages = [(pages[i], pr[i]) for i in range(N)]
ranked_pages.sort(key=lambda x: x[1], reverse=True)
return ranked_pages
# 使用示例
if __name__ == "__main__":
# 示例链接关系
web_links = {
'A': ['B', 'C'],
'B': ['C'],
'C': ['A'],
'D': ['C']
}
# 计算PageRank
results = pagerank(web_links)
print("PageRank结果:")
for page, rank in results:
print(f"{page}: {rank:.4f}")
这个实现包含了PageRank算法的所有关键要素：
- 链接关系的矩阵表示
- 阻尼系数的处理
- 悬挂节点(没有出链的页面)的处理
- 迭代计算直到收敛
在实际应用中，对于大规模网页图，我们通常会使用稀疏矩阵表示和分布式计算来优化性能。
4.2 BM25相关性评分
BM25(Okapi BM25)是信息检索领域最成功的相关性评分算法之一，它基于概率检索模型，比传统的TF-IDF方法更能准确反映文档与查询的相关性。
算法原理：
BM25评分公式如下：
score(D, Q) = Σ IDF(qi) × (TF(qi, D) × (k1 + 1)) / (TF(qi, D) + k1 × (1 - b + b × |D|/avgdl))
其中：
- D：文档
- Q：查询，由词项qi组成
- TF(qi, D)：词项qi在文档D中的词频
- |D|：文档D的长度(词项数)
- avgdl：所有文档的平均长度
- k1和b：可调参数，通常k1∈[1.2,2.0]，b=0.75
- IDF(qi)：词项qi的逆文档频率，计算方式为：
IDF(qi) = log((N - n(qi) + 0.5) / (n(qi) + 0.5) + 1)
其中N是文档总数，n(qi)是包含qi的文档数。
Python实现：
python复制import math
from collections import defaultdict
class BM25:
def __init__(self, documents, k1=1.5, b=0.75):
"""
初始化BM25评分器
参数:
documents -- 文档列表，每个文档是字典 {'id': ..., 'text': ...}
k1, b -- BM25参数
"""
self.k1 = k1
self.b = b
self.documents = documents
self.N = len(documents)
self.avgdl = sum(len(doc['text'].split()) for doc in documents) / self.N
self.doc_lengths = [len(doc['text'].split()) for doc in documents]
# 构建词项统计
self.term_stats = defaultdict(dict)
self.doc_freq = defaultdict(int) # 包含每个词项的文档数
for doc_id, doc in enumerate(documents):
text = doc['text']
terms = text.split()
term_counts = defaultdict(int)
for term in terms:
term_counts[term] += 1
for term, count in term_counts.items():
self.term_stats[term][doc_id] = count
self.doc_freq[term] += 1
def idf(self, term):
"""计算词项的IDF值"""
if term not in self.doc_freq:
return 0
n_qi = self.doc_freq[term]
return math.log((self.N - n_qi + 0.5) / (n_qi + 0.5) + 1)
def score(self, doc_id, query_terms):
"""计算文档对查询的BM25得分"""
score = 0.0
doc_length = self.doc_lengths[doc_id]
for term in query_terms:
if term not in self.term_stats or doc_id not in self.term_stats[term]:
continue
tf = self.term_stats[term][doc_id]
idf = self.idf(term)
# BM25计算
numerator = tf * (self.k1 + 1)
denominator = tf + self.k1 * (1 - self.b + self.b * doc_length / self.avgdl)
score += idf * numerator / denominator
return score
def rank(self, query):
"""对查询结果进行排序"""
query_terms = query.split()
scores = []
for doc_id in range(self.N):
doc_score = self.score(doc_id, query_terms)
scores.append((doc_id, doc_score))
# 按分数降序排序
scores.sort(key=lambda x: x[1], reverse=True)
return scores
# 使用示例
if __name__ == "__main__":
# 示例文档
documents = [
{'id': 0, 'text': '人工智能是研究开发用于模拟延伸和扩展人的智能的理论方法'},
{'id': 1, 'text': '机器学习是人工智能的核心是使计算机具有智能的根本途径'},
{'id': 2, 'text': '深度学习是机器学习的分支是一种以人工神经网络为架构的算法'}
]
# 初始化BM25
bm25 = BM25(documents, k1=1.5, b=0.75)
# 执行查询
query = "人工智能 机器学习"
results = bm25.rank(query)
print(f"查询: '{query}' 的结果:")
for doc_id, score in results:
print(f"文档ID: {doc_id}, 得分: {score:.4f}, 内容: {documents[doc_id]['text'][:30]}...")
BM25相比TF-IDF有几个关键优势：
- 对词频进行饱和处理，避免高频词过度影响
- 考虑文档长度，惩罚过长的文档
- 参数可调，可以针对不同数据集优化
在实际应用中，BM25通常与其他特征(如PageRank)结合使用，形成最终的排序分数。
4.3 深度学习排序模型
随着深度学习的发展，Google等搜索引擎开始采用神经网络模型来改进排序效果。RankBrain是Google于2015年推出的深度学习排序系统，它能够更好地理解查询意图和文档语义。
核心思想：
- 查询理解：将查询转换为语义向量，理解用户真实意图
- 文档表示：将文档内容也转换为语义向量
- 相关性计算：计算查询向量与文档向量的相似度
- 特征融合：将语义相似度与传统特征(如BM25、PageRank)结合
BERT模型应用：
Google