先把话说在前面如今RAG框架一抓一大把LangChain、LlamaIndex、各种向量数据库开箱即用为什么还要用Python基础库从零手撕一个RAG内核我的回答很直接——框架让你跑通demo但不会让你真正懂RAG。等你被检索错位坑过一次、等你需要把检索逻辑改到贴合自己业务的时候你就明白亲手写过一遍底层链路有多值钱。这篇文章不依赖任何第三方算法库只用Python标准库和手写逻辑把文档切分、向量化、相似度检索、提示词拼装、大模型接入整个Pipeline拆开揉碎。适合刚入门RAG、被框架黑盒劝退以及想把检索逻辑做深度定制的同学。读完你能得到一个可运行的迷你RAG同时拥有自己改内核的能力。1. 为什么要手撕RAG内核1.1 框架黑盒带来的三个痛点用LangChain跑一个RAG demo快的时候十分钟就能看到一个会引用的ChatBot。但问题在于框架把太多关键决策写死在内部文本怎么切、向量怎么算、相似度怎么排、Top-K怎么取都是默认的正确答案。一旦你换了一个垂直领域的数据比如合同条款、产品说明书、客服工单这些默认参数往往会失灵而你连从哪里调都未必找得清。第二个痛点是排查困难。RAG出问题大家习惯先怀疑大模型最后换模型、调temperature折腾一圈才发现是切分环节把关键句子拦腰截断了。没有底层实现的概念你连这个问题属于哪一层都判断不了只能黑盒调参靠感觉碰运气。第三个痛点是定制受限。真实项目里RAG不是固定流水线经常要动态调整检索范围、给不同来源赋予不同权重、在检索结果上做二次过滤。这些需求到框架里通常要改源码或写一堆Callback而你一旦手写过一遍会发现这类改造就是改两个函数的事。1.2 手撕RAG的最短学习路径我不建议你一开始就追求高精度、海量数据、混合检索。最务实的路径是先用纯基础库把能跑通、能看到中间结果、能随手改的迷你RAG做出来再逐步向生产级方案迁移。这条路径可以分为四个阶段第一理解RAG由哪几个独立环节组成每个环节解决什么问题第二用标准库分别实现切分、向量化、检索和提示词拼装第三用一个本地大模型或兼容API把整条链路跑起来第四观察检索失败案例理解每个参数为什么影响结果。这个过程看起来比套框架慢但收获是框架给不了的。等你以后再接触LangChain、向量数据库看到的不再是黑盒而是一层层可拆的组件定位问题会快很多。2. RAG内核的整体设计2.1 RAG不是向量搜索是开卷考试很多人把RAG等同于向量数据库加相似度检索这是最常见的误区。检索只是手段RAG的目标是给大模型提供证据让它基于给定资料输出而不是基于训练时记住的知识瞎猜。我习惯用一个开卷考试的类比。普通大模型是闭卷考试训练时没见过的私有知识、最新数据它只能靠猜测作答所以容易一本正经地编造。RAG相当于给大模型配了一份参考资料考试时可以翻书。翻书的前提是你知道书在哪一章这就是检索。但翻到正确的页码和答出正确的答案是两回事后者仍然由大模型完成。所以RAG的内核不是向量搜索而是检索 增强的完整链路检索负责把相关资料捞出来增强负责把资料组织成有用上下文。2.2 五个模块的职责划分一个不依赖框架的RAG内核至少要有五个模块文档加载读入本地文本、PDF本文先以纯文本为主或数据库里的内容把原始数据变成干净字符串。文本切分把长文档切成能放进上下文窗口的块同时控制块与块之间的重叠避免语义被切断。向量化把每个块转换成一个可计算相似度的向量。这里要选一种不需要深度学习也能实现的方案。检索对用户query做同样的向量化然后在所有块向量里找最相似的结果返回Top-K。生成把检索出来的块拼成提示词连同原始问题一起交给大模型由大模型生成最终答案。每个模块之间最好用函数隔离输入输出清晰。这样以后你想替换任何一环只需要换函数而不影响其他环节。我后面给的代码就是这个思路。2.3 为什么我用TF-IDF而不是向量数据库市面上主流方案是用Embedding模型生成稠密向量再用FAISS、Milvus这类向量库做近似检索。这类方案效果确实好但不符合基础库手撕的主题。我用的是TF-IDF加余弦相似度它本质上是把文本变成稀疏向量再用点积算相似度整个过程只需要标准库里的math和re。TF-IDF的优点有三个一是零依赖不用装torch、transformers二是可解释性强你能清楚看到每个词在向量里占多少权重三是在特定领域、专业术语多但语料量不大的场景下它不一定比预训练embedding差多少。缺点是它对语义近义词无能为力用户问怎么退货知识库里写的是退款流程字面上匹配不上这是后期需要升级embedding的动机。但先用TF-IDF把整条链路跑通、看清RAG的逻辑骨架再替换向量化方案学习成本比一上来就调大模型低得多。3. 用Python基础库实现RAG内核3.1 文档加载与清洗第一个环节是读文档。真实场景里文档可能是PDF、Word、Excel但先从纯文本开始最能看清问题。文件编码是常见坑中文文本最好指定utf-8。import re def read_document(path: str) - str: with open(path, r, encodingutf-8) as f: text f.read() # 把不同换行符统一 text text.replace(\r\n, \n).replace(\r, \n) # 清理多余空行和首尾空白 text re.sub(r\n{3,}, \n\n, text).strip() return text清洗这一步往往被低估。如果文档里全是空行、页码、页眉页脚切出来的块会包含大量噪声检索权重被无关词干扰。我一般还会根据文档类型做额外处理比如合同类文本保留条款编号说明书类文本去掉重复的品牌名。清洗原则是尽量去掉不参与语义判断的模板化内容但别破坏正文结构。3.2 分块策略重叠窗口不能省接下来是切分。一个常见的笨办法是固定长度硬切不考虑语义边界。在基础版里可以接受但必须加重叠窗口。原因很简单如果某段关键内容横跨两个块的边界没有重叠这段内容会被切成两半检索时无论哪一块都不包含完整信息大模型拿到的就是残缺证据。def chunk_text(text: str, chunk_size: int 400, overlap: int 100) - list: if chunk_size overlap: raise ValueError(chunk_size必须大于overlap) chunks [] start 0 while start len(text): end start chunk_size chunks.append(text[start:end]) if end len(text): break start end - overlap return chunks这里的chunk_size指字符数英文可能按token算中文用字符数也够用。重叠率通常取10%到30%之间。比如chunk_size400、overlap100那么每移动300字符就产生一个新块相邻块之间有100字符重复这样一段300字符左右的语义内容至少能完整落在某一个块里。如果知识库文档结构很强比如每个条款本身就是语义单元则应该优先按段落切再对过长段落做二次固定长度切分这种混合策略比纯固定长度靠谱。3.3 中文分词与停用词的一个折中方案分词是中文RAG绕不开的问题。Python标准库里没有中文分词器但我们可以用字和相邻双字组合做一个轻量方案既避免引入jieba又能保留部分词语边界信息。STOPWORDS set(的了是在我你有他她它这那一也不就都而及与和或) def tokenize(text: str) - list: text re.sub(r[^\w\u4e00-\u9fff], , text.lower()) tokens [] # 单字 tokens.extend([ch for ch in text if ch not in STOPWORDS]) # 相邻双字包含一个非常小的自定义词典放行 tokens.extend([text[i:i2] for i in range(len(text) - 1)]) return tokens这个方案的核心是bigram。单个汉字有时候歧义太大比如银行拆成银和行跟行走的行在向量空间里会打架银行作为bigram出现时区分度就高很多。停用词表我故意写得很短因为中文里大量单字词去掉后反而更准。缺点很明显bigram数量多词表膨胀而且人工智这类残缺组合也会出现。想提高质量可以放行一组高频自定义词或者直接换jieba。这一步是允许用第三方库的第一个合理升级点。3.4 TF-IDF权重计算有了分词结果就可以计算每个词在语料库中的权重。TF-IDF由两部分组成。TF部分也就是词频表示一个词在当前块里出现多少次。直观理解一块文本里反复出现退货说明这块内容很可能是讲退货的。IDF部分也就是逆文档频率表示一个词在所有块中有多常见。比如我们在每块都出现区分度很低质粒只在少数生物类块里出现区分度就很高。两者相乘就是词的权重。import math def build_idf(chunks: list) - dict: df {} for chunk in chunks: seen set(tokenize(chunk)) for token in seen: df[token] df.get(token, 0) 1 total len(chunks) idf {} for token, freq in df.items(): # 加1平滑防止分母为0 idf[token] math.log((total 1) / (freq 1)) 1.0 return idf注意这里的应用有个微妙点IDF应该基于你在检索时能看到的全部候选块计算而不是只基于某一篇文档。如果每篇文档都单独算IDF跨文档检索时权重尺度会不一致。另外IDF公式里的1.0是让即使高频词也保留基本权重不至于让普通词彻底消失。这个微小设计在后续调优中会体现价值。3.5 向量化与余弦相似度检索TF-IDF的权重算出来后每个块可以表示成一个固定长度的稀疏向量向量维度等于全量词汇表大小。具体做法是先统计所有文档里出现过的token建立一个词表索引然后对每个块把token出现的次数转成TF-IDF值再对向量做L2归一化。def build_vocab(chunks: list) - dict: vocab {} for chunk in chunks: for token in tokenize(chunk): if token not in vocab: vocab[token] len(vocab) return vocab def vectorize(text: str, vocab: dict, idf: dict) - list: vec [0.0] * len(vocab) for token in tokenize(text): if token in vocab: idx vocab[token] vec[idx] 1.0 # 乘IDF for idx, val in enumerate(vec): if val: vec[idx] val * idf.get(list(vocab.keys())[idx], 1.0) # L2归一化 norm math.sqrt(sum(x * x for x in vec)) if norm 0: return vec return [x / norm for x in vec]这段代码有一个效率问题vocab.keys()在循环里反复调用很慢实际写的时候单独建一个token列表就行。但逻辑很直白适合理解。归一化是为了让后续计算简化为点积。余弦相似度衡量的是两个向量方向是否一致与向量长度无关。归一化后每个向量的模都是1两个向量的余弦相似度正好等于点积省去每次算模的运算。检索时对query做同样的向量化然后和所有块向量点积排序取前K。def search(query: str, chunk_vecs: list, vocab: dict, idf: dict, top_k: int 3): q_vec vectorize(query, vocab, idf) scored [] for idx, c_vec in enumerate(chunk_vecs): score sum(q_vec[d] * c_vec[d] for d in range(len(q_vec))) scored.append((score, idx)) scored.sort(keylambda x: x[0], reverseTrue) return scored[:top_k]这段代码看着简单它已经是完整的暴力检索内核。数据量几千块以内这种全表扫描完全够用没必要上索引。我发现很多初学者一上来就上FAISS结果是索引概念没理解排查还多一层复杂度。先把暴力检索跑通再谈优化。3.6 提示词拼装与大模型接入检索这一步出结果后要把它转成上下文。提示词设计直接决定大模型会不会胡说、会不会忽视你给的材料。我的拼装习惯是先声明只能根据资料回答再列出检索块最后给出用户问题。还要把块来源带上哪怕只是块编号模型也能建立一个参考文献的感觉。def build_prompt(context: str, query: str) - str: return ( 请根据以下资料回答用户问题如果资料中找不到答案 请直接回答资料中未找到相关信息不要编造。\n\n f资料\n{context}\n\n f用户问题{query}\n 请用中文回答。 )大模型接入有两种典型方式本地Ollama或兼容OpenAI协议的服务。为了让文章保持零第三方库我用urllib.request发HTTP请求不装requests也能跑。import urllib.request import json import os def call_ollama(prompt: str, model: str qwen2.5:7b, host: str http://localhost:11434) - str: payload json.dumps({ model: model, prompt: prompt, stream: False, temperature: 0.2 }).encode(utf-8) req urllib.request.Request( f{host}/api/generate, datapayload, headers{Content-Type: application/json} ) with urllib.request.urlopen(req, timeout120) as resp: data json.loads(resp.read().decode(utf-8)) return data[response]如果用的是OpenAI兼容接口代码也差不多。密钥建议走环境变量别硬编码在代码里。def call_openai_compatible(prompt: str, model: str gpt-4o-mini) - str: api_key os.environ.get(OPENAI_API_KEY) base_url os.environ.get(OPENAI_BASE_URL, https://api.openai.com/v1/chat/completions) payload json.dumps({ model: model, messages: [ {role: system, content: 你是严谨的资料助手。}, {role: user, content: prompt} ], temperature: 0.2 }).encode(utf-8) req urllib.request.Request( base_url, datapayload, headers{Content-Type: application/json, Authorization: fBearer {api_key}} ) with urllib.request.urlopen(req, timeout60) as resp: data json.loads(resp.read().decode(utf-8)) return data[choices][0][message][content]3.7 一条命令跑通全流程把前面的函数串起来整个RAG内核就完整了。我专门写了一个入口函数方便你在命令行里测不同的query。def run_rag(query: str, doc_path: str, chunk_size: int 400, overlap: int 100, top_k: int 3, use_llm: bool True): doc read_document(doc_path) chunks chunk_text(doc, chunk_size, overlap) vocab build_vocab(chunks) idf build_idf(chunks) chunk_vecs [vectorize(c, vocab, idf) for c in chunks] scored search(query, chunk_vecs, vocab, idf, top_k) print(检索结果) for score, idx in scored: print(fscore{score:.4f} chunk#{idx}: {chunks[idx][:80]}...) context \n\n---分割线---\n\n.join(chunks[idx] for _, idx in scored) prompt build_prompt(context, query) if use_llm: return call_ollama(prompt) return prompt, scored现在的RAG内核已经具备全部基本要素加载、切分、词汇表、IDF、向量化、暴力检索、提示词构造、大模型生成。建议你先把use_llm设为False跑几次看搜索结果和prompt长什么样再打开大模型开关。这个先看检索再看生成的调试习惯能省掉大量排查时间。4. 参数调优与问题排查4.1 chunk_size和overlap怎么调没有万能参数但有一个靠谱的调试方法拿三个真实query做测试集固定其他参数分别跑chunk_size100、300、500看检索命中的chunk是否真的覆盖了答案。chunk_size太小比如50个字符切出来的块可能只有一两个短句虽然聚焦但上下文不完整大模型拿到碎片信息容易抓不住因果。chunk_size太大比如2000个字符每块包含大量无关噪声向量的主题不突出相似度分数会被高频词带偏。经验值我建议先从300到500起调。overlap的直接作用是把边界信息带进相邻块。如果overlap设为0两块边界处的事实会丢失如果overlap太大比如占到chunk_size的一半会产生大量冗余块检索时重复内容多索引体积也膨胀。我通常按chunk_size的15%到25%设置重叠。4.2 空向量与零分检索的兜底策略最典型的问题是query做向量化后可能是一个全零向量。为什么会这样因为query中的词一个都没出现在词汇表里比如用户用英文缩写SKU提问知识库文里全写的是商品编号。此时点积全是0排序等于随机检索直接失效。遇到这种情况有两个兜底策略。第一检索得分低于阈值的块直接丢弃不能把弱相关块硬塞给大模型。第二建立一个同义词或别名表在query进入检索前做归一化比如SKU自动扩展成SKU 商品编号。还有更简单的一招对query做字符级拆分而不是只做bigram。这样即使没见过退货这个词也至少能匹配到退和货的单字不至于完全失效。基础版里我用单字bigram混合分词很大程度就是为了应对这种冷启动问题。4.3 从结果反推切分与权重的bugRAG调试最有价值的技巧是不要只盯着最终答案要把检索分数、块内容原样打印出来看。我在3.7里就特意保留了print这一步不是可选项而是必须看的中间产物。有几种异常表现对应不同病因检索分数普遍很低比如全部低于0.1说明query与知识库词汇重叠度太低优先检查分词和清洗。分数很高但捞上来的块内容跟问题不相关可能是chunk_size太大导致向量被高频词主导也可能是停用词放行了一些泛化词。多次query检索下来命中的chunk总是同一段往往是文档里存在大量重复模板内容IDF没能把这类重复词压下去。有一次我遇到检索分值都在0.9以上但答案完全错误打印chunk才发现整份合同里甲方和乙方密布TF-IDF分词把这两个词当成强特征可问题是用户问的是付款条件。后来我把类似甲方乙方的泛称加入停用词并把IDF的计算单位从全部块调整为过滤重复段落后的块问题立刻改善。4.4 常见错误速查表现象原因排查/解决方式检索所有分数都是0query词表与知识库词汇完全不重合检查query分词考虑同义词扩展或改用embedding建词表时内存爆炸bigram词表过大且没去停用词过滤单字噪声控制最短词长或改用jiebachunk之间大量重复内容overlap设置过大降低overlap到chunk_size的15%左右大模型引用不存在的细节检索块被硬切语义不完整增加overlap或按段落边界优先切分Ollama调用超时本地模型首次加载冷启动先提前跑一次模型加载再调用API答案太长乱编提示词没有禁止编造在build_prompt里明确限定范围并强制声明资料外不回答5. 从基础库版本升级到生产环境的路线5.1 替换向量化从TF-IDF到语义Embedding手写基础版跑通后你会发现整套代码真正的短板不是相似度计算而是TF-IDF不懂语义。想升级最好保留前面的文档加载、切分、检索排序和提示词构建逻辑只替换vectorize这一个环节。现在用sentence-transformers或OpenAI的Embedding接口都行。核心改动很小把chunk和query分别编码成稠密向量再用同样的余弦检索逻辑。关键是要保持先切分后向量化再做Top-K的架构不变。这样你既享受embedding的语义匹配能力又保留了自己对切分和上下文拼接的完全控制权。有人会问既然要换embedding当初为什么不直接上因为直接上有两个坏处一是你会把所有注意力放在调模型上完全没理解切分和检索权重对结果的影响二是embedding模型的输出是黑盒出了问题你很难判断是模型问题还是分块问题。基础版跑通后你已经能把问题定位到具体环节再换embedding就是真正的优化而不是换一个更大的黑盒。5.2 索引升级与重排数据量超过几万块时暴力检索的线性扫描会明显变慢。这个时候再引入ANN索引才有意义比如FAISS。我建议把索引作为独立模块抽象出来输入仍然是chunk向量列表接口继续暴露search(query, top_k)。这样你的业务代码完全不需要感知底层是暴力扫描还是HNSW图索引。生产级RAG通常还会加一层重排。Top-K检索出来的结果大概率已经把相关块捞出来了但顺序不一定理想。这时用一个更强的模型对Top-K再做一次精排取前N个进入提示词。基础版里这一步对应的是scored.sort(...)后面的环节你完全可以在这里插入一个重排函数而不影响其他逻辑。5.3 给你的RAG加个持久化缓存最后提一个很实用的扩展我觉得RAG的持久化经常被忽略。基础版每次启动都要重新切分、建立IDF、向量化全部文本数据量一大就很浪费。最简单的做法是把vocab、idf、chunk列表和chunk_vec用json/文件保存下来下次启动直接加载。这一步能让手搓RAG从玩具变成一个可日常使用的工具。我实测过一个小知识库文档大约10万字全流程向量化大约一秒多本机内存完全够用。再加上缓存后启动时间缩短到0.2秒左右。很多在线的知识库工具其实也就是这个内核外面套了一层Web界面核心原理和你刚手撕的这部分没有本质区别。个人在实际项目里踩坑最多的永远是分块和权重而不是代码逻辑。我强烈建议你做一个调试小工具输入一个query同时打印检索分数、命中的块原文和答案然后反复调整chunk_size、overlap、停用词。别急着上大模型和向量数据库先用一两个真实问题把检索调到哪怕不看答案也能猜到它该命中哪一块再谈模型效果。如果你动手写完这套基础版再去翻任何RAG框架那些概念就不再是抽象黑盒了。遇到问题你也会第一个想到去查自己最熟的那一层逻辑。 
长河君 · 2026/10/3 9:58:39 · 阅读 1.2k 从零手撕RAG内核：用Python标准库实现检索增强生成全流程
先把话说在前面如今RAG框架一抓一大把LangChain、LlamaIndex、各种向量数据库开箱即用为什么还要用Python基础库从零手撕一个RAG内核我的回答很直接——框架让你跑通demo但不会让你真正懂RAG。等你被检索错位坑过一次、等你需要把检索逻辑改到贴合自己业务的时候你就明白亲手写过一遍底层链路有多值钱。这篇文章不依赖任何第三方算法库只用Python标准库和手写逻辑把文档切分、向量化、相似度检索、提示词拼装、大模型接入整个Pipeline拆开揉碎。适合刚入门RAG、被框架黑盒劝退以及想把检索逻辑做深度定制的同学。读完你能得到一个可运行的迷你RAG同时拥有自己改内核的能力。1. 为什么要手撕RAG内核1.1 框架黑盒带来的三个痛点用LangChain跑一个RAG demo快的时候十分钟就能看到一个会引用的ChatBot。但问题在于框架把太多关键决策写死在内部文本怎么切、向量怎么算、相似度怎么排、Top-K怎么取都是默认的正确答案。一旦你换了一个垂直领域的数据比如合同条款、产品说明书、客服工单这些默认参数往往会失灵而你连从哪里调都未必找得清。第二个痛点是排查困难。RAG出问题大家习惯先怀疑大模型最后换模型、调temperature折腾一圈才发现是切分环节把关键句子拦腰截断了。没有底层实现的概念你连这个问题属于哪一层都判断不了只能黑盒调参靠感觉碰运气。第三个痛点是定制受限。真实项目里RAG不是固定流水线经常要动态调整检索范围、给不同来源赋予不同权重、在检索结果上做二次过滤。这些需求到框架里通常要改源码或写一堆Callback而你一旦手写过一遍会发现这类改造就是改两个函数的事。1.2 手撕RAG的最短学习路径我不建议你一开始就追求高精度、海量数据、混合检索。最务实的路径是先用纯基础库把能跑通、能看到中间结果、能随手改的迷你RAG做出来再逐步向生产级方案迁移。这条路径可以分为四个阶段第一理解RAG由哪几个独立环节组成每个环节解决什么问题第二用标准库分别实现切分、向量化、检索和提示词拼装第三用一个本地大模型或兼容API把整条链路跑起来第四观察检索失败案例理解每个参数为什么影响结果。这个过程看起来比套框架慢但收获是框架给不了的。等你以后再接触LangChain、向量数据库看到的不再是黑盒而是一层层可拆的组件定位问题会快很多。2. RAG内核的整体设计2.1 RAG不是向量搜索是开卷考试很多人把RAG等同于向量数据库加相似度检索这是最常见的误区。检索只是手段RAG的目标是给大模型提供证据让它基于给定资料输出而不是基于训练时记住的知识瞎猜。我习惯用一个开卷考试的类比。普通大模型是闭卷考试训练时没见过的私有知识、最新数据它只能靠猜测作答所以容易一本正经地编造。RAG相当于给大模型配了一份参考资料考试时可以翻书。翻书的前提是你知道书在哪一章这就是检索。但翻到正确的页码和答出正确的答案是两回事后者仍然由大模型完成。所以RAG的内核不是向量搜索而是检索 增强的完整链路检索负责把相关资料捞出来增强负责把资料组织成有用上下文。2.2 五个模块的职责划分一个不依赖框架的RAG内核至少要有五个模块文档加载读入本地文本、PDF本文先以纯文本为主或数据库里的内容把原始数据变成干净字符串。文本切分把长文档切成能放进上下文窗口的块同时控制块与块之间的重叠避免语义被切断。向量化把每个块转换成一个可计算相似度的向量。这里要选一种不需要深度学习也能实现的方案。检索对用户query做同样的向量化然后在所有块向量里找最相似的结果返回Top-K。生成把检索出来的块拼成提示词连同原始问题一起交给大模型由大模型生成最终答案。每个模块之间最好用函数隔离输入输出清晰。这样以后你想替换任何一环只需要换函数而不影响其他环节。我后面给的代码就是这个思路。2.3 为什么我用TF-IDF而不是向量数据库市面上主流方案是用Embedding模型生成稠密向量再用FAISS、Milvus这类向量库做近似检索。这类方案效果确实好但不符合基础库手撕的主题。我用的是TF-IDF加余弦相似度它本质上是把文本变成稀疏向量再用点积算相似度整个过程只需要标准库里的math和re。TF-IDF的优点有三个一是零依赖不用装torch、transformers二是可解释性强你能清楚看到每个词在向量里占多少权重三是在特定领域、专业术语多但语料量不大的场景下它不一定比预训练embedding差多少。缺点是它对语义近义词无能为力用户问怎么退货知识库里写的是退款流程字面上匹配不上这是后期需要升级embedding的动机。但先用TF-IDF把整条链路跑通、看清RAG的逻辑骨架再替换向量化方案学习成本比一上来就调大模型低得多。3. 用Python基础库实现RAG内核3.1 文档加载与清洗第一个环节是读文档。真实场景里文档可能是PDF、Word、Excel但先从纯文本开始最能看清问题。文件编码是常见坑中文文本最好指定utf-8。import re def read_document(path: str) - str: with open(path, r, encodingutf-8) as f: text f.read() # 把不同换行符统一 text text.replace(\r\n, \n).replace(\r, \n) # 清理多余空行和首尾空白 text re.sub(r\n{3,}, \n\n, text).strip() return text清洗这一步往往被低估。如果文档里全是空行、页码、页眉页脚切出来的块会包含大量噪声检索权重被无关词干扰。我一般还会根据文档类型做额外处理比如合同类文本保留条款编号说明书类文本去掉重复的品牌名。清洗原则是尽量去掉不参与语义判断的模板化内容但别破坏正文结构。3.2 分块策略重叠窗口不能省接下来是切分。一个常见的笨办法是固定长度硬切不考虑语义边界。在基础版里可以接受但必须加重叠窗口。原因很简单如果某段关键内容横跨两个块的边界没有重叠这段内容会被切成两半检索时无论哪一块都不包含完整信息大模型拿到的就是残缺证据。def chunk_text(text: str, chunk_size: int 400, overlap: int 100) - list: if chunk_size overlap: raise ValueError(chunk_size必须大于overlap) chunks [] start 0 while start len(text): end start chunk_size chunks.append(text[start:end]) if end len(text): break start end - overlap return chunks这里的chunk_size指字符数英文可能按token算中文用字符数也够用。重叠率通常取10%到30%之间。比如chunk_size400、overlap100那么每移动300字符就产生一个新块相邻块之间有100字符重复这样一段300字符左右的语义内容至少能完整落在某一个块里。如果知识库文档结构很强比如每个条款本身就是语义单元则应该优先按段落切再对过长段落做二次固定长度切分这种混合策略比纯固定长度靠谱。3.3 中文分词与停用词的一个折中方案分词是中文RAG绕不开的问题。Python标准库里没有中文分词器但我们可以用字和相邻双字组合做一个轻量方案既避免引入jieba又能保留部分词语边界信息。STOPWORDS set(的了是在我你有他她它这那一也不就都而及与和或) def tokenize(text: str) - list: text re.sub(r[^\w\u4e00-\u9fff], , text.lower()) tokens [] # 单字 tokens.extend([ch for ch in text if ch not in STOPWORDS]) # 相邻双字包含一个非常小的自定义词典放行 tokens.extend([text[i:i2] for i in range(len(text) - 1)]) return tokens这个方案的核心是bigram。单个汉字有时候歧义太大比如银行拆成银和行跟行走的行在向量空间里会打架银行作为bigram出现时区分度就高很多。停用词表我故意写得很短因为中文里大量单字词去掉后反而更准。缺点很明显bigram数量多词表膨胀而且人工智这类残缺组合也会出现。想提高质量可以放行一组高频自定义词或者直接换jieba。这一步是允许用第三方库的第一个合理升级点。3.4 TF-IDF权重计算有了分词结果就可以计算每个词在语料库中的权重。TF-IDF由两部分组成。TF部分也就是词频表示一个词在当前块里出现多少次。直观理解一块文本里反复出现退货说明这块内容很可能是讲退货的。IDF部分也就是逆文档频率表示一个词在所有块中有多常见。比如我们在每块都出现区分度很低质粒只在少数生物类块里出现区分度就很高。两者相乘就是词的权重。import math def build_idf(chunks: list) - dict: df {} for chunk in chunks: seen set(tokenize(chunk)) for token in seen: df[token] df.get(token, 0) 1 total len(chunks) idf {} for token, freq in df.items(): # 加1平滑防止分母为0 idf[token] math.log((total 1) / (freq 1)) 1.0 return idf注意这里的应用有个微妙点IDF应该基于你在检索时能看到的全部候选块计算而不是只基于某一篇文档。如果每篇文档都单独算IDF跨文档检索时权重尺度会不一致。另外IDF公式里的1.0是让即使高频词也保留基本权重不至于让普通词彻底消失。这个微小设计在后续调优中会体现价值。3.5 向量化与余弦相似度检索TF-IDF的权重算出来后每个块可以表示成一个固定长度的稀疏向量向量维度等于全量词汇表大小。具体做法是先统计所有文档里出现过的token建立一个词表索引然后对每个块把token出现的次数转成TF-IDF值再对向量做L2归一化。def build_vocab(chunks: list) - dict: vocab {} for chunk in chunks: for token in tokenize(chunk): if token not in vocab: vocab[token] len(vocab) return vocab def vectorize(text: str, vocab: dict, idf: dict) - list: vec [0.0] * len(vocab) for token in tokenize(text): if token in vocab: idx vocab[token] vec[idx] 1.0 # 乘IDF for idx, val in enumerate(vec): if val: vec[idx] val * idf.get(list(vocab.keys())[idx], 1.0) # L2归一化 norm math.sqrt(sum(x * x for x in vec)) if norm 0: return vec return [x / norm for x in vec]这段代码有一个效率问题vocab.keys()在循环里反复调用很慢实际写的时候单独建一个token列表就行。但逻辑很直白适合理解。归一化是为了让后续计算简化为点积。余弦相似度衡量的是两个向量方向是否一致与向量长度无关。归一化后每个向量的模都是1两个向量的余弦相似度正好等于点积省去每次算模的运算。检索时对query做同样的向量化然后和所有块向量点积排序取前K。def search(query: str, chunk_vecs: list, vocab: dict, idf: dict, top_k: int 3): q_vec vectorize(query, vocab, idf) scored [] for idx, c_vec in enumerate(chunk_vecs): score sum(q_vec[d] * c_vec[d] for d in range(len(q_vec))) scored.append((score, idx)) scored.sort(keylambda x: x[0], reverseTrue) return scored[:top_k]这段代码看着简单它已经是完整的暴力检索内核。数据量几千块以内这种全表扫描完全够用没必要上索引。我发现很多初学者一上来就上FAISS结果是索引概念没理解排查还多一层复杂度。先把暴力检索跑通再谈优化。3.6 提示词拼装与大模型接入检索这一步出结果后要把它转成上下文。提示词设计直接决定大模型会不会胡说、会不会忽视你给的材料。我的拼装习惯是先声明只能根据资料回答再列出检索块最后给出用户问题。还要把块来源带上哪怕只是块编号模型也能建立一个参考文献的感觉。def build_prompt(context: str, query: str) - str: return ( 请根据以下资料回答用户问题如果资料中找不到答案 请直接回答资料中未找到相关信息不要编造。\n\n f资料\n{context}\n\n f用户问题{query}\n 请用中文回答。 )大模型接入有两种典型方式本地Ollama或兼容OpenAI协议的服务。为了让文章保持零第三方库我用urllib.request发HTTP请求不装requests也能跑。import urllib.request import json import os def call_ollama(prompt: str, model: str qwen2.5:7b, host: str http://localhost:11434) - str: payload json.dumps({ model: model, prompt: prompt, stream: False, temperature: 0.2 }).encode(utf-8) req urllib.request.Request( f{host}/api/generate, datapayload, headers{Content-Type: application/json} ) with urllib.request.urlopen(req, timeout120) as resp: data json.loads(resp.read().decode(utf-8)) return data[response]如果用的是OpenAI兼容接口代码也差不多。密钥建议走环境变量别硬编码在代码里。def call_openai_compatible(prompt: str, model: str gpt-4o-mini) - str: api_key os.environ.get(OPENAI_API_KEY) base_url os.environ.get(OPENAI_BASE_URL, https://api.openai.com/v1/chat/completions) payload json.dumps({ model: model, messages: [ {role: system, content: 你是严谨的资料助手。}, {role: user, content: prompt} ], temperature: 0.2 }).encode(utf-8) req urllib.request.Request( base_url, datapayload, headers{Content-Type: application/json, Authorization: fBearer {api_key}} ) with urllib.request.urlopen(req, timeout60) as resp: data json.loads(resp.read().decode(utf-8)) return data[choices][0][message][content]3.7 一条命令跑通全流程把前面的函数串起来整个RAG内核就完整了。我专门写了一个入口函数方便你在命令行里测不同的query。def run_rag(query: str, doc_path: str, chunk_size: int 400, overlap: int 100, top_k: int 3, use_llm: bool True): doc read_document(doc_path) chunks chunk_text(doc, chunk_size, overlap) vocab build_vocab(chunks) idf build_idf(chunks) chunk_vecs [vectorize(c, vocab, idf) for c in chunks] scored search(query, chunk_vecs, vocab, idf, top_k) print(检索结果) for score, idx in scored: print(fscore{score:.4f} chunk#{idx}: {chunks[idx][:80]}...) context \n\n---分割线---\n\n.join(chunks[idx] for _, idx in scored) prompt build_prompt(context, query) if use_llm: return call_ollama(prompt) return prompt, scored现在的RAG内核已经具备全部基本要素加载、切分、词汇表、IDF、向量化、暴力检索、提示词构造、大模型生成。建议你先把use_llm设为False跑几次看搜索结果和prompt长什么样再打开大模型开关。这个先看检索再看生成的调试习惯能省掉大量排查时间。4. 参数调优与问题排查4.1 chunk_size和overlap怎么调没有万能参数但有一个靠谱的调试方法拿三个真实query做测试集固定其他参数分别跑chunk_size100、300、500看检索命中的chunk是否真的覆盖了答案。chunk_size太小比如50个字符切出来的块可能只有一两个短句虽然聚焦但上下文不完整大模型拿到碎片信息容易抓不住因果。chunk_size太大比如2000个字符每块包含大量无关噪声向量的主题不突出相似度分数会被高频词带偏。经验值我建议先从300到500起调。overlap的直接作用是把边界信息带进相邻块。如果overlap设为0两块边界处的事实会丢失如果overlap太大比如占到chunk_size的一半会产生大量冗余块检索时重复内容多索引体积也膨胀。我通常按chunk_size的15%到25%设置重叠。4.2 空向量与零分检索的兜底策略最典型的问题是query做向量化后可能是一个全零向量。为什么会这样因为query中的词一个都没出现在词汇表里比如用户用英文缩写SKU提问知识库文里全写的是商品编号。此时点积全是0排序等于随机检索直接失效。遇到这种情况有两个兜底策略。第一检索得分低于阈值的块直接丢弃不能把弱相关块硬塞给大模型。第二建立一个同义词或别名表在query进入检索前做归一化比如SKU自动扩展成SKU 商品编号。还有更简单的一招对query做字符级拆分而不是只做bigram。这样即使没见过退货这个词也至少能匹配到退和货的单字不至于完全失效。基础版里我用单字bigram混合分词很大程度就是为了应对这种冷启动问题。4.3 从结果反推切分与权重的bugRAG调试最有价值的技巧是不要只盯着最终答案要把检索分数、块内容原样打印出来看。我在3.7里就特意保留了print这一步不是可选项而是必须看的中间产物。有几种异常表现对应不同病因检索分数普遍很低比如全部低于0.1说明query与知识库词汇重叠度太低优先检查分词和清洗。分数很高但捞上来的块内容跟问题不相关可能是chunk_size太大导致向量被高频词主导也可能是停用词放行了一些泛化词。多次query检索下来命中的chunk总是同一段往往是文档里存在大量重复模板内容IDF没能把这类重复词压下去。有一次我遇到检索分值都在0.9以上但答案完全错误打印chunk才发现整份合同里甲方和乙方密布TF-IDF分词把这两个词当成强特征可问题是用户问的是付款条件。后来我把类似甲方乙方的泛称加入停用词并把IDF的计算单位从全部块调整为过滤重复段落后的块问题立刻改善。4.4 常见错误速查表现象原因排查/解决方式检索所有分数都是0query词表与知识库词汇完全不重合检查query分词考虑同义词扩展或改用embedding建词表时内存爆炸bigram词表过大且没去停用词过滤单字噪声控制最短词长或改用jiebachunk之间大量重复内容overlap设置过大降低overlap到chunk_size的15%左右大模型引用不存在的细节检索块被硬切语义不完整增加overlap或按段落边界优先切分Ollama调用超时本地模型首次加载冷启动先提前跑一次模型加载再调用API答案太长乱编提示词没有禁止编造在build_prompt里明确限定范围并强制声明资料外不回答5. 从基础库版本升级到生产环境的路线5.1 替换向量化从TF-IDF到语义Embedding手写基础版跑通后你会发现整套代码真正的短板不是相似度计算而是TF-IDF不懂语义。想升级最好保留前面的文档加载、切分、检索排序和提示词构建逻辑只替换vectorize这一个环节。现在用sentence-transformers或OpenAI的Embedding接口都行。核心改动很小把chunk和query分别编码成稠密向量再用同样的余弦检索逻辑。关键是要保持先切分后向量化再做Top-K的架构不变。这样你既享受embedding的语义匹配能力又保留了自己对切分和上下文拼接的完全控制权。有人会问既然要换embedding当初为什么不直接上因为直接上有两个坏处一是你会把所有注意力放在调模型上完全没理解切分和检索权重对结果的影响二是embedding模型的输出是黑盒出了问题你很难判断是模型问题还是分块问题。基础版跑通后你已经能把问题定位到具体环节再换embedding就是真正的优化而不是换一个更大的黑盒。5.2 索引升级与重排数据量超过几万块时暴力检索的线性扫描会明显变慢。这个时候再引入ANN索引才有意义比如FAISS。我建议把索引作为独立模块抽象出来输入仍然是chunk向量列表接口继续暴露search(query, top_k)。这样你的业务代码完全不需要感知底层是暴力扫描还是HNSW图索引。生产级RAG通常还会加一层重排。Top-K检索出来的结果大概率已经把相关块捞出来了但顺序不一定理想。这时用一个更强的模型对Top-K再做一次精排取前N个进入提示词。基础版里这一步对应的是scored.sort(...)后面的环节你完全可以在这里插入一个重排函数而不影响其他逻辑。5.3 给你的RAG加个持久化缓存最后提一个很实用的扩展我觉得RAG的持久化经常被忽略。基础版每次启动都要重新切分、建立IDF、向量化全部文本数据量一大就很浪费。最简单的做法是把vocab、idf、chunk列表和chunk_vec用json/文件保存下来下次启动直接加载。这一步能让手搓RAG从玩具变成一个可日常使用的工具。我实测过一个小知识库文档大约10万字全流程向量化大约一秒多本机内存完全够用。再加上缓存后启动时间缩短到0.2秒左右。很多在线的知识库工具其实也就是这个内核外面套了一层Web界面核心原理和你刚手撕的这部分没有本质区别。个人在实际项目里踩坑最多的永远是分块和权重而不是代码逻辑。我强烈建议你做一个调试小工具输入一个query同时打印检索分数、命中的块原文和答案然后反复调整chunk_size、overlap、停用词。别急着上大模型和向量数据库先用一两个真实问题把检索调到哪怕不看答案也能猜到它该命中哪一块再谈模型效果。如果你动手写完这套基础版再去翻任何RAG框架那些概念就不再是抽象黑盒了。遇到问题你也会第一个想到去查自己最熟的那一层逻辑。