爬取、提取、索引、服务四阶段分离;用内容哈希判断增量,五千页中仅索引变化的二十页,而非每次全量重跑。保留原始 HTML 是关键决策,提取器迭代时可直接从磁盘重跑。
站搜索的难点不在搜索本身,而在第二次运行:五千个页面,其中二十个发生了变化,而索引任务不能花一个小时和一张账单去重新嵌入另外四千九百八十个内容块。对每个 chunk 做内容哈希,可以把这件事变成一个几秒钟就能完成的任务。
把它们拆开,因为它们的失败模式不同,运行节奏也不同。
保留原始 HTML 是人们最后悔没做的事。当你更换提取器的时候——你一定会换的,第一次模板改动把 Cookie 弹窗塞进所有 chunk 时——你从磁盘重新提取,而不是重新爬取整个网站。
尽早决定"一个页面"在你的站点里意味着什么,因为这并不显而易见,而且会改变结果。一篇分页文章是一个文档拆分到五个 URL;一个带锚点的文档页面可以说是一个 URL 上的五个文档;一个列表页面根本不应该被索引,因为它包含了一百个其他页面的碎片,每个查询都会微弱地匹配到它。把包含规则写下来——一个 URL 模式列表就够了——把它放在爬虫旁边,否则它只会活在写它的那个人脑子里。
你自己的网站仍然值得发一个条件请求并加上延迟。标准库同时提供了这两样东西。
# crawl.py
import time, urllib.request, urllib.robotparser
from urllib.parse import urljoin, urlparse
UA = "yoursite-search/1.0 (+https://yoursite.example/about-our-crawler)"
DELAY = 0.5
rp = urllib.robotparser.RobotFileParser()
rp.set_url("https://yoursite.example/robots.txt")
rp.read()
def fetch(url, etag=None):
if not rp.can_fetch(UA, url):
return None, None, "disallowed"
req = urllib.request.Request(url, headers={"User-Agent": UA})
if etag:
req.add_header("If-None-Match", etag)
try:
with urllib.request.urlopen(req, timeout=30) as r:
return r.read().decode("utf-8", "replace"), r.headers.get("ETag"), "ok"
except urllib.error.HTTPError as e:
if e.code == 304:
return None, etag, "unchanged"
return None, None, "http-" + str(e.code)
finally:
time.sleep(DELAY)
304 Not Modified 是可能的最低成本回答,大多数行为良好的静态站点托管都会主动发送 ETag。在一个行为良好的站点上,仅此一项就能把夜间爬取削减到只有几次真实下载——这还没有算下面任何嵌入优化的收益。
目标是文章,不是外壳。导航、页脚和 Cookie 弹窗在每个页面上都重复出现,如果它们最终进入了你的 chunk,那么每个查询都会轻微地匹配到每个页面,这就是那个返回看似合理实则胡说八道的搜索系统的典型症状。
优先使用模板已经给你的容器:<main>、<article>,或者一个已知的内容 class。在你自己的站点上你是知道这一点的——当 CSS 选择器可以精确命中时,不要用通用的可读性启发式方法。
剥离 <script>、<style>、<nav>、<header>、<footer> 以及任何 aria-hidden 的内容。
保留 <h1>–<h3> 作为每个 chunk 的标题路径,和 PDF 构建的做法一样。这是同一个技巧,而且出于同样的原因有效。
对结果做断言:如果一个你知道很长的页面提取出来的文本低于 200 字符,说明选择器坏了。大声让任务失败,而不是索引空页面。
Python 标准库的 html.parser 就够用了,而且没有依赖项;对 HTML 的内容提取在你对模板的控制权越少的时候就越难。
这就是机制,而且只有几行代码。对每个 chunk 做哈希;把哈希存在向量旁边;重新索引时,只嵌入那些哈希在表中不存在的 chunk,并删除那些在该 URL 下不再出现的哈希对应的行。
# index.py
import hashlib, json, sqlite3
DB = sqlite3.connect("search.db")
DB.executescript("""
CREATE TABLE IF NOT EXISTS chunk (
hash TEXT PRIMARY KEY, -- sha256 of the normalised chunk text
url TEXT NOT NULL,
ord INTEGER NOT NULL,
title TEXT NOT NULL,
text TEXT NOT NULL,
vec TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS chunk_url ON chunk(url);
""")
def chunk_hash(text):
return hashlib.sha256(" ".join(text.split()).encode()).hexdigest()
def reindex(url, title, pieces):
wanted = {chunk_hash(p): (i, p) for i, p in enumerate(pieces)}
have = {h for (h,) in DB.execute(
"SELECT hash FROM chunk WHERE url = ?", (url,))}
stale = have - wanted.keys()
if stale:
DB.executemany("DELETE FROM chunk WHERE hash = ?",
[(h,) for h in stale])
fresh = [h for h in wanted if h not in have]
if fresh:
vecs = embed([wanted[h][1] for h in fresh]) # the only paid call
DB.executemany(
"INSERT OR REPLACE INTO chunk (hash,url,ord,title,text,vec) "
"VALUES (?,?,?,?,?,?)",
[(h, url, wanted[h][0], title, wanted[h][1], json.dumps(v))
for h, v in zip(fresh, vecs)],
)
DB.commit()
return len(fresh), len(stale)
两个细节完成了这项工作。哈希是基于 " ".join(text.split()) 计算的,所以重新换行的段落或更改的缩进不算作变更——否则一次模板调整就会重新嵌入整个站点。而 ord 被存储但不被哈希,所以把一个段落移到页面上方只会重排行而不会重新嵌入任何东西。
需要针对的相反失败模式是这样的:两个 URL 上文本完全相同的 chunk 共享一个哈希,而以哈希为主键,第二个 URL 会覆盖第一个。如果你的站点确实有重复的样板段落,改用 sha256(url + text) 作为主键,并接受每个 URL 会重复嵌入一次这个事实。
每夜任务处理 5,000 个页面,每个约 6 个 chunk = 30,000 个 chunk。
全量重新嵌入: 30,000 x 300 tokens = 9.0M tokens
增量处理,20 个页面变化:
120 x 300 tokens = 36,000 tokens
= 全量运行的 0.4%。
嵌入价格为每百万 tokens $0.02,那就是 $0.18 对比 $0.0007 ——
无论哪个都很小。真正的原因在于时钟:全量运行是 30,000 次嵌入
的网络往返和速率限制,而增量运行在你还没泡好咖啡的时候就结束了。
一次嵌入调用用于查询,然后和任何其他向量搜索一样的点积——但有两样东西属于服务路径,而不属于批处理脚本。
缓存查询嵌入。站点搜索查询遵循极端的幂律分布;一个以规范化查询为键、控制在几千条目的字典,可以省去大多数搜索的一次网络往返。
按 URL 分组结果。用户想要的是页面,不是 chunk。取每个 URL 得分最高的 chunk,用它的文本作为摘要,并且永远不要返回同一个页面两次。
要有降级方案。如果嵌入调用失败或超时,降级到关键词搜索而不是显示错误。能降级的搜索才是可用的搜索。
第一个投诉会是这样:搜索一个精确术语——产品代码、人的姓氏、版本号——返回的是关于这个一般性主题的页面,而不是包含该术语的页面。这是嵌入按设计工作的表现:它们编码的是语义,而序列号没有语义可以编码。
解决方案不是更好的模型。让关键词搜索并行运行——SQLite 的 FTS5 给你提供数据库中已有的 BM25——然后合并两个结果列表。倒数排名融合是标准的合并方式,而且只有三行代码:对每个文档的评分是它在各列表中 1 / (60 + rank) 的总和,然后排序。混合检索在几乎所有真实语料库上都胜过任一半身,而且两种方法在互补的查询上失败,这正是合并有效的根本原因。
站点搜索的评判标准是浏览器自身的自动补全,那可是即时的,所以预算很紧,这些术语值得在发现它们之前先写下来。
目标:用户停止输入后 300 毫秒内渲染出结果。
防抖 150 ms <- 你的选择,它吃掉了二分之一的预算
请求到你的服务器 20 ms
查询嵌入调用 80-250 ms <- 一个网络调用,而且是风险所在
向量扫描,30,000 个 chunk 15 ms (NumPy:一次 30,000 x 1,536 矩阵乘法)
分组、摘要、序列化 5 ms
渲染 15 ms
两个结论随之而来。
1. 嵌入调用是你唯一无法控制的项,而且它在整个预算中占比相当。
把它缓存起来,并考虑用一个更小的嵌入模型来处理查询——仅当
查询和文档使用同一个模型时才可以——它们必须如此,否则向量
是不可比的。
2. 把防抖设在 150 ms 而不是 300 ms 可以免费节省 150 ms,代价是
对分burst输入的人发出更多请求。由于缓存会吸收重复的前缀,
这个代价比看起来要小。
RAG 构建中的纯 Python 扫描在这个语料库规模上需要 1–3 秒,根本塞不进这个预算;用 NumPy 做同样的运算作为单次矩阵乘法就可以了,绰绰有余。这才是这里使用数值库的真正原因,而且值得精确说明——不是"NumPy 更快",而是"预算是 300 毫秒,而其中一项是 1,000 毫秒"。
如果嵌入调用不可用或很慢,返回关键词结果并标记为降级结果,而不是显示一个加载动画。一个有时只是不错的搜索框比一个有时完全不工作的搜索框更好;能降级而不是失败,这才是让一个功能在第一个糟糕的星期里仍然存活下来的模式。
按照站点变化的频率来安排爬取和索引的计划。对于大多数站点每夜一次是对的;每小时发布一次的文档站点想要的是部署时的 webhook。
把索引写到一个新的 SQLite 文件上,等任务成功后再 rename 覆盖旧文件。原子 rename 意味着搜索永远不会从一个半成品的索引中提供服务。
记录得分高于某个阈值但零结果的查询。这个日志是你站点能产生的最高价值的内容研究:它是一个列表,用你访问者自己的话写出了他们期望你写过但你没有写的东西。每周读一遍。第二有价值的日志是有人在十秒内再次搜索的查询,这是结果乍一看就不对的标志。
当你更换嵌入模型时,所有内容都必须重新嵌入——来自两个模型的向量是不可比的。把那个迁移计划为双写而不是一次切换。
Build a RAG Pipeline From Scratch Without a Framework
Build a Recommendation Feed With Embeddings