私有知识库实战：查询改写、混合检索与重排优化 RAG

✍ Ando🕐 2026-07-08📦 21.2 KB 🟢 已读 𝕏 文章列表

本文是私有知识库构建系列的第 3 篇，旨在解决基础 RAG 系统在短问题召回、关键词匹配及结果排序上的缺陷。作者详细介绍了通过 LLM 进行查询改写以补全上下文，利用 PostgreSQL 结合 Jieba 实现混合检索（向量+BM25），以及使用 RRF 算法融合检索结果，从而显著提升知识库的日常可用性。

RAG 混合检索查询改写向量数据库 PostgreSQL BM25 搜索优化知识库 LLM Python

# 私有知识库第3篇：查询改写 + 混合检索 + 重排，让 RAG 真的扛得住日常 **作者**: Ando **日期**: 2026-07-06T03:55:58.000Z **来源**: [https://x.com/ando_w/status/2074321836117037349](https://x.com/ando_w/status/2074321836117037349) --- ![Image](/doc/2819/image/img_9070b059f574.jpg) 第 2 篇跑通了最小工程，5 个文件，一个聊天框，能问能答。但你真往里灌 200 份文档自己问两天，会发现它有几个病，藏不住。这一篇治这些病。 > **Ando@ando_w**: [原文链接](https://x.com/ando_w/status/2073979240962228582) > # 一、第 2 篇留下的 3 个病你自己跑一遍就能撞上。病 1：短问题直接废。用户问「报多少」。系统不知道他问的是报销报多少、报价报多少、还是报表报多少。「报多少」这三个字的 embedding 几乎是个平均向量，谁都像又谁都不像。返回的 5 个 chunk 里，可能 3 个讲报销、1 个讲报表、1 个讲报价单，模型只能瞎猜一个答。病 2：关键词召不回来。向量检索擅长语义相似，但不擅长精确匹配。文档里写的是「ISO/IEC 27001:2022」，用户问「27001 怎么做」，向量不一定能把那条召回来。编号、人名、产品代号、专有名词，全是向量检索的盲区。你可能会说，27001 和 ISO 语义上不相关吗？相关。但在 1536 维或 1024 维的空间里，一个编号的距离，可能比一段无关的「信息安全概述」还远。这就是向量检索的现实。病 3：top-k 里全是噪声。向量返回 top-5，里面可能有 3 个是「语义相关但其实没用」的。比如问「报销流程」，召回了「公司差旅制度」「财务审批规范」「去年团建费用公示」——前两个有用，第三个没用但语义相关。全塞进 prompt，模型会被第三个带偏，答着答着开始讲团建。第 2 篇你只能调 top-k、调 chunk 大小、调 prompt，治不了这三个病。要治得加三个东西： - 病 1 → 查询改写 - 病 2 → 混合检索（向量 + 关键词） - 病 3 → 重排下面一个一个做。代码接着第 2 篇的项目结构，新增 3 个文件，改 2 个。 # 二、查询改写：把短问题说完整 ## 思路用户问「报多少」，系统不知道上下文。但如果前面聊过「我上周出差，报销怎么走」，那「报多少」明显是问报销报多少。查询改写就是：在检索之前，先让 LLM 把用户的短问题、代词、口语，结合历史对话补全成一个完整的检索查询。这一步看着多余，但它是 RAG 从「能 demo」到「能扛日常」的分水岭。没有它，多轮对话基本没法用。 ## 单查询改写最简单的版本，把原问题 + 最近几轮历史丢给 LLM，让它输出一个补全后的查询。新建 app/query_rewrite.py： ``` from openai import OpenAI from app.config import get_settings settings = get_settings() client = OpenAI( api_key=settings.llm_api_key, base_url=settings.llm_base_url, ) REWRITE_PROMPT = """你在帮一个私有知识库改写用户的检索查询。用户的问题可能太短、有代词、或者口语化，直接拿去向量检索效果很差。请结合历史对话，把它改写成一个完整的、能独立检索的查询。要求： - 补全代词和省略的内容 - 保留用户的核心意图 - 不要回答问题，只改写成查询 - 输出 3 个不同的改写，每行一个，不要编号历史对话： {history} 用户当前问题：{question} 3 个改写查询：""" def rewrite_query(question: str, history: list[dict] = None) -> list[str]: """把一个问题改写成多个检索变体""" history_text = "" if history: history_text = "\n".join( f"{'用户' if m['role'] == 'user' else '助手'}：{m['content']}" for m in history[-4:] # 最近 2 轮 ) resp = client.chat.completions.create( model=settings.llm_model, # qwen3.7-max messages=[{ "role": "user", "content": REWRITE_PROMPT.format( history=history_text or "（无）", question=question, ), }], temperature=0.3, max_tokens=200, ) text = resp.choices[0].message.content.strip() queries = [q.strip() for q in text.split("\n") if q.strip()] # 保留原始问题，放在第一位 queries.insert(0, question) return queries[:4] # 原始 + 3 个改写 ``` 返回的是一个列表，第一个是原问题，后面 3 个是改写变体。 ## 为什么要多个变体一个改写只覆盖一个理解方向。用户问「27001」，改写可能补成「ISO 27001 信息安全管理体系怎么做」，但用户其实可能想问「27001 认证要花多少钱」。一个改写猜不全。生成 3 个变体，每个并行检索，结果合并去重，召回率会高不少。这个思路叫 RAG-Fusion，本质是用 LLM 的发散性去对冲单次检索的窄。代价是多调 3 次检索，但向量检索很便宜，可以接受。LLM 改写这一步也只要 200 token，几分钱。 # 三、混合检索：向量 + 关键词 ## 为什么不能只用向量向量检索的盲区在第 1 节讲过了——编号、人名、产品代号、专有名词。这些 token 在向量空间里位置不稳定。关键词检索（BM25）正好相反，它强在精确匹配，弱在语义。用户问「报销流程」，BM25 能把所有含「报销」的文档按词频召回来，但「差旅审批」这种语义相同、用词不同的文档它召不回。两个各有盲区，合起来盲区最小。这就是混合检索。 ## BM25 是什么 BM25 是 TF-IDF 的升级版，关键词检索的事实标准。公式不用记，你只要知道：一个词在当前文档出现越多、在整个语料出现越少，这个词对这篇文档就越重要。实现上有两条路： - PostgreSQL 自带全文检索（tsvector + ts_rank），近似 BM25，不引新依赖 - rank_bm25 这个 Python 库，标准 BM25，但每次全量计算，文档多了会慢生产环境我建议用 PostgreSQL 的方案，索引走数据库，不用在 Python 里扛。但 PostgreSQL 默认的全文检索对中文不友好——它按空格和标点分词，中文会整段当成一个词。 ## 中文分词的处理解法是入库前先用 jieba 分好词，存到一个字段里，检索时也用 jieba 分词后再去匹配。这样 PostgreSQL 的 simple 分词配置就能用了。先改一下数据模型。打开 app/models.py： ``` from sqlalchemy import Column, Integer, String, Text, DateTime, func from sqlalchemy.dialects.postgresql import Vector from pgvector.sqlalchemy import Vector from app.db import Base class Document(Base): __tablename__ = "documents" id = Column(Integer, primary_key=True) content = Column(Text, nullable=False) source = Column(String(500)) permission_group = Column(String(50), default="public") embedding = Column(Vector(1024)) # Qwen3-Embedding 维度 # 新增：jieba 分词后的文本，用于全文检索 tokens = Column(Text) # 新增：全文检索字段，基于 tokens 自动生成 tsv = Column( Text, Computed("to_tsvector('simple', tokens)", persisted=True), ) created_at = Column(DateTime, server_default=func.now()) ``` Computed(..., persisted=True) 让 PostgreSQL 自动维护这个字段，每次 tokens 变了 tsv 跟着变，不用你操心。注意 Computed 要从 sqlalchemy 导入： ``` from sqlalchemy import Computed ``` 然后改入库逻辑，加一步分词。打开 app/ingest.py（第 2 篇里写文档入库那个文件），在写入前加： ``` import jieba def tokenize(text: str) -> str: """jieba 分词，返回空格分隔的词，用于全文检索""" words = jieba.cut_for_search(text) # 过滤单字和纯标点，留有意义的词 return " ".join(w.strip() for w in words if len(w.strip()) > 1) ``` 写入文档时把 tokens=tokenize(chunk_text) 一起塞进去。记得装 jieba： ``` pip install jieba ``` 还要给 tsv 建索引，不然全文检索会全表扫： ``` CREATE INDEX idx_documents_tsv ON documents USING gin(tsv); ``` ## 混合检索的代码新建 app/hybrid_search.py： ``` from sqlalchemy import text as sql_text from app.db import SessionLocal from app.embedding import get_embedding from app.config import get_settings import jieba settings = get_settings() def tokenize(text: str) -> str: words = jieba.cut_for_search(text) return " ".join(w.strip() for w in words if len(w.strip()) > 1) def vector_search(query: str, top_k: int = 10, perm: str = "public") -> list[dict]: """向量召回""" emb = get_embedding(query) db = SessionLocal() try: rows = db.execute( sql_text(""" SELECT id, content, source, embedding <=> :emb AS distance FROM documents WHERE permission_group = :perm ORDER BY embedding <=> :emb LIMIT :k """), {"emb": str(emb), "perm": perm, "k": top_k}, ).fetchall() # 把距离转成一个 0-1 的分数，越小越相关 return [ {"id": r[0], "content": r[1], "source": r[2], "score": 1.0 - float(r[3])} for r in rows ] finally: db.close() def keyword_search(query: str, top_k: int = 10, perm: str = "public") -> list[dict]: """关键词召回（PostgreSQL 全文检索）""" tokens = tokenize(query) if not tokens: return [] db = SessionLocal() try: rows = db.execute( sql_text(""" SELECT id, content, source, ts_rank(tsv, plainto_tsquery('simple', :tokens)) AS rank FROM documents WHERE permission_group = :perm AND tsv @@ plainto_tsquery('simple', :tokens) ORDER BY rank DESC LIMIT :k """), {"tokens": tokens, "perm": perm, "k": top_k}, ).fetchall() return [ {"id": r[0], "content": r[1], "source": r[2], "score": float(r[3])} for r in rows ] finally: db.close() def rrf_fuse(vector_results: list[dict], keyword_results: list[dict], k: int = 60, top_k: int = 10) -> list[dict]: """RRF 融合两路结果 RRF 公式：score(d) = sum( 1 / (k + rank_i) ) k 是平滑常数，一般取 60。 """ scores = {} content_map = {} for rank, r in enumerate(vector_results): doc_id = r["id"] scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1) content_map[doc_id] = r for rank, r in enumerate(keyword_results): doc_id = r["id"] scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1) if doc_id not in content_map: content_map[doc_id] = r sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True) result = [] for doc_id in sorted_ids[:top_k]: item = dict(content_map[doc_id]) item["rrf_score"] = scores[doc_id] result.append(item) return result def hybrid_search(query: str, top_k: int = 10, perm: str = "public") -> list[dict]: v = vector_search(query, top_k=top_k, perm=perm) kw = keyword_search(query, top_k=top_k, perm=perm) return rrf_fuse(v, kw, top_k=top_k) ``` ## RRF 是什么两路检索返回的分数不在一个量纲上——向量是余弦距离，关键词是 ts_rank，没法直接比。RRF（Reciprocal Rank Fusion）绕开这个问题，不看绝对分数，只看排名。公式就一行：每个文档的融合分数 = 它在每路结果里排名的倒数之和。排第 1 的贡献 1/61，排第 2 的贡献 1/62，依此类推。k=60 是个平滑常数，让排名靠后的文档不至于完全没权重。 RRF 的好处是不用调权重。你不用纠结向量占 70% 还是关键词占 30%，它自己平衡。这是工程上最省心的混合检索方案。 # 四、重排：从 10 个里挑出真正能用的 3 个 ## 为什么还要重排混合检索返回 10 个候选，但里面还是有噪声。向量召回靠语义距离，关键词召回靠词频，两者都不知道这个 chunk 到底能不能回答用户的问题。重排就是把这 10 个候选再过一遍，用一个更准的模型逐个打分，挑出真正能用的 3 个。 ## 两种方案方案 A：Cross-encoder 重排模型。用 BGE-reranker、Cohere Rerank 这种专门的 cross-encoder。它把「问题 + 候选」一起喂进去，输出一个相关性分数。准，但要么本地跑模型吃显存，要么调外部 API。方案 B：直接用 LLM 打分。让千问给每个候选打 0-10 分。简单，不用引新依赖，准头够用。这一篇先用方案 B，跑通再说。新建 app/rerank.py： ``` from openai import OpenAI from app.config import get_settings settings = get_settings() client = OpenAI(api_key=settings.llm_api_key, base_url=settings.llm_base_url) RERANK_PROMPT = """你在给检索结果打分。用户问题：{question} 候选文档内容（截取前 500 字）： {content} 打分标准： - 10：完全能回答问题 - 7-9：能回答主要部分 - 4-6：部分相关，需要补充 - 1-3：几乎无关 - 0：完全无关只输出一个数字，不要解释。""" def rerank(question: str, candidates: list[dict], top_k: int = 3) -> list[dict]: scored = [] for c in candidates: resp = client.chat.completions.create( model=settings.llm_model, messages=[{ "role": "user", "content": RERANK_PROMPT.format( question=question, content=c["content"][:500], # 截断省 token ), }], temperature=0.0, max_tokens=5, ) try: score = float(resp.choices[0].message.content.strip()) except (ValueError, TypeError): score = 0.0 c["rerank_score"] = score scored.append(c) scored.sort(key=lambda x: x["rerank_score"], reverse=True) return scored[:top_k] ``` temperature=0.0 是为了让打分稳定。max_tokens=5 是因为它只要吐一个数字，给多了它可能开始解释，浪费 token。 10 个候选逐个打分，要调 10 次 LLM。这一步是整个 pipeline 里最慢的。后面要优化的话，可以换成 cross-encoder 批量打分，或者只对 RRF 分数前 5 的打分。先用对的，再用快的。 # 五、串成完整 pipeline 把三步串起来。改 app/retrieval.py（第 2 篇里那个简单的 retrieve 函数）： ``` from app.query_rewrite import rewrite_query from app.hybrid_search import hybrid_search from app.rerank import rerank def retrieve(question: str, history: list[dict] = None, top_k: int = 3, perm: str = "public") -> list[dict]: # 1. 查询改写：原问题 + 3 个变体 queries = rewrite_query(question, history) # 2. 每个查询都做一次混合检索，结果合并去重 all_candidates = {} for q in queries: results = hybrid_search(q, top_k=10, perm=perm) for r in results: # 同一个文档被多个查询召回，RRF 分数累加 if r["id"] in all_candidates: all_candidates[r["id"]]["rrf_score"] += r["rrf_score"] else: all_candidates[r["id"]] = r candidates = list(all_candidates.values()) # 按累计 RRF 分数排一下，取前 10 进重排 candidates.sort(key=lambda x: x.get("rrf_score", 0), reverse=True) candidates = candidates[:10] # 3. 重排，挑出 top_k top = rerank(question, candidates, top_k=top_k) return top ``` 第 2 篇的 app/main.py 里那个 /ask 接口不用改，它调的还是 retrieve，只不过 retrieve 内部变强了。这就是分层的好处——上层接口不动，底层换实现。完整流程现在是： ``` 用户问题 ↓ 查询改写（原问题 + 3 个变体） ↓ 每个查询做向量+关键词混合检索（RRF 融合） ↓ 合并去重，按累计 RRF 分数取前 10 ↓ LLM 重排，挑出 top 3 ↓ 塞进 prompt 生成答案 ``` 第 2 篇是「问一句 → 向量召回 5 个 → 生成」。第 3 篇是「问一句 → 改写 4 个 → 每个混合检索 → 融合 → 重排 → 生成」。环节多了 3 倍，但召回质量也上了个台阶。 # 六、踩过的 5 个坑 ## 坑 1：查询改写把用户原意改没了 LLM 改写的时候容易自作主张，把「报销报多少」改成「公司报销标准和流程说明」。查询是完整了，但用户可能只想知道一个数字。解法是 prompt 里写死「保留用户的核心意图」，而且原始问题永远放第一个参与检索。改写是补充，不是替代。 ## 坑 2：中文分词没做，全文检索全废 PostgreSQL 的 simple 配置按空格分词。中文文档没有空格，整段会被当成一个词，tsv @@ plainto_tsquery 永远匹配不上。必须入库前 jieba 分词。如果你不想改入库逻辑，也可以装 zhparser 这个 PostgreSQL 扩展，让数据库自己分词。但装扩展要权限，云数据库不一定让你装，jieba 预处理更通用。 ## 坑 3：重排太慢，用户等不了 10 个候选逐个调 LLM 打分，每个 300ms，加起来 3 秒。加上前面改写和检索，一个请求要 5-6 秒。 demo 能忍，生产不能忍。短期解法：只对 RRF 前 5 的打分，砍掉一半时间。长期解法：换 BGE-reranker 本地跑，批量打分，10 个一起算只要 200ms。 ## 坑 4：多查询改写放大了 token 消耗 4 个查询 × 每个 top 10 = 40 次向量召回。好在向量召回便宜，但 embedding 要调 4 次（每个改写都要算 embedding）。如果用 OpenAI 的 embedding，4 次还好；用本地模型要注意并发。 ## 坑 5：没有评估集，改了不知道好不好这是最大的坑。你加了查询改写、混合检索、重排，感觉上「好像准了」，但说不出来准了多少。没有评估集，所有优化都是拍脑袋。下一篇会专门讲怎么建评估集，这一篇先记住一件事：改任何参数之前，先攒 50 个真实问题和标准答案。不然你改完没法验证，等于白改。 # 七、怎么知道改好了最朴素的方法：拿 50 个真实问题，跑改之前的第 2 篇版本，记录每个问题召回了什么、答得怎么样。再跑改之后的第 3 篇版本，对比。关注两个数： - 召回率（Recall@k）：标准答案所在的 chunk，有没有出现在 top-k 里。改之前可能是 60%，改之后应该能到 80% 以上。 - 答案准确率：最终生成的答案对不对。这个要人看，或者用 LLM 当裁判打个分。如果你公司有标注人力，让人标 50 个问题的标准答案和「相关 chunk」。如果没有，自己花一个下午标，值得。这 50 个问题会陪着你调很久。一个偷懒的办法：把用户真实问过的问题记下来，攒一周就有 50 个了。比你自己拍脑袋编的问题真实得多。 # 八、下一步到这一篇，你的 RAG 已经能扛日常使用了。但还有几块没做： - 评估集和自动评测（这一篇留的坑） - 增量更新（文档改了怎么只更新变化的部分） - 权限模型（多对多，不是第 2 篇那种简单分组） - Agent 编排（多轮检索，工具调用）第 4 篇会做评估集 + 增量更新。评估集是让你从「感觉变好了」变成「确实变好了」，增量更新是让知识库能长期跑下去不用全量重建。如果你想先看到哪一块，评论区说一下，我调整顺序。参考资料： - pgvector 官方文档：https://github.com/pgvector/pgvector - PostgreSQL 全文检索文档：https://www.postgresql.org/docs/current/textsearch.html - RRF 论文：https://plg.uwaterloo.ca/~gvcormac/cormacksigir09-rrf.pdf - 阿里云百炼 Qwen3.7-Max 文档：https://help.aliyun.com/zh/model-studio/ - jieba 中文分词：https://github.com/fxsjy/jieba 上一篇：私有知识库第 2 篇：用 FastAPI + pgvector 跑通最小可运行工程 ## 相关链接 - [Ando](https://x.com/ando_w) - [@ando_w](https://x.com/ando_w) - [869](https://x.com/ando_w/status/2074321836117037349/analytics) - [Jul 6](https://x.com/ando_w/status/2073979240962228582) - [490](https://x.com/ando_w/status/2073979240962228582/analytics) - [https://github.com/pgvector/pgvector](https://github.com/pgvector/pgvector) - [https://www.postgresql.org/docs/current/textsearch.html](https://www.postgresql.org/docs/current/textsearch.html) - [https://plg.uwaterloo.ca/~gvcormac/cormacksigir09-rrf.pdf](https://plg.uwaterloo.ca/~gvcormac/cormacksigir09-rrf.pdf) - [https://help.aliyun.com/zh/model-studio/](https://help.aliyun.com/zh/model-studio/) - [https://github.com/fxsjy/jieba](https://github.com/fxsjy/jieba) - [私有知识库第 2 篇：用 FastAPI + pgvector 跑通最小可运行工程](https://x.com/ando_w/status/2073070705789329561) - [Upgrade to Premium](https://x.com/i/premium_sign_up) - [10:37 AM · Jul 7, 2026](https://x.com/ando_w/status/2074321836117037349) - [869 Views](https://x.com/ando_w/status/2074321836117037349/analytics) - [View quotes](https://x.com/ando_w/status/2074321836117037349/quotes) --- *导出时间: 2026/7/8 15:53:50*

🔗 相关文档