前面我们已经学习了 Embedding、文档切分和向量检索。现在把这些环节串起来:用户提出问题,程序从知识库找出相关片段,再把片段作为上下文交给回答器。这个流程通常称为 RAG(Retrieval-Augmented Generation,检索增强生成)。为了让实验不依赖网络、密钥或特定模型,本文用纯 Python 写一个字符 n-gram 检索器,并用“从证据中抽取句子”的方式完成最小回答。它不是要替代大模型,而是先把 RAG 的数据流验证清楚。

RAG 的核心数据流

一个最小 RAG 程序可以拆成四步。第一步准备知识库,把原文切成若干条有来源的片段;第二步把问题和片段转换成可比较的表示,并召回最相关的几条;第三步按固定格式把问题、证据和约束组装成上下文;第四步由生成模型根据上下文作答,并保留来源引用。

这里的“增强”不是给模型凭空增加能力,而是把外部资料临时放进本次请求。检索阶段负责找资料,生成阶段负责组织语言,两者应分别测试。若正确片段没有被召回,继续修改提示词通常没有帮助;若证据已经正确但回答出现臆测,则应检查上下文边界和回答约束。

最小可运行的本地示例

把下面代码保存为 local_rag.py,直接使用 Python 3.11 运行,不需要安装第三方包:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
from dataclasses import dataclass
from math import log, sqrt
from re import findall


@dataclass(frozen=True)
class Chunk:
source: str
text: str


KNOWLEDGE = [
Chunk("deploy.md", "部署前应把数据库连接配置放入环境变量,不要写进代码或提交到仓库。"),
Chunk("logging.md", "排查接口异常时,先查看应用日志、请求时间和最近一次配置变更。"),
Chunk("cache.md", "缓存适合保存短时间内会重复读取的数据,并应设置合理的过期时间。"),
]


def grams(text: str) -> list[str]:
"""用相邻两个字符构造简单特征;仅用于教学,不代表生产级分词。"""
text = "".join(text.lower().split())
return [text[i:i + 2] for i in range(len(text) - 1)] or [text]


def vector(text: str) -> dict[str, float]:
counts: dict[str, int] = {}
for gram in grams(text):
counts[gram] = counts.get(gram, 0) + 1
return {key: log(1 + value) for key, value in counts.items()}


def cosine(left: dict[str, float], right: dict[str, float]) -> float:
common = left.keys() & right.keys()
dot = sum(left[key] * right[key] for key in common)
size_left = sqrt(sum(value * value for value in left.values()))
size_right = sqrt(sum(value * value for value in right.values()))
return dot / (size_left * size_right) if size_left and size_right else 0.0


def retrieve(question: str, limit: int = 2) -> list[tuple[float, Chunk]]:
question_vector = vector(question)
ranked = [(cosine(question_vector, vector(chunk.text)), chunk)
for chunk in KNOWLEDGE]
return sorted(ranked, key=lambda item: item[0], reverse=True)[:limit]


def answer(question: str, evidence: list[tuple[float, Chunk]]) -> str:
"""模拟生成阶段:只从召回证据中抽取包含问题特征的句子。"""
question_grams = set(grams(question))
candidates = []
for score, chunk in evidence:
for sentence in findall(r"[^。!?.!?]+[。!?.!?]", chunk.text):
overlap = len(question_grams & set(grams(sentence)))
candidates.append((overlap, score, sentence, chunk.source))
if not candidates or max(item[0] for item in candidates) == 0:
return "知识库中没有足够证据回答这个问题。"
_, _, sentence, source = max(candidates)
return f"{sentence}(来源:{source})"


question = input("请输入问题:").strip()
results = retrieve(question)
print("\n召回片段:")
for score, chunk in results:
print(f"- {score:.3f} | {chunk.source} | {chunk.text}")
print("\n回答:")
print(answer(question, results))

运行:

1
python local_rag.py

可以输入“数据库连接配置应该放在哪里?”观察程序先打印候选片段,再从证据中给出回答。这里没有虚构运行结果:相似度和排序由你本地的 Python 实际计算。Chunk 同时保存正文和来源,检索返回 (score, chunk),回答阶段因此能够展示引用,而不是只返回一段无法追溯的文本。

逐段理解实现

grams 把中文文本转换为相邻双字符特征,例如“环境变量”会产生“环境”“境变”“变量”。这是为了在没有分词器和 Embedding API 的情况下演示“文本表示”;vector 使用对数频次降低重复字符的影响,cosine 用余弦相似度比较问题和片段。真实项目通常改用同一个 Embedding 模型为入库文本和查询生成向量,并把向量保存到向量数据库。

retrieve 只做召回和排序,不回答问题。limit=2 表示把前两条候选交给下一阶段;数量并非越大越好,因为无关内容会占用上下文。answer 是一个确定性的回答器:从召回片段中选择与问题共享特征最多的句子。它的限制很明显,但也恰好说明了生成模型需要什么输入——问题、证据和“只能依据证据作答”的约束。

接入大模型时,可以保留 retrieve,将其结果格式化为带来源的上下文,再通过当前模型 SDK 的文本生成接口提交。例如上下文可以是:[来源: deploy.md]\n部署前应把...。提示词应明确要求“只使用提供的证据;证据不足时明确说不知道;在结论后保留来源”。密钥只能从环境变量读取,不应写入脚本、日志或 Markdown。模型和 SDK 版本变化较快,实际接入前应以所选服务商的官方文档为准。

常见问题

这算真正的 RAG 吗? 算一个可验证的最小闭环,但回答阶段是抽取式的,不是大模型生成式回答。它适合验证检索、上下文和引用;要获得自然语言答案,再替换 answer,不要同时改动所有环节。

为什么问题明明相关,却没有召回正确片段? 先打印每条分数,检查切分、特征表示和 limit。生产系统还要确认入库与查询使用同一个 Embedding 模型,并检查元数据过滤是否误排除了证据。

把所有文档拼接给模型不行吗? 文档少时可以做实验,但数据变大后会增加延迟、费用和噪声,也可能超过上下文窗口。RAG 的价值就在于先缩小证据范围。

能否相信最高分片段? 不能。相似度是召回信号,不是事实正确率。关键业务应保留来源、设置最低分数、测试已知问题,并对无证据问题允许拒答。

小结

RAG 的最小闭环是“问题表示—检索片段—组装上下文—基于证据回答”。本文用纯 Python 完成了一个离线、可重复运行的版本,并把检索和回答明确分层。字符特征只是教学替身,下一步工程化时应接入真实 Embedding、向量数据库和模型生成,同时保留来源、权限过滤与召回评估。先验证数据流,再逐步替换组件,通常比一开始引入完整框架更容易定位问题。