从零搭建 RAG 系统用 LangChain ChromaDB 给自己做一个私有知识库前言上周我把自己两年来积累的 200 多篇技术笔记丢进了一个 RAG 系统。现在我可以直接问它我之前写过关于 Redis 缓存穿透的笔记吗怎么解决的它会翻遍我所有的笔记找到最相关的几篇然后用 AI 帮我组织成一段清晰的回答。这种跟自己的知识库对话的感觉太棒了。今天完整记录搭建过程。从零开始跟着做就能跑起来。一、RAG 是什么1.1 一句话解释RAGRetrieval-Augmented Generation 先检索再生成。大模型不知道你的私有数据。RAG 的做法是先从你的知识库里检索出最相关的文档片段然后把这些片段塞给大模型当参考资料让它基于你的数据来回答问题。1.2 工作流程graph TD subgraph 离线阶段 A[原始文档] -- B[文本分割] B -- C[向量化 Embedding] C -- D[存入向量数据库] end subgraph 在线阶段 E[用户提问] -- F[问题向量化] F -- G[向量相似度检索] D -- G G -- H[取出 Top-K 相关片段] H -- I[拼接 Prompt] I -- J[大模型生成回答] end style J fill:#8b5cf6,color:#fff就两个阶段离线把你的文档切成小块转成向量存起来在线用户提问时找出最相关的块让 AI 据此回答二、环境准备2.1 安装依赖pip install langchain langchain-openai langchain-community chromadb tiktoken包作用langchainRAG 流程编排框架langchain-openaiOpenAI 模型接入chromadb轻量级向量数据库本地运行无需部署tiktokenToken 计数控制上下文长度2.2 项目结构my-rag/ ├── docs/ # 放你的文档.txt, .md, .pdf ├── chroma_db/ # ChromaDB 持久化目录自动生成 ├── ingest.py # 文档导入脚本 ├── query.py # 查询脚本 └── rag_engine.py # RAG 核心引擎三、核心实现3.1 第一步文档加载与分割# ingest.py — 文档导入 import os from langchain_community.document_loaders import ( TextLoader, DirectoryLoader, ) from langchain.text_splitter import RecursiveCharacterTextSplitter def 加载文档(文档目录: str): 从目录中加载所有 .md 和 .txt 文件 loader DirectoryLoader( 文档目录, glob**/*.md, # 支持 Markdown loader_clsTextLoader, loader_kwargs{encoding: utf-8}, ) 文档列表 loader.load() print(f 加载了 {len(文档列表)} 个文档) return 文档列表 def 分割文档(文档列表): 把长文档切成小块 splitter RecursiveCharacterTextSplitter( chunk_size500, # 每块最多 500 字 chunk_overlap50, # 相邻块重叠 50 字保证语义连续 separators[\n\n, \n, 。, , , ], ) chunks splitter.split_documents(文档列表) print(f✂️ 分割成 {len(chunks)} 个文本块) return chunkschunk_size 怎么选太小100字语义不完整检索结果碎片化太大2000字检索精度下降噪声多500 字是一个不错的起点。你可以根据自己的文档特点调整⚠️chunk_overlap 不能省如果不设重叠一句话被切成两半两半都检索不到。50 字重叠能有效缓解这个问题。3.2 第二步向量化存储# rag_engine.py — RAG 核心引擎 import os from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate class RAG引擎: def __init__(self, 数据库目录: str ./chroma_db): self.数据库目录 数据库目录 # 向量化模型把文本变成数字向量 self.embedding OpenAIEmbeddings( modeltext-embedding-3-small, # 便宜又够用 openai_api_keyos.getenv(OPENAI_API_KEY), ) # 大语言模型负责最终回答 self.llm ChatOpenAI( modelgpt-4o-mini, temperature0.3, # 低温度 回答更稳定 openai_api_keyos.getenv(OPENAI_API_KEY), ) # 向量数据库 self.向量库 None def 导入文档(self, 文档块列表): 将文档块向量化并存入 ChromaDB self.向量库 Chroma.from_documents( documents文档块列表, embeddingself.embedding, persist_directoryself.数据库目录, ) print(f 已存入 {len(文档块列表)} 个文档块到向量数据库) def 加载已有数据库(self): 加载之前已经构建好的向量数据库 self.向量库 Chroma( persist_directoryself.数据库目录, embedding_functionself.embedding, ) 数量 self.向量库._collection.count() print(f 已加载向量数据库共 {数量} 个文档块)3.3 第三步检索与生成# 接上面的 RAG引擎 类 def 查询(self, 问题: str, 返回数量: int 3) - str: RAG 查询先检索再生成 if not self.向量库: raise RuntimeError(请先导入文档或加载数据库) # 构建检索器 retriever self.向量库.as_retriever( search_typesimilarity, # 相似度检索 search_kwargs{k: 返回数量}, # 返回最相关的 k 个片段 ) # 自定义提示词模板 提示模板 PromptTemplate( template你是一个知识库助手。请根据以下参考资料回答用户的问题。 要求 1. 只根据参考资料回答不要编造信息 2. 如果参考资料中没有相关内容请如实说在知识库中没有找到相关信息 3. 回答要简洁清晰用大白话 参考资料 {context} 用户问题{question} 回答, input_variables[context, question], ) # 构建 RAG 链 qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, # 把所有检索结果塞进一个 prompt retrieverretriever, chain_type_kwargs{prompt: 提示模板}, return_source_documentsTrue, # 返回来源文档 ) # 执行查询 result qa_chain.invoke({query: 问题}) # 打印来源 print(\n 参考来源) for i, doc in enumerate(result[source_documents], 1): 来源 doc.metadata.get(source, 未知) print(f {i}. {来源}) return result[result]3.4 完整使用流程# 导入文档只需执行一次 from ingest import 加载文档, 分割文档 from rag_engine import RAG引擎 # 第一步加载并分割文档 docs 加载文档(./docs) chunks 分割文档(docs) # 第二步导入向量数据库 engine RAG引擎() engine.导入文档(chunks) # 第三步开始提问 answer engine.查询(我之前写过关于 Redis 缓存穿透的笔记吗怎么解决的) print(f\n 回答{answer})运行效果 加载了 217 个文档 ✂️ 分割成 1842 个文本块 已存入 1842 个文档块到向量数据库 参考来源 1. docs/redis/缓存穿透解决方案.md 2. docs/redis/布隆过滤器实战.md 3. docs/架构/缓存设计模式.md 回答你之前写过两篇相关笔记。缓存穿透的核心问题是查询一个数据库里 也不存在的数据导致每次请求都穿透到数据库。你记录了两种解决方案一是 用布隆过滤器在缓存层拦截不存在的 key二是对空结果也做短时间缓存设置 较短的 TTL。你推荐在高并发场景优先用布隆过滤器。四、进阶优化4.1 检索质量提升多路召回单一的向量检索有时会遗漏关键信息。加一路关键词检索做兜底。from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever def 创建混合检索器(向量库, 文档块列表, k3): 向量检索 BM25 关键词检索 混合召回 # 向量检索器语义匹配 向量检索器 向量库.as_retriever(search_kwargs{k: k}) # BM25 检索器关键词匹配 bm25检索器 BM25Retriever.from_documents(文档块列表) bm25检索器.k k # 混合检索各占 50% 权重 混合检索器 EnsembleRetriever( retrievers[向量检索器, bm25检索器], weights[0.5, 0.5], ) return 混合检索器为什么要混合向量检索擅长语义匹配缓存雪崩 和 大量 key 同时过期 能关联上BM25 擅长精确匹配搜 Redis 就一定能匹配到包含 Redis 的文档两者互补召回率更高4.2 对话记忆让 RAG 系统记住上下文支持多轮对话。from langchain.memory import ConversationBufferWindowMemory # 只保留最近 5 轮对话节省 Token memory ConversationBufferWindowMemory( k5, memory_keychat_history, return_messagesTrue, output_keyresult, ) # 在 RetrievalQA 中注入 memory qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverretriever, memorymemory, return_source_documentsTrue, ) # 多轮对话示例 qa_chain.invoke({query: 我写过哪些关于 Redis 的笔记}) qa_chain.invoke({query: 其中关于性能优化的是哪篇}) # 会带上上文五、避坑指南5.1 中文分割的坑⚠️ 默认的RecursiveCharacterTextSplitter对中文不太友好。记得在separators里加上中文标点splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[ \n\n, # 段落 \n, # 换行 。, # 句号 , # 感叹号 , # 问号 , # 分号 , # 空格 ], )5.2 Embedding 模型选择模型维度价格中文效果text-embedding-3-small1536$0.02/百万Token够用text-embedding-3-large3072$0.13/百万Token更好对于个人知识库text-embedding-3-small完全够用。省钱。5.3 Token 超限检索出来的文档块太多总 Token 数超过模型上下文窗口怎么办# 方案限制检索数量 截断 retriever 向量库.as_retriever( search_kwargs{ k: 3, # 最多返回 3 个块 } ) # 3 个块 × 500 字/块 1500 字 ≈ 2000 Token # 加上提示词和问题总共约 2500 Token # GPT-4o-mini 支持 128K完全够用六、总结整个 RAG 系统的核心就四步加载把文档读进来分割切成 500 字的小块向量化用 Embedding 模型转成数字检索生成找到最相关的块让 AI 据此回答代码总量不到 150 行。一个下午就能搭完。自己的知识库自己做主。不用担心数据泄露不用付月租费。这大概就是技术带来的温柔吧。