# 私有知识库第4篇：评估集 + 增量更新，让 RAG 能长期跑下去
**作者**: Ando
**日期**: 2026-07-07T02:37:19.000Z
**来源**: [https://x.com/ando_w/status/2074699344129789985](https://x.com/ando_w/status/2074699344129789985)
---
第 3 篇把检索做扎实了，召回率上来了，重排也加了。你调了一晚上参数，感觉好多了。
但"感觉好多了"这五个字，是 RAG 项目里最贵的五个字。
这一篇填第 3 篇留的两个坑：你怎么知道真的变好了，文档改了怎么办。
> **Ando@ando_w**: [原文链接](https://x.com/ando_w/status/2074321836117037349)
>
# 一、第 3 篇留下的两个坑
坑 1：没有评估集，调参靠手感。
你把 top_k 从 5 改到 10，又把 rerank 的权重从 0.6 调到 0.4，问了几句话感觉答得还行，就上线了。下周有人反馈答非所问，你也不知道是哪次改的锅。
没有评估集，所有调参都是在赌。赌赢了不知道为什么赢，输了不知道输在哪。
坑 2：文档一改就要全量重建。
运营改了一份产品手册，你把 200 份文档全部重新切分、重新算 embedding、重新灌库。跑一次 40 分钟，烧掉几块钱 token。更要命的是，这 40 分钟里知识库是停的。
这两个坑不填，你的 RAG 就是个"能 demo，不能长期跑"的玩具。
# 二、评估集是什么
一句话：50 个问题，每个问题带标准答案和应该召回的 chunk。
它不是考试，是体检表。你每次改参数、换模型、调检索策略，都拿这 50 个问题跑一遍，看分数涨了还是跌了。
听起来很重，其实标一个下午就够。
评估集长这样（一个 JSON 文件）：
```
[
{
"question": "ISO/IEC 27001:2022 的附录A有多少个控制项？",
"gold_answer": "93个控制项，分为4个主题",
"gold_chunk_ids": ["doc003_chunk012", "doc003_chunk013"]
},
{
"question": "报销流程需要哪些审批节点？",
"gold_answer": "直属主管、部门负责人、财务复核",
"gold_chunk_ids": ["doc007_chunk004"]
}
]
```
三个字段：
- question：用户真实会问的问题
- gold_answer：标准答案（简短就行，不用长篇）
- gold_chunk_ids：这个问题应该召回哪些 chunk（这是关键）
gold_chunk_ids 是评估检索的尺子。如果系统召回了这些 chunk，检索就是对的；如果召不回来，检索就有问题。
# 三、怎么标（偷懒版）
别自己编问题。
自己编的问题有两个毛病：一是太"干净"，不像真实用户问的；二是你会下意识挑系统能答的，评估集就失真了。
正确做法：把用户真实问过的问题记下来，攒一周就有 50 个。没有用户？找同事让他们问，问完请杯奶茶。
标 gold_chunk_ids 的偷懒办法：
1. 让你现在的 RAG 系统答一遍这 50 个问题
2. 看它召回的 chunk，人工判断哪些是对的
3. 对的记下来当 gold，错的标出来
4. 没有对的全靠你翻文档找
这一步最累，但最值。这 50 个问题会陪着你调很久，是你整个项目的地基。
gold_answer 可以后标。 前期只标 question + goldchunkids 就能评估检索质量。答案准确率那一步用到 gold_answer 时再补。
# 四、三个核心指标
评估集跑一遍，看三个数。
## 1. Recall@k（召回率）
该回来的 chunk，回来了吗？
```
Recall@k = |系统召回的 ∩ 应该召回的| / |应该召回的|
```
举例：goldchunkids 有 2 个，系统 top_k=5 召回了 5 个，其中 1 个在 gold 里。Recall@5 = 1/2 = 0.5。
这个指标最重要。Recall 低，说明该回来的没回来，后面重排再牛也救不回来——东西根本没进候选池。
## 2. Precision@k（精确率）
回来的 chunk，有多少是垃圾？
```
Precision@k = |系统召回的 ∩ 应该召回的| / |系统召回的k个|
```
接上例：top_k=5，回来了 1 个对的，4 个没用。Precision@5 = 1/5 = 0.2。
Precision 低，说明候选池里噪声多，会干扰模型生成答案。
Recall 和 Precision 的取舍： 一般优先保 Recall。候选池大一点没关系，反正有重排兜底。但如果 Recall 高、Precision 很低，说明检索在召回一堆垃圾，得查检索策略。
## 3. 答案准确率（LLM 当裁判）
前两个指标只看检索。但用户在乎的是答案对不对。
人工判断答案对不对太慢。让 LLM 当裁判：
```
你是评估员。判断下面答案是否正确回答了问题。
问题：{question}
标准答案：{gold_answer}
待评估答案：{system_answer}
只输出 "正确" 或 "错误"，再附一句理由。
```
跑 50 个问题，统计"正确"的比例，就是答案准确率。
为什么用 LLM 裁判而不是字符串匹配？ 因为答案表述可以不一样。"93个控制项" 和 "共93项" 都对，字符串匹配会判错，LLM 能判断语义。
# 五、自动化评测脚本
把上面三个指标串成一个脚本。跑一次，输出报告。
```
# eval_rag.py
import json
import os
import psycopg2
from openai import OpenAI
# 复用第3篇的配置
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
EMBED_MODEL = "text-embedding-v3"
LLM_MODEL = "qwen-max"
client = OpenAI(
api_key=DASHSCOPE_API_KEY,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 数据库连接（复用第3篇）
conn = psycopg2.connect(
host="localhost", port=5432,
dbname="ragdb", user="rag", password="rag"
)
def get_embedding(text):
resp = client.embeddings.create(
model=EMBED_MODEL, input=text, dimensions=1024
)
return resp.data[0].embedding
def retrieve(question, top_k=5):
"""简化版检索：只用向量检索。第3篇的混合检索可替换进来。"""
emb = get_embedding(question)
with conn.cursor() as cur:
cur.execute("""
SELECT id, content, embedding <=> %s::vector AS dist
FROM chunks
ORDER BY embedding <=> %s::vector
LIMIT %s
""", (str(emb), str(emb), top_k))
return [{"id": r[0], "content": r[1]} for r in cur.fetchall()]
def generate_answer(question, chunks):
context = "\n\n".join(c["content"] for c in chunks)
resp = client.chat.completions.create(
model=LLM_MODEL,
messages=[
{"role": "system", "content": f"根据以下资料回答问题。\n\n资料：\n{context}"},
{"role": "user", "content": question}
]
)
return resp.choices[0].message.content
def llm_judge(question, gold_answer, sys_answer):
"""LLM当裁判，判断答案对不对"""
resp = client.chat.completions.create(
model=LLM_MODEL,
messages=[
{"role": "system", "content": "你是评估员。判断待评估答案是否正确回答了问题。只输出一行：先输出正确或错误，空格后附一句理由。"},
{"role": "user", "content": f"问题：{question}\n标准答案：{gold_answer}\n待评估答案：{sys_answer}"}
]
)
line = resp.choices[0].message.content.strip()
return line.startswith("正确"), line
def evaluate(eval_file="evalset.json", top_k=5):
with open(eval_file, "r", encoding="utf-8") as f:
evalset = json.load(f)
total = len(evalset)
sum_recall = 0
sum_precision = 0
correct = 0
for item in evalset:
q = item["question"]
gold_ids = set(item["gold_chunk_ids"])
gold_answer = item.get("gold_answer", "")
# 检索
chunks = retrieve(q, top_k=top_k)
retrieved_ids = set(c["id"] for c in chunks)
# Recall / Precision
hit = retrieved_ids & gold_ids
recall = len(hit) / len(gold_ids) if gold_ids else 1
precision = len(hit) / top_k if top_k else 0
sum_recall += recall
sum_precision += precision
# 答案准确率
if gold_answer:
answer = generate_answer(q, chunks)
is_correct, reason = llm_judge(q, gold_answer, answer)
if is_correct:
correct += 1
print(f"=== 评估报告（top_k={top_k}）===")
print(f"问题数：{total}")
print(f"Recall@{top_k}：{sum_recall/total:.3f}")
print(f"Precision@{top_k}：{sum_precision/total:.3f}")
print(f"答案准确率：{correct}/{total} = {correct/total:.3f}")
if __name__ == "__main__":
evaluate(top_k=5)
```
跑一遍：
```
=== 评估报告（top_k=5）===
问题数：50
Recall@5：0.720
Precision@5：0.380
答案准确率：38/50 = 0.760
```
这三个数就是你的基线。之后每次改参数，都和这个基线比。
怎么用：
```
# 改 top_k 前先跑一次
python eval_rag.py # Recall@5: 0.720
# 把 top_k 改成 10
python eval_rag.py # Recall@10: 0.810, Precision@10: 0.250
# Recall 涨了，Precision 跌了——正常，池子大了垃圾也多了
# 看答案准确率：0.78，比之前 0.76 高一点点，可以接受
```
没有这个脚本，你根本不知道 top_k 该设几。
# 六、增量更新：文档改了别全量重建
评估集搞定了，看第二个坑：文档更新。
全量重建为什么不行：
- 200 份文档重新切分 + embedding，40 分钟
- token 钱：200 份 × 平均 30 chunk × 1024 token，烧掉几块到十几块
- 这 40 分钟里，知识库要么停服，要么返回旧数据
- 一周改 3 次文档，一周就要重建 3 次
增量更新的思路： 记住每份文档的"指纹"，只重算指纹变了的。
## 三种操作
关键：怎么判断"修改了"？ 用文档内容的 hash。内容没变就不动，变了就重算。
## 表结构改造
第 2 篇的 documents 表加两个字段：
```
ALTER TABLE documents
ADD COLUMN content_hash VARCHAR(64),
ADD COLUMN updated_at TIMESTAMP DEFAULT NOW();
```
content_hash 存整份文档内容的 SHA256。下次同步时，算新文档的 hash，和库里比，一样就跳过。
# 七、代码实现增量更新
```
# incremental_update.py
import hashlib
import psycopg2
from openai import OpenAI
import os
import glob
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
EMBED_MODEL = "text-embedding-v3"
client = OpenAI(
api_key=DASHSCOPE_API_KEY,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
conn = psycopg2.connect(
host="localhost", port=5432,
dbname="ragdb", user="rag", password="rag"
)
def get_embeddings(texts):
"""批量embedding，省钱省时间"""
resp = client.embeddings.create(
model=EMBED_MODEL, input=texts, dimensions=1024
)
return [d.embedding for d in resp.data]
def split_text(text, chunk_size=500, overlap=50):
"""简易切分，第2篇同款"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunks.append(text[start:end])
start = end - overlap
return [c for c in chunks if c.strip()]
def content_hash(text):
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def upsert_document(doc_id, title, text):
"""增量更新一份文档：新增/修改/跳过"""
new_hash = content_hash(text)
with conn.cursor() as cur:
# 查老状态
cur.execute(
"SELECT content_hash FROM documents WHERE id = %s",
(doc_id,)
)
row = cur.fetchone()
if row and row[0] == new_hash:
print(f"[跳过] {title} 内容没变")
return
if row:
# 修改：先删旧chunk
cur.execute("DELETE FROM chunks WHERE doc_id = %s", (doc_id,))
print(f"[更新] {title} 重新切分")
else:
print(f"[新增] {title}")
# 写/更新 documents 表
cur.execute("""
INSERT INTO documents (id, title, content_hash, updated_at)
VALUES (%s, %s, %s, NOW())
ON CONFLICT (id) DO UPDATE
SET title = EXCLUDED.title,
content_hash = EXCLUDED.content_hash,
updated_at = NOW()
""", (doc_id, title, new_hash))
# 切分 + 批量 embedding
pieces = split_text(text)
embeddings = get_embeddings(pieces)
# 插入 chunks
for i, (piece, emb) in enumerate(zip(pieces, embeddings)):
chunk_id = f"{doc_id}_chunk{i:03d}"
cur.execute("""
INSERT INTO chunks (id, doc_id, content, embedding)
VALUES (%s, %s, %s, %s::vector)
""", (chunk_id, doc_id, piece, str(emb)))
conn.commit()
print(f" 完成：{len(pieces)} 个 chunk")
def delete_document(doc_id):
"""文档下线：删干净"""
with conn.cursor() as cur:
cur.execute("DELETE FROM chunks WHERE doc_id = %s", (doc_id,))
cur.execute("DELETE FROM documents WHERE id = %s", (doc_id,))
conn.commit()
print(f"[删除] {doc_id}")
def sync_directory(dir_path):
"""同步整个目录：扫描文件，新增/更新/删除"""
# 当前库里的文档
with conn.cursor() as cur:
cur.execute("SELECT id FROM documents")
db_ids = {r[0] for r in cur.fetchall()}
# 磁盘上的文档
disk_files = {}
for f in glob.glob(os.path.join(dir_path, "*.md")):
doc_id = os.path.splitext(os.path.basename(f))[0]
disk_files[doc_id] = f
disk_ids = set(disk_files.keys())
# 新增 + 修改
for doc_id, f in disk_files.items():
with open(f, "r", encoding="utf-8") as fh:
text = fh.read()
upsert_document(doc_id, doc_id, text)
# 删除（库里有、磁盘上没有的）
for doc_id in (db_ids - disk_ids):
delete_document(doc_id)
print("\n同步完成")
if __name__ == "__main__":
sync_directory("./docs")
```
跑一次：
```
[跳过] 产品手册 内容没变
[更新] 报销制度 重新切分
完成：8 个 chunk
[新增] 入职指南
完成：12 个 chunk
[删除] 过期公告_2024
同步完成
```
200 份文档里只有 3 份变了，只重算这 3 份。从 40 分钟降到 1 分钟，token 钱省了 98%。
这个脚本挂个定时任务（crontab / Windows 计划任务），每天凌晨跑一次，知识库就永远是新的。
# 八、把两件事串起来
评估集和增量更新不是两个独立功能，是配套的。
流程：
1. 凌晨 2 点，定时任务触发 sync_directory
2. 增量更新跑完，知识库是新的
3. 自动跑一遍 eval_rag.py
4. 如果 Recall 或答案准确率比基线跌了 5 个百分点以上，发个告警（邮件/钉钉/飞书）
```
# watch.py —— 更新后自动评测 + 告警
import subprocess
BASELINE = {"recall": 0.72, "precision": 0.38, "accuracy": 0.76}
THRESHOLD = 0.05 # 跌5个点就报警
def run_eval():
"""跑评估脚本，解析结果"""
out = subprocess.check_output(["python", "eval_rag.py"]).decode()
# 简化解析，实际用正则提取 Recall/Precision/accuracy
return {"recall": 0.69, "precision": 0.37, "accuracy": 0.74}
def check():
result = run_eval()
drops = []
for k, v in result.items():
if BASELINE[k] - v > THRESHOLD:
drops.append(f"{k}: {BASELINE[k]:.2f} 降到 {v:.2f}")
if drops:
msg = "RAG 指标下降告警：\n" + "\n".join(drops)
print(msg)
# 发告警（邮件/钉钉/飞书 webhook）
# send_alert(msg)
else:
print("指标正常")
if __name__ == "__main__":
check()
```
这样你的 RAG 就有了"自我体检"能力。文档更新引入了问题，第二天就能发现，不用等用户投诉。
# 九、这一篇你得到了什么
- 评估集：50 个问题 + gold chunk，三个指标（Recall、Precision、答案准确率），自动化评测脚本
- 增量更新：content_hash 追踪，只重算变化的文档，200 份从 40 分钟降到 1 分钟
- 监控闭环：更新后自动评测，指标下跌自动告警
到这一篇，你的 RAG 不只是"能跑"，而是"能长期跑、还能知道自己在变好还是变坏"。
# 十、下一步
还有两块没做：
- 权限模型：第 2 篇做了简单分组，但企业里是"这个人能看这 3 个库，那个人能看那 5 个"的多对多
- Agent 编排：多轮检索、工具调用、自己决定要不要再查一次
第 5 篇做哪个，评论区说一声。
参考资料：
- pgvector 官方文档：https://github.com/pgvector/pgvector
- RAG 评估方法论（Azure AI）：https://learn.microsoft.com/zh-cn/azure/ai-studio/concepts/evaluation-metrics-built-in
- LLM-as-a-Judge 论文：https://arxiv.org/abs/2306.05685
- 阿里云百炼 text-embedding-v3 文档：https://help.aliyun.com/zh/model-studio/
- Precision 与 Recall 解释：https://en.wikipedia.org/wiki/Precisionandrecall
上一篇：私有知识库第3篇：查询改写 + 混合检索 + 重排，让 RAG 真的扛得住日常
## 相关链接
- [Ando](https://x.com/ando_w)
- [@ando_w](https://x.com/ando_w)
- [254](https://x.com/ando_w/status/2074699344129789985/analytics)
- [Jul 7](https://x.com/ando_w/status/2074321836117037349)
- [869](https://x.com/ando_w/status/2074321836117037349/analytics)
- [https://github.com/pgvector/pgvector](https://github.com/pgvector/pgvector)
- [https://arxiv.org/abs/2306.05685](https://arxiv.org/abs/2306.05685)
- [https://help.aliyun.com/zh/model-studio/](https://help.aliyun.com/zh/model-studio/)
- [https://en.wikipedia.org/wiki/Precisionandrecall](https://en.wikipedia.org/wiki/Precisionandrecall)
- [私有知识库第3篇：查询改写 + 混合检索 + 重排，让 RAG 真的扛得住日常](https://x.com/ando_w/status/2074321836117037349)
- [Upgrade to Premium](https://x.com/i/premium_sign_up)
- [11:37 AM · Jul 8, 2026](https://x.com/ando_w/status/2074699344129789985)
- [254 Views](https://x.com/ando_w/status/2074699344129789985/analytics)
---
*导出时间: 2026/7/8 15:53:40*