第8章|校园生活证据问答助手¶
核心问题: 如果模型并不知道本校的菜单和制度,它为什么仍可能给出一段流畅的答案?
本 Notebook 用 6 道固定题拆开观察检索增强生成(RAG)的完整链路:资料分块 → 文本表示 → 相似度检索 → 阈值判断 → 证据拼接 → 基于证据回答 → 引用核查。
你最终要完成一个离线可运行的最小 RAG,并亲自解释:
- 检索器是否找到了正确原文;
- 生成器是否真的依据原文回答;
- 资料中没有答案时,系统为什么应当拒答;
- 只改变
TOP_K后,结果为何发生变化。
这不是“点一下看效果”的演示。所有核心代码都能查看,实验结论必须由你根据输出和原文作出。
学习目标与成果¶
完成本项目后,你应该能够:
- 用自己的话说明 RAG 的“检索、拼接、生成”三步;
- 读懂字符 n-gram TF-IDF、余弦相似度、Top-K 和相似度阈值在代码中的作用;
- 区分“检索命中”和“回答有依据”两级验收;
- 比较无检索猜测与有检索回答,但不把一次演示冒充正式模型评测;
- 只修改一个主变量,保留可复现的实验记录;
- 说明本案例为什么优先考虑 RAG,而不是把动态制度写入模型参数。
交付物包括:运行后的 Notebook、TOP_K=1 与 TOP_K=3 对照、至少 5 条引用核查、知识库数据卡和 AI 协同记录。
0. 运行前先预测¶
先打开 chapters/ch08-rag/student/experiment_log.md,在运行代码之前写下你的判断:
- Q6“学校游泳馆几点开门?”在资料库没有答案时,普通聊天模型可能怎样回答?
TOP_K=1改成TOP_K=3后,检索到的资料更多,回答一定会更好吗?- 如果检索器找错了原文,但回答碰巧正确,系统能否通过工程验收?为什么?
使用 Cline 时,可以让它先解释将要读取和修改的文件,再等待你批准。不要让 AI 代写实验结论。
1. 找到本章资源并加载完整实现¶
Notebook 可能从 teaching_resources/ 或 notebooks/ 启动。下面的代码向上寻找资源根目录,再加载兼容入口 scripts_optional/ch08/minimal_rag.py。
该脚本是本章完整、可测试的参考实现,没有使用 LangChain 等大型编排框架。
from pathlib import Path
import importlib.util
import inspect
import json
import re
import sys
start = Path.cwd().resolve()
RESOURCE_ROOT = next(
path for path in [start, *start.parents]
if (path / "scripts_optional/ch08/minimal_rag.py").exists()
)
SCRIPT = RESOURCE_ROOT / "scripts_optional/ch08/minimal_rag.py"
CHAPTER_DIR = RESOURCE_ROOT / "chapters/ch08-rag"
spec = importlib.util.spec_from_file_location("ch08_minimal_rag", SCRIPT)
rag = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = rag
spec.loader.exec_module(rag)
print("资源根目录:", RESOURCE_ROOT)
print("完整算法文件:", SCRIPT.relative_to(RESOURCE_ROOT))
print("离线检索器:", rag.CharNgramTfidfProvider.name)
2. 先看资料:知识库质量决定回答上限¶
本章使用三套课程虚构资料:
- 食堂菜单;
- 图书馆制度;
- 社团章程。
每个 Markdown 二级标题对应一个语义完整的知识片段,标题开头是稳定段落编号。例如 LIBRARY-02 不只是显示标签,它还必须能够返回 library.md 中的真实原文。
三份资料不代表任何真实学校,也不包含个人信息。
chunks = rag.load_markdown_chunks(CHAPTER_DIR / "data")
print(f"共读取 {len(chunks)} 个知识片段:")
for chunk in chunks:
print(f"{chunk.id:12}|{chunk.title:8}|来源文件:{chunk.source}")
检查一段原文¶
先不要提问。直接查看 LIBRARY-02,确认“最多借几本书”这个事实在资料中怎样表述。后面引用核查必须回到这里,而不是只相信模型给出的编号。
chunk_by_id = {chunk.id: chunk for chunk in chunks}
sample = chunk_by_id["LIBRARY-02"]
print("段落编号:", sample.id)
print("段落标题:", sample.title)
print("来源文件:", sample.source)
print("原文:", sample.text)
3. 文本怎样变成可以比较的数字¶
离线基线不调用网络模型,而是使用字符 n-gram TF-IDF:
- 把中文文本拆成连续的 1~3 字符片段;
- 计算片段在当前文档中的突出程度(TF);
- 降低所有文档都常见的片段权重(IDF);
- 把结果组成向量并单位化;
- 查询向量与资料向量做点积,得到余弦相似度。
这个方法便于阅读和复现,但主要依赖文字重叠,不等同于语义 embedding。它是断网时用于观察 RAG 机制的确定性基线。
print(inspect.getsource(rag.char_ngrams))
print(inspect.getsource(rag.CharNgramTfidfProvider))
provider = rag.CharNgramTfidfProvider()
document_texts = [f"{chunk.title}。{chunk.text}" for chunk in chunks]
provider.fit(document_texts)
document_vectors = provider.encode(document_texts)
print("知识库向量形状:", document_vectors.shape)
print("词表大小:", len(provider.vocabulary))
print("第1段向量长度:", round(float((document_vectors[0] @ document_vectors[0]) ** 0.5), 3))
4. 检索:排序结果不等于可靠证据¶
检索器总能排出“最相近”的 Top-K 段,即使所有资料都不能回答问题。因此本项目把过程分成两步:
- 候选排序: 取相似度最高的 Top-K;
- 证据接受: 只有分数达到固定阈值的候选才能进入生成提示词。
相似度分数不是“答案正确概率”。它只能表示:在当前知识库、当前文本表示方法下,这两段文字有多接近。
查看完整检索与回答代码¶
重点阅读 retrieve()、accepted_evidence()、build_prompt() 和 answer()。你会看到检索与生成是两个明确分开的环节。
print(inspect.getsource(rag.MinimalRAG.retrieve))
print(inspect.getsource(rag.MinimalRAG.accepted_evidence))
print(inspect.getsource(rag.MinimalRAG.build_prompt))
print(inspect.getsource(rag.MinimalRAG.mock_generate))
print(inspect.getsource(rag.MinimalRAG.answer))
运行一次检索¶
先用 Q2 观察:正确段落应当是 LIBRARY-02。默认取 2 个候选,但不代表两个候选都会成为证据。
assistant = rag.MinimalRAG(chunks, min_score=rag.MIN_SCORE)
question = "借阅者一次最多借几本书?"
hits = assistant.retrieve(question, top_k=2)
for hit in hits:
decision = "通过阈值" if hit.score >= assistant.min_score else "未采用"
print(f"{hit.id}|score={hit.score:.3f}|{decision}|{hit.title}")
5. 拼接与生成:只让模型看到已接受证据¶
build_prompt() 把通过阈值的资料和问题组合起来,并明确要求:
- 只能依据给定资料;
- 没有证据时固定回答“资料中未提到该信息。”;
- 每个事实附段落编号。
离线 mock 生成器只摘录证据中的句子并附编号。它不是大语言模型,也不追求自然对话;它用于建立全班断网时仍可复现、可核查的最低基线。
result = assistant.answer(question, top_k=2)
print("发送给生成器的提示词:")
print(result["prompt"])
print("\n离线回答:")
print(result["answer"])
可选:接入经教师校验的 Qwen 模型¶
参考实现预留了两个小接口:
embedding = rag.CallableEmbeddingProvider(my_embed_function, "approved-qwen-embedding")
assistant = rag.MinimalRAG(chunks, provider=embedding)
result = assistant.answer(
question,
generation_provider=my_qwen_chat_function,
)
my_embed_function(list[str])为每段文字返回一个等长向量;my_qwen_chat_function(prompt)接收已经拼好的证据提示词并返回文本;- API Key 只能从环境变量或学校配置读取,不能写入 Notebook;
- 不得向云端上传真实个人资料或未经授权的校务文档;
- 接入真实模型后,必须重新运行 6 道固定题和两级验收。
网页端 Qwen、DeepSeek 可以用于讨论提示词和核验思路,但手工复制的回答也要记录平台显示的模型名称、日期和完整提示词。
6. 六道固定题:形成无检索与有检索对照¶
Q1~Q5 在资料内有明确答案,Q6 是库外题。
“无检索回答”使用明确标注的课程模拟猜测,只用于说明没有证据仍强行作答的风险,不是对 Qwen、DeepSeek 或其他模型的正式评测结果。“有检索回答”则来自上面的实际离线程序。
questions = rag.load_questions(CHAPTER_DIR / "data/questions.json")
baseline_rows = rag.run_fixed_questions(top_k=2)
for item, row in zip(questions, baseline_rows):
print(f"\n{item['id']}|{item['question']}")
print("无检索模拟:", rag.naive_no_retrieval_answer(item["id"]))
print("有检索回答:", row["answer"])
print("通过阈值证据:", row["accepted_ids"] or "无(应拒答)")
for row in baseline_rows:
retrieval_label = "通过" if row["retrieval_pass"] else "未通过"
grounding_label = "通过" if row["grounding_pass"] else "未通过"
print(
f"{row['id']}|检索层={retrieval_label}|"
f"生成层={grounding_label}|证据={row['accepted_ids'] or '无,执行拒答'}"
)
assert all(row["retrieval_pass"] for row in baseline_rows)
assert all(row["grounding_pass"] for row in baseline_rows)
单独观察库外题¶
Q6 与图书馆“开馆时间”有部分文字重叠,因此检索器仍会把图书馆段落排在前面。但分数没有达到阈值,它不能进入证据区,系统必须拒答。
这个例子说明:Top-K 排名靠前不等于资料真正能回答问题。
outside = assistant.answer("学校游泳馆几点开门?", top_k=2)
for hit in outside["hits"]:
decision = "通过阈值" if hit["id"] in outside["accepted_ids"] else "未采用"
print(f"{hit['id']}|score={hit['score']:.3f}|{decision}")
print("最终回答:", outside["answer"])
assert outside["accepted_ids"] == []
assert outside["answer"] == rag.REFUSAL
8. 学生主实验:只改变 TOP_K¶
这是本章唯一必改变量。
- 先令
STUDENT_TOP_K = 1,运行并记录; - 在实验日志写下改成 3 后的预测;
- 再令
STUDENT_TOP_K = 3,重新运行; - 保持分块、表示法、阈值和六道题完全不变。
如果同时修改多个变量,就无法判断结果究竟由什么引起。
# TODO(学生必改):先用 1 完成记录,再改成 3。
STUDENT_TOP_K = 1
student_rows = rag.run_fixed_questions(top_k=STUDENT_TOP_K)
for row in student_rows:
candidate_ids = [hit["id"] for hit in row["hits"]]
print(
f"{row['id']}|候选={candidate_ids}|"
f"通过阈值={row['accepted_ids'] or '无'}|回答={row['answer']}"
)
用同一函数并排查看 1 与 3¶
程序只整理客观输出,不会替你完成“为什么”的解释。你要特别检查:Top-K 增大后,是否出现更多同主题但不能回答当前问题的候选。
def compact_results(top_k):
rows = rag.run_fixed_questions(top_k=top_k)
return {
row["id"]: {
"candidates": [hit["id"] for hit in row["hits"]],
"accepted": row["accepted_ids"],
"retrieval_pass": row["retrieval_pass"],
"grounding_pass": row["grounding_pass"],
}
for row in rows
}
comparison = {
"TOP_K=1": compact_results(1),
"TOP_K=3": compact_results(3),
}
print(json.dumps(comparison, ensure_ascii=False, indent=2))
9. 引用核查:编号必须回到真实 Markdown 原文¶
至少抽查 5 条回答。在实验记录中把每条引用标为:
- A:原文完全支持;
- B:原文部分支持,但回答有简化、合并或改写;
- C:原文没有该事实,或者编号不存在。
下面的代码只帮助定位原文,等级和处理意见必须由你本人判断。
for row in student_rows[:5]:
print(f"\n{row['id']}|回答:{row['answer']}")
for source_id in row["accepted_ids"]:
source = chunk_by_id[source_id]
print(f" {source_id}({source.source}):{source.text}")
10. 新增资料之前先写数据卡¶
在教师允许下,可以复制一份知识库到个人工作目录,新增一个有唯一编号的课程自编段落和一道验证题。不要直接覆盖教师核准基准。
数据卡至少记录:
- 资料名称与来源;
- 是课程自编、公开许可还是得到授权;
- 版本日期和更新责任人;
- 是否包含姓名、学号、联系方式等个人信息;
- 可以回答什么、不能回答什么;
- 一道验证题及期望段落编号。
新增资料后要重新建立索引并运行全部固定题,确认新资料没有让旧题退化。
11. 为什么本项目优先 RAG,而不是微调¶
食堂菜单、借阅制度和社团章程都可能更新,回答还要求回到原文核验。本项目因此优先更新外部知识库和检索索引:
- RAG: 模型参数不变,回答前检索可更新资料;
- 微调: 调整模型参数,更适合相对稳定的任务方式、格式、术语或表达风格;
- 组合使用: 复杂系统可以用微调建立行为能力,再用 RAG 提供动态事实。
这不是“RAG 永远优于微调”的结论,而是根据更新频率、可追溯要求、数据成熟度和预算做出的工程选择。
12. 完成检查与结论¶
提交前确认:
- 我在运行前写了预测;
- Notebook 已从第一格运行到最后一格;
TOP_K=1与TOP_K=3各运行一次,其他变量不变;- 6 道题分别完成检索层和生成层验收;
- Q6 没有证据时明确拒答;
- 至少 5 条引用回到 Markdown 原文核查;
- 我保存了“原始意图—AI建议—人工选择—实际改动—验证”记录;
- 新增资料有数据卡,不含未经许可的个人信息;
- 我能用不超过 200 字说明:RAG 改善了什么,仍不能保证什么。
工程底线: RAG 可以降低无依据生成风险,但不能自动修正错误资料,也不能保证检索、生成和引用永不出错。可靠系统必须保留拒答、日志、测试和人工核验。