问模型一个你私有代码库的问题,它只能瞎编或认怂——参数化知识覆盖不到你的数据。本文是「从零手写 Agent」系列第 9 篇:手写最小 RAG 管线,全程只靠 numpy,把"回答前先查资料"的完整链路走一遍,并实测玩具向量与真 embedding 的差距在哪。上一篇的子 Agent 解决的是人手不够,这一篇解决原料不够:知识进不了上下文,再强的循环也只能空转。
本文按"为什么是 RAG → 四步管线 → 玩具向量实测 → 检索质量 → 总结"组织。
1、概述:为什么是 RAG
让模型掌握私有知识有两条路。对比如下:
| 方案 | 原理 | 代价 |
|---|---|---|
| 微调 | 把知识烧进模型参数 | 贵、慢、更新一次重训一次 |
| RAG | 回答前检索资料塞进上下文 | 便宜、即时更新、可溯源 |
解读:RAG(Retrieval-Augmented Generation,检索增强生成)是知识库问答的绝对主流。它和 Agent 的关系同样简单:检索可以做成一个工具,Agent 循环自主决定何时查资料——静态管线升级为动态的检索型 Agent,只是一次工具注册的事。
两条路的适用面也能一句话分清:知识每天都在变的(文档、工单、公告)走 RAG,改一个字立刻生效;要改的是风格、口吻、领域语感这类"写进骨子里"的能力,才值得微调。两者不互斥,生产系统常见的组合是微调打底、RAG 供料。
两条路之间还有个常被忽略的中间选项:文档不多时干脆全文塞进上下文。本项目约三十个文档全文合计几万 token,长上下文模型一顿吃得下,检索这一步可以整体省掉。但文档量再翻十倍,或者每次提问只真正用到其中百分之一,全文塞入的 token 账单和干扰噪声会把你推回检索路线——RAG 的本质是用检索精度换上下文预算。经验分界:语料几十万 token 以内且每次问答密度高,直接塞;再往上,检索。
2、四步管线
最小 RAG 由"建库两步、查询两步"构成:
解读:向量是把文本压成定长数组,语义相近的文本向量方向相近;余弦相似度衡量方向的夹角,检索就是在向量空间里找邻居。把 Top-K 检索块与问题一起塞进提示词,模型据实作答并标注来源——溯源能力是 RAG 相对微调的独特优势。
建库两步是离线的一次性成本,查询两步每次提问在线执行。四步拆开各不到二十行代码,从最影响质量的一步说起。
2.1、切块:最影响质量的一步
切块是管线里最影响质量的一步,按语义单元切块的参考实现如下:
CHUNK_LINES = 40 # 每块约 40 行:太大带噪声,太碎丢上下文
def chunk_file(path) -> list[dict]:
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
chunks = []
for start in range(0, len(lines), CHUNK_LINES):
block = lines[start:start + CHUNK_LINES]
if any(line.strip() for line in block): # 跳过纯空白块
chunks.append({"source": f"{path}:{start + 1}",
"text": "\n".join(block)})
return chunks
解读:每个块自带来源标注(文件名加起始行),答案引用才有据可查;块的大小是精度与噪声的折中,进阶做法是按段落聚合并在相邻块之间留重叠,防止语义被边界斩断。
重叠切块的升级版按段落聚合攒块,块间保留尾部重叠,实现如下:
def chunk_by_paragraph(text: str, max_chars=1200, overlap=200):
paras = [p for p in text.split("\n\n") if p.strip()]
chunks, buf = [], ""
for p in paras:
if buf and len(buf) + len(p) > max_chars: # 攒够一块就落盘
chunks.append({"text": buf})
buf = buf[-overlap:] + "\n\n" # 尾部重叠续到下块
buf += p + "\n\n"
if buf.strip():
chunks.append({"text": buf})
return chunks
解读:overlap 让被边界斩断的句子在相邻两块里各存一份,命中任何一块都能拿到完整语义。代价是重复内容的 embedding 成本,200 字重叠摊在 1200 字的块上约多花六分之一,值得付。
2.2、检索函数:余弦取 Top-K
库里每块存好向量后,检索就是打分排序,暴力扫描版如下:
def top_k_search(query: str, chunks: list[dict], k: int = 3):
qv = embed(query) # 查询向量在线计算
scored = [(cosine(qv, ch["vector"]), ch) for ch in chunks]
scored.sort(key=lambda x: -x[0]) # 相关度降序
return [{"score": round(s, 4), **ch} for s, ch in scored[:k]]
解读:三十个文档切出一百二十块,建库产物是一百二十乘一千零二十四的浮点数组,不到 1 MB;每次检索算一百二十个余弦,实测笔记本上不到 1 毫秒——这个规模根本轮不到向量数据库出场。上千块之前,列表加排序就是正确方案;真到百万块,把这段换成 faiss 或 Milvus 的 ANN 索引,管线其余三步一行不改。
2.3、拼提示:让资料进上下文
最后一步把检索结果拼进提示词并调用模型,代码如下:
def build_prompt(question: str, hits: list[dict]) -> str:
docs = "\n\n".join(
f"[资料{i}] 来源 {h['source']},相关度 {h['score']}:\n{h['text']}"
for i, h in enumerate(hits, 1))
return ("只依据以下资料回答,注明来源,资料不足就直说:\n\n"
f"{docs}\n\n问题:{question}")
def answer(question: str, chunks: list[dict]) -> str:
hits = top_k_search(question, chunks, k=3) # 检索 Top-3
resp = client.messages.create(
model=MODEL, max_tokens=1024,
messages=[{"role": "user",
"content": build_prompt(question, hits)}])
return "".join(b.text for b in resp.content if b.type == "text")
解读:提示词里两句话是质量护栏:只依据资料回答,压住模型的脑补;资料不足就直说,给模型认怂的台阶。资料带着来源和相关度一起展示,模型引用有抓手,你排查时也能立刻分清是检索的锅还是生成的锅——可溯源让两类错误可以分家问责。至此四步齐活:建库脚本跑一次,answer 函数随叫随到。
K 值的选法没有玄学:K 越大,正确块进入上下文的概率越高,但噪声与 token 账单同步上涨。实测从 1 起步逐档往上加,本项目语料上答案质量在 K 等于 3 到 5 之间到顶,再加大就开始被无关资料带偏——先小后大,到顶即收。
3、玩具向量实测
不依赖任何 embedding 服务,用"字符二元组哈希"就能造出可用的玩具向量:
import numpy as np
DIM = 1024
def embed(text: str) -> np.ndarray:
v = np.zeros(DIM)
for i in range(len(text) - 1):
v[hash(text[i:i + 2]) % DIM] += 1 # 相邻字符对决定维度
norm = np.linalg.norm(v)
return v / norm if norm > 0 else v
def cosine(a, b) -> float:
return float(a @ b) # 归一化后点积即余弦
解读:每个相邻字符对哈希到某一维累加,最后归一化。它只懂字面重合,不懂语义——这正是教学价值。实测语义鸿沟:搜"安全阀"命中相关文档,搜同义的"失控保护"则完全跑偏;换成真 embedding(如智谱的 embedding-3),同义词会映射到相近方向,鸿沟消失。玩具与真品的差距一行见底:字面统计对语义编码。注意 Anthropic 不提供 embedding 接口,生态上通常用其他厂商或开源模型补这一块,智谱开放平台与numpy是本文依赖。
建库成本还能再砍一刀:embedding 结果落盘缓存,文档没变就不重算。玩具向量零成本无所谓,真 API 按量计费,三十个文档几分钱看着不多,语料库每天全量重嵌就是白烧——以文件内容哈希做缓存键,只有新增或变更的块才重新计算,这是工程版与玩具版的分水岭之一。
3.1、字面查询的实测记录
用第 2 节的管线给本项目约三十个文档建库,两组查询的真实检索记录如下:
查询「安全阀在哪个文件定义」
命中1 config.py:1 相关度 0.83 ← 五行入口文件
命中2 safety.py:40 相关度 0.71 ← 真正的定义块
查询「失控保护怎么触发」(上一问的同义改写)
命中1 router.py:12 相关度 0.44 ← 完全跑偏
命中2 utils.py:88 相关度 0.41 ← 完全跑偏
解读:第一组字面对得上,正确块虽被入口文件压了一头,好歹留在 Top-3 里;第二组一个字都对不上,向量立刻翻脸——玩具向量没有"同义词方向相近"的性质,字面重合为零就是零。这份记录还预告了第 4 节的主角:首位命中是五行小文件,不是巧合。
3.2、换真 embedding 补语义
真 embedding 的接入只是把 embed 函数换掉,其余三步不动。以智谱为例,批量调用如下:
from zhipuai import ZhipuAI
zc = ZhipuAI(api_key=API_KEY)
def embed_api(texts: list[str]) -> list[np.ndarray]:
vecs = []
for i in range(0, len(texts), 64): # 单请求上限 64 条
resp = zc.embeddings.create(model="embedding-3",
input=texts[i:i + 64])
vecs += [np.array(d.embedding) for d in resp.data]
return vecs
解读:embedding-3 默认输出 2048 维,支持 256 到 2048 的套娃式降维自选,单条最长 8192 token,标价 0.5 元每百万 token——本项目三十个文档全量向量化一遍约几万 token,几分钱。换上真向量重跑 3.1 的两组查询:字面查询命中不变,同义改写"失控保护"的正确块相关度从查无此人升到 0.68 上下、稳居首位,语义鸿沟的消融一步到位。参数细节以智谱 Embedding-3 文档为准。
4、检索质量是主要矛盾
把这套管线架在本项目约三十个文档上做对比实验:同一个私有知识问题,裸问时模型诚实认怂;带检索问时,答案质量完全取决于检索命中。实测暴露的第一个缺陷很典型:又短又小的块总排在前面——小块字符分布集中,与查询字面重合度天然偏高,五行的入口文件抢过了真正的正文块。加证据条数后正确块进入上下文,模型给出了带来源引用的准确回答。
4.1、小块偏好与对策
小块偏好可以算明白:玩具向量按字符对计数,五行入口文件只有几十个字符对,向量能量集中在极少维度,与任何查询的归一化点积都容易虚高;上千行的正文块把字符对摊薄在一千零二十四维上,单维能量低,字面得分被稀释。对策按侵入性排序有三个:
- 过滤:建库时丢弃少于五行或五十字的碎块
- 惩罚:得分除以块长的平方根,压掉短块的天然优势
- 加量:Top-K 从 3 提到 5,让正确块挤进上下文
三种口径在同一问题上的首位命中对比如下:
| 口径 | 首位命中 | 判读 |
|---|---|---|
| 裸余弦 | 入口文件块 | 字面重合虚高 |
| 长度惩罚 | 正文定义块 | 真相关块浮出 |
| 加大 K 值 | 入口文件块 | 首位仍偏但正确块入榜 |
解读:实测里最省事的是第三种——K 提到 5,多付两个块的上下文 token,正确块进上下文后生成侧自会取舍;长度惩罚更彻底但要调系数。真 embedding 下小块偏好同样存在,只是幅度收窄——机理从字符对计数换成语义向量,短文本与查询的表面相似度偏高是通病。
由此得出三条工程直觉。其一切块要按语义单元(段落、函数)而非定长硬切,相邻块留重叠防止语义被边界斩断;其二检索质量比生成质量更值得投入,喂错资料神仙也答不对;其三评估要单独做检索层命中率,而不是只看最终答案。规模化时把暴力扫描换成向量数据库即可,管线结构不变。
5、总结
RAG 的最小骨架四步走完:切块、向量化、余弦检索、塞上下文作答;玩具向量足以讲清原理,真 embedding 解决语义鸿沟,而检索质量才是生产系统的主要矛盾。对于动手派,建议拿自己的一批文档建库,分别用字面与语义两种查询词实测命中差异,再把检索包装成工具挂到 Agent 上。下一篇解决两个工程化硬需求:结构化输出与评估。RAG 模式的系统讨论可参考 Anthropic RAG 文档。

