给博客加一个能搜正文的搜索：FastAPI 倒排索引 + Vue 3 搜索框
站内搜索只能匹配标题，正文完全搜不到。这篇是重写实录：一个两百行不到的 Python 内存倒排索引、中文 bigram 分词、FastAPI 接口，以及 Vue 3 那边必须处理的防抖与竞态。
博客的搜索框上线很久了，一直是个半成品：它只匹配标题。我以为够了，直到上周想找自己写的讲前端安全默认值那篇，我搜的是"CSP"，那是文章的核心。零条结果。标题里确实没有 CSP，正文里有，但我的搜索根本不看正文。
那天晚上我把它重写了。下面是从建索引到前端联调的过程，代码是线上跑着的版本，删掉业务细节，照着抄能跑。
先划边界
写代码前先定死约束：48 篇文章、正文合计三十万字、数据在 JSON 里没有数据库、一台 2GB 内存的小机器、一个 uvicorn 进程。
基于这些，三个方案是这么排除的。
- SQL 的
LIKE '%词%'要求把 JSON 迁进 SQLite，为一个搜索功能迁移整站存储不划算，不加索引还要全表扫描。 - SQLite 的 FTS5 功能最全，但它绑定 SQLite，等于要多维护一份影子库并和 JSON 同步，这类同步几乎一定会出错。
- 进程内自己写倒排索引，数据量就这么点，一个字典装得下，且不用改动存储层任何一行。
我选了第三个。代价是排序得自己写，而我本来也只要求"能搜到"，不要求搜得准。
中文分词用 bigram
倒排索引的核心是词，中文没有空格。我试过 jieba，切分质量好，但首次加载要一秒、词典常驻几十 MB，而且它不认识这站里最高频的词：FastAPI、uvicorn、Nginx、CSP。
最后用了最土的办法，英文按词切，中文切成相邻两字的组合。
# backend/app/search/tokenizer.py
import re
_TOKEN_RE = re.compile(r"[a-z0-9]+|[\u4e00-\u9fa5]+")
def tokenize(text: str) -> list[str]:
tokens: list[str] = []
for chunk in _TOKEN_RE.findall((text or "").lower()):
if chunk[0].isascii() or len(chunk) == 1:
tokens.append(chunk)
else:
tokens.extend(chunk[i : i + 2] for i in range(len(chunk) - 1))
return tokens正则先把连续的英数字和连续的汉字抠出来，标点和 Markdown 符号全丢掉。中文块切成 chunk[i:i+2] 的滑动窗口，"静态资源缓存"变成"静态、态资、资源、源缓、缓存"，所以搜"缓存"能命中，搜"资源"也能。
代价是误召回，"源缓"这种跨词边界的组合也被索引了，但没人会搜它。真正要防的是单字查询，一个汉字会命中所有含它的文章，所以接口里限定两字起搜。顺带一个好处：代码块里的内容也被索引，搜 pytest、lru_cache 这类只出现在代码里的词能命中。
索引和打分
倒排索引一句话能说完：一个字典，键是词，值是这个词出现在哪些文档里。要排序得再记两样东西，字段权重和文章长度。
# backend/app/search/index.py
import math
from .tokenizer import tokenize
TITLE_WEIGHT, SUMMARY_WEIGHT, CONTENT_WEIGHT = 5.0, 2.0, 1.0
class InvertedIndex:
def __init__(self) -> None:
self.lengths: dict[str, int] = {}
self.postings: dict[str, dict[str, float]] = {}
def add(self, post_id: str, content: str, title: str = "", summary: str = "") -> None:
content_tokens = tokenize(content)
self.lengths[post_id] = len(content_tokens) or 1
weights: dict[str, float] = {}
for token in content_tokens:
weights[token] = weights.get(token, 0.0) + CONTENT_WEIGHT
for token in tokenize(summary):
weights[token] = weights.get(token, 0.0) + SUMMARY_WEIGHT
for token in tokenize(title):
weights[token] = weights.get(token, 0.0) + TITLE_WEIGHT
for token, weight in weights.items():
self.postings.setdefault(token, {})[post_id] = weight
def search(self, query: str, limit: int = 200) -> list[tuple[str, float]]:
scores: dict[str, float] = {}
for token in tokenize(query):
bucket = self.postings.get(token)
if not bucket:
continue
idf = math.log(1 + len(self.lengths) / (1 + len(bucket)))
for post_id, weight in bucket.items():
norm = 1.0 + math.log(1 + weight)
scores[post_id] = scores.get(post_id, 0.0) + idf * norm / math.sqrt(self.lengths[post_id])
ranked = sorted(scores.items(), key=lambda item: -item[1])
return [(pid, round(score, 4)) for pid, score in ranked[:limit]]add 把正文、摘要、标题的 token 投进同一个字典，但给不同权重：标题一个词贡献 5 分，摘要 2 分，正文 1 分。这样"缓存"出现在标题里只算一次，也比在正文里出现十次更能说明主题。
打分公式不是标准 BM25，是我自己拼的，三项各有理由。idf 用 log(1 + N / df)，词越稀有分越高，搜"CSP"只有三篇包含，分自然高，搜"的"几乎每篇都包含，分接近零。norm 取 1 + log(1 + weight) 是对权重取对数，因为万字长文里某个词可能出现二十次，线性累加会让长文永远排第一。最后除以 sqrt(length) 再压一次长度差异。这几个系数是拿站内十来个查询试出来的，要相关性准就上现成的 BM25 库。
挂到 FastAPI 上
索引是内存里的副本，需要一个服务层持有它，并在文章变更后失效。
# backend/app/services/search_service.py
import re
from threading import Lock
from ..search.index import InvertedIndex
WIDTH = 70
class SearchService:
def __init__(self, posts) -> None:
self._posts = posts
self._lock = Lock()
self._index = None
def invalidate(self) -> None:
with self._lock:
self._index = None
def _build(self) -> InvertedIndex:
index = InvertedIndex()
for post in self._posts.list_published():
index.add(post["id"], post.get("content", ""), post.get("title", ""))
return index
def search(self, keyword: str) -> list[dict]:
with self._lock:
if self._index is None:
self._index = self._build()
hits = self._index.search(keyword)
return [self._payload(hit, keyword) for hit in hits]
def _payload(self, hit, keyword: str) -> dict:
post = self._posts.get(hit[0])
return {"id": post["id"], "title": post["title"], "slug": post["slug"],
"snippet": snippet(post.get("content", ""), keyword)}
def snippet(content: str, keyword: str) -> str:
flat = re.sub(r"\s+", " ", content or "").strip()
at = flat.lower().find(keyword.lower())
if at == -1:
return flat[: WIDTH * 2]
start, end = max(0, at - WIDTH), min(len(flat), at + len(keyword) + WIDTH)
return f"{'…' if start else ''}{flat[start:end]}{'…' if end < len(flat) else ''}"构建放在锁外完成、进锁只换引用，这样 180 毫秒的构建过程不会让搜索请求排队，读到的也永远是完整的索引。
snippet 是给前端高亮用的，把正文压成单行、找到关键词第一次出现的位置、往两边各取 70 字。find 之前两边都转小写，否则搜 fastapi 命中不了正文里的 FastAPI。
路由层保持项目一贯风格，业务接口全部 POST，响应统一 envelope。
# backend/app/api/routes/search.py
class SearchPayload(BaseModel):
keyword: str = Field(default="", max_length=64)
page: int = 1
page_size: int = 10
@router.post("/search")
def search(payload: SearchPayload, service=Depends(get_search_service)):
keyword = payload.keyword.strip()
if len(keyword) < 2:
return envelope({"keyword": keyword, "total": 0, "items": []})
items = service.search(keyword)
return envelope({"keyword": keyword, "total": len(items), "items": items[: payload.page_size]})max_length=64 的校验和 len(keyword) < 2 的判断都是防滥用用的，搜索接口公开，我不希望有人拿 1MB 的字符串来让我分词。
前端搜索框：防抖和竞态
这个项目刻意不用 vue-router 和 Pinia，搜索状态得自己在 composable 里管。下面这段里有两处删掉任何一个都会出线上 bug。
// frontend/src/composables/usePostSearch.js
import { onScopeDispose, ref, watch } from "vue";
import { searchPosts } from "../services/blogService";
export function usePostSearch(delay = 260) {
const keyword = ref("");
const results = ref([]);
const loading = ref(false);
let timer = null;
let seq = 0;
async function run(value) {
const current = ++seq;
loading.value = true;
const data = await searchPosts({ keyword: value, page: 1, page_size: 10 });
if (current !== seq) return;
results.value = data.items;
loading.value = false;
}
watch(keyword, (value) => {
clearTimeout(timer);
const trimmed = value.trim();
if (trimmed.length < 2) {
results.value = [];
return;
}
timer = setTimeout(() => run(trimmed), delay);
});
onScopeDispose(() => clearTimeout(timer));
return { keyword, results, loading };
}第一处是 seq，用来丢弃过期响应。我输入"缓存"发出请求 A，删一个字再输回来发出请求 B，如果 A 比 B 晚返回，界面会先显示 B 的结果再被 A 覆盖，用户看到结果闪一下变回旧的。自增序号比对能解决，代价是无效请求仍然发出去了，只是响应被忽略。用 AbortController 能连请求一起取消，但我的 blogService 把错误处理和 envelope 解包统一包在 request 里，传信号要改公共层，对一个搜索框不值。
第二处是 onScopeDispose，组件卸载时清掉定时器。否则用户在 260 毫秒内离开页面，定时器照样触发，往一个已卸载的组件里写 ref，而且会留下一批野请求。这类问题开发时几乎发现不了，因为没人会一边输入一边秒退。
delay 设 260 毫秒是试出来的：100 毫秒在中文输入法下会发出一堆请求，500 毫秒以上又觉得卡。用中文输入法时还要管 compositionstart/compositionend，否则拼音字母也会触发搜索，我这里偷懒只做了长度判断。
上线之后
- 索引内存 12MB，比我预想的大，bigram 让 token 数膨胀到汉字数的两倍左右。涨到 500 篇时我会考虑只索引正文前 3000 字。
- 冷启动构建 180 毫秒，我把它从请求路径挪到了应用启动事件里。
- 搜索延迟 P95 约 6 毫秒，绝大部分时间花在分词而不是查表上。
- 召回：搜
CSP命中 3 篇，达到目的。搜"缓存"命中 5 篇，两篇只提了一句排在最后。搜"性能"命中 7 篇，噪音偏多，这词太泛了。
还缺三样：拼音、同义词（缓存和 cache 是两个独立词）、英文词干还原（搜 caching 命中不了 cache）。都该加，只是没到痛点。
边界也得说清：这套东西适合"数据在内存里、量不大、能搜到就行"的场景。文章上万篇，或者要精准排序和拼写纠错，就别学我，Meilisearch 和 Typesense 部署起来也就一个 Docker 命令。我不用它们的原因是这站已经有一个进程了，不想为搜索再养一个。