RAG 是解决 AI 胡说八道问题最有效的方案之一。这篇文章带你从零用 Python 实现一个完整的 RAG 系统搞清楚向量数据库、Embedding、检索、增强生成每一步在做什么。为什么 LLM 会胡说八道两个根本原因1. 知识截止日期LLM 的训练数据有截止日期最新信息它不知道。2. 不知道你的私有数据你公司内部的文档、你自己的笔记、你的代码库——这些 LLM 从未见过。传统做法是微调Fine-tuning把你的数据重新训练进模型。但这代价极高算力成本、时间成本而且数据更新后又要重训。RAGRetrieval-Augmented Generation检索增强生成是一种更轻量的方案传统做法把知识塞进模型里一次性更新困难 RAG查询时实时检索相关知识注入 Prompt灵活易更新RAG 的完整架构┌─── 离线索引阶段 ───┐ 文档 → 切片 → Embedding → 向量数据库 └────────────────────┘ ┌─── 在线查询阶段 ───┐ 用户问题 → Embedding → 向量检索 → 相关文档片段 ↓ 注入 Prompt → LLM → 回答 └────────────────────┘两个阶段离线索引把文档切片、向量化、存入向量数据库一次性在线查询把用户问题向量化检索相似片段拼进 Prompt让 LLM 回答动手实现第一步文档切片文档不能整篇扔给 LLM太长需要切成合适大小的片段Chunk。# chunker.pyfrompathlibimportPathdefread_document(file_path:str)-str:读取文档支持 .txt 和 .mdwithopen(file_path,r,encodingutf-8)asf:returnf.read()defchunk_text(text:str,chunk_size:int500,# 每个 Chunk 的字符数chunk_overlap:int50# 相邻 Chunk 的重叠字符数保留上下文)-list[str]: 按字符数切片带重叠 为什么要重叠 如果一个答案恰好跨越两个 Chunk 的边界 没有重叠的话会被切断检索时可能找不到。 chunks[]start0whilestartlen(text):endstartchunk_size chunktext[start:end]# 尝试在句子边界切割不在单词中间断开ifendlen(text):last_periodchunk.rfind(。)last_newlinechunk.rfind(\n)break_pointmax(last_period,last_newline)ifbreak_pointchunk_size*0.5:# 切割点不能太靠前chunkchunk[:break_point1]endstartlen(chunk)ifchunk.strip():chunks.append(chunk.strip())startend-chunk_overlapreturnchunks# 测试textread_document(company_docs.md)chunkschunk_text(text)print(f文档共{len(text)}字符切成{len(chunks)}个 Chunk)print(f\n第一个 Chunk 预览\n{chunks[0][:200]}...)Chunk 大小的选择场景推荐 Chunk 大小精确问答需要精确片段200-400 字符通用问答400-800 字符需要更多上下文800-1500 字符第二步Embedding向量化Embedding 是把文本转换成数值向量的过程。语义相似的文本向量距离更近。# embedder.pyfromanthropicimportAnthropicimportnumpyasnp clientAnthropic()defembed_text(text:str)-list[float]: 调用 Voyage AIAnthropic 推荐的 Embedding 模型 生成文本的向量表示 importvoyageai vovoyageai.Client()resultvo.embed([text],modelvoyage-3,# 当前最强通用 Embedding 模型input_typedocument# 索引文档时用 document查询时用 query)returnresult.embeddings[0]defembed_batch(texts:list[str],batch_size:int128)-list[list[float]]:批量 Embedding效率更高importvoyageai vovoyageai.Client()all_embeddings[]foriinrange(0,len(texts),batch_size):batchtexts[i:ibatch_size]resultvo.embed(batch,modelvoyage-3,input_typedocument)all_embeddings.extend(result.embeddings)print(f已处理{min(ibatch_size,len(texts))}/{len(texts)}个 Chunk)returnall_embeddingsdefcosine_similarity(v1:list[float],v2:list[float])-float:计算两个向量的余弦相似度-1 到 1越大越相似v1,v2np.array(v1),np.array(v2)returnnp.dot(v1,v2)/(np.linalg.norm(v1)*np.linalg.norm(v2))为什么用余弦相似度而不是欧氏距离向量的方向代表语义长度代表强度。余弦相似度只比较方向不受向量长度影响更适合语义匹配。第三步构建向量数据库# vector_store.pyimportjsonimportnumpyasnpfrompathlibimportPathclassSimpleVectorStore: 轻量级向量存储适合小数据集生产环境建议用 ChromaDB/Pinecone/Milvus def__init__(self,store_path:strvector_store.json):self.store_pathstore_path self.documents[]# [{text, embedding, metadata}, ...]ifPath(store_path).exists():self.load()defadd(self,text:str,embedding:list[float],metadata:dictNone):self.documents.append({text:text,embedding:embedding,metadata:metadataor{}})defsearch(self,query_embedding:list[float],top_k:int5)-list[dict]:暴力搜索计算与所有文档的相似度返回最相似的 top_k 个ifnotself.documents:return[]similarities[]query_arrnp.array(query_embedding)fori,docinenumerate(self.documents):doc_arrnp.array(doc[embedding])simnp.dot(query_arr,doc_arr)/(np.linalg.norm(query_arr)*np.linalg.norm(doc_arr))similarities.append((i,float(sim)))# 按相似度降序排列similarities.sort(keylambdax:x[1],reverseTrue)results[]foridx,scoreinsimilarities[:top_k]:results.append({text:self.documents[idx][text],score:score,metadata:self.documents[idx][metadata]})returnresultsdefsave(self):withopen(self.store_path,w,encodingutf-8)asf:json.dump(self.documents,f,ensure_asciiFalse)print(f已保存{len(self.documents)}个文档向量到{self.store_path})defload(self):withopen(self.store_path,r,encodingutf-8)asf:self.documentsjson.load(f)print(f已加载{len(self.documents)}个文档向量)第四步完整的 RAG 查询流程# rag.pyimportanthropicimportvoyageaifromchunkerimportchunk_text,read_documentfromvector_storeimportSimpleVectorStoredefbuild_index(doc_paths:list[str],store_path:strvector_store.json):离线索引阶段读取文档、切片、向量化、存储vovoyageai.Client()storeSimpleVectorStore(store_path)fordoc_pathindoc_paths:print(f正在处理{doc_path})textread_document(doc_path)chunkschunk_text(text)# 批量 Embeddingembeddings[]foriinrange(0,len(chunks),128):batchchunks[i:i128]resultvo.embed(batch,modelvoyage-3,input_typedocument)embeddings.extend(result.embeddings)forchunk,embeddinginzip(chunks,embeddings):store.add(chunk,embedding,metadata{source:doc_path})store.save()print(f\n索引构建完成共{len(store.documents)}个 Chunk)returnstoredefquery(user_question:str,store:SimpleVectorStore,top_k:int5)-str:在线查询阶段检索相关 Chunk注入 Prompt调用 LLMvovoyageai.Client()claudeanthropic.Anthropic()# Step 1: 对问题做 Embeddingquery_embeddingvo.embed([user_question],modelvoyage-3,input_typequery# 查询时用 query与文档 Embedding 区分).embeddings[0]# Step 2: 检索最相关的 Chunkresultsstore.search(query_embedding,top_ktop_k)# 过滤低相似度的结果相似度 0.6 通常是噪声relevant_chunks[rforrinresultsifr[score]0.6]ifnotrelevant_chunks:return抱歉在提供的文档中没有找到与您问题相关的信息。# Step 3: 构建增强后的 Promptcontext\n\n---\n\n.join([f[来源{r[metadata].get(source,未知)}]\n{r[text]}forrinrelevant_chunks])promptf请根据以下文档内容回答用户的问题。 ## 相关文档内容{context}## 用户问题{user_question}## 回答要求 - 只使用上述文档中的信息回答不要添加文档中没有的内容 - 如果文档中没有足够信息回答问题请明确说明 - 回答要简洁准确如有必要可以引用原文 # Step 4: 调用 LLM 生成答案responseclaude.messages.create(modelclaude-3-5-sonnet-20241022,max_tokens1024,messages[{role:user,content:prompt}])returnresponse.content[0].text# 使用示例if__name____main__:# 构建索引只需运行一次storebuild_index([docs/product_manual.md,docs/faq.md])# 查询questions[这个产品支持哪些操作系统,如何重置密码,退款政策是什么]forqinquestions:print(f\n问题{q})print(f回答{query(q,store)})print(-*50)关键优化点优化一更好的切片策略按语义段落、章节切割而不是按固定字符数切割defsemantic_chunk(text:str,max_chunk_size:int500)-list[str]:按段落切割大段落再按句子切割paragraphs[p.strip()forpintext.split(\n\n)ifp.strip()]chunks[]forparainparagraphs:iflen(para)max_chunk_size:chunks.append(para)else:# 大段落按句子切割sentencespara.split(。)currentforsentinsentences:iflen(current)len(sent)max_chunk_size:currentsent。else:ifcurrent:chunks.append(current.strip())currentsent。ifcurrent:chunks.append(current.strip())returnchunks优化二混合检索纯向量检索对精确关键词不敏感混合 BM25关键词检索效果更好# 向量检索 关键词检索结果融合vector_resultsstore.search(query_embedding,top_k10)keyword_resultsbm25_search(query_text,top_k10)final_resultsreciprocal_rank_fusion(vector_results,keyword_results)[:5]总结RAG 的核心链路文档 → 切片 → Embedding → 向量数据库 ↑ 查询 → Embedding → 相似度检索 ──┘ ↓ 相关 Chunk 注入 Prompt → LLM → 回答关键决策点Chunk 大小太小丢失上下文太大稀释相关性Embedding 模型决定语义检索的质量相似度阈值过滤噪声避免把不相关内容注入 Prompttop_k 选择相关 Chunk 太少可能遗漏信息太多会超出 Context Window
从零构建一个 RAG 系统：用 Python 实现“让 AI 读懂你的私有文档“
RAG 是解决 AI 胡说八道问题最有效的方案之一。这篇文章带你从零用 Python 实现一个完整的 RAG 系统搞清楚向量数据库、Embedding、检索、增强生成每一步在做什么。为什么 LLM 会胡说八道两个根本原因1. 知识截止日期LLM 的训练数据有截止日期最新信息它不知道。2. 不知道你的私有数据你公司内部的文档、你自己的笔记、你的代码库——这些 LLM 从未见过。传统做法是微调Fine-tuning把你的数据重新训练进模型。但这代价极高算力成本、时间成本而且数据更新后又要重训。RAGRetrieval-Augmented Generation检索增强生成是一种更轻量的方案传统做法把知识塞进模型里一次性更新困难 RAG查询时实时检索相关知识注入 Prompt灵活易更新RAG 的完整架构┌─── 离线索引阶段 ───┐ 文档 → 切片 → Embedding → 向量数据库 └────────────────────┘ ┌─── 在线查询阶段 ───┐ 用户问题 → Embedding → 向量检索 → 相关文档片段 ↓ 注入 Prompt → LLM → 回答 └────────────────────┘两个阶段离线索引把文档切片、向量化、存入向量数据库一次性在线查询把用户问题向量化检索相似片段拼进 Prompt让 LLM 回答动手实现第一步文档切片文档不能整篇扔给 LLM太长需要切成合适大小的片段Chunk。# chunker.pyfrompathlibimportPathdefread_document(file_path:str)-str:读取文档支持 .txt 和 .mdwithopen(file_path,r,encodingutf-8)asf:returnf.read()defchunk_text(text:str,chunk_size:int500,# 每个 Chunk 的字符数chunk_overlap:int50# 相邻 Chunk 的重叠字符数保留上下文)-list[str]: 按字符数切片带重叠 为什么要重叠 如果一个答案恰好跨越两个 Chunk 的边界 没有重叠的话会被切断检索时可能找不到。 chunks[]start0whilestartlen(text):endstartchunk_size chunktext[start:end]# 尝试在句子边界切割不在单词中间断开ifendlen(text):last_periodchunk.rfind(。)last_newlinechunk.rfind(\n)break_pointmax(last_period,last_newline)ifbreak_pointchunk_size*0.5:# 切割点不能太靠前chunkchunk[:break_point1]endstartlen(chunk)ifchunk.strip():chunks.append(chunk.strip())startend-chunk_overlapreturnchunks# 测试textread_document(company_docs.md)chunkschunk_text(text)print(f文档共{len(text)}字符切成{len(chunks)}个 Chunk)print(f\n第一个 Chunk 预览\n{chunks[0][:200]}...)Chunk 大小的选择场景推荐 Chunk 大小精确问答需要精确片段200-400 字符通用问答400-800 字符需要更多上下文800-1500 字符第二步Embedding向量化Embedding 是把文本转换成数值向量的过程。语义相似的文本向量距离更近。# embedder.pyfromanthropicimportAnthropicimportnumpyasnp clientAnthropic()defembed_text(text:str)-list[float]: 调用 Voyage AIAnthropic 推荐的 Embedding 模型 生成文本的向量表示 importvoyageai vovoyageai.Client()resultvo.embed([text],modelvoyage-3,# 当前最强通用 Embedding 模型input_typedocument# 索引文档时用 document查询时用 query)returnresult.embeddings[0]defembed_batch(texts:list[str],batch_size:int128)-list[list[float]]:批量 Embedding效率更高importvoyageai vovoyageai.Client()all_embeddings[]foriinrange(0,len(texts),batch_size):batchtexts[i:ibatch_size]resultvo.embed(batch,modelvoyage-3,input_typedocument)all_embeddings.extend(result.embeddings)print(f已处理{min(ibatch_size,len(texts))}/{len(texts)}个 Chunk)returnall_embeddingsdefcosine_similarity(v1:list[float],v2:list[float])-float:计算两个向量的余弦相似度-1 到 1越大越相似v1,v2np.array(v1),np.array(v2)returnnp.dot(v1,v2)/(np.linalg.norm(v1)*np.linalg.norm(v2))为什么用余弦相似度而不是欧氏距离向量的方向代表语义长度代表强度。余弦相似度只比较方向不受向量长度影响更适合语义匹配。第三步构建向量数据库# vector_store.pyimportjsonimportnumpyasnpfrompathlibimportPathclassSimpleVectorStore: 轻量级向量存储适合小数据集生产环境建议用 ChromaDB/Pinecone/Milvus def__init__(self,store_path:strvector_store.json):self.store_pathstore_path self.documents[]# [{text, embedding, metadata}, ...]ifPath(store_path).exists():self.load()defadd(self,text:str,embedding:list[float],metadata:dictNone):self.documents.append({text:text,embedding:embedding,metadata:metadataor{}})defsearch(self,query_embedding:list[float],top_k:int5)-list[dict]:暴力搜索计算与所有文档的相似度返回最相似的 top_k 个ifnotself.documents:return[]similarities[]query_arrnp.array(query_embedding)fori,docinenumerate(self.documents):doc_arrnp.array(doc[embedding])simnp.dot(query_arr,doc_arr)/(np.linalg.norm(query_arr)*np.linalg.norm(doc_arr))similarities.append((i,float(sim)))# 按相似度降序排列similarities.sort(keylambdax:x[1],reverseTrue)results[]foridx,scoreinsimilarities[:top_k]:results.append({text:self.documents[idx][text],score:score,metadata:self.documents[idx][metadata]})returnresultsdefsave(self):withopen(self.store_path,w,encodingutf-8)asf:json.dump(self.documents,f,ensure_asciiFalse)print(f已保存{len(self.documents)}个文档向量到{self.store_path})defload(self):withopen(self.store_path,r,encodingutf-8)asf:self.documentsjson.load(f)print(f已加载{len(self.documents)}个文档向量)第四步完整的 RAG 查询流程# rag.pyimportanthropicimportvoyageaifromchunkerimportchunk_text,read_documentfromvector_storeimportSimpleVectorStoredefbuild_index(doc_paths:list[str],store_path:strvector_store.json):离线索引阶段读取文档、切片、向量化、存储vovoyageai.Client()storeSimpleVectorStore(store_path)fordoc_pathindoc_paths:print(f正在处理{doc_path})textread_document(doc_path)chunkschunk_text(text)# 批量 Embeddingembeddings[]foriinrange(0,len(chunks),128):batchchunks[i:i128]resultvo.embed(batch,modelvoyage-3,input_typedocument)embeddings.extend(result.embeddings)forchunk,embeddinginzip(chunks,embeddings):store.add(chunk,embedding,metadata{source:doc_path})store.save()print(f\n索引构建完成共{len(store.documents)}个 Chunk)returnstoredefquery(user_question:str,store:SimpleVectorStore,top_k:int5)-str:在线查询阶段检索相关 Chunk注入 Prompt调用 LLMvovoyageai.Client()claudeanthropic.Anthropic()# Step 1: 对问题做 Embeddingquery_embeddingvo.embed([user_question],modelvoyage-3,input_typequery# 查询时用 query与文档 Embedding 区分).embeddings[0]# Step 2: 检索最相关的 Chunkresultsstore.search(query_embedding,top_ktop_k)# 过滤低相似度的结果相似度 0.6 通常是噪声relevant_chunks[rforrinresultsifr[score]0.6]ifnotrelevant_chunks:return抱歉在提供的文档中没有找到与您问题相关的信息。# Step 3: 构建增强后的 Promptcontext\n\n---\n\n.join([f[来源{r[metadata].get(source,未知)}]\n{r[text]}forrinrelevant_chunks])promptf请根据以下文档内容回答用户的问题。 ## 相关文档内容{context}## 用户问题{user_question}## 回答要求 - 只使用上述文档中的信息回答不要添加文档中没有的内容 - 如果文档中没有足够信息回答问题请明确说明 - 回答要简洁准确如有必要可以引用原文 # Step 4: 调用 LLM 生成答案responseclaude.messages.create(modelclaude-3-5-sonnet-20241022,max_tokens1024,messages[{role:user,content:prompt}])returnresponse.content[0].text# 使用示例if__name____main__:# 构建索引只需运行一次storebuild_index([docs/product_manual.md,docs/faq.md])# 查询questions[这个产品支持哪些操作系统,如何重置密码,退款政策是什么]forqinquestions:print(f\n问题{q})print(f回答{query(q,store)})print(-*50)关键优化点优化一更好的切片策略按语义段落、章节切割而不是按固定字符数切割defsemantic_chunk(text:str,max_chunk_size:int500)-list[str]:按段落切割大段落再按句子切割paragraphs[p.strip()forpintext.split(\n\n)ifp.strip()]chunks[]forparainparagraphs:iflen(para)max_chunk_size:chunks.append(para)else:# 大段落按句子切割sentencespara.split(。)currentforsentinsentences:iflen(current)len(sent)max_chunk_size:currentsent。else:ifcurrent:chunks.append(current.strip())currentsent。ifcurrent:chunks.append(current.strip())returnchunks优化二混合检索纯向量检索对精确关键词不敏感混合 BM25关键词检索效果更好# 向量检索 关键词检索结果融合vector_resultsstore.search(query_embedding,top_k10)keyword_resultsbm25_search(query_text,top_k10)final_resultsreciprocal_rank_fusion(vector_results,keyword_results)[:5]总结RAG 的核心链路文档 → 切片 → Embedding → 向量数据库 ↑ 查询 → Embedding → 相似度检索 ──┘ ↓ 相关 Chunk 注入 Prompt → LLM → 回答关键决策点Chunk 大小太小丢失上下文太大稀释相关性Embedding 模型决定语义检索的质量相似度阈值过滤噪声避免把不相关内容注入 Prompttop_k 选择相关 Chunk 太少可能遗漏信息太多会超出 Context Window
