在上一篇文章中,我们完成了文档的加载和分割,但这只是第一步。想象一下,你的知识库里有 10000 个文档片段,当用户问"如何优化数据库性能?"时,系统如何在海量信息中快速找到最相关的内容?这就是检索的价值——它是连接问题和答案的桥梁。检索质量的好坏,直接决定了 RAG 系统能否给出准确答案。
一、概念速览
在[《构建你的第一个知识库(下)》]中,介绍了 Embedding 的基本概念:将文本转换为高维向量,让计算机理解语义相似度。我们也学习了向量数据库的基础用法。
混合检索
本文重点介绍混合检索(Hybrid Search),它结合了两种检索方式的优势:
| 检索方式 | 原理 | 优势 | 劣势 |
|---|---|---|---|
| 向量检索 | 语义相似度计算 | 理解同义词、上下文 | 对精确关键词不敏感 |
| 关键词检索(BM25) | 词频统计 | 精确匹配专业术语 | 无法理解语义 |
| 混合检索 | 两者加权融合 | 兼顾语义和精确性 | 配置稍复杂 |
示例:用户搜索"Python 爬虫 403 错误",向量检索能找到"网页爬取被拒绝"的相关内容,关键词检索则精确定位"403 状态码"的技术文档。
二、实战
Chroma 向量数据库配置
Chroma 是轻量级向量数据库,非常适合快速原型开发:
from langchain_community.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader import os # 加载文档(延续上一篇的示例) loader = TextLoader("enterprise_docs.txt", encoding="utf-8") documents = loader.load() # 文本分割 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, length_function=len, ) chunks = text_splitter.split_documents(documents) # 创建向量数据库 persist_directory = "./chroma_db" vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory=persist_directory, collection_name="enterprise_kb" ) print(f"✅ 已向量化 {len(chunks)} 个文档片段")配置要点:
persist_directory:数据持久化目录,重启后数据不丢失collection_name:集合名称,可创建多个知识库- Chroma 会自动处理向量索引,无需手动优化
基础向量检索
先看最简单的向量检索效果:
# 创建检索器 retriever_vector = vectorstore.as_retriever( search_type="similarity", # 相似度检索 search_kwargs={"k": 3} # 返回前 3 个结果 ) # 测试查询 query = "如何提升数据库查询速度?" results = retriever_vector.get_relevant_documents(query) print(f"查询:{query} ") for i, doc in enumerate(results, 1): print(f"结果{i}:{doc.page_content[:100]}...") print(f"相关度评分:{doc.metadata.get('score', 'N/A')} ")实现混合检索
混合检索结合向量检索和 BM25 关键词检索:
from langchain.retrievers import BM25Retriever, EnsembleRetriever # 准备文档内容(BM25 需要原始文本) doc_texts = [doc.page_content for doc in chunks] # 创建 BM25 检索器 retriever_bm25 = BM25Retriever.from_texts( doc_texts, metadatas=[doc.metadata for doc in chunks] ) retriever_bm25.k = 3 # 返回前 3 个结果 # 创建混合检索器 ensemble_retriever = EnsembleRetriever( retrievers=[retriever_vector, retriever_bm25], weights=[0.6, 0.4] # 向量检索权重 0.6,BM25 权重 0.4 ) # 测试混合检索 query = "PostgreSQL 索引优化" results_hybrid = ensemble_retriever.get_relevant_documents(query) print(f"混合检索结果:") for i, doc in enumerate(results_hybrid, 1): print(f"{i}. {doc.page_content[:80]}...")完整代码示例
from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain_community.document_loaders import DirectoryLoader, TextLoader import os class EnterpriseKnowledgeBase: """企业知识库检索系统""" def __init__(self, docs_path, persist_dir="./chroma_db"): self.docs_path = docs_path self.persist_dir = persist_dir self.embeddings = None self.vectorstore = None self.chunks = None def load_documents(self): """加载文档""" print("📂 正在加载文档。..") loader = DirectoryLoader( self.docs_path, glob="**/*.txt", loader_cls=TextLoader, loader_kwargs={'encoding': 'utf-8'} ) documents = loader.load() # 文本分割 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["。", "!", "?", ";", " "] ) self.chunks = text_splitter.split_documents(documents) print(f"✅ 已加载 {len(documents)} 个文档,分割为 {len(self.chunks)} 个片段") def initialize_embeddings(self, model_name="all-MiniLM-L6-v2"): """初始化 Embedding 模型""" print(f"🔧 正在加载 Embedding 模型: {model_name}") self.embeddings = HuggingFaceEmbeddings( model_name=f"sentence-transformers/{model_name}", model_kwargs={'device': 'cpu'}, encode_kwargs={'normalize_embeddings': True} ) print("✅ Embedding 模型加载完成") def create_vectorstore(self): """创建向量数据库""" print("💾 正在创建向量数据库。..") self.vectorstore = Chroma.from_documents( documents=self.chunks, embedding=self.embeddings, persist_directory=self.persist_dir, collection_name="enterprise_kb" ) print("✅ 向量数据库创建完成") def create_hybrid_retriever(self, vector_weight=0.6, bm25_weight=0.4, k=3): """创建混合检索器""" # 向量检索器 retriever_vector = self.vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": k} ) # BM25 检索器 doc_texts = [doc.page_content for doc in self.chunks] retriever_bm25 = BM25Retriever.from_texts( doc_texts, metadatas=[doc.metadata for doc in self.chunks] ) retriever_bm25.k = k # 混合检索器 ensemble_retriever = EnsembleRetriever( retrievers=[retriever_vector, retriever_bm25], weights=[vector_weight, bm25_weight] ) print(f"✅ 混合检索器创建完成(向量:{vector_weight}, BM25:{bm25_weight})") return ensemble_retriever def search(self, query, retriever, top_k=3): """执行检索""" results = retriever.get_relevant_documents(query) return results[:top_k] def compare_retrievers(self, query): """对比不同检索器效果""" # 向量检索 retriever_vector = self.vectorstore.as_retriever( search_kwargs={"k": 3} ) results_vector = retriever_vector.get_relevant_documents(query) # BM25 检索 doc_texts = [doc.page_content for doc in self.chunks] retriever_bm25 = BM25Retriever.from_texts(doc_texts) retriever_bm25.k = 3 results_bm25 = retriever_bm25.get_relevant_documents(query) # 混合检索 retriever_hybrid = self.create_hybrid_retriever() results_hybrid = retriever_hybrid.get_relevant_documents(query) # 打印对比结果 print(f"{'='*60}") print(f"查询:{query}") print(f"{'='*60}") print("📊 向量检索结果:") for i, doc in enumerate(results_vector, 1): print(f" {i}. {doc.page_content[:60]}...") print("📊 BM25 检索结果:") for i, doc in enumerate(results_bm25, 1): print(f" {i}. {doc.page_content[:60]}...") print("📊 混合检索结果:") for i, doc in enumerate(results_hybrid, 1): print(f" {i}. {doc.page_content[:60]}...") # 使用示例 if __name__ == "__main__": # 初始化知识库 kb = EnterpriseKnowledgeBase(docs_path="./docs") # 加载文档并创建向量库 kb.load_documents() kb.initialize_embeddings() kb.create_vectorstore() # 创建混合检索器 hybrid_retriever = kb.create_hybrid_retriever( vector_weight=0.6, bm25_weight=0.4, k=3 ) # 测试查询 test_queries = [ "如何优化数据库查询性能?", "PostgreSQL B-tree 索引的原理", "Python 内存泄漏如何排查" ] for query in test_queries: print(f"🔍 查询:{query}") results = kb.search(query, hybrid_retriever) for i, doc in enumerate(results, 1): print(f"结果{i}:") print(f" 内容:{doc.page_content[:100]}...") print(f" 来源:{doc.metadata.get('source', 'N/A')}") # 对比不同检索器 kb.compare_retrievers("PostgreSQL 索引优化最佳实践")三、总结
通过本文的实战演练,我们掌握了知识库检索的核心技术。
混合检索是王道:向量检索+BM25 的组合能覆盖更多场景,权重建议从 0.6:0.4 开始调优。
持续优化:建立测试查询集,定期评估检索准确率,根据业务场景调整策略。
如何学习大模型 AI ?
由于新岗位的生产效率,要优于被取代岗位的生产效率,所以实际上整个社会的生产效率是提升的。
但是具体到个人,只能说是:
“最先掌握AI的人,将会比较晚掌握AI的人有竞争优势”。
这句话,放在计算机、互联网、移动互联网的开局时期,都是一样的道理。
我在一线互联网企业工作十余年里,指导过不少同行后辈。帮助很多人得到了学习和成长。
我意识到有很多经验和知识值得分享给大家,也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑,所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限,很多互联网行业朋友无法获得正确的资料得到学习提升,故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。
第一阶段(10天):初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识,对大模型 AI 的理解超过 95% 的人,可以在相关讨论时发表高级、不跟风、又接地气的见解,别人只会和 AI 聊天,而你能调教 AI,并能用代码将大模型和业务衔接。
- 大模型 AI 能干什么?
- 大模型是怎样获得「智能」的?
- 用好 AI 的核心心法
- 大模型应用业务架构
- 大模型应用技术架构
- 代码示例:向 GPT-3.5 灌入新知识
- 提示工程的意义和核心思想
- Prompt 典型构成
- 指令调优方法论
- 思维链和思维树
- Prompt 攻击和防范
- …
第二阶段(30天):高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习,学会构造私有知识库,扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架,抓住最新的技术进展,适合 Python 和 JavaScript 程序员。
- 为什么要做 RAG
- 搭建一个简单的 ChatPDF
- 检索的基础概念
- 什么是向量表示(Embeddings)
- 向量数据库与向量检索
- 基于向量检索的 RAG
- 搭建 RAG 系统的扩展知识
- 混合检索与 RAG-Fusion 简介
- 向量模型本地部署
- …
第三阶段(30天):模型训练
恭喜你,如果学到这里,你基本可以找到一份大模型 AI相关的工作,自己也能训练 GPT 了!通过微调,训练自己的垂直大模型,能独立训练开源多模态大模型,掌握更多技术方案。
到此为止,大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗?
- 为什么要做 RAG
- 什么是模型
- 什么是模型训练
- 求解器 & 损失函数简介
- 小实验2:手写一个简单的神经网络并训练它
- 什么是训练/预训练/微调/轻量化微调
- Transformer结构简介
- 轻量化微调
- 实验数据集的构建
- …
第四阶段(20天):商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知,可以在云端和本地等多种环境下部署大模型,找到适合自己的项目/创业方向,做一名被 AI 武装的产品经理。
- 硬件选型
- 带你了解全球大模型
- 使用国产大模型服务
- 搭建 OpenAI 代理
- 热身:基于阿里云 PAI 部署 Stable Diffusion
- 在本地计算机运行大模型
- 大模型的私有化部署
- 基于 vLLM 部署大模型
- 案例:如何优雅地在阿里云私有部署开源大模型
- 部署一套开源 LLM 项目
- 内容安全
- 互联网信息服务算法备案
- …
学习是一个过程,只要学习就会有挑战。天道酬勤,你越努力,就会成为越优秀的自己。
如果你能在15天内完成所有的任务,那你堪称天才。然而,如果你能完成 60-70% 的内容,你就已经开始具备成为一名大模型 AI 的正确特征了。