💡 RAG 的锅，很多时候不是模型背得不够熟，而是你把一整本 PDF 切成了“知识碎纸片”，还要求它自己拼回目录。
本文最终效果
我们不使用云端数据库，直接在本地完成：
Markdown 文档
↓ 标题感知切块
BM25 关键词召回 + 向量语义召回
↓ 加权合并 + 轻量重排
Top-K 片段
↓
Ollama 生成带 [S1] 引用的答案
文章里的脚本可以直接运行。向量模型第一次运行会下载到本地，之后离线可用。
为什么“能搜到”仍然答不对
一个 RAG 答案要同时满足三件事：
- 召回：正确片段进入候选集。
- 排序：最有用的片段排在前面。
- 约束生成：模型只能根据片段回答，并给出来源。
只做向量检索会漏掉精确的错误码、参数名和版本号；只做关键词检索又理解不了同义表达。因此本文使用混合检索：BM25 保留精确匹配，向量相似度补充语义匹配。
准备环境
winget install --id=astral-sh.uv -e
mkdir rag-quality-lab
cd rag-quality-lab
uv init --python 3.12
uv add sentence-transformers rank-bm25 numpy
ollama pull qwen2.5:7b
mkdir docs
新建 docs/cache.md：
# Redis 缓存
## 缓存击穿
热点 Key 失效时，大量请求会同时访问数据库。可以使用互斥锁、逻辑过期或提前刷新，避免请求瞬间打满数据库。
新建 docs/deploy.md：
# 部署检查
## 灰度发布
灰度期间先把少量流量切到新版本，观察错误率、P95 延迟和业务指标，再逐步扩大流量。
先把文档切对
新建 rag.py：
from __future__ import annotations
import json
import os
import re
import urllib.request
from dataclasses import asdict, dataclass
from pathlib import Path
import numpy as np
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
MODEL_NAME = os.getenv("EMBEDDING_MODEL", "BAAI/bge-small-zh-v1.5")
DOCS_DIR = Path("docs")
@dataclass
class Chunk:
chunk_id: str
source: str
heading: str
text: str
def tokens(text: str) -> list[str]:
"""同时保留中文单字、英文单词、数字和下划线。"""
return re.findall(r"[\u4e00-\u9fff]|[a-zA-Z0-9_]+", text.casefold())
def split_markdown(path: Path) -> list[Chunk]:
chunks: list[Chunk] = []
heading = path.stem
buffer: list[str] = []
def flush() -> None:
text = "\n".join(buffer).strip()
if not text:
return
# 让每块足够短，模型才能准确引用；真实项目可按 token 数改成滑动窗口。
for offset in range(0, len(text), 800):
piece = text[offset : offset + 800].strip()
if piece:
chunks.append(
Chunk(f"{path.name}#{len(chunks) + 1}", path.name, heading, piece)
)
for line in path.read_text(encoding="utf-8").splitlines():
if line.startswith("#"):
flush()
buffer.clear()
heading = line.lstrip("# ").strip() or path.stem
else:
buffer.append(line)
flush()
return chunks
def load_chunks() -> list[Chunk]:
chunks: list[Chunk] = []
for path in sorted(DOCS_DIR.rglob("*.md")):
chunks.extend(split_markdown(path))
if not chunks:
raise RuntimeError("docs 目录没有 Markdown 文件")
return chunks
def normalise(values: np.ndarray) -> np.ndarray:
low, high = values.min(), values.max()
if high == low:
return np.zeros_like(values)
return (values - low) / (high - low)
def search(query: str, limit: int = 5) -> list[dict]:
chunks = load_chunks()
model = SentenceTransformer(MODEL_NAME)
documents = [chunk.text for chunk in chunks]
bm25 = BM25Okapi([tokens(text) for text in documents])
query_tokens = tokens(query)
keyword_scores = normalise(np.asarray(bm25.get_scores(query_tokens), dtype=float))
vectors = model.encode(documents, normalize_embeddings=True, show_progress_bar=False)
query_vector = model.encode([query], normalize_embeddings=True, show_progress_bar=False)[0]
vector_scores = normalise(np.asarray(vectors @ query_vector, dtype=float))
query_set = set(query_tokens)
candidates: list[dict] = []
for index, chunk in enumerate(chunks):
overlap = len(query_set & set(tokens(chunk.text))) / max(len(query_set), 1)
# 语义分数解决同义表达，BM25 保留错误码/参数名等精确匹配。
score = 0.60 * vector_scores[index] + 0.30 * keyword_scores[index] + 0.10 * overlap
candidates.append(
{
**asdict(chunk),
"score": round(float(score), 4),
"vector_score": round(float(vector_scores[index]), 4),
"keyword_score": round(float(keyword_scores[index]), 4),
}
)
return sorted(candidates, key=lambda item: item["score"], reverse=True)[:limit]
def answer(query: str, hits: list[dict]) -> str:
context = "\n\n".join(
f"[{item['chunk_id']}] 来源={item['source']} 标题={item['heading']}\n{item['text']}"
for item in hits
)
prompt = f"""
你是一个严谨的技术文档助手。只根据下面的资料回答问题，不要补充资料中没有的事实。
每个结论后必须附来源标记，例如 [deploy.md#1]。如果资料不足，直接说“资料不足”。
问题：{query}
资料：
{context}
"""
body = json.dumps(
{
"model": os.getenv("OLLAMA_MODEL", "qwen2.5:7b"),
"messages": [{"role": "user", "content": prompt}],
"stream": False,
}
).encode("utf-8")
request = urllib.request.Request(
"http://127.0.0.1:11434/api/chat",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=120) as response:
result = json.loads(response.read().decode("utf-8"))
return result["message"]["content"]
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("query")
parser.add_argument("--limit", type=int, default=5)
args = parser.parse_args()
hits = search(args.query, max(1, min(args.limit, 10)))
print("\n候选片段：")
for hit in hits:
print(f"{hit['chunk_id']} score={hit['score']} {hit['text'][:80]}")
print("\n答案：")
print(answer(args.query, hits))
运行一次检索：
uv run python rag.py "缓存击穿怎么处理？"
第一次会下载 BAAI/bge-small-zh-v1.5，之后模型保存在本地缓存。你应该先看到候选片段，再看到带 [cache.md#...] 引用的答案。若只想检查检索、不启动 Ollama，可以把 print(answer(...)) 临时改成 print(json.dumps(hits, ensure_ascii=False, indent=2))。
为什么要保留 BM25
假设问题是：
FIN-03010 设置为 0 怎么办？
向量模型可能理解“参数配置异常”，但不一定把包含精确字符串 FIN-03010 的片段排第一。BM25 对错误码、版本号、类名和配置项非常敏感；混合分数可以兼顾两类查询：
最终分数 = 0.60 × 语义相似度
+ 0.30 × BM25 关键词分数
+ 0.10 × 查询词重叠率
权重不是宇宙真理，应该用自己的评测集调整。重要的是把权重写出来，别让“感觉搜得不错”冒充评测。
加一个最小评测集
新建 eval.jsonl：
{"query":"缓存击穿怎么处理？","expected":"cache.md"}
{"query":"灰度发布先看哪些指标？","expected":"deploy.md"}
新建 evaluate.py：
import json
from rag import search
total = 0
hit_at_3 = 0
with open("eval.jsonl", encoding="utf-8") as file:
for line in file:
case = json.loads(line)
hits = search(case["query"], limit=3)
sources = [item["source"] for item in hits]
ok = case["expected"] in sources
hit_at_3 += int(ok)
total += 1
print("OK" if ok else "MISS", case["query"], sources)
print(f"Recall@3 = {hit_at_3 / total:.2%}")
运行：
uv run python evaluate.py
只有两个样本时，Recall@3 没有统计意义；实际项目至少准备几十个问题，并记录正确来源、版本、是否需要多个片段。每次修改切块或权重后重新运行，才能知道改动是否真的有效。
切块的三个实用规则
1. 标题必须跟着正文走
cache.md#1 比一段孤零零的文字更有用。标题、文件名、版本号和更新时间应该作为元数据一起进入上下文。
2. 不要只按固定字符数硬切
本文先按 Markdown 标题分段，再按 800 字符做上限，是一个可运行的起点。生产环境还要处理代码块、表格、列表和跨段落引用，最好改成 token 计数和小幅 overlap。
3. 召回片段不能无限多
Top-K 越大不一定越好。无关片段会稀释上下文，增加模型引用错误。先用评测集比较 K=3/5/8，再决定上下文长度。
引用不是装饰
让模型输出 [source#chunk] 只是第一步。应用层还应该：
- 保存命中的 chunk ID、分数和模型版本。
- 点击引用时能定位到原文标题或页码。
- 当最高分低于阈值时返回“资料不足”，不要强行回答。
- 对生成答案做引用存在性检查，发现不存在的标记就拒绝发布。
可以加一个简单阈值：
hits = search(query, limit=5)
if not hits or hits[0]["score"] < 0.35:
print("资料不足，请补充文档或换个问法。")
else:
print(answer(query, hits))
阈值同样要通过评测集确定，不要照抄 0.35。
从本地脚本升级到生产服务
文档入库任务 → 解析/切块 → embedding → 向量库
↘ 关键词索引
查询 → 混合召回 → 重排 → 阈值判断 → LLM + 引用
生产化时可把 vectors、BM25 索引和元数据持久化到 Qdrant、OpenSearch 或 PostgreSQL；把模型和索引版本写入元数据；对文档更新做增量索引。本文用内存索引是为了让读者先看懂评分和评测，而不是被数据库配置挡在门外。
常见坑
把扫描 PDF 当成文本 PDF
扫描件没有文本层，解析结果可能是空字符串。先 OCR，再进入切块流程，并保留页码。
只看生成答案，不看候选片段
答案错了，先判断是召回错、排序错还是生成错。候选片段和分数必须可观测。
用一个问题证明系统很好
演示问题应该命中，不能代表系统可靠。准备“相似问题、否定问题、版本问题、错误码问题、资料不存在问题”五类评测。
总结
RAG 的质量不是由“换一个更大的模型”单独决定的，而是由数据和流程共同决定：
结构化切块 → 混合召回 → 可解释排序 → 评测集 → 引用约束 → 持续迭代
先让检索结果可解释，再让模型负责表达；当答案不对时，你才能知道该修文档、修切块、修召回，还是修 Prompt。
参考资料
作者：牛马便利店一号店员
文档信息
- 本文作者：牛马
- 本文链接：https://geekhappy.com/2026/08/25/rag-retrieval-quality/
- 版权声明：自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）