使用 Python 构建无需支付 API 费用的本地 RAG 流水线

一个完整的检索增强生成流程完全可以在个人电脑上构建并运行，仅需CPU、免费的开源模型，无需OpenAI密钥、信用卡，也没有使用限制。关键在于理解每个组成部分，而非直接复制黑盒系统。

最佳学习方式是动手实践：打开终端和名为rag.py的Python文件，逐段运行代码并在执行前预测输出结果。大多数步骤之后都会有简短的练习题，正是通过这些练习才能真正掌握知识。

RAG解决的问题

语言模型仅知晓其训练数据中的内容。私人笔记、公司维基、上周的会议记录或任何未被收录的文档对它来说都是不可见的。当被问及这些内容时，它要么承认不知情，更糟糕的是会自信地编造答案。

RAG将任务分为两个部分：

1. 检索：找出与问题相关的文档片段。这属于搜索问题。
2. 生成：将那些片段交给模型，并指示其根据这些片段给出答案。

模型会停止猜测，开始阅读。其余的都是为了顺利完成这两步而需要的操作方法。

question ──► [embed] ──► [search your docs] ──► top chunks ──┐
                                                             ▼
                                          [LLM: "answer using this context"] ──► answer

免费工具包

整个系统由三个组件构成。sentence-transformers利用可在CPU上运行且只需下载一次的小型模型将文本转换为向量。numpy负责处理相似度计算，而整个“向量数据库”仅用约五行代码即可实现，便于理解其工作原理。Flan-T5版本的transformers用于生成答案，因为它经过指令微调、可在CPU上运行且无需调用API。

只需安装一次所有组件：

pip install sentence-transformers transformers torch numpy

首次加载时，每个模型都会从 Hugging Face 获取其权重。嵌入器的大小约为 80 MB，而 flan-t5-base 的大小则约为 1 GB。之后这些权重会被缓存，从而支持离线使用。在磁盘或内存资源有限的情况下，flan-t5-small（约 300 MB）可以替代基础模型使用，虽然质量稍低，但速度会快很多。

第一步：模型无法知晓的语料库

通过那些模型从未见过的文档，最容易理解 RAG 的作用，因此脚本从一个小型的虚构公司维基开始。创建 rag.py 文件，然后将相关文档粘贴进去：

# rag.py
DOCUMENTS = [
    """Nimbus is a fictional note-taking app launched in 2023. The free plan,
    called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
    collaboration features on the free plan.""",

该维基网站还有更多条目：

    """Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
    monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
    real-time collaboration with up to 5 people per note.""",    """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
    encryption is only available on the Pro plan and must be enabled manually in
    Settings > Security. Once enabled it cannot be turned off for that note.""",    """The Nimbus mobile app supports offline editing. Changes made offline are
    queued and sync automatically the next time the device is online. If two
    devices edit the same note offline, Nimbus keeps both versions and flags a
    conflict for the user to resolve.""",    """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
    Refunds are processed to the original payment method within 5 business days.
    Annual plans cancelled after 30 days are not refundable but stay active until
    the end of the billing period.""",    """Nimbus support is available via email at help@nimbus.example and live chat.
    Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
    Free-plan users receive email support with a typical 48-hour response time.""",
]

在开始构建任何东西之前，先证明问题确实存在。在 Python shell 中，向普通模型询问关于 Nimbus 的情况：

from transformers import pipeline
gen = pipeline("text2text-generation", model="google/flan-t5-base")
print(gen("How much does Nimbus Pro cost?", max_new_tokens=50)[0]["generated_text"])

它会生成某些内容，但由于只是猜测，这些内容要么有误要么为空。请记住这个结果，因为教程的后续部分会对其进行修正。

第2步：分块

整个文档无法直接输入模型。出于两个原因，它们会被切割成更小的分块：其一，这样检索会更精确，因为模型能找到回答问题的段落而非整份手册；其二，模型的上下文窗口容量有限。

添加一个分块函数：

def chunk_text(text, chunk_size=60, overlap=15):
    """Split text into overlapping chunks of `chunk_size` words."""
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        if end >= len(words):
            break
        start = end - overlap   # step back by `overlap` so context isn't cut
    return chunks

然后在记录每个数据块来自哪份文档的同时构建数据块列表：

# Build our chunk list, remembering which doc each chunk came from
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
    for c in chunk_text(doc):
        chunks.append({"doc_id": doc_id, "text": c})print(f"{len(DOCUMENTS)} documents -> {len(chunks)} chunks")
for c in chunks[:3]:
    print("-", c["text"][:70], "...")

出现重叠现象是因为某个事实可能跨越边界，比如退款政策的时间范围介于“5天内”和“工作日”之间。如果没有重叠，那么任一片段都无法涵盖全部事实。在每个分割点重复一些词语，就能让这些事实在分割后依然完整。

练习：使用chunk_size=20生成更多更小的片段，然后再尝试chunk_size=500，此时这些短文档会各自成为一个独立的片段。在运行之前，先推测哪种设置能更精确地提取信息，但同时又可能丢失周围的上下文。小片段虽然精确度较高，但缺乏上下文信息，这种矛盾正是分块处理的核心问题。并不存在一种普遍适用的最佳大小。

第3步：嵌入向量

嵌入向量是一种用数字表示的向量，用于编码文本的含义。意义相近的文本片段即便没有共同的词汇，也会形成指向相同方向的向量，因此“它多少钱？”会与“价格是多少？”被归为同类。

导入嵌入模型：

from sentence_transformers import SentenceTransformer

然后嵌入每个数据块：

embedder = SentenceTransformer("all-MiniLM-L6-v2")# Embed every chunk. normalize_embeddings=True makes the vectors unit-length,
# which lets us measure similarity with a simple dot product later.
chunk_texts = [c["text"] for c in chunks]
chunk_vectors = embedder.encode(chunk_texts, normalize_embeddings=True)print("vector shape:", chunk_vectors.shape)   # (num_chunks, 384)

现在的每个数据块都对应一个384维向量，其中384就是该模型用于表达含义的数值数量。

练习：通过一个小实验观察含义是如何体现的：

import numpy as np
pairs = embedder.encode(
    ["the price of the pro plan", "how much does it cost", "the weather in Paris"],
    normalize_embeddings=True,
)
print("price vs cost :", round(float(pairs[0] @ pairs[1]), 3))   # should be HIGH
print("price vs weather:", round(float(pairs[0] @ pairs[2]), 3)) # should be LOW

a @ b这一表达式表示点积，对于单位向量而言其值等同于余弦相似度。数值为1.0表示方向完全相同，而0则表示毫无关联。尽管这两组短语没有共用任何词汇，但第一组的相似度应远高于第二组，这便是语义搜索的体现。在运行程序之前，试着在0.2的误差范围内猜出这两个数值。

第4步：检索

在搜索阶段，需将查询语句进行向量化处理，再与每个分块向量进行比较，保留相似度最高的几个向量。由于这些向量已经过归一化处理，因此“比较”操作实际上就相当于矩阵乘法。

import numpy as np

检索功能：

def retrieve(question, k=3):
    q_vec = embedder.encode([question], normalize_embeddings=True)[0]
    scores = chunk_vectors @ q_vec              # cosine similarity to every chunk
    top_idx = np.argsort(scores)[::-1][:k]      # indices of the k highest scores
    return [(chunks[i]["text"], float(scores[i])) for i in top_idx]for text, score in retrieve("How much does Nimbus Pro cost?"):
    print(f"[{score:.3f}] {text[:80]}...")

尽管问题中提到的是“成本”，而文档中写的是“每月8美元”，但定价相关内容仍应因得分最高而显示在顶部。这属于语义匹配，而非关键词匹配。

练习：尝试三个问题，看看会显示什么内容：

• retrieve("Can I get my money back?")应返回退款相关内容，尽管问题中从未出现“refund”这个词。
• retrieve("Is my data safe?")应显示加密相关内容。
• retrieve("Does it still work without an internet connection?")应返回离线同步相关内容。

许多人低估了这个阶段的重要性。当检索返回错误的片段时，再巧妙的提示也无法得到正确答案。输入有误则输出必错，因此调试工作应首先从这里开始。

第5步：生成

现在将这两部分结合起来。设计一个提示语，将检索到的片段传递给Flan-T5，并要求它仅基于这些片段来回答问题。

from transformers import pipeline

生成函数：

generator = pipeline("text2text-generation", model="google/flan-t5-base")def rag_answer(question, k=3):
    retrieved = retrieve(question, k=k)
    context = "\n".join(text for text, _ in retrieved)    prompt = f"""Answer the question using only the context below.
If the answer is not in the context, say you don't know.Context:
{context}Question: {question}
Answer:"""    out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
    return out.strip(), retrievedanswer, sources = rag_answer("How much does Nimbus Pro cost?")
print("ANSWER:", answer)
print("\nBased on:")
for text, score in sources:
    print(f"  [{score:.3f}] {text[:70]}...")

将结果与第一步中的错误答案或空答案进行比较。模型和问题完全相同，唯一的区别在于提供了相关文本供阅读。这一差异正是RAG的全部价值所在。

练习：提出文档中未涉及的问题，例如关于Nimbus的Linux桌面客户端的相关问题，然后使用该问题调用rag_answer函数。设计合理的系统会承认自己不知道答案；如果你的系统编造了回复，那就是一种被称为“检索后幻觉”的真正故障模式，后续章节将对此进行探讨。

完整的脚本

以下是完整的rag.py代码，它是一个包含交互式循环的可运行程序。代码开头是导入语句：

# rag.py — a complete, local, no-API RAG system
import numpy as np
from sentence_transformers import SentenceTransformer
from transformers import pipeline

完整内容如下：

DOCUMENTS = [
    """Nimbus is a fictional note-taking app launched in 2023. The free plan,
    called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
    collaboration features on the free plan.""",
    """Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
    monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
    real-time collaboration with up to 5 people per note.""",
    """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
    encryption is only available on the Pro plan and must be enabled manually in
    Settings > Security. Once enabled it cannot be turned off for that note.""",
    """The Nimbus mobile app supports offline editing. Changes made offline are
    queued and sync automatically the next time the device is online. If two
    devices edit the same note offline, Nimbus keeps both versions and flags a
    conflict for the user to resolve.""",
    """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
    Refunds are processed to the original payment method within 5 business days.
    Annual plans cancelled after 30 days are not refundable but stay active until
    the end of the billing period.""",
    """Nimbus support is available via email at help@nimbus.example and live chat.
    Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
    Free-plan users receive email support with a typical 48-hour response time.""",
]def chunk_text(text, chunk_size=60, overlap=15):
    words = text.split()
    chunks, start = [], 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        if end >= len(words):
            break
        start = end - overlap
    return chunksprint("Loading models (first run downloads them)...")
embedder = SentenceTransformer("all-MiniLM-L6-v2")
generator = pipeline("text2text-generation", model="google/flan-t5-base")# Index the documents once at startup
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
    for c in chunk_text(doc):
        chunks.append({"doc_id": doc_id, "text": c})
chunk_vectors = embedder.encode(
    [c["text"] for c in chunks], normalize_embeddings=True
)def retrieve(question, k=3):
    q_vec = embedder.encode([question], normalize_embeddings=True)[0]
    scores = chunk_vectors @ q_vec
    top_idx = np.argsort(scores)[::-1][:k]
    return [(chunks[i]["text"], float(scores[i])) for i in top_idx]def rag_answer(question, k=3):
    retrieved = retrieve(question, k=k)
    context = "\n".join(text for text, _ in retrieved)
    prompt = (
        "Answer the question using only the context below. "
        "If the answer is not in the context, say you don't know.\n\n"
        f"Context:\n{context}\n\nQuestion: {question}\nAnswer:"
    )
    out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
    return out.strip()if __name__ == "__main__":
    print("RAG ready. Ask about Nimbus (or type 'quit').\n")
    while True:
        q = input("You: ").strip()
        if q.lower() in {"quit", "exit", ""}:
            break
        print("Nimbus bot:", rag_answer(q), "\n")

从以下内容开始：

python rag.py

可以尝试这样的提示语：退款期限是多久、免费用户是否可以使用实时聊天功能，以及免费存储空间有多大。

通过A/B测试证明价值

有一种实验方法可以通过在有上下文信息和无上下文信息的情况下提出相同问题来清晰地说明问题。首先是一个基准函数：

def no_rag(question):
    out = generator(f"Question: {question}\nAnswer:", max_new_tokens=80)
    return out[0]["generated_text"].strip()

接着是对比：

q = "Can free-plan Nimbus users use live chat support?"
print("WITHOUT context:", no_rag(q))
print("WITH context   :", rag_answer(q))

没有上下文的答案只是猜测，而有了上下文的话就应该说明实时聊天功能仅面向专业版用户，免费用户则只能通过邮件获取回复。可以尝试三种不同的问题。这个A/B测试是目前最能体现RAG优势的演示，建议保留下来用于向他人解释这一概念。

按收益排序的升级选项

在系统已经可以正常运行的基础上，以下这些升级按其带来的收益从高到低排列：

1. 显示信息来源。该功能已经会返回retrieved状态，因此应在每个答案下方打印出相关内容片段。真正的RAG产品成败就在于这一点——有来源标注的答案是可信的，而没有标注的则不过是格式好看的猜测而已。

先获取更多结果，再过滤。将k的值设为5，并设定最低分值，丢弃得分低于约0.25的项。如果没有任何片段通过筛选，回复“我不知道”并完全跳过模型调用。这样就能消除类似第5步练习中出现的那种幻觉现象。

真正的向量存储库。chunk_vectors @ q_vec会扫描每一个片段，对于数百个片段来说还可以，但面对数百万个片段时效率就很低。Chroma（通过pip install chromadb安装）可以对向量进行索引以实现快速搜索，同时将其保存到磁盘上，从而避免在启动时重复嵌入向量。嵌入、存储和余弦搜索的概念保持不变，只是Chroma能在更大规模上实现这些操作。

更强大的免费生成器。Flan-T5体积较小且有时表达较为简略。若需获得更好的本地化答案，可安装Ollama，它在本地运行Llama 3.2或Phi等模型，并将generator(...)调用替换为对本地API的请求。使用Google的Gemini或Groq提供的免费API层级，只需替换rag_answer中的生成代码即可实现相同功能。此时仅有一个函数会发生变化，而检索部分则保持不变。

评估。编写十个已知答案的问题，通过rag_answer进行处理并检查正确性。这虽然看似枯燥，却是区分玩具与真正产品的关键——因为无法衡量的事物就无法得到改进。

问题诊断

RAG实际上是一种搜索功能与写作辅助工具的结合体，几乎所有问题都可归因于这两部分中的某一方。如果检索到了正确的信息片段但答案仍错误，那是生成或提示词方面的问题；而如果根本未检索到正确的信息片段导致答案错误，那就是涉及信息分段、嵌入模型或参数k的检索问题。在做出任何修改之前，先确定是哪一部分出了故障。将retrieve()和rag_answer()设为独立的函数，正是为了能够分别检查这两部分的功能，而保持这种分离正是此处最宝贵的习惯。

故障排除

• 下载速度慢或失败：首次运行时会从Hugging Face获取模型，并将其缓存到~/.cache/huggingface目录下。一旦被缓存，即可离线使用。

内存压力或运行缓慢：可将模型降级为 google/flan-t5-small，并将 chunk_size 设为 40。该嵌入模型体积足够小，不会成为性能瓶颈。

torch 安装问题：请从 PyTorch 的索引中获取仅支持 CPU 的版本，使用命令 pip install torch --index-url https://download.pytorch.org/whl/cpu 进行安装。

回复内容过短：应提高 max_new_tokens 的值，因为 Flan-T5 往往会过早停止生成。使用如 Ollama 提供的更大规模的生成器，比任何参数调整都更有效。

总是只生成一个片段：可能是文档内容几乎相同，或是参数 k 设定得太低。建议查看原始评分，因为它们能反映模型对内容的排序方式。

通过实际练习，人们对 RAG 的理解会比仅使用库封装功能的人更为深入。