一句话记住 RAG
RAG 的全称是 Retrieval-Augmented Generation，中文通常称为“检索增强生成”。
Load → Chunk → Embed → Retrieve → Augment → Generate在本次实验中，它对应：
security_安全事件上报速查.md
→ 读取为字符串
→ 切成 4 个 Chunk
→ bge-m3 转成 1024 维向量
→ 余弦相似度排序、Top-K、证据门槛
→ 用户问题 + 检索证据 + 来源组成 Prompt
→ qwen3:8b 依据证据生成
→ 带来源回答，或在证据不足时拒答两个模型各自负责什么
bge-m3：负责生成语义向量
bge-m3 是 Embedding 模型。它把文本转换成向量，让程序可以比较两段文字在语义上是否接近。
它不负责：
直接回答用户问题；
自动打开知识库文件；
把知识库永久记到模型参数中；
输出一段自然语言解释。
qwen3:8b：负责读证据并组织回答
qwen3:8b 是生成模型。它接收我们组装好的 Prompt，再把问题和已检索证据整理成自然语言回答。
它不负责：
在当前脚本里主动搜索文件；
计算 Chunk 相似度；
决定知识库中哪一段排名第一。
所以当前程序的分工是：
bge-m3 生成向量 → Python 计算相似度、排序和把关 → qwen3:8b 依据证据回答为什么先只用一个文件
正式盲盒目录里有 25 个文件和多种格式。如果一开始同时处理 PDF、Word、Excel、旧 Office 文件、版本冲突和脏数据，一旦结果不对，很难判断问题来自解析、清洗、分块、检索还是生成。
因此第一阶段只使用：
data/raw/00_所有待分类原始盲盒/security_安全事件上报速查.md这个文件只有 236 个字符，结构清楚，但已经足够跑通完整 RAG 链路。
第 1 次迭代：Load
对应脚本：learning/minimal_rag.py
from pathlib import Path
file_path = Path(
r"F:\AI\AgentMake\Projects\EnterpriseKnowledgeAgent"
r"\data\raw\00_所有待分类原始盲盒"
r"\security_安全事件上报速查.md"
)
text = file_path.read_text(encoding="utf-8")
print("数据类型：", type(text))
print("字符数量：", len(text))实际结果：
数据类型： <class 'str'>
字符数量： 236这里已经完成了 Load：磁盘上的文件变成了 Python 可以继续处理的字符串。
需要记住：
文件路径和编码属于 Load 层；
文件能打开，不代表内容已经适合检索；
正式多文件系统还要保留文件名、版本、页码、章节等元数据。
第 2 次迭代：Chunk
chunks = [
part.strip()
for part in text.split("\n\n")
if part.strip()
]
print("分块数量：", len(chunks))
for index, chunk in enumerate(chunks, start=1):
print(f"\n--- Chunk {index} ---")
print(chunk)实际得到 4 个 Chunk：
这里按空行分块恰好保留了四步流程的完整性，所以本次小实验可用。但这不代表 split("\n\n") 是生产级分块器。以后面对长段落、表格、跨页内容和多级标题时，还需要研究：
chunk_size；overlap；标题和正文的结构关系；
页码、章节、表格行等来源定位；
一个 Chunk 是否保留了完整语义。
第 3 次迭代：手工关键词检索
最早的检索逻辑是：给每个 Chunk 计算命中了几个关键词。
question = "公司的电脑不见了该如何处理？"
keywords = ["设备"]
scored_chunks = []
for index, chunk in enumerate(chunks, start=1):
score = sum(
1
for keyword in keywords
if keyword in chunk
)
scored_chunks.append((score, index, chunk))
scored_chunks.sort(reverse=True)
best_score, best_index, best_chunk = scored_chunks[0]它能找到 Chunk 3，但这次成功有一个关键问题：真正参与评分的是写死的 keywords，不是 question。
也就是说，即使把问题改成“今天午饭吃什么”，只要仍然写着：
keywords = ["设备"]它依然会命中同一个 Chunk。这不是可靠的问句检索，只是我们人工提前告诉程序答案里应该出现什么词。
第 4 次迭代：从问题自动生成二元词
为了让 question 真正参与检索，我们从问题和 Chunk 中提取连续的两个中文字符，再计算交集。
def chinese_bigrams(value):
chinese_text = "".join(
character
for character in value
if "\u4e00" <= character <= "\u9fff"
)
return {
chinese_text[index:index + 2]
for index in range(len(chinese_text) - 1)
}
question_tokens = chinese_bigrams(question)
automatic_results = []
for index, chunk in enumerate(chunks, start=1):
chunk_tokens = chinese_bigrams(chunk)
matched_tokens = question_tokens & chunk_tokens
score = len(matched_tokens)
automatic_results.append((score, index, chunk))它比手写关键词前进了一步，因为检索单位确实来自问题：
当问题使用“设备丢失”时，可以通过共同字面片段“设备”找到 Chunk 3；
当问题改写为“公司的电脑不见了该如何处理？”时，问题与原文的“设备”没有共同二元词，当前脚本会全部得 0 分并拒答。
所以二元词检索仍然只是字符重叠检索，不是语义理解。它不明白：
电脑不见了 ≈ 设备丢失这个失败实验正是引入 Embedding 的原因。
第 5 次迭代：用 bge-m3 验证语义相似度
对应脚本：learning/embedding_demo.py
我们先不做完整 RAG，只验证 Embedding 是否能识别同义表达。
1. 准备三段文本
texts = [
"设备丢失",
"电脑不见了",
"今天午饭吃什么",
]2. 调用 Ollama 的 Embedding 接口
import json
from urllib.request import Request, urlopen
ollama_url = "http://127.0.0.1:11434/api/embed"
request_body = {
"model": "bge-m3",
"input": texts,
}
request = Request(
url=ollama_url,
data=json.dumps(
request_body,
ensure_ascii=False,
).encode("utf-8"),
headers={
"Content-Type": "application/json",
},
method="POST",
)
with urlopen(request, timeout=60) as response:
response_body = json.load(response)
embeddings = response_body["embeddings"]结果中有 3 个向量，每个向量都是 1024 维：
文本数量： 3
向量数量： 3
每个向量的维度： 10243. 计算余弦相似度
本质上有点像高中数学的求余弦值
1. 基础向量形式（高中数学课本标准公式）
设向量为 \vec{a}和 \vec{b}，它们的夹角为 \theta 。
根据向量数量积公式：
移项变形得到余弦值 \cos \theta ：
2. 二维坐标系形式（高中平面向量）
设平面向量坐标为：
向量点积：
\vec{a} \cdot \vec{b} = x_1 x_2 + y_1 y_2
向量 \vec{a}的模长：
|\vec{a}| = \sqrt{x_1^2 + y_1^2}
向量 \vec{b} 的模长：
|\vec{b}| = \sqrt{x_2^2 + y_2^2}
代入余弦公式：
同理，三维空间坐标形式（高中空间向量）就是：
设空间向量坐标为 \vec{a} = (x_1, y_1, z_1)和\vec{b} = (x_2, y_2, z_2) ：
$$
\cos \theta = \frac{x_1 x_2 + y_1 y_2 + z_1 z_2}{\sqrt{x_1^2 + y_1^2 + z_1^2} \sqrt{x_2^2 + y_2^2 + z_2^2}}
$$
4. n 维向量通用形式（AI 算法/代码计算公式）
在 AI embedding 中，向量维度增加到 n 维。
设 n 维向量 \vec{a} = (a_1, a_2, \dots, a_n)$ 和 $\vec{b} = (b_1, b_2, \dots, b_n) ：
n 维点积： \vec{a} \cdot \vec{b} = \sum_{i=1}^{n} a_i b_i
n 维模长： |\vec{a}| = \sqrt{\sum_{i=1}^{n} a_i^2}， |\vec{b}| = \sqrt{\sum_{i=1}^{n} b_i^2}
余弦相似度计算公式：
$$
\text{Similarity} = \cos \theta = \frac{\sum_{i=1}^{n} a_i b_i}{\sqrt{\sum_{i=1}^{n} a_i^2} \sqrt{\sum_{i=1}^{n} b_i^2}}
$$
简易代码如下：
from math import sqrt
def cosine_similarity(left_vector, right_vector):
dot_product = sum(
left * right
for left, right in zip(
left_vector,
right_vector,
strict=True,
)
)
left_length = sqrt(
sum(value * value for value in left_vector)
)
right_length = sqrt(
sum(value * value for value in right_vector)
)
return dot_product / (left_length * right_length)通过Embedding模型计算出来的实际结果：
设备丢失 ↔ 电脑不见了 0.8017
设备丢失 ↔ 今天午饭吃什么 0.4334
电脑不见了 ↔ 今天午饭吃什么 0.4304结论：字面不同但含义相近的“设备丢失”和“电脑不见了”明显更接近。这就是语义检索可以解决同义改写的基础。
注意：
余弦相似度不是“答案正确率”，
0.8017也不是 80.17% 正确。分数只适合在当前模型、当前语料和当前处理方式下做相对比较。这个值主要是基于当前已有数据集做出的计算，比如当前
设备丢失 ↔ 电脑不见了 0.8017这对值计算出来的是最高的，如果加入一个“设备遗失，电脑丢失”，那么最高的一定会变成
设备丢失 ↔ 设备遗失 0.9817（假设）
设备丢失 ↔ 电脑丢失 0.8017（假设）所以同样的之前“最相似的0.8017”的Ranking排名也会下降
第 6 次迭代：Retrieve 和 Top-K
对应脚本：learning/vector_retrieval.py
1. 问题和 Chunk 使用同一个模型向量化
question = "公司的电脑不见了该如何处理？"
# 第一项是问题，后面四项是待检索的 Chunk。
texts = [question, *chunks]把这 5 段文本一起发送给 bge-m3 后：
embeddings = response_body["embeddings"]
question_embedding = embeddings[0]
chunk_embeddings = embeddings[1:]这里依赖一个简单对应关系：输入第 1 项是问题，所以返回第 1 个向量也是问题向量；后面的向量按顺序对应 4 个 Chunk。
2. 问题向量逐个比较 Chunk 向量
ranked_chunks = []
for index, (chunk, chunk_embedding) in enumerate(
zip(chunks, chunk_embeddings, strict=True),
start=1,
):
similarity = cosine_similarity(
question_embedding,
chunk_embedding,
)
ranked_chunks.append(
(similarity, index, chunk)
)
ranked_chunks.sort(
key=lambda item: item[0],
reverse=True,
)实际相似度：
完整处置流程所在的 Chunk 3 排到了第一名。
3. 观察 Top-K
top_k = 2
for rank, (similarity, index, chunk) in enumerate(
ranked_chunks[:top_k],
start=1,
):
print(
f"\n排名 {rank}｜Chunk {index}"
f"｜相似度 {similarity:.4f}"
)
print(chunk)Top-K 的意思是：排序后取前 K 个候选证据。本实验设置 top_k = 2，用于观察前两名。
当前代码虽然打印了 Top-2，但 Augment 阶段实际只取
ranked_chunks[0]，也就是 Top-1 的 Chunk 3。不要把当前实现误解为“Top-2 全部送给了 Qwen”。以后可以再学习怎样去重、控制 Token，并把多个合格 Chunk 一起放进 Prompt。
第 7 次迭代：证据门槛和拒答
Top-K 永远能从候选列表中排出第一名，但“第一名”不等于“真的相关”。即使问年假，安全事件文档里也总会有一个分数最高的 Chunk。
因此在调用 Qwen 之前加入临时门槛：
best_similarity, best_index, best_chunk = ranked_chunks[0]
min_similarity = 0.55
if best_similarity < min_similarity:
print(
"\n根据当前知识库证据不足，"
"无法可靠回答该问题。"
)
raise SystemExit(0)两次实验结果：
这里有两个重要结论：
拒答应尽量在检索层先把关，不能只依赖 LLM 看见无关证据后自觉说不知道。
0.55只是教学值。它只来自一个正例和一个负例，不能复制到其他模型、语料或项目中当生产阈值。以后必须用固定评测集校准。
第 8 次迭代：Augment
Augment 的意思不是训练模型，而是把刚检索到的外部证据加入 Prompt。
System Prompt
system_prompt = """
你是企业知识库问答助手。
只能依据用户提供的证据回答，不得补充证据之外的事实。
如果证据不足，请明确回答“根据当前证据无法回答”。
回答必须标注证据来源。
""".strip()User Prompt
user_prompt = f"""
用户问题：
{question}
检索证据：
[来源：Chunk {best_index}]
{best_chunk}
请根据检索证据回答用户问题。
""".strip()此时 Qwen 收到的不是整个文件夹，而是我们明确交给它的：
回答规则 + 用户问题 + Top-1 证据 + 来源标签这一步中的“检索增强”，就是 RAG 名字里的 Augmented。
第 9 次迭代：Generate
最后调用 Ollama 的聊天接口：
chat_url = "http://127.0.0.1:11434/api/chat"
chat_request_body = {
"model": "qwen3:8b",
"stream": False,
"messages": [
{
"role": "system",
"content": system_prompt,
},
{
"role": "user",
"content": user_prompt,
},
],
"options": {
"temperature": 0,
},
}实际调用信息证明这次确实使用了 qwen3:8b：
实际模型： qwen3:8b
是否完成： True
结束原因： stop最终回答覆盖了检索证据中的四个步骤：
断开可疑网络或停止操作，但不要关机、重装或删除日志；
保留邮件、聊天、截图、时间和设备编号等证据；
15 分钟内联系安全团队邮箱；
按安全团队指示隔离设备、修改凭证或通知相关人员。
回答中标注了 Chunk 3 来源，至此完成了最小闭环。
最终完整代码
下面是当前 vector_retrieval.py 的等价整理版。它完整串起：
Load → Chunk → Embed → Retrieve → 门槛 → Augment → Generateimport json
from math import sqrt
from pathlib import Path
from urllib.request import Request, urlopen
embed_url = "http://127.0.0.1:11434/api/embed"
chat_url = "http://127.0.0.1:11434/api/chat"
file_path = Path(
r"F:\AI\AgentMake\Projects\EnterpriseKnowledgeAgent"
r"\data\raw\00_所有待分类原始盲盒"
r"\security_安全事件上报速查.md"
)
def cosine_similarity(left_vector, right_vector):
dot_product = sum(
left * right
for left, right in zip(
left_vector,
right_vector,
strict=True,
)
)
left_length = sqrt(
sum(value * value for value in left_vector)
)
right_length = sqrt(
sum(value * value for value in right_vector)
)
return dot_product / (left_length * right_length)
# 1. Load
document_text = file_path.read_text(encoding="utf-8")
# 2. Chunk
chunks = [
part.strip()
for part in document_text.split("\n\n")
if part.strip()
]
question = "公司的电脑不见了该如何处理？"
# 3. Embed：第一个输入是问题，后面是待检索 Chunk。
texts = [question, *chunks]
embed_request_body = {
"model": "bge-m3",
"input": texts,
}
embed_request = Request(
url=embed_url,
data=json.dumps(
embed_request_body,
ensure_ascii=False,
).encode("utf-8"),
headers={
"Content-Type": "application/json",
},
method="POST",
)
with urlopen(embed_request, timeout=60) as response:
embed_response_body = json.load(response)
embeddings = embed_response_body["embeddings"]
question_embedding = embeddings[0]
chunk_embeddings = embeddings[1:]
# 4. Retrieve：计算相似度并排序。
ranked_chunks = []
for index, (chunk, chunk_embedding) in enumerate(
zip(chunks, chunk_embeddings, strict=True),
start=1,
):
similarity = cosine_similarity(
question_embedding,
chunk_embedding,
)
ranked_chunks.append((similarity, index, chunk))
ranked_chunks.sort(
key=lambda item: item[0],
reverse=True,
)
top_k = 2
print(f"问题：{question}")
print(f"Top-{top_k} 检索结果：")
for rank, (similarity, index, chunk) in enumerate(
ranked_chunks[:top_k],
start=1,
):
print(
f"\n排名 {rank}｜Chunk {index}"
f"｜相似度 {similarity:.4f}"
)
print(chunk)
# 5. 证据门槛：不相关问题不进入生成阶段。
best_similarity, best_index, best_chunk = ranked_chunks[0]
min_similarity = 0.55
print("\n最高相似度：", f"{best_similarity:.4f}")
print("证据门槛：", f"{min_similarity:.4f}")
if best_similarity < min_similarity:
print(
"\n根据当前知识库证据不足，"
"无法可靠回答该问题。"
)
raise SystemExit(0)
# 6. Augment：把规则、问题、证据和来源组成 Prompt。
system_prompt = """
你是企业知识库问答助手。
只能依据用户提供的证据回答，不得补充证据之外的事实。
如果证据不足，请明确回答“根据当前证据无法回答”。
回答必须标注证据来源。
""".strip()
user_prompt = f"""
用户问题：
{question}
检索证据：
[来源：Chunk {best_index}]
{best_chunk}
请根据检索证据回答用户问题。
""".strip()
# 7. Generate：调用 qwen3:8b 依据证据回答。
chat_request_body = {
"model": "qwen3:8b",
"stream": False,
"messages": [
{
"role": "system",
"content": system_prompt,
},
{
"role": "user",
"content": user_prompt,
},
],
"options": {
"temperature": 0,
},
}
chat_request = Request(
url=chat_url,
data=json.dumps(
chat_request_body,
ensure_ascii=False,
).encode("utf-8"),
headers={
"Content-Type": "application/json",
},
method="POST",
)
with urlopen(chat_request, timeout=120) as response:
chat_response_body = json.load(response)
print("\n=== 模型调用信息 ===")
print("实际模型：", chat_response_body.get("model"))
print("是否完成：", chat_response_body.get("done"))
print("结束原因：", chat_response_body.get("done_reason"))
answer = chat_response_body["message"]["content"].strip()
print("\n=== RAG 最终回答 ===")
print(answer)怎样运行这三个实验
前提：Ollama 已运行，并且本机已经有 bge-m3 和 qwen3:8b。
在当前项目的根目录依次运行：
python .\learning\minimal_rag.py
python .\learning\embedding_demo.py
python .\learning\vector_retrieval.py三个文件的职责不要混淆：
以后每次做 RAG 都按这个顺序检查
1. Load
文件是否真的读到了？
编码和解析器是否正确？
是否保留文件名、版本、页码、章节等来源元数据？
原始文件是否保持不变？
2. Chunk
Chunk 数量是否合理？
关键事实是否被切断？
Chunk 是否过长、过短或重复过多？
来源定位是否跟着 Chunk 一起保存？
3. Embed
文档和问题是否使用同一个 Embedding 模型？
向量数量是否与文本数量一致？
向量维度是否一致？
模型是否适合当前语言和领域？
4. Retrieve
Top-K 中是否出现了正确证据？
正确证据排名第几，分数是多少？
是否打印了候选内容、来源和分数？
不可回答问题是否也被强行匹配？
阈值是否来自评测，而不是凭感觉写死？
5. Augment
Prompt 是否包含用户原问题？
是否只放入检索得到的证据？
是否包含清楚的来源标记？
是否明确要求证据不足时拒答？
多个 Chunk 是否需要去重、排序和控制 Token？
6. Generate
实际调用的是哪个模型？
回答是否只使用证据中的事实？
引用能否定位回原文？
没有证据时是否拒答？
问题出在检索，还是出在模型没有正确使用已召回证据？
当前最小版和正式 RAG 的区别
当前脚本为了看清原理，把全部操作放在一次运行里：
每次提问 → 重新读取文件 → 重新切块 → 重新计算全部向量 → 回答正式 RAG 通常拆成两条链路。
离线建库
文件 → Load → 清洗 → Chunk → 文档 Embed → 保存到向量索引这条链路只在文件新增、更新或索引重建时运行。
在线问答
用户问题 → 问题 Embed → Retrieve Top-K → 门槛/重排
→ Augment Prompt → Generate → 回答和引用这样提问时不需要重新计算所有文档向量。
当前边界
这次已经完成：
单 Markdown 的 Load；
按空行切成 4 个 Chunk；
从手写关键词、二元词检索迭代到语义向量检索；
bge-m3的 1024 维 Embedding；余弦相似度、排序和 Top-K；
教学用证据门槛；
Prompt 增强；
qwen3:8b生成；带 Chunk 来源回答；
不可回答问题在生成前拒答。
这次尚未完成：
25 个原始文件的盘点与分类；
PDF、Word、PPT、Excel 等多格式解析；
由你亲手完成的数据清洗；
更合理的分块与 overlap；
文件名、页码、章节、表格行和版本级引用；
向量持久化和向量数据库；
固定评测集与正式阈值校准；
混合检索、Reranker、权限过滤；
Agent 路由和业务流程。
下一阶段从空文件 learning/01_inventory.py 开始，只读盘点盲盒中的 25 个文件。数据分类和清洗判断由你亲手完成，教程和助手只负责逐步解释、出题与复核。
最后再背一遍
Load → Chunk → Embed → Retrieve → Augment → Generate文件 → Chunk → bge-m3 → Top-K/门槛 → Prompt → qwen3:8b → 带来源回答以后无论换成 LangChain、LlamaIndex、Qdrant、Dify，还是加入 Reranker 和 Agent，这条主链路都不会消失。框架只是替你封装其中的步骤；判断每一步输入、输出和效果，仍然是你真正掌握 RAG 的标准。
评论