ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

手写最小RAG应用:Embedding + Chroma + DeepSeek 完整实现

手写最小RAG应用:Embedding + Chroma + DeepSeek 完整实现 如果你也在跟着一份 Day 1 到 Day 30 的学习计划走那第 8 天应该是第一个让你有“我竟然真的能徒手搭一个 AI 应用”的节点。这一天的任务很具体不用 LangChain 全家桶不用一键脚本只用三个最朴素的组件——Embedding 模型、Chroma 向量库、DeepSeek API把一个最小 RAG 从零到手写出来。这个项目适合刚学完 Python 基础、对 LLM API 有一点点了解但还不太清楚“RAG 到底怎么串起来”的朋友。做完之后你会明白RAG 不是魔法它就是“先把资料变成索引再让大模型开卷考试”。下面我把整个实现过程和踩坑经历原原本本写下来尽量做到你照着敲就能跑通。1. 整体设计思路拆解1.1 为什么 RAG 非要自己手写一遍RAG 的全称是 Retrieval-Augmented Generation检索增强生成。网上 90% 的 RAG 教程都会给你一张架构图然后告诉你“先向量化再检索再生成”。这句话听得多了但真到写代码的时候很多人还是会卡在“向量化到底怎么切”“检索结果怎么塞进 Prompt”“向量库选哪个”这些细节上。我强烈建议 Day 8 不要直接上 LangChain 或者 LlamaIndex。不是它们不好而是框架把太多细节藏起来了。你调用一个Retriever、一个QAChain表面上跑通了但出了问题根本不知道是切分的问题、embedding 的问题还是 Prompt 的问题。手写一遍最小系统等于把每一个环节都亲手解了一遍以后再回去用框架就稳得多。类比一下开卷考试。大模型是那个记忆力不错的考生但它的“脑子”只到训练截止日期你手里的私有文档它没见过。RAG 干的事就是考前帮它把教材目录和章节页码整理好考试时它知道去哪翻书翻到之后照着念。这个“翻书”的过程就是向量检索而不是搜索引擎那种关键词匹配因为向量检索能理解语义。1.2 为什么偏偏选 Embedding、Chroma、DeepSeek 这三个选型原则很简单能本地跑就本地跑能少花钱就少花钱能少装依赖就少装依赖。Embedding 模型选了一个本地小模型BAAI/bge-small-zh-v1.5。参数规模小CPU 也能跑中文效果在同级别里算能打向量维度 512 维存储和计算开销都不大。向量库选 Chroma。它是嵌入式向量数据库不需要单独启动服务数据直接存本地目录对新手极其友好。相比 Faiss、Milvus 那些Chroma 的上手成本低一个数量级。生成模型选 DeepSeek API。国内直连延迟可以接受价格便宜而且接口兼容 OpenAI SDK代码写起来非常顺。它支持deepseek-chat和deepseek-reasoner两个模型Day 8 用deepseek-chat就够。这三个组合下来你的电脑只需要装一个 PyTorch 的 CPU 版本、一个 Chroma 库、一个 OpenAI SDK连 GPU 都不需要。这是最小成本验证 RAG 全链路的最佳方案。1.3 最小架构就是一条直线整个系统不复杂数据流是单向的读入一个或多个文本文件。按固定长度切块每块之间留一点重叠。用本地 Embedding 模型把每一块转成向量。向量和原始文本一起写入 Chroma。用户提问时把问题也转成向量。在 Chroma 里做相似度检索拿回最相关的 3 个文本块。把文本块作为上下文拼进 Prompt发给 DeepSeek。DeepSeek 返回答案。这个流程看起来简单但每一步都有坑。我后面按顺序拆开讲每一步都给可直接复制的代码。2. 环境准备与 Embedding 选型实战2.1 Python 环境怎么配最省事我建议用 Python 3.10兼容性最好。如果你机器上默认是 3.12遇到 Chroma 或 PyTorch 的版本兼容问题不要硬刚直接建一个 3.10 的虚拟环境最省事。python3.10 -m venv rag_env source rag_env/bin/activate # Windows 用 rag_env\Scripts\activate然后安装依赖pip install sentence-transformers chromadb openai python-dotenv这里sentence-transformers会自动带上来 PyTorch默认可能是 CPU 版。如果它给你装了个巨大的 CUDA 版本也不必惊慌只是下载慢。想要明确使用 CPU 版本可以先去 PyTorch 官网用 pip 装好 CPU 版再装 sentence-transformers。2.2 安装过程中最容易翻车的点前几次我在这步踩了不少坑列几个重点chromadb和 Python 3.12 的某些版本组合会报cannot import name html5lib from google.protobuf或者sqlite3版本问题。降回 Python 3.10 通常能解决。如果下载模型非常慢不要反复删缓存重试先用sentence-transformers跑一次让它把模型信息下载下来再手动去 Hugging Face 文件列表里把缺失的权重文件放进缓存目录。或者干脆找一个已经有国内镜像的模型渠道但注意别用任何需要额外配置的代理方式。openai库版本别太老建议 1.30 以上。太老版本没有OpenAI()这种新式客户端。2.3 Embedding 模型排行怎么理解网上经常能搜到“Embedding 模型排行”“MTEB 中文榜单”这类热词。排行榜能帮你少踩坑但千万别只看排名。榜单测的是通用任务的平均分你的数据集长什么样、你的业务问题属于什么领域只有自己跑过才知道。我给三个常见的中文小模型做过对比模型参数量向量维度中文能力适合场景BAAI/bge-small-zh-v1.5约 24M512好性价比极高最小项目、低资源环境BAAI/bge-base-zh-v1.5约 102M768更好但显存/内存占用增加追求效果、硬件允许m3e-small约 24M512中上通用中文语义text2vec-base-chinese约 102M768中上长文本分类、语义匹配Day 8 直接选 bge-small-zh-v1.5。它还有一点好官方建议在查询语句前加一句指令“为这个句子生成表示以用于检索相关文章”短查询效果有提升。这个我们在检索阶段会用到。2.4 先跑通 Embedding 再谈别的写一段最简代码验证模型能加载from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-small-zh-v1.5, devicecpu) vec model.encode(今天天气怎么样, normalize_embeddingsTrue) print(vec.shape)如果输出里有个(512,)说明模型没问题。这里normalize_embeddingsTrue很关键它让向量归一化后面用余弦距离就不会被文本长度干扰。3. 把文档拆成可检索的块文本切分策略3.1 切分质量直接决定 RAG 上限很多人以为 RAG 的瓶颈在大模型实际操作后你会发现检索的那一步才是决定回答质量的命门。如果你把一整篇 8000 字的文章直接塞进去做 embedding查询时和整篇文章的相似度都会被“平均”掉真正相关的那个段落反而不突出。RAG 教程里讲得最多的切块策略本质上就是解决这个问题。切块太短比如 100 字语义信息不完整检索出来也容易上下文残缺切块太长比如 2000 字向量混合了多个主题检索精度下降。一个合理的起点是 500 字左右再让相邻块之间有 50 到 100 字的重叠。3.2 最小实现按字符长度硬切为了把逻辑讲清楚Day 8 先不做按句号、按 Markdown 标题的智能切分直接用固定长度切。这个方案足够应付大多数纯文本笔记。def split_text(text, chunk_size500, chunk_overlap80): if len(text) chunk_size: return [text] chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) chunks.append(text[start:end]) if end len(text): break start end - chunk_overlap return chunks为什么要有重叠如果一句话正好被切在两块中间没有重叠就会导致这句话在两边都是残缺的。有了重叠即使关键句被拦腰切断后面一块的开头也会带着这句话的尾巴检索命中概率会高很多。3.3 一个比较现实的改进优先按段落切字符硬切虽然能用但偶尔会把列表项、代码块切碎。你可以在切块前先按“换行符”做一次粗切再把小于一定长度的小段合并。这样既保留了段落的语义边界又不会产生太多碎片化空块。def smart_split(text, min_chunk200, max_chunk700, overlap80): paragraphs [p.strip() for p in text.split(\n) if p.strip()] chunks [] current for para in paragraphs: if len(current) len(para) max_chunk: current \n para else: if current: chunks.append(current) current para if current: chunks.append(current) return chunks这种方案对文本文件、Markdown 笔记、会议纪要都够用。等以后资料变成 PDF 和 Word再考虑按段落结构识别或者引入专门的文档解析工具。3.4 “RAG 知识库能存图片吗”这个问题经常看到有人搜“rag知识库能存储图片嘛”。答案是能但最小 RAG 暂时不需要。Chroma 本身只存向量和文本它不关心你的向量是从文字算出来的还是从图片算出来的。你要真想存图片需要把图片送入一个多模态模型比如 CLIP转成图片向量再把向量和图片路径一起存进去。检索时拿文本向量去比对图片向量返回图片路径。这条路复杂度高不少Day 8 的文本 RAG 先不要碰免得失焦。4. Chroma 向量库的搭建与检索4.1 Chroma 的两种用法我只推荐持久化模式Chroma 可以当内存数据库用也可以保持本地持久化。最小项目一定要用PersistentClient否则你辛辛苦苦向量化完程序一退数据全没了。import chromadb client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection( nameminimal_rag, metadata{hnsw:space: cosine} )创建 Collection 时可以指定相似度空间为cosine。bge 模型向量做了归一化用余弦距离比较合适。没有指定的话 Chroma 默认是 L2也能用但余弦在很多场景下更符合我们对“语义相似度”的直觉。4.2 写入向量时必须保证 id 唯一Chroma 的add要求每个 id 唯一重复 id 会直接报错。更稳妥的方式是直接使用upsert有则更新无则插入。def index_document(path): text path.read_text(encodingutf-8) chunks split_text(text) ids [f{path.stem}_{i} for i in range(len(chunks))] embeddings embedder.encode(chunks, normalize_embeddingsTrue).tolist() metadatas [{source: str(path)} for _ in chunks] collection.upsert( idsids, documentschunks, embeddingsembeddings, metadatasmetadatas )这里documents是原始文本embeddings是我们自己算好的向量。Chroma 也支持你传一个内置的 embedding 函数让它自己计算但默认函数走的是all-MiniLM-L6-v2处理中文效果一般而且会和我们的 bge 模型不一致。所以最可控的做法永远是“你自己算好向量再传进去”。4.3 检索与 Hit Rate别只看效果要量化检索时把用户 query 转成向量然后查最相似的 top_kdef search(query, top_k3): query_embedding embedder.encode( [为这个句子生成表示以用于检索相关文章 query], normalize_embeddingsTrue ).tolist() result collection.query( query_embeddingsquery_embedding, n_resultstop_k, include[documents, distances, metadatas] ) return result[documents][0], result[distances][0], result[metadatas][0]这里BGE模型查询时加指令我记得论文里明确建议对短查询这么做。可以对比一下不加指令也能用但加了之后中文短查询的命中率明显稳定。怎么评估检索质量我一直用一个很朴素的“RAG hit rate”方法准备 10 个问题每个问题都人工标注它应该命中的文件或文本块然后跑检索看正确答案有没有出现在 top 3 里最后统计命中比例。比如 10 个问题里有 7 个在 top 3 中出现了正确片段那 hit rate 就是 70%。这是最粗糙但最直观的评估后续再想深入可以用召回率、精确率、MRR 这些指标。4.4 重启后重新加载数据PersistentClient的好处是第二次运行只需要重新连接同一个目录不需要重新解析文档client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(minimal_rag, metadata{hnsw:space: cosine})注意如果你在第一次创建时没有指定余弦空间第二次指定也不会生效因为 Collection 已经存在了。所以一开始就要想好。想要换空间把./chroma_db目录删掉重建就行。4.5 向量化入库的完整流程把所有.txt文件批量入库from pathlib import Path docs_dir Path(./docs) for file_path in docs_dir.glob(*.txt): index_document(file_path) print(f已入库{file_path})入库后我习惯打印每个文件切出来的块数。如果切出来 1 块说明文件很短如果切出来 200 块说明内容很多。块数异常时说明切分逻辑可能需要调整。5. DeepSeek 接入与提示词工程实战5.1 用 OpenAI SDK 调 DeepSeek最小代码DeepSeek 开放平台提供了 OpenAI 兼容接口所以我们不需要单独装 DeepSeek 的 SDK直接用openai库把base_url指过去就行。from openai import OpenAI import os llm OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com, timeout30, )调用方式response llm.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: 你好}], temperature0.2, ) print(response.choices[0].message.content)这里有个容易踩的坑api_key不要写到代码里也不要提交到 Git 仓库。我是放在.env文件里然后用python-dotenv加载DEEPSEEK_API_KEYsk-xxxxfrom dotenv import load_dotenv load_dotenv()5.2 把检索结果拼进 PromptRAG 的灵魂是 Prompt 里面必须有明确的“资料区”和“问题区”。我常用的模板def build_prompt(question, context_chunks): context \n\n.join(context_chunks) prompt f你是知识库助手。请只根据下面的资料回答问题。 如果资料里没有相关内容请直接说“根据现有资料无法回答”不要编造。 资料 {context} 问题 {question} 回答 return prompt注意context不能太长。DeepSeek 上下文窗口虽然大但你把 20 个块全塞进去既浪费 token又会引入噪声。Day 8 的检索先取 top 3 到 top 5 就够了。后面想优化可以加一个重排序模型先粗召回 20 个再精排选 5 个。5.3 设定“不知道”的边界刚开始跑的时候我发现一个现象当检索到的上下文和问题完全无关时DeepSeek 还是会硬答而且答得像模像样。原因是大模型本身有很强的“讨好倾向”你给了它一堆材料它总觉得应该从中找点什么。所以 Prompt 里必须明确一句“如果资料里没有相关内容请直接说不知道”。这句提示能显著减少幻觉。另外可以把temperature调低一点。知识问答我一般设置在0.1到0.3之间。温度太高回答会发散容易自由发挥太低则机械好处是稳定。5.4 本地部署和 API我为什么先选 API最近“deepseek 本地部署”“DeepSeek 本地部署 Jetson Orin”这类话题很热。本地部署确实能保护隐私、离线可用但一个最小 RAG 项目里本地部署模型的成本远比想象中高。你要准备 GPU、考虑量化、处理推理服务一个晚上根本搞不完。我更推荐的做法是Day 8 先用 DeepSeek API 验证 RAG 流程跑通之后再考虑把生成模型换成本地模型。很多社区里提到的deepseek harness之类的封装工具我也建议先不要碰。新手期最忌讳的是工具比业务还复杂。官方 SDK 是最稳的起点等你理解 RAG 全链路之后再去找更高阶的封装会轻松很多。6. 组装成可用的最小 RAG 应用6.1 完整代码骨架我把上面所有片段整合成一个文件rag.py你复制后把路径改成自己的就能跑。import os from pathlib import Path from dotenv import load_dotenv from sentence_transformers import SentenceTransformer import chromadb from openai import OpenAI load_dotenv() DATA_DIR Path(./docs) CHROMA_PATH ./chroma_db COLLECTION_NAME minimal_rag TOP_K 3 embedder SentenceTransformer(BAAI/bge-small-zh-v1.5, devicecpu) client chromadb.PersistentClient(pathCHROMA_PATH) collection client.get_or_create_collection( nameCOLLECTION_NAME, metadata{hnsw:space: cosine} ) llm OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com, timeout30, ) def split_text(text, chunk_size500, overlap80): if len(text) chunk_size: return [text] chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) chunks.append(text[start:end]) if end len(text): break start end - overlap return chunks def embed_texts(texts): return embedder.encode(texts, normalize_embeddingsTrue).tolist() def index_document(path): text path.read_text(encodingutf-8) chunks split_text(text) ids [f{path.stem}_{i} for i in range(len(chunks))] collection.upsert( idsids, documentschunks, embeddingsembed_texts(chunks), metadatas[{source: str(path)} for _ in chunks] ) print(f入库 {path.name}切分为 {len(chunks)} 块) def index_all(): for path in DATA_DIR.glob(*.txt): index_document(path) def search(query, top_kTOP_K): query_emb embed_texts([为这个句子生成表示以用于检索相关文章 query]) result collection.query( query_embeddingsquery_emb, n_resultstop_k, include[documents, metadatas, distances] ) return result[documents][0], result[metadatas][0], result[distances][0] def ask(question): docs, metas, distances search(question) context \n\n.join(docs) prompt f你是知识库助手。请只根据下面的资料回答问题。 如果资料里没有相关内容请直接说“根据现有资料无法回答”不要编造。 资料 {context} 问题 {question} 回答 response llm.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: prompt}], temperature0.2, ) return response.choices[0].message.content if __name__ __main__: index_all() while True: q input(\n问题: ).strip() if q in (exit, quit): break print(f\n答案: {ask(q)})我把入口做成了命令行问答运行python rag.py后直接在终端聊天。它适用于任何纯文本笔记你只需要把docs目录里的 txt 替换成自己的资料。6.2 如果问答结果不满意先查中间变量这个最小 RAG 的好处是每一步都能单独打印出来排查。比如用户问“我们上次开会结论是什么”如果回答不对可以先不调 DeepSeek先把search返回的三个documents打出来看看检索到的到底是不是会议纪要那段。如果检索到的内容已经是驴唇不对马嘴那问题大概率在切分、embedding、或者文件写入阶段和 DeepSeek 无关。千万不要直接跳到大模型 Prompt 去调。我用这个思路省了无数时间。6.3 从笔记到知识库的落地玩法第 8 天做完之后我立刻把平时记录的几十篇 Markdown 笔记放了进去。之后遇到类似“我之前说的那个部署方案是什么来着”直接问终端就行效率比手动翻文档高很多。这种玩法可以推广到产品文档、会议纪要、收藏文章、面试复习资料。把.txt文件丢进docs再跑一次index_all()知识库就更新了。后面想继续升级可以加一个“增量索引”每次只处理修改过的文件不过那是后话。7. 常见问题排查与避坑清单7.1 问题速查表症状大概率原因解决办法检索出来的文本完全不相关切块太大或太小embedding 模型不适合领域调整 chunk_size / overlap换 bge-base 或针对领域微调入库时报 repeated id同一个文件名重复加内容不同的块用upsert或 id 里加上文件修改时间戳重启后查询不到数据使用了内存模式EphemeralClient改用PersistentClient(path...)Chroma 报 sqlite3 错误Python 版本太高换 Python 3.10 虚拟环境DeepSeek 返回 401API Key 错误或没加载 .env检查环境变量确认.env和脚本同目录DeepSeek 返回超时网络波动或上下文过长增加timeout60减小 top_k回答总是编造Prompt 里没强调“不知道”加入“如果资料里没有请直接说不知道”每个 query 都要几秒才返回本地 embedding 推理慢把 top_k 调小换更小的模型或批量预计算向量7.2 检索不准的排查路径我遇到最多的不是“API 报错”而是“检索结果不准”。这个问题要按顺序排查先打印result[documents]确认命中的文本块内容。如果第一块内容就已经不对属于召回问题。调整切分策略。默认 500 字如果资料有很多小标题建议用smart_split按段落切。调整重叠长度。实体名如果总是横跨两块重叠太少检索时就会丢信息。换 embedding 模型。bge-small 不行就换 bge-base还不行再考虑更大的模型。调整查询指令。BGE 模型查询时加了指令后效果差异很明显别忘了。如果检索已经准了但大模型回答依然不对那问题才在 Prompt 或模型参数。7.3 DeepSeek API 报错汇总这里把 DeepSeek 接入阶段常见的报错写清楚AuthenticationErrorAPI Key 不合法或者环境变量没加载。RateLimitError请求频率超限可以先 sleep 一下再重试。BadRequestError多半是 messages 格式不对比如 role 写成了system之外的非法值。APIConnectionError网络不通或超时。可以缩短上下文、降低 top_k同时在OpenAI()里把timeout调到 60。InvalidRequestError提示prompt is too long检索结果太多把n_results调小。所有 API 报错第一步永远是先打印原始返回信息。不要瞎猜错误信息里有非常明确的修复线索。7.4 RAG 的瓶颈到底在哪网上搜“rag瓶颈”能看到很多讨论。我自己实践下来的体会是大多数瓶颈不在大模型而在“检索质量”。很多项目跑起来后用户觉得回答傻第一反应是换更好的大模型。但如果你把一个已经检索错的片段交给再强的模型它也只能基于错误上下文硬答。所以 Day 8 之后如果你有余力优先优化三件事切分策略让文本块语义完整。检索策略增加 hit rate做重排序。评估集积累 20 个业务问题每次改动都测一遍避免“感觉变好了其实只是这次运气好”。8. 一个小项目的实际收获做完这个最小 RAG 之后我最大的感受是以后再看到“RAG 智能体”“多轮问答知识库”“企业级 RAG 框架”这些概念不再心虚了。因为你已经知道最底层的那条链路由哪些零件组成所谓智能体无非是在这条链路上再加记忆、再加工、再加工具本质是一样的。我建议你也把 Day 8 的代码保留好往docs里塞一份自己最熟悉的资料多问几个问题然后把回答效果记录下来。这是下一步优化的原始数据。等第 9 天做重排序或者增量更新的时候你就知道改进到底有没有用了。
返回列表