☰
手搓一个最小RAG应用:五段式流程拆解与代码实现
2026/10/20 8:57:54 网站建设 项目流程

1. 先搞懂RAG:它解决什么问题,凭什么值得学

1.1 大模型应用落地的三道坎

做应用的人应该都有这种体会:模型API调得再溜,真放到业务场景里,总会碰到几个绕不过去的坎。

第一道坎是知识过时。大模型的训练数据有一个截止时间,你问它三个月前刚发布的新政策、昨天刚上线的接口文档,它要么答不上来,要么用旧知识硬撑。第二道坎是幻觉。模型在它自己都没把握的领域里,会极其流畅地编造出一整套看起来合情合理的答案。闲聊无所谓,放到企业客服、政策问答、设备维修指导这些场景,一次幻觉可能就造成一次实打实的损失。第三道坎是私有数据。你公司内部的规章制度、产品说明书、过往项目总结,模型根本没见过。想让模型“学会”这些内容,传统思路是微调,但微调周期长、成本高,业务一变就得重新来一遍。

这三道坎凑在一起,成了很多团队“模型很厉害,产品上不了线”的尴尬原因。RAG的出现,恰好就是冲着这三道坎来的。

1.2 开卷考试思路:检索增强生成到底在做什么

RAG的完整叫法是Retrieval-Augmented Generation,检索增强生成。我每次给朋友讲这个东西,都喜欢用一个类比:它相当于给大模型开卷考试。

你不再指望模型“背”下所有知识,而是每次用户提问时,先跑到一个外部知识库里,把和问题最相关的几段材料翻出来,然后把这些材料和用户的问题一起交到大模型手上:“你看,这是参考答案,照着它回答。”模型的工作重心从“回忆知识”变成了“阅读理解+归纳表达”,这个转变非常关键。

这套设计带来三个很直接的好处。一是知识实时可控,知识库里的文档是随时可以增删改的,文档更新完,系统立刻就能用上新知识,不用重新训练。二是答案可追溯,每一句话都能对应到知识库的某个片段,出了问题能查源头。三是落地门槛低,相比微调需要准备训练数据、算力、调参,RAG只需要你做好文档处理和检索,大部分工作量是数据工程,对纯算法功底的要求不高,工程经验足够的人很快就能上手。

我见过太多团队在“RAG还是微调”这个问题上纠结大半年,最后发现真正的问题是连自己的文档质量都没整理干净。这里我的建议很明确:先RAG跑通,再评估要不要上微调。因为RAG的每一个环节都能单独优化、单独排查,而微调一旦效果不好,你很难定位是数据问题还是参数问题。

注意:RAG适合“知识密集型”场景,比如问答客服、政策解读、企业内部知识库。如果你要做的是“让模型学会某种输出格式”或者“固化某种能力”,那微调更合适。别迷信“万物皆可RAG”,先分清场景再选方案。

2. 拆开流水线:RAG的五个核心环节

2.1 五个环节与各自的角色

RAG看起来玄乎,但拆开来看就是一条非常标准的流水线,五个环节,一个都不能少:

  • 文档加载与解析:把PDF、Word、Markdown、网页等原始素材读进来,清洗成干净的纯文本。
  • 文本切块:把长文档切成一段一段的“小卡片”。这是RAG里最容易被小看、但对效果影响最大的一步。
  • 向量化:把每一段文字通过嵌入模型转成一组数字向量。这组向量可以理解为“这段话的语义坐标”。
  • 存储与检索:向量存进向量数据库,用户提问时先把问题也转成向量,然后在库里找最接近的若干段。
  • 生成与提示词组装:把检索到的片段和用户问题拼成一个Prompt,交给大模型生成最终答案。

把这五个环节串起来看,RAG的核心难点其实不在“生成”,而在“检索”。检索这一步决定了模型能看到什么材料,材料不对,后面再厉害的模型也白搭。所以后面调试的时候,你的主要精力应该花在文档处理和检索质量上,而不是纠结要不要换一个更大的模型。

2.2 三个关键参数:切块大小、重叠、Top-K

入门RAG,有三个参数你必须先搞明白,因为它们直接决定检索效果的上限。

切块大小很好理解,就是每段文本多长。切太大,一段里混了多个主题,检索时容易把不相关的内容一起带进来;切太小,语义不完整,模型看到的是一堆碎片。以中文文档为例,我建议起步值设在200到500字之间。重叠则解决“一句话被从中间切开”的问题,让下一块带上上一块的尾部,默认50字左右,遇到强依赖上下文的场景可以加大到100。

Top-K是检索后返回的片段数量。K值太小,模型看到的材料不够;K值太大,Prompt里塞满无关内容,既浪费token,又会干扰生成。我通常用5到8之间。注意这几个参数是一组联动项,切块500字配Top-K 5,模型大约能看到2500字的上下文;如果你改用切块200字,那Top-K 建议同步提到8到10,否则视野会突然变小。

这里想先把一个底层概念讲透,就是“向量检索到底是怎么找到相似内容的”。嵌入模型会把文本转成一个几百维甚至上千维的向量,一段文本和另一段文本的语义相似度,就近似等于这两个向量在“语义空间”里的夹角大小。夹角越小,方向越接近,语义就越相关。这个计算量化的指标叫余弦相似度,取值范围在-1到1之间,越接近1越相似。我们后面写代码时,核心检索逻辑就是算这个东西。

3. 完整代码:手搓一个最小RAG应用

3.1 先说清楚我的教学路线

网上绝大多数RAG教程,一上来就甩LangChain加向量数据库全家桶。我的看法恰恰相反:新手直接上框架,只会得到一个“黑盒”——报错不知道错在哪,效果差不知道调哪里。

所以这篇文章我采用“手搓”路线。我会用最精简的Python代码,把RAG的五个环节一个一个写出来,总代码量不超过150行,没有任何复杂框架依赖,你只需要装两个库。等你彻底理解了每一步在干嘛,再去用LangChain、Chroma这些工具,就是水到渠成的事,而且你还能反过来评价它们做得好不好。

3.2 环境准备与依赖清单

你只需要一个Python 3.9以上的环境,然后安装openai和numpy这两个库。openai这个SDK现在被很多大模型服务用作标准接口格式,绝大多数模型服务都兼容它,你用哪家服务就把base_url换成哪家的就行。

pip install openai numpy

然后创建配置文件。这里的重点是我把API地址和模型名都抽成了变量,换服务商时只需要改这一处,不用动后面的逻辑。

import os import numpy as np from openai import OpenAI # ====== 配置区:只需要改这一块 ====== API_KEY = "sk-你的密钥" # 换成你自己的API Key BASE_URL = "https://你的服务地址/v1" # 兼容OpenAI格式的服务地址 EMBEDDING_MODEL = "text-embedding-你的嵌入模型名" # 嵌入模型 LLM_MODEL = "你的对话模型名" # 生成模型 # ===================================== client = OpenAI(api_key=API_KEY, base_url=BASE_URL)

嵌入模型和对话模型通常不是同一个,这点新手很容易搞混。嵌入模型负责把文字变成向量,对话模型负责读材料、写答案,它们的参数、定价、更新频率都不一样,分开配置是行业里约定俗成的做法。

3.3 文档处理:加载与切块

第一步是把原始文档读进来。为了不把示例搞复杂,我先用txt和Markdown格式做演示,PDF、Word的解析后面单独讲。读进来之后,先做基本清洗:把多余空行、无意义的制表符去掉,保留段落结构。

def load_document(file_path: str) -> str: with open(file_path, "r", encoding="utf-8") as f: text = f.read() # 简单清洗:压缩多余空行 lines = [line.strip() for line in text.splitlines() if line.strip()] return "\n".join(lines) def chunk_text(text: str, chunk_size: int = 300, overlap: int = 50) -> list[str]: chunks = [] start = 0 text_len = len(text) while start < text_len: end = min(start + chunk_size, text_len) # 尽量在句子边界切分,避免从词语中间硬切 if end < text_len: last_break = max(text.rfind("。", start, end), text.rfind("\n", start, end)) if last_break > start + chunk_size * 0.6: end = last_break + 1 chunks.append(text[start:end]) start = end - overlap return chunks

切块的代码我做了两个细节处理,这里拆开讲一下。

第一,重叠机制。start不是从end开始的,而是从end减去overlap开始的,这保证了前一块的尾部和后一块的开头有50个字符的重叠,语义不至于断在中间。第二,更聪明的切法。我不会硬按300字符切,而是优先找离目标位置最近的句号或换行符作为实际切分点,条件是它不能太靠前,否则这块会太短。这样切出来的片段基本都在完整句子处断开,检索和阅读体验都好很多。

3.4 向量化与存储检索

接下来实现嵌入和检索。先写一个统一的嵌入函数,后面建库和查库都用它:

def get_embedding(text: str) -> list[float]: resp = client.embeddings.create(model=EMBEDDING_MODEL, input=[text]) return resp.data[0].embedding

然后是向量存储。教学阶段我不用任何数据库,直接用Python列表在内存里存,配合numpy算余弦相似度。这样你能亲眼看到检索的全部计算过程。

class VectorStore: def __init__(self): self.texts = [] self.vectors = [] def add(self, text: str, vector: list[float]): self.texts.append(text) self.vectors.append(np.array(vector, dtype=np.float32)) def search(self, query_vector: list[float], top_k: int = 5): query = np.array(query_vector, dtype=np.float32) scores = [] for vec in self.vectors: # 余弦相似度:点积除以两个向量模长的乘积 cos = np.dot(query, vec) / (np.linalg.norm(query) * np.linalg.norm(vec) + 1e-9) scores.append(cos) top_idx = np.argsort(scores)[-top_k:][::-1] return [(self.texts[i], scores[i]) for i in top_idx]

这段代码里那个+1e-9是个极小值,作用是防止向量模长为0时除零报错。实际业务里模长为零的极端情况很少,但防御性写法永远是值得的。

接着写建库流程。遍历每个切块、调嵌入接口、插入存储,打印进度方便观察:

def build_index(file_path: str) -> VectorStore: store = VectorStore() text = load_document(file_path) chunks = chunk_text(text, chunk_size=300, overlap=50) total = len(chunks) for i, chunk in enumerate(chunks): vec = get_embedding(chunk) store.add(chunk, vec) if (i + 1) % 20 == 0: print(f"进度: {i + 1}/{total}") print(f"建库完成,共 {total} 个片段") return store

3.5 组装Prompt并生成回答

最后一步是查询和生成。这里Prompt怎么写,直接决定了回答质量,我把我常用的模板贴出来并且解释每一行是干嘛的。

def ask(store: VectorStore, question: str, top_k: int = 5) -> str: # 1. 先把问题转成向量 q_vec = get_embedding(question) # 2. 检索最相关的片段 results = store.search(q_vec, top_k=top_k) # 3. 把片段拼成参考资料 context = "\n\n".join( [f"[片段{i + 1}]\n{text}" for i, (text, score) in enumerate(results)] ) # 4. 组装Prompt,把上下文和问题一起交给模型 prompt = f"""你是一个知识库问答助手。请仅根据以下参考资料回答用户的问题,并遵守规则: 1. 参考资料中没有的信息,明确回答“资料中未提及”,不要编造。 2. 回答尽可能引用参考资料中的原文,保持严谨。 3. 如果资料中存在矛盾,在回答中指出并说明差异。 【参考资料】 {context} 【用户提问】 {question} """ resp = client.chat.completions.create( model=LLM_MODEL, messages=[{"role": "user", "content": prompt}], temperature=0.3, ) return resp.choices[0].message.content

选temperature=0.3是有讲究的。RAG场景我们要的是“忠于资料”,不是“天马行空”,所以把温度降下来,让模型更保守、更贴原文。如果把它调到0.9,模型就可能会开始“自由发挥”,幻觉概率明显上升。

整个流程串起来就是这几行代码:

if __name__ == "__main__": store = build_index("product_manual.md") # 建库 answer = ask(store, "产品的保修政策是什么?") # 提问 print(answer)

我拿一份真实风格的内部产品手册测试过,问“设备支持的最大湿度范围”,它能直接引用手册里的“相对湿度10%到90%”原话,并且不额外发挥。这个最小实现,已经足够作为很多轻量级工具的原型了。

4. 从手搓到工程化:向量数据库与框架方案

4.1 什么时候该换掉内存版存储

手搓版本有一个明显瓶颈:所有向量存在内存里,进程一关数据就没了,而且数据量一大检索就慢。如果你的目标是做一个一次性的脚本工具,这个版本完全够用。但如果你要做成常驻服务、知识库上千条文档,就得换成真正的向量数据库。

主流的向量数据库方案,我整理了一个对比表格:

方案部署难度适合场景特点
Chroma低中小型项目、学习原型纯Python库,代码量极小,适合从手搓版本平滑迁移
FAISS中百万级向量的离线检索底层是索引算法库,速度快,但不自带服务能力
Milvus高大规模线上系统分布式架构,功能全,运维成本高
Elasticsearch中已有ES体系、需要全文检索混合既能向量检索又能关键词检索,适合已有ES的团队

入门阶段我推荐Chroma,因为它最接近“能跑的本地缓存数据库”,安装和使用都简单。把上面的VectorStore换成Chroma,代码如下:

import chromadb client_db = chromadb.Client() collection = client_db.create_collection(name="knowledge_base") # 建库 collection.add( ids=[str(i) for i in range(len(chunks))], documents=chunks, embeddings=[get_embedding(c) for c in chunks], ) # 查询 result = collection.query(query_embeddings=[get_embedding(question)], n_results=5)

注意,我不建议在真实生产系统里每条数据都现场调嵌入接口,那既慢又费钱。规范做法是把嵌入结果缓存起来,或者用离线脚本批量算好再入库。

4.2 LangChain这类框架要不要学

我的立场是:框架要学,但排在手搓之后。LangChain的价值在于把加载、切块、向量库、模型调用里面的大量样板代码封装成了标准组件,你在手搓阶段理解了原理,再看框架就不会被抽象的类和方法绕晕。

一个典型的LangChain RAG流程,核心也就这么几行:

from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma loader = TextLoader("product_manual.md") documents = loader.load() splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50) docs = splitter.split_documents(documents) vectorstore = Chroma.from_documents(docs, embedding=你的嵌入函数) retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

你仔细看,RecursiveCharacterTextSplitter的chunk_size和chunk_overlap,就是我手搓版本里的那两个参数,只是换了皮。这也是我一直坚持手搓先行的原因:框架只是帮你省了打字的功夫,没有帮你解决任何“思考”的问题。参数怎么设、为什么这样设,还是得靠你对原理的理解。

4.3 文档格式多样的处理方案

很多新手跑完我的示例后,第一句话是“我的文档是PDF怎么办”。这里给出最常见三种格式的处理思路:PDF这类复杂排版文档,优先用专门的解析工具(比如把PDF转成文本或按页提取,老式扫描件还得走OCR);Word文档可以直接按原生格式提取段落;网页内容则要抓取后清洗HTML标签和脚本代码,只保留正文。

在这个环节上最容易犯的错误是“贪全”。有的同事上来就上一个重量级解析框架,结果部署环境依赖冲突查了一整天。我建议按需引入,80%的场景其实是txt、Markdown和结构化JSON,先把这三类搞定,再逐步加复杂格式。

5. 实操中的坑与排查技巧

5.1 检索效果差:从四个环节逐个排查

新手跑通RAG后,第一个反馈往往是“问几个问题还行,但有些问题答得很奇怪”。这时候别急着怪模型,先按下面的顺序排查。

先看检索到的片段是不是真的相关。你可以把检索返回的原文和相似度分数打印出来,这是最快的一次定位。如果片段相关度分数普遍低于0.65,那问题就出在检索环节:检查是不是切块太小导致每块都只含半句话;再看看是不是嵌入了太多无意义内容;最后确认嵌入模型和你的领域是否匹配。

再看是不是问题里包含大量专业行话。行业缩写、产品型号、内部黑话,通用嵌入模型可能没学过,这时候可以试试往问题里加几个“常见别名”,或者在建库时就把同义词写进文档里。这些做法听着土,但实测对中文行业场景特别有效。

如果检索出来的片段本身是相关的,但答案是错的,那就是生成环节的问题。先说结论:把Prompt里的“仅根据资料回答”改成“如果资料不足,请明说”,这一句话通常就能挽救一半的错误回答。剩下的可能是Top-K太小,模型只能看到部分材料,遇到复杂问题把K调大一点。

5.2 常见问题的速查表

我把过去半年在多个项目里踩过的坑整理成一张表,这张表你在调RAG时可以直接当排查手册用:

现象可能原因处理办法
检索的片段牛头不对马嘴切块太大或切点乱调小chunk_size,按语义边界切分
相似度分数普遍很低嵌入模型选型差换专门针对中文优化的嵌入模型
回答内容对但引用错误模型混淆了多个片段Prompt要求标注[片段编号],引用来源
回答编造资料外的内容温度过高或Prompt约束弱调低temperature,加强“资料不足请明说”
大文档检索慢全量扫描换FAISS或Chroma,支持索引加载
更新文档后回答没变化没有重新建库/查了旧缓存检查入库流程是否有缓存逻辑
相同问题答案经常变温度偏高RAG场景建议temperature不超过0.5

这里我想重点强调一个前面没细讲的概念:文本切块不是一次性就能定好的,它需要根据文档语气和问题类型反复调。我做过一个内部设备手册案例,第一版切块400字,问“设备报警后怎么复位”这种跨多段的复合问题时,模型回答总是少步骤。我把切块改成200字且增加重叠到80,并把Top-K从5调到8后,模型能看到更细碎的步骤片段,回答的完整度明显提升。数据工程就是这样,不是说参数一遍跑通就不用管了,而是要回到“模型到底需要看到什么”这个根本问题。

5.3 关于成本的一个提醒

RAG的成本大头是嵌入接口的调用,而不是对话生成。很多人不知道,嵌入一个300字的片段,和嵌入一篇5000字的长文,消耗的token量差十几倍,但实际质量未必更好。我的经验是先切块再嵌入,逐块计费,你会发现成本比直接全文嵌入省得多。

另外,如果知识库内容很少变化,可以把嵌入结果持久化到本地文件,启动时直接加载,不用每次都重新调接口。我试过最狠的方案是把1000条片段的向量序列化成一个npy文件,加载只需要几秒钟,成本几乎为零。这个技巧对个人项目和中小团队非常友好,可以放心抄走。

6. 我的几点体会

技术文章写到这里,按惯例该“总结”了。但我想说点更实在的:项目做多了之后,我对RAG最大的体会是“检索为王”这四个字。

模型本身的能力、Prompt的写法当然重要,但你回想一下整个流程,决定答案上限的,永远是模型“看到了什么”。同样一个大模型,给它的材料如果精准,它可以表现得像专家;材料一旦给错了,它会用非常专业的语气说出完全错误的东西。这个认知帮我解决了许多调试难题——效果不好,我第一反应永远先去看检索日志,而不是换模型、调样例。

第二个体会是:做RAG项目,数据整理至少占六成时间。很多同事抱怨RAG效果不行,我一看他的知识库,几百个文件堆在一起,命名混乱、格式五花八门、重复内容到处都是。这种数据,换任何参数都救不回来。RAG学得快,但背地里拼的是你把文档当作产品在做的那份耐心。

最后分享一个小技巧:在你把系统交给别人用之前,先准备一套“基准问题集”。挑十个有代表性的问题,每改一次参数,就跑一遍这十个问题,记录回答质量变化。这是最简单、最有效的回归测试方式,比任何评测框架都实用。我的建议是永远保留这十问的成绩单,你会发现调参从此不再是玄学,而是有数据支撑的工程决策。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询