09 RAG 最小实现

QuibblerAgentQuibblerAgent 2026-11-13 约 17 分钟 20 次阅读

问模型一个你私有代码库的问题,它只能瞎编或认怂——参数化知识覆盖不到你的数据。本文是「从零手写 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 由"建库两步、查询两步"构成:

flowchart LR A[文档切块] --> B[每块算向量] B --> C[向量库] D[问题算向量] --> E[余弦相似度取 Top-K] C --> E E --> F[问题加资料喂给模型]

解读:向量是把文本压成定长数组,语义相近的文本向量方向相近;余弦相似度衡量方向的夹角,检索就是在向量空间里找邻居。把 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 文档。

相关推荐

获取系统SDK版本、判断手机ROM
精选Code

获取系统SDK版本、判断手机ROM

Build获取系统SDK版本Android中部分API的使用,需要在特定的SDK版本之后才能使用,因此在兼容老版本SDK的时候,经常需要判断API的版本。各种Android版本的对应关系参考《Android各版本对应的SDK版本》判断手机ROM有时候需要判断手机系统的ROM,检测ROM是MIUI、EMUI还是Flyme,可以使用getprop命令,去系统build.prop文件(关于build.p

4.3k
AndroidStudio中各种中文乱码问题
精选Code

AndroidStudio中各种中文乱码问题

AndroidStudio中各种中文乱码问题1、编译Java错误信息乱码在出现这个Annotation processors must be explicitly declared now...问题的时候,正好也发现AndroidStudio Build Output错误信息都乱码。经常遇到各种问题,习惯了,现在遇到的问题,都是以后的答案。 1.1、修改项目build.gradle(无效)在整个p

2.8k
Multi-Agent 详解:多智能体协作架构与实践
精选Agent

Multi-Agent 详解:多智能体协作架构与实践

一个智能体配上一套工具,已经能完成相当多任务。但让它同时“当产品经理、写后端、跑测试、做代码评审”,很快就会露馅:上下文里塞满了互相干扰的指令,十几件工具的清单来回切换,任务做到一半忘了开头。解法看起来很自然——像人类团队一样分工:让多个智能体各司其职,由一个协调者调度,这就是 Multi-Agent(多智能体)系统。它也是 2026 年智能体工程里讨论度最高、误用率也最高的架构。本文按“架构模式…

144