当前位置： 首页 > news >正文

从零搭建 RAG 系统：用 LangChain + ChromaDB 给自己做一个私有知识库

news 2026/5/28 1:13:13

从零搭建 RAG 系统：用 LangChain + ChromaDB 给自己做一个私有知识库

前言

上周我把自己两年来积累的 200 多篇技术笔记丢进了一个 RAG 系统。现在我可以直接问它："我之前写过关于 Redis 缓存穿透的笔记吗？怎么解决的？"

它会翻遍我所有的笔记，找到最相关的几篇，然后用 AI 帮我组织成一段清晰的回答。

这种"跟自己的知识库对话"的感觉，太棒了。

今天完整记录搭建过程。从零开始，跟着做就能跑起来。

一、RAG 是什么

1.1 一句话解释

RAG（Retrieval-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

就两个阶段：

  1. 离线：把你的文档切成小块，转成向量，存起来
  2. 在线：用户提问时，找出最相关的块，让 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_cls=TextLoader, loader_kwargs={"encoding": "utf-8"}, ) 文档列表 = loader.load() print(f"📄 加载了 {len(文档列表)} 个文档") return 文档列表 def 分割文档(文档列表): """把长文档切成小块""" splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每块最多 500 字 chunk_overlap=50, # 相邻块重叠 50 字（保证语义连续） separators=["\n\n", "\n", "。", "！", "？", " "], ) chunks = splitter.split_documents(文档列表) print(f"✂️ 分割成 {len(chunks)} 个文本块") return chunks

💡chunk_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( model="text-embedding-3-small", # 便宜又够用 openai_api_key=os.getenv("OPENAI_API_KEY"), ) # 大语言模型（负责最终回答） self.llm = ChatOpenAI( model="gpt-4o-mini", temperature=0.3, # 低温度 = 回答更稳定 openai_api_key=os.getenv("OPENAI_API_KEY"), ) # 向量数据库 self.向量库 = None def 导入文档(self, 文档块列表): """将文档块向量化并存入 ChromaDB""" self.向量库 = Chroma.from_documents( documents=文档块列表, embedding=self.embedding, persist_directory=self.数据库目录, ) print(f"💾 已存入 {len(文档块列表)} 个文档块到向量数据库") def 加载已有数据库(self): """加载之前已经构建好的向量数据库""" self.向量库 = Chroma( persist_directory=self.数据库目录, embedding_function=self.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_type="similarity", # 相似度检索 search_kwargs={"k": 返回数量}, # 返回最相关的 k 个片段 ) # 自定义提示词模板 提示模板 = PromptTemplate( template="""你是一个知识库助手。请根据以下参考资料回答用户的问题。 要求： 1. 只根据参考资料回答，不要编造信息 2. 如果参考资料中没有相关内容，请如实说"在知识库中没有找到相关信息" 3. 回答要简洁清晰，用大白话 参考资料： {context} 用户问题：{question} 回答：""", input_variables=["context", "question"], ) # 构建 RAG 链 qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", # 把所有检索结果塞进一个 prompt retriever=retriever, chain_type_kwargs={"prompt": 提示模板}, return_source_documents=True, # 返回来源文档 ) # 执行查询 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 创建混合检索器(向量库, 文档块列表, k=3): """向量检索 + 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( k=5, memory_key="chat_history", return_messages=True, output_key="result", ) # 在 RetrievalQA 中注入 memory qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, memory=memory, return_source_documents=True, ) # 多轮对话示例 qa_chain.invoke({"query": "我写过哪些关于 Redis 的笔记？"}) qa_chain.invoke({"query": "其中关于性能优化的是哪篇？"}) # 会带上上文

五、避坑指南

5.1 中文分割的坑

⚠️ 默认的RecursiveCharacterTextSplitter对中文不太友好。记得在separators里加上中文标点：

splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, 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 系统的核心就四步：

  1. 加载：把文档读进来
  2. 分割：切成 500 字的小块
  3. 向量化：用 Embedding 模型转成数字
  4. 检索+生成：找到最相关的块，让 AI 据此回答

代码总量不到 150 行。一个下午就能搭完。

自己的知识库自己做主。不用担心数据泄露，不用付月租费。这大概就是技术带来的温柔吧。

http://www.jsqmd.com/news/900086/

相关文章：

  • tesla P100显卡使用体验AI部署小结
  • 2026年 集成房屋/临时用房/移动房厂家推荐榜：装配式房屋/打包箱房屋/快拼箱房屋/工地临建房/模块化房屋源头厂家综合实力深度解析与选购指南 - 品牌企业推荐师（官方）
  • 使用curl命令快速测试taotoken大模型api连通性与返回格式
  • CCS链接警告剖析：SECTIONS缺失导致输出段‘XXXXXXX’未定义的修复策略
  • 有哪些AI写作辅助平台是真的贴合学术规范，而不是模板套话？
  • 13 - 异常处理
  • 从零到一：MobileNet V1/V2 核心架构解析与轻量级模型实战搭建
  • 告别自签名警告：为Proxmox VE管理界面配置域名与SSL证书
  • LoongSon——PMON实战命令手册：从启动到调试
  • 2026年Q2云南厨电工厂深度解析：家园优品如何引领区域产业升级？ - 2026年企业资讯
  • 3分钟学会Windows 11终极优化：Win11Debloat免费系统清理完整指南
  • 告别手写定位符！用 Appium Inspector 的录制和搜索功能快速生成 Python/Java 测试脚本
  • 68_《智能体微服务架构企业级实战教程》运维与部署之编写docker-compose部署脚本
  • LeagueAkari：英雄联盟玩家的智能效率革命，告别传统低效操作
  • 2026年Q2苏州的经济合同纠纷法律服务深度解析与选择指南 - 2026年企业资讯
  • 从Linux到SPDK：NVMe Namespace的创建、绑定与高性能存储实践
  • SAP FICO 集成场景下GL_ACCT_MASTER_SAVE的实战应用与BAPI封装
  • AI 基础概念卡片
  • ChatGPT客服话术设计全链路拆解，从客户投诉归因→话术颗粒度分级→AB测试验证→实时迭代机制
  • 工期紧张时的救星：哪些HC-276厂商能做到灵活排产并按时交付？ - 品牌2025
  • Cortex-R4处理器nCPUHALT信号原理与应用解析
  • Pearcleaner：Mac应用清理的终极解决方案，彻底释放存储空间
  • Notepad++ 详细下载安装全流程指南
  • 2026年 热电阻/铠装热电阻/温度传感器厂家推荐榜：TKWZPK-24-440/WZPK-24-440型号精度与耐用性深度解析 - 品牌企业推荐师（官方）
  • 边缘计算安全最佳实践：保护边缘环境中的数据和应用
  • 第06篇｜module.json5 深读：设备类型、权限、Ability 与智能体配置
  • 【Qt】QModbusRtuSerialMaster：串行Modbus客户端实战与帧时序调优
  • 被低估的超级不锈钢：为什么高端装备都在悄悄使用UNS S21800？ - 品牌2025
  • Go语言timer源码：时间调度实现深度解析
  • 航空发动机叶盘系统的多场耦合振动特性及优化设计【附程序】