|
**Faiss简介
!pip install faiss-cpu
FAISS（Facebook AI Similarity Search）
是一个由Facebook AI Research开发的库，用于高效地处理和搜索大规模的向量数据。
在许多AI和机器学习应用中，我们经常需要处理高维度的向量数据，
并需要快速地 找到与给定向量最相似的向量 。
例如，
在推荐系统中，我们可能需要找到与用户的兴趣向量最相似的商品向量；
在自然语言处理中，我们可能需要找到与给定词向量最相似的词向量。
这种操作被称为“最近邻搜索”。
全部代码
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv(filename="env.txt"))
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAI
from langchain_openai import OpenAIEmbeddings
!echo "肺主呼吸之气、一身之气。肺司呼吸的功能正常，则气道通畅，呼吸调匀，面有色泽。若病邪犯肺，影响其呼吸功能，则出现胸满咳嗽、喘促、呼吸不利、面色紫绀等症状。">/tmp/a.txt
!echo "from langchain_community.document_loaders import TextLoader">>/tmp/a.txt
!echo "from langchain_community.vectorstores import FAISS">>/tmp/a.txt
loader = TextLoader('/tmp/a.txt')
documents = loader.load()
# 准备文本分割器
text_splitter = RecursiveCharacterTextSplitter(chunk_size=50, chunk_overlap=0)
# 将大文档分成小块
texts = text_splitter.split_documents(documents)
print(texts[0].page_content) # str
#选择嵌入模型
#embeddings = OpenAIEmbeddings(deployment='text-embedding-ada-002')
#model = OpenAIEmbeddings(model="text-embedding-3-large")
embeddings = OpenAIEmbeddings()
#将文本保存到FAISS 向量数据库中
db = FAISS.from_documents(texts, embeddings)
#大小写影响不大
docs = retriever.invoke("FAISS")
print("\n\n".join([x.page_content[:200] for x in docs[:2]]))
FAISS
TextLoader
能，则出现胸满咳嗽、喘促、呼吸不利、面色紫绀等症状。
#大小写影响不大
docs = retriever.invoke("FAISS")
print("\n\n".join([x.page_content[:200] for x in docs[:2]]))
FAISS
TextLoader
docs = retriever.invoke("呼吸")
print("\n\n".join([x.page_content[:200] for x in docs[:2]]))
肺主呼吸之气、一身之气。肺司呼吸的功能正常，则气道通畅，呼吸调匀，面有色泽。若病邪犯肺，影响其呼吸功
能，则出现胸满咳嗽、喘促、呼吸不利、面色紫绀等症状。
FAISS.from_documents
FAISS.from_documents(
documents: 'List[Document]',
embedding: 'Embeddings',
**kwargs: 'Any',
)
Docstring:
Return VectorStore initialized from documents and embeddings.
#将文本保存到FAISS 向量数据库中
db = FAISS.from_documents(texts, embeddings)
# 准备文件搜索器
retriever = db.as_retriever()
#返回一个文档数组，其元素按向量相似度从高到低排列
docs = retriever.invoke("faiss")
print("\n\n".join([x.page_content[:200] for x in docs[:3]]))
FAISS.from_texts
FAISS.from_texts(
texts: 'List[str]',
embedding: 'Embeddings',
metadatas: 'Optional[List[dict]]' = None,
ids: 'Optional[List[str]]' = None,
**kwargs: 'Any',
) -> 'FAISS'
Docstring:
Construct FAISS wrapper from raw documents.
This is a user friendly interface that:
1. Embeds documents.
2. Creates an in memory docstore
3. Initializes the FAISS database
db_text =FAISS.from_texts(texts, embeddings)
retriever_text = db_text.as_retriever()
docs = retriever_text.invoke("FAISS")
print("\n\n".join([x.page_content[:200] for x in docs[:2]]))
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv(filename="env.txt"))
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAI
from langchain_openai import OpenAIEmbeddings
# 准备文本分割器
text_splitter = RecursiveCharacterTextSplitter(chunk_size=50, chunk_overlap=10,length_function = len)
text="""
肺主呼吸之气、一身之气。肺司呼吸的功能正常，则气道通畅，呼吸调匀，面有色泽。若病邪犯肺，影响其呼吸功能，则出现胸满咳嗽、喘促、呼吸不利、面色紫绀等症状。
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
"""
texts = text_splitter.split_text(text)
texts
['肺主呼吸之气、一身之气。肺司呼吸的功能正常，则气道通畅，呼吸调匀，面有色泽。若病邪犯肺，影响其呼吸',
'病邪犯肺，影响其呼吸功能，则出现胸满咳嗽、喘促、呼吸不利、面色紫绀等症状。',
'from langchain_community.document_loaders import',
'import TextLoader',
'from langchain_community.vectorstores import',
'import FAISS']
# 准备文本分割器
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=10,length_function = len)
text="""
肺主呼吸之气、一身之气。肺司呼吸的功能正常，则气道通畅，呼吸调匀，面有色泽。若病邪犯肺，影响其呼吸功能，则出现胸满咳嗽、喘促、呼吸不利、面色紫绀等症状。
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
"""
texts = text_splitter.split_text(text)
texts
['肺主呼吸之气、一身之气。肺司呼吸的功能正常，则气道通畅，呼吸调匀，面有色泽。若病邪犯肺，影响其呼吸功能，则出现胸满咳嗽、喘促、呼吸不利、面色紫绀等症状。',
'from langchain_community.document_loaders import TextLoader',
'from langchain_community.vectorstores import FAISS']
db_text =FAISS.from_texts(texts, embeddings)
retriever_text = db_text.as_retriever()
docs = retriever_text.invoke("FAISS")
print("\n\n".join([x.page_content[:200] for x in docs[:2]]))
FAISS
TextLoader
from_embeddings：覆盖,新加载的向量会覆盖内存数据库中的向量
# 定义一个长文本。
text = """
肺与美容的关系
肺主气主呼吸
肺主呼吸之气、一身之气。肺司呼吸的功能正常，则气道通畅，呼吸调匀，面有色泽。若病邪犯肺，影响其呼吸功能，则出现胸满咳嗽、喘促、呼吸不利、面色紫绀等症状。
肺主宣发与肃降
肺主宣发是指肺气向上升宣和向外布散的功能。
若肺气不能宣发而凝滞，清阳不升，则见胸满、鼻塞、咳喘、面色晦暗;
浊气不能肃降，也能引起呼吸短促、喘促、咳痰等症状;
肺与大肠相表里，肺失肃降，则大肠传导失常，肠道实热，表里失和，面部就易出黄褐斑或生痤疮、酒渣鼻。
肺主皮毛通调水道
肺主皮毛，是指肺脏通过它的宣发作用把水谷精微输布于皮毛，以滋养周身皮肤、毛发、肌肉，也能温分肉、充皮肤、肥腠理、司开阖，防御外邪入侵。肺气足则皮肤滋润光滑有弹性，毫毛浓密光泽等;肺气虚，不能行气与津液以温养毫毛，毫毛之营养不足，就会憔悴枯槁，皮肤干燥，毛发暗淡枯稿，面色淡白;卫外不固则易发风疹过敏等症;肺热上熏则发痤疮、酒渣鼻、皮炎等症。
肺开窍于鼻
鼻是肺呼吸的通道，所以称“鼻为肺窍”。肺与大肠相表里，肺失肃降，则大肠传导失常，粪便排出不畅。临床上，肺失清肃，则大便困难;大肠实热，又引起肺气不利而咳喘胸满;表里失和，患者面部就易出黄褐斑或生痤疮。
"""
from langchain.text_splitter import CharacterTextSplitter
text_splitter = CharacterTextSplitter(
separator = "\n",
chunk_size = 200,
chunk_overlap = 50
)
texts = text_splitter.split_text(text)
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())
embeddings = OpenAIEmbeddings()
#1. Embeds documents.
#2. Creates an in memory docstore
#3. Initializes the FAISS database
faiss = FAISS.from_texts(texts, embeddings)
embeddings = OpenAIEmbeddings()
text_embeddings = embeddings.embed_documents(texts)
text_embedding_pairs = zip(texts, text_embeddings)
type(text_embeddings)
list
len(text_embeddings[0])
1536
#1. Embeds documents.
#2. Creates an in memory docstore
#3. Initializes the FAISS database
faiss = FAISS.from_embeddings(text_embedding_pairs, embeddings)
mytab('ulfaiss', 'tbfaiss');
FAISS检索
Retriever（检索器）
Retriever（检索器）
在使用FAISS（Facebook AI Similarity Search）库进行大规模向量搜索时，
通常会遇到直接使用FAISS的API进行搜索（如faiss.search()或faiss.knn_query()等）
和将FAISS封装为Retriever（检索器）并通过invoke()方法调用的两种方式。
虽然两者都能实现向量搜索的功能，但将FAISS转化为Retriever后使用，
在某些场景下具有一些明显的优势，主要体现在以下几个方面：
更高的抽象级别和灵活性：
将FAISS封装为Retriever后，可以在更高层次的抽象上操作，
这意味着你可以更容易地与其他组件（如文本处理、数据处理流水线等）集成。
这种封装使得FAISS的使用更加灵活，尤其是在复杂的机器学习或数据科学项目中。
统一的接口：
如果你的项目中使用了多个检索器（如基于Elasticsearch、向量数据库等），
将FAISS也封装为Retriever可以提供一个统一的接口来调用它们。
这有助于简化代码结构，使得在不同检索器之间切换或组合使用变得更加容易。
易于扩展和维护：
随着项目的发展，可能需要添加新的检索器或修改现有的检索逻辑。
将FAISS封装为Retriever后，你可以通过继承或组合的方式轻松地扩展或修改检索逻辑，
而无需修改大量底层代码。
更好的错误处理和日志记录：
在Retriever的实现中，你可以添加自定义的错误处理和日志记录逻辑，
以便更好地监控和调试检索过程。
这对于大型项目或生产环境尤为重要，因为它们需要更高的稳定性和可维护性。
支持异步和并行处理：
某些Retriever实现可能支持异步或并行处理，这可以显著提高检索性能，
尤其是在处理大量查询或大型数据集时。
虽然FAISS本身也支持多线程搜索，但将FAISS封装为Retriever后，
你可以更容易地将其集成到支持异步或并行处理的框架中。
与框架和库的兼容性：
在某些情况下，将FAISS封装为Retriever可能是为了满足特定框架或库的要求。
例如，在构建基于Jina AI的神经网络搜索系统时，
将FAISS作为后端搜索引擎并封装为Retriever是推荐的做法，因为这样可以更容易地与其他Jina组件集成。
综上所述，将FAISS转化为Retriever后使用的主要优势在于提高了
抽象级别、灵活性、可扩展性、可维护性以及与其他组件的兼容性。
然而，这也取决于你的具体需求和项目上下文。
如果项目规模较小且不需要复杂的集成和扩展，直接使用FAISS的API可能更为简单直接。
mytab('ulfaiss_research', 'tbfaiss_research');
FAISS持久化
index存储与加载
index注意事项
index只是index
index存储与加载
import faiss
import numpy as np
# 创建一些随机数据
np.random.seed(42)
data = np.random.rand(1000, 64).astype('float32') # 假设有1000个64维的向量
# 创建一个平面索引（IndexFlatL2），使用L2距离
index = faiss.IndexFlatL2(64)
# 将数据添加到索引中
index.add(data)
# 保存索引到文件
faiss.write_index(index, "my_index.index")
# 从文件中加载索引
loaded_index = faiss.read_index("my_index.index")
# 验证加载的索引是否包含相同的数据（这里仅作为示例，通常不会直接比较索引对象）
# 实际上，你应该通过查询来验证索引的有效性
query = np.random.rand(1, 64).astype('float32')
dis, ind = loaded_index.search(query, 5) # 查询最相似的5个向量
dis
array([[6.099057 , 6.650052 , 6.803897 , 6.9097624, 6.952726 ]],
dtype=float32)
ind
array([[502, 606, 234, 85, 162]])
query
array([[0.15736598, 0.02078138, 0.32708594, 0.8243158 , 0.01537996,
0.10943393, 0.80685484, 0.3058517 , 0.9382135 , 0.38024414,
0.79670393, 0.25878984, 0.06183124, 0.8815545 , 0.92729646,
0.70809966, 0.7700678 , 0.00712762, 0.7538235 , 0.4721964 ,
0.45279047, 0.54006994, 0.27121022, 0.6962885 , 0.29558888,
0.5586066 , 0.13692723, 0.67507565, 0.18961276, 0.17041399,
0.02428316, 0.12178577, 0.23257025, 0.22230937, 0.32767144,
0.05441642, 0.935198 , 0.11917588, 0.7019639 , 0.53507316,
0.5527888 , 0.10551683, 0.24321494, 0.7798813 , 0.46368647,
0.45706782, 0.5459118 , 0.43380588, 0.80613595, 0.7487052 ,
0.3170042 , 0.8120811 , 0.5186064 , 0.6277368 , 0.86617583,
0.47599325, 0.75626034, 0.62808734, 0.42674494, 0.7890484 ,
0.10515912, 0.5812202 , 0.04641076, 0.15616913]], dtype=float32)
data[ind]
确保在调用write_index之前，你的索引已经包含了所有需要的数据，包括向量和可能的ID映射等。
read_index加载的索引将是一个与原来内存中相同的索引对象的副本，所以你可以直接使用它来进行搜索等操作。
FAISS索引持久化功能依赖于磁盘I/O，因此可能会比内存操作慢一些。但在需要节省内存或持久化索引以供后续使用时，这是一个非常有用的特性。
综上所述，FAISS通过write_index和read_index函数提供了索引的持久化功能，允许你将内存中的索引结构及其向量数据保存到磁盘上，并在需要时重新加载它们。
index注意事项
索引类型：
示例中使用了IndexFlatL2，这是一种简单的索引类型，
适用于小规模数据集或作为其他更复杂索引（如IndexIVFFlat、IndexIVFPQ）的基准。
对于大规模数据集，建议使用更高效的索引类型。
数据规模：
当处理大规模数据集时，索引的保存和加载可能会消耗较多的磁盘空间和时间。
因此，在实际应用中，需要根据具体情况选择合适的索引类型和持久化策略。
异常处理：
在保存和加载索引时，应添加适当的异常处理逻辑以处理可能出现的错误（如文件权限问题、磁盘空间不足等）。
FAISS当数据量大时，应该使用哪种索引
IVF（Inverted File）索引
inverted 英/ɪnˈvɜːtɪd/ 美/ɪnˈvɜːrtɪd/ adj. 反向的，倒转的，颠倒的，（尤指）倒置的 v.（使）倒转，颠倒，倒置
特点：
IVF索引是FAISS中应用最广泛的索引类型之一，
它将向量空间划分为多个子空间（聚类），并为每个子空间构建一个倒排索引（哈希表）。
搜索时，首先根据查询向量找到对应的子空间，然后在该子空间中进行相似度计算和搜索。
这种方法通过减少搜索的空间范围，显著提高了搜索效率。
适用场景：
适用于大规模数据集，能够显著加快搜索速度，同时保持较高的召回率。
PQ（Product Quantization）索引
特点：
PQ索引是一种用于大规模向量量化的索引类型。
它将向量划分为若干个子向量，并对每个子向量进行量化，
然后将量化后的子向量拼接起来形成新的向量表示。
在搜索时，通过比较量化后的向量来计算相似度。
PQ索引通过减小向量的维度，降低了计算和存储的复杂度。
适用场景：
适用于需要量化向量以节省存储空间和计算资源的情况，尤其适用于大规模数据集。
index只是index：存储的是向量，搜索的输入也是向量，而不是文本**
# 定义一个长文本。
text = """
肺与美容的关系
肺主气主呼吸
肺主呼吸之气、一身之气。肺司呼吸的功能正常，则气道通畅，呼吸调匀，面有色泽。若病邪犯肺，影响其呼吸功能，则出现胸满咳嗽、喘促、呼吸不利、面色紫绀等症状。
肺主宣发与肃降
肺主宣发是指肺气向上升宣和向外布散的功能。
若肺气不能宣发而凝滞，清阳不升，则见胸满、鼻塞、咳喘、面色晦暗;
浊气不能肃降，也能引起呼吸短促、喘促、咳痰等症状;
肺与大肠相表里，肺失肃降，则大肠传导失常，肠道实热，表里失和，面部就易出黄褐斑或生痤疮、酒渣鼻。
肺主皮毛通调水道
肺主皮毛，是指肺脏通过它的宣发作用把水谷精微输布于皮毛，以滋养周身皮肤、毛发、肌肉，也能温分肉、充皮肤、肥腠理、司开阖，防御外邪入侵。肺气足则皮肤滋润光滑有弹性，毫毛浓密光泽等;肺气虚，不能行气与津液以温养毫毛，毫毛之营养不足，就会憔悴枯槁，皮肤干燥，毛发暗淡枯稿，面色淡白;卫外不固则易发风疹过敏等症;肺热上熏则发痤疮、酒渣鼻、皮炎等症。
肺开窍于鼻
鼻是肺呼吸的通道，所以称“鼻为肺窍”。肺与大肠相表里，肺失肃降，则大肠传导失常，粪便排出不畅。临床上，肺失清肃，则大便困难;大肠实热，又引起肺气不利而咳喘胸满;表里失和，患者面部就易出黄褐斑或生痤疮。
"""
from langchain.text_splitter import CharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())
```python
def get_embeddings(text):
text_splitter = CharacterTextSplitter(
separator = "\n",
chunk_size = 1000,
chunk_overlap = 50
)
texts = text_splitter.split_text(text)
embeddings = OpenAIEmbeddings()
text_embeddings = embeddings.embed_documents(texts)
return texts,text_embeddings
# get_embeddings(text)
```
```
texts,embeddings = get_embeddings(text)
```
```python
import os
import faiss
import numpy as np
```
```
index_path = "my_index.index"
# if os.path.exists(index_path):
# 创建一个平面索引（IndexFlatL2），使用L2距离
dimensions = len(embeddings[0])
index = faiss.IndexFlatL2(dimensions)
# 将数据添加到索引中
index.add(np.array(embeddings))
# 保存索引到文件
faiss.write_index(index, "my_index.index")
# 从文件中加载索引
loaded_index = faiss.read_index("my_index.index")
```
```
query = "肺与美容的关系"
_,embeddings_query = get_embeddings(query)
```
```
dis, ind = loaded_index.search(np.array(embeddings_query),k=1)
dis
array([[0.23161677]], dtype=float32)
ind
array([[0]])
```
|
|
|
|
|
|
### 从一个真实场景说起
想象一下：
- 你所在的公司有大量 **内部文档、产品手册、技术资料**
- 用户提问：**“你们的产品支持哪些操作系统？”**
- 直接问大模型，得到：**“抱歉，我不知道你们产品的具体信息”**
数据明明就在公司文档里，模型却答不上来——这就是 RAG 要解决的核心问题。
### 大模型的两大固有局限
| 局限 | 原因 | 典型表现 |
|------|------|----------|
| **知识不是实时的** | 训练数据有截止日期，重新训练成本极高 | 无法回答“2024 年最新的 AI 发展趋势是什么” |
| **不知道私有领域知识** | 训练语料里没有你的内部数据 | 无法回答“我们公司的请假政策是什么” |
### RAG 的解题思路
**RAG（Retrieval-Augmented Generation，检索增强生成）= 检索（Retrieval）+ 生成（Generation）**
在“用户提问”和“模型回答”之间插入一步检索：
```text
用户提问
↓
① 从你的知识库中检索最相关的内容片段
↓
② 把「检索到的内容 + 用户问题」一起放入提示词
↓
③ 大模型基于给定内容作答（而不是凭记忆编造）
```
由于回答有据可查，RAG 能从根源上 **缓解幻觉**、**突破知识时效限制**，还能让回答 **带引用、可追溯**。
|
|
### 两种解决方案
让大模型掌握垂直领域知识，天然能想到两条路：
| 方案 | 类比 | 一句话描述 |
|------|------|------------|
| **微调 Fine-tuning** | 闭卷考试 | 考试前先学习资料，把知识“背”进模型参数里 |
| **RAG 检索增强生成** | 开卷考试 | 考试时允许翻资料，查完再组织语言作答 |
### 详细对比
| 维度 | 微调 | RAG |
|------|------|-----|
| 成本 | 高（算力 + 标注数据，万美元级起步） | 低（几乎只需向量库 + 检索服务） |
| 知识更新 | 困难，需重新训练 | 灵活，实时增删文档即可 |
| 可解释性 | 弱，知识融在参数里 | 强，能给出引用出处 |
| 对模型原有能力 | 可能造成“灾难性遗忘” | 不影响 |
| 擅长 | 改变风格/语气/格式、固化专业能力 | 注入事实性、时效性知识 |
| 短板 | 贵、慢、难更新 | 依赖检索质量，上下文长度受限 |
### 怎么选？
- 需要 **事实知识问答、知识频繁更新、要求可追溯** → 优先 RAG
- 需要 **特定风格输出、领域术语表达、复杂指令遵循** → 优先微调
- 生产系统常见做法：**RAG 打底 + 少量微调调风格**，两者并不互斥
|
|
### RAG 关键术语表（先混个脸熟）
| 术语 | 英文 | 一句话解释 |
|------|------|-----------|
| 向量工程 | Vector Engineering | 把文本/图像等非结构化数据转为向量并加以利用的工程实践 |
| 文本向量 / 嵌入 | Embeddings | 把文本映射成一串浮点数（如 1536 维），语义相近则向量相近 |
| 文档切块 | Chunking | 把长文档切成合适大小的片段（chunk），是检索的基本单位 |
| 向量数据库 | Vector Database | 专门存储向量并支持高效相似度检索的数据库 |
| 检索器 | Retriever | 封装“给定 query 返回相关文档”的统一接口 |
| 相似度 | Similarity | 衡量两个向量语义接近程度：余弦相似度、欧氏距离、点积 |
| Top-K | — | 检索时返回相似度最高的 K 个文档块 |
| 重排 | Rerank | 用更精细的模型对初筛结果重新排序，提升精度 |
| 倒排索引 | Inverted Index | 关键词检索的底层数据结构：词 → 出现该词的文档列表 |
| 提示词模板 | Prompt Template | 把「检索结果 + 用户问题」填入固定格式的提示词骨架 |
### 三个核心角色
```text
┌─────────────┐ 向量化+入库 ┌─────────────┐
│ 原始文档 │ ────────────▶ │ 向量数据库 │ ← 离线：建立索引
└─────────────┘ └─────────────┘
▲ 相似度检索 Top-K
┌─────────────┐ 组装提示词 ┌─────┴───────┐
│ 大模型 LLM │ ◀──────────── │ 用户问题 │ ← 在线：检索 + 生成
└─────────────┘ └─────────────┘
```
### 向量工程的技术特点
- **高维表示**：通常 512~1536 维向量
- **语义保持**：相似内容的向量在空间中距离较近
- **多模态支持**：文本、图像、音频都能向量化
|
|
### 典型应用场景
**企业级应用**
| 场景 | 说明 |
|------|------|
| 智能客服 | 基于企业文档的自动问答、FAQ 匹配、多轮对话 |
| 文档检索 | 法律条文/案例检索、技术文档搜索、产品手册查询 |
| 知识管理 | 企业知识库整合、研发文献管理、培训资料问答 |
**个人应用**：个人笔记智能检索、学习助手（基于教材问答）、代码助手（基于代码库问答）
### RAG vs 传统搜索
| 特性 | 传统搜索 | RAG 系统 |
|------|----------|----------|
| 输出 | 相关文档链接列表 | 直接的自然语言回答 |
| 理解 | 关键词匹配 | 语义理解 |
| 整合 | 用户自己阅读整合 | 自动整合多文档信息 |
| 体验 | 多步操作 | 一步到位 |
### 给初学者的学习路径（本文档即按此顺序组织）
1. **理解问题**：为什么需要 RAG？→ 见「RAG入门导读」（本章）
2. **掌握流程**：离线建索引 + 在线检索生成两大阶段 → 见「RAG核心流程」
3. **动手实战**：跑通课上代码 rag-embeddings → 见「rag-embeddings实战」
4. **认识工具**：常用向量数据库与 LangChain → 见「常见向量数据库」「LangChain索引与图」
5. **进阶提升**：GraphRAG、检索优化与评估 → 见「GraphRAG进阶」「RAG优化与评估」
|
|
### 一图看懂 RAG
RAG 系统由 **两个阶段** 构成：**离线的“建立索引”** 和 **在线的“检索生成”**。
```text
━━━━━━ 阶段一：建立索引（离线，做一次或增量做） ━━━━━━
原始文档 文本切分 向量化 入库
(PDF/Word/HTML) ──▶ (chunk 分块) ──▶ (Embedding 模型) ──▶ (向量数据库)
│ │ │
保留原文与元数据 每块得到一个向量 「向量 + 原文」成对存储
━━━━━━ 阶段二：检索生成（在线，每次提问执行） ━━━━━━
用户提问 ──▶ 问题向量化 ──▶ 相似度检索 Top-K ──▶ 组装提示词 ──▶ LLM 生成回答
│ │ │ │
queryEmbedding 与库中向量比对 「上下文+问题」 基于给定资料作答
```
### 阶段一：建立索引（4 步）
1. **文档加载**：把 PDF、Docx、HTML 等不同格式解析为纯文本
2. **文本切分**：分割成更小的片段（chunk）
3. **向量化（Embedding）**：片段经嵌入模型转成向量
4. **存储**：原始语料块与向量以“键值对”形式存入向量数据库，供后续快速且频繁的搜索
### 阶段二：检索生成（5 步）
1. 把用户 Query 转化为 embedding（queryEmbedding）
2. 与向量库中的文档向量计算相似度，取最相似的 **K 个**（K 值可自己设置）
3. 取回这 K 个文档的原文文本
4. 把文本与问题合并进提示词模板
5. 发给大模型“包装”，生成最终回复返回用户
> 记忆口诀：**离线“切块入库”，在线“查料、拼提示词、问模型”**。
|
|
### 第 1 步：文档加载与预处理
不同格式的数据，构建 RAG 前都要 **统一转化为文本**：
| 数据类型 | 处理方式 |
|----------|----------|
| 文本（txt/md） | 直接读取 |
| PDF / Word | 解析器提取正文（pdfminer.six、PyPDF2、docx2txt） |
| 图片 | OCR 识别文字，或多模态大模型生成文本描述 |
| 表格 | 转成文字描述，或导入数据库后走 nl2sql 路线 |
> 目前 **文本检索的质量最好**，所以一切数据都尽量先转成文本。
预处理还包括：去噪（页眉页脚/水印）、格式清理、按空行重组段落等。
### 第 2 步：文档切分（Chunking）
**为什么要切成小块？**
- **精确匹配**：小片段更容易与问题语义对齐
- **提高效率**：减少送入 LLM 的无关文本量
- **长度限制**：大模型输入有上限（上下文窗口）
- **检索准确**：避免长文档稀释语义、引入噪声
**怎么切？两个关键参数：**
```python
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=200, # 每块最大字符数
chunk_overlap=50, # 相邻块重叠字符数，防止关键信息被切断
length_function=len # 中文按字符数计量
)
chunks = splitter.split_text(long_text)
```
**chunk_size 与 chunk_overlap 的权衡：**
| 参数 | 偏小 | 偏大 |
|------|------|------|
| chunk_size | 检索精准但上下文不足，答案片面 | 上下文全但语义被稀释、检索变粗 |
| chunk_overlap | 过小：边界信息被切断丢失 | 过大：重复内容多、浪费 token |
经验值：chunk_size 取 200~1000 字符，overlap 取 chunk_size 的 10%~20%。
### 第 3 步：向量化（Embedding）
调用嵌入模型，把每个 chunk 变成一个高维向量（如 1536 维）：
```python
text_embeddings = embeddings.embed_documents(texts) # 批量文档向量化
query_embedding = embeddings.embed_query("用户问题") # 单条查询向量化
```
### 第 4 步：入库存储
「chunk 原文 + 向量 + 元数据」成对写入向量数据库：
```python
db = FAISS.from_documents(chunks_docs, embeddings) # LangChain 一步建库
```
此后检索只跟向量打交道，原文通过 id 映射取回。
|
|
### 第 1 步：问题向量化
用户的问题同样要用 **同一个 Embedding 模型** 转成向量。
> ⚠️ 关键原则：**建库用什么模型，查询就必须用什么模型**。不同模型的向量空间不同，混用会导致相似度计算完全失效。
### 第 2~3 步：相似度检索 Top-K
```python
retriever = db.as_retriever(search_kwargs={"k": 3})
docs = retriever.invoke("Llama 2 有多少个参数？")
# 返回一个文档数组，元素按向量相似度从高到低排列
```
- **K 的取舍**：K 太小可能漏掉关键证据；K 太大则噪声多、token 费用高。常见取 3~5。
- 检索返回的是 **chunk 原文**（向量库里存着 id → 原文的映射）。
### 第 4 步：组装提示词
把「检索到的资料 + 用户问题」填入模板：
```text
你是一个专业的 AI 问答助手。
任务：根据下述给定的文档信息回答用户问题。
约束：
- 完全基于给定信息回答，不要添加额外内容
- 如果信息不足，直接回复"根据提供的信息无法回答此问题"
参考文档：
__INFO__
用户问题：
__QUERY__
```
### 第 5 步：LLM 生成回答
```python
prompt = build_prompt(prompt_template, info=search_results, query=user_query)
answer = get_completion(prompt) # 调用大模型
```
### 一个重要认知
**RAG 的效果上限由检索质量决定。** 课上实测：问“Llama 2 有多少个参数？”，若检索没召回含 "7B/13B/70B" 的段落，模型只能回答“无法回答”——不是模型不会，而是资料没给到。这也是后面「RAG优化与评估」要解决的核心问题。
|
|
### 关键词检索的底层：倒排索引（Inverted Index）
- **正排索引**：文档 → 内容（key 是文档 id，value 是全文）
- **倒排索引**：词 → 文档列表（key 是分词后的 term，value 是包含该词的文档 id 列表，即 Posting List）
```text
分词（Tokenization）
doc 1 ────────────▶ "阳" ──▶ [doc1, doc2] ┐
doc 2 ────────────▶ "光" ──▶ [doc1, doc3] │ 倒排列表
doc 3 ────────────▶ "手机" ──▶ [doc2] ┘ (Posting List)
```
搜索“阳光”时先分词成 ["阳", "光"]，取两个 posting list 求交集/并集，再按相关度（如 BM25 打分）排序返回。
**ElasticSearch** 就是基于倒排索引的企业级分布式搜索与分析引擎，RAG 第一版常用它做关键词检索。
### 查询前的文本预处理
为提高召回，入库与查询前通常做：
- **分词**：句子切成词（英文 word_tokenize，中文 jieba）
- **去停用词**：去掉 a / the / is 等无意义词
- **词干提取**：running / runs 统一为 run
- **格式清理**：去标点、统一大小写
效果示例：`"How many parameters does Llama 2 have?"` → `"many parameters Llama 2"`
### 关键词检索的局限（为什么还要向量检索？）
**词汇鸿沟问题**：只认“字面”，不懂“语义”。
| 问题对 | 能否互相召回 |
|--------|--------------|
| "chat version" vs "conversational variant" | ❌ 语义相同、字面不同，召回失败 |
| 中文提问 vs 英文文档 | ❌ 跨语言无法匹配 |
| “参数量” vs "7B/13B/70B" | ❌ 概念相关、字面无关 |
结论：关键词检索适合 **精确术语匹配**（型号、法条编号），语义理解必须交给 **向量检索**，生产中常将两者 **混合使用**。
|
|
### 什么是文本向量（Embeddings）？
1. 将文本转成 **一组浮点数**：每个下标 i 对应一个维度
2. 整个数组对应 n 维空间中的一个 **点**，文本向量又叫 Embeddings
3. 向量之间可以 **计算距离**，距离远近对应 **语义相似度** 大小
```text
"How many parameters?" ──▶ [0.12, -0.33, 0.81, ..., 0.20] ┐ 语义相近
"How much params?" ──▶ [0.11, -0.30, 0.85, ..., 0.18] ┘ → 向量相近
"今天天气不错" ──▶ [-0.72, 0.15, -0.44, ..., 0.91] ── 语义无关 → 距离很远
```
**神奇之处**：即使中英文混合，向量也能识别语义相关性——课上实测英文查询 "global conflicts" 与中文新闻“联合国就苏丹达尔富尔地区大规模暴力事件发出警告”相似度最高。
### 相似度怎么算？
- **余弦相似度**（最常用）：取值 -1 到 1，越大越相似；只看方向不看长度。公式：cos(θ) = A·B / ( \|A\| × \|B\| )
- **欧氏距离**：值越小越相似；对向量长度敏感。公式：d = √Σ(ai-bi)²
- **点积**：计算最快；向量归一化后等价于余弦相似度
```python
import numpy as np
from numpy import dot
from numpy.linalg import norm
def cos_sim(a, b):
"""余弦相似度：值越大越相似，范围 -1 到 1"""
return dot(a, b) / (norm(a) * norm(b))
def euclidean_distance(a, b):
"""欧氏距离：值越小越相似"""
return norm(np.asarray(a) - np.asarray(b))
```
### 向量检索 vs 关键词检索
| 特性 | 关键词检索 | 向量检索 |
|------|------------|----------|
| 理解 | 字面匹配 | 语义理解 |
| 同义词 | 无法处理 | 天然支持 |
| 跨语言 | 不支持 | 支持（多语言模型） |
| 精确术语 | 强（编号/型号精确命中） | 可能有偏差 |
| 成本 | 低（倒排索引成熟） | 高（向量化 + 向量运算） |
### 为什么需要向量数据库？
1000 万个文档 × 1536 维，逐个 for 循环算相似度太慢。向量数据库用 **近似最近邻（ANN）算法**（HNSW、IVF 等）实现毫秒级检索、亿级扩展——详见「常见向量数据库」章节。
|
|
### RAG 提示词模板的五要素
1. **角色定义**：明确 AI 的身份与任务（“你是一个专业的 AI 问答助手”）
2. **任务说明**：根据给定文档回答用户问题
3. **约束条件**：禁止编造、信息不足时明确拒答（**抑制幻觉的关键**）
4. **信息注入**：检索到的文档片段占位符（如 `__INFO__`）
5. **输出要求**：回答语言、格式、简洁度
### 课上使用的模板（可直接复用）
```python
prompt_template = """
你是一个专业的AI问答助手。
任务：根据下述给定的文档信息回答用户问题。
重要约束：
- 完全基于给定信息回答，不要添加额外内容
- 如果信息不足，直接回复"根据提供的信息无法回答此问题"
- 回答要准确、简洁、有用
- 请用中文回答
参考文档：
__INFO__
用户问题：
__QUERY__
"""
```
### 模板填充函数
```python
def build_prompt(template, **kwargs):
"""将变量填入模板：列表自动按行拼接"""
prompt = template
for key, value in kwargs.items():
if isinstance(value, str):
val = value
elif isinstance(value, list) and all(isinstance(e, str) for e in value):
val = '\n'.join(value)
else:
val = str(value)
prompt = prompt.replace(f"__{key.upper()}__", val)
return prompt
```
### 实用技巧
- **拒答兜底**：强制“信息不足就明确说”，比硬编更可信
- **引用编号**：要求“标注引用了第几条文档”，便于溯源
- **分节清晰**：用明确的分隔符隔开“资料区”与“问题区”
- **控制长度**：Top-K 资料过多时截断或摘要，避免超出上下文窗口
|
|
### 课上代码：rag_tutorial_improved.ipynb
课上代码位于 `课上代码/rag-embeddings/` 目录，核心是 `rag_tutorial_improved.ipynb`——通过构建一个 **能回答《Llama 2 论文》问题的智能机器人**，完整走一遍 RAG 流程。
**配套资料**：
| 文件 | 用途 |
|------|------|
| llama2.pdf | Llama 2 论文，知识库原料 |
| rag.gif / RAG.mp4 / rag-flow.jpg | RAG 流程演示 |
| mteb.png | MTEB Embedding 模型评测榜 |
| raptor.jpg / rag-fusion.jpeg / sbert-rerank.png | 进阶优化参考（RAPTOR / RAG-Fusion / 重排） |
| sbert.png / embeddings.png / vectordb.png / sim.png | 向量与相似度原理图 |
### 实战路线图（由浅入深）
```text
① PDF 文档解析 从论文中提取段落文本
↓
② ElasticSearch 索引 关键词检索版 RAG（第一版）
↓
③ LLM + 提示词封装 让检索结果变成自然语言回答
↓
④ 关键词检索的局限 词汇鸿沟实验：同义改写召回失败
↓
⑤ 文本向量 + ChromaDB 语义检索升级（第二版）
↓
⑥ AdvancedRAG_Bot 向量检索版完整系统 + 中英文测试
```
### 依赖安装
```bash
pip install pdfminer.six elasticsearch7 nltk chromadb openai numpy
```
|
|
### 提取文字：extract_text_from_pdf
```python
from pdfminer.high_level import extract_pages
from pdfminer.layout import LTTextContainer
def extract_text_from_pdf(filename, page_numbers=None, min_line_length=1):
'''从 PDF 文件中（按指定页码）提取文字'''
paragraphs = []
buffer = ''
full_text = ''
for i, page_layout in enumerate(extract_pages(filename)):
if page_numbers is not None and i not in page_numbers:
continue
for element in page_layout:
if isinstance(element, LTTextContainer):
full_text += element.get_text() + '\n'
# 按空行分隔，将文本重新组织成段落
lines = full_text.split('\n')
for text in lines:
if len(text) >= min_line_length:
buffer += (' ' + text) if not text.endswith('-') else text.strip('-')
elif buffer:
paragraphs.append(buffer)
buffer = ''
if buffer:
paragraphs.append(buffer)
return paragraphs
paragraphs = extract_text_from_pdf("llama2.pdf", min_line_length=10)
# 共提取 983 个段落，每个段落将成为一个可检索单元
```
**代码要点**：
- `LTTextContainer`：只取文本容器，自动忽略图形/图片元素
- `endswith('-')`：处理英文单词跨行连字符（如 "embed-ding" 拼回 "embedding"）
- `min_line_length=10`：过滤页码、单字符等噪声行
- 按空行重组：PDF 里的“行”重组为语义“段落”
### 提取图片与表格（扩展）
- **图片**：递归遍历 `LTFigure` / `LTImage` 元素，`element.stream.get_rawdata()` 取原始字节，PIL 打开后保存为 png（无法识别的格式存 .bin）
- **表格**：按文本坐标启发式还原——Y 坐标相近的文本行分组成“行”，X 坐标对齐的判断为“列”，导出 CSV/Excel
### 万物归文本
不管什么格式的数据，构建 RAG 之前都要转化成文本：
- **图片**：对图片的文本描述（OCR / 多模态大模型）
- **表格**：对表格数据的描述，或导入数据库用 nl2sql 处理
> 目前文本搜索的质量最好，所以要都转化为（文本）搜索引擎。
|
|
### 文本预处理：to_keywords
```python
import re, nltk
from nltk.tokenize import word_tokenize
from nltk.corpus import stopwords
nltk.download('punkt', quiet=True) # 分词器
nltk.download('stopwords', quiet=True) # 停用词表
def to_keywords(input_string):
"""将文本转换为搜索关键词：去符号 → 分词 → 去停用词"""
no_symbols = re.sub(r'[^a-zA-Z0-9\s]', ' ', input_string)
word_tokens = word_tokenize(no_symbols)
stop_words = set(stopwords.words('english'))
return ' '.join([
w for w in word_tokens if w.lower() not in stop_words
])
# "How many parameters does Llama 2 have?" → "many parameters Llama 2"
```
这样 "parameters count" 与 "parameter number" 能匹配到相同文档。
### 连接 ElasticSearch 并灌入文档
```python
from elasticsearch7 import Elasticsearch, helpers
es = Elasticsearch(
hosts=['http://your-es-host:9200'], # 阿里云 serverless ES 示例
http_auth=("user", "password"),
)
index_name = "llama2_knowledge_base"
if es.indices.exists(index=index_name):
es.indices.delete(index=index_name) # 删除旧索引（演示用）
es.indices.create(index=index_name)
actions = [
{
"_index": index_name,
"_id": f"doc_{i}",
"_source": {
"keywords": to_keywords(para), # 关键词版本（用于匹配）
"text": "original:" + para, # 原文（用于返回）
},
}
for i, para in enumerate(paragraphs)
]
helpers.bulk(es, actions) # ✅ 成功导入 983 个文档
```
**设计要点**：**双字段设计**——`keywords` 存关键词版用于检索匹配，`text` 存原文用于返回给 LLM。
### 封装搜索接口
```python
def search(query_string, top_n=3):
"""在 ElasticSearch 中搜索相关文档"""
search_query = {
"match": {
"keywords": to_keywords(query_string)
}
}
res = es.search(index=index_name, query=search_query, size=top_n)
return [hit["_source"]["text"] for hit in res["hits"]["hits"]]
```
`match` 查询会自动分词后匹配倒排索引，按相关度打分排序返回 Top-N。
|
|
### 第 3 步：封装大模型函数
```python
from openai import OpenAI
client = OpenAI(
api_key="你的 API Key",
base_url="https://ark.cn-beijing.volces.com/api/v3", # 火山方舟示例
)
def get_completion(prompt, model="deepseek-v3-250324", temperature=0):
"""调用大模型生成回答；temperature=0 输出最稳定"""
messages = [{"role": "user", "content": prompt}]
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
)
return response.choices[0].message.content
```
**要点**：
- `temperature=0`：RAG 问答要“稳”不要“创”，取 0 最确定
- OpenAI 兼容接口：换 `base_url` 即可接入 DeepSeek、豆包等国产模型
### 提示词模板（与「RAG核心流程·提示词工程」同款）
```python
prompt_template = """
你是一个专业的AI问答助手。
任务：根据下述给定的文档信息回答用户问题。
重要约束：
- 完全基于给定信息回答，不要添加额外内容
- 如果信息不足，直接回复"根据提供的信息无法回答此问题"
- 请用中文回答
参考文档：
__INFO__
用户问题：
__QUERY__
"""
```
模板设计原则回顾：**角色定义 → 任务说明 → 约束条件（拒答兜底）→ 信息注入 → 输出要求**。
### 模板填充：build_prompt
```python
def build_prompt(template, **kwargs):
"""将变量填入模板：__INFO__ / __QUERY__ 占位符替换"""
prompt = template
for key, value in kwargs.items():
if isinstance(value, str):
val = value
elif isinstance(value, list) and all(isinstance(e, str) for e in value):
val = '\n'.join(value) # 文档列表按行拼接
else:
val = str(value)
prompt = prompt.replace(f"__{key.upper()}__", val)
return prompt
```
至此，RAG 三大件就绪：**检索（search）+ 生成（get_completion）+ 粘合剂（build_prompt）**。
|
|
### 先看关键词检索的局限（动机实验）
```python
# 这两个问题语义相同，但用词不同
query1 = "Does Llama 2 have a chat version?"
query2 = "Does Llama 2 have a conversational variant?"
# 关键词检索：query1 命中 "chat version" 原文；query2 关键词完全对不上 → 召回失败
```
**词汇鸿沟**：相同语义的问题，关键词检索结果可能完全不同——因为关键词匹配无法理解语义相似性。
### 调用 Embedding API 向量化
```python
def get_embeddings(texts, model="doubao-embedding-large-text-250515", dimensions=None):
"""获取文本的向量表示（豆包 embedding，也支持 OpenAI text-embedding 系列）"""
if dimensions:
data = client.embeddings.create(
input=texts, model=model, dimensions=dimensions
).data
else:
data = client.embeddings.create(input=texts, model=model).data
return [x.embedding for x in data]
```
**语义相似度实验**：英文查询 "global conflicts" 对 5 条中文新闻算余弦相似度——达尔富尔暴力事件（最相关）得分最高，水上乐园（不相关）得分最低。**向量能跨语言理解语义**。
### 向量数据库封装：VectorDatabase（ChromaDB）
```python
import chromadb
from chromadb.config import Settings
class VectorDatabase:
"""向量数据库封装：存储 + 语义检索"""
def __init__(self, collection_name, embedding_function):
self.client = chromadb.Client(Settings(allow_reset=True))
self.collection = self.client.get_or_create_collection(
name=collection_name
)
self.embedding_fn = embedding_function
def add_documents(self, documents):
embeddings = self.embedding_fn(documents) # ① 文档向量化
ids = [f"doc_{i}" for i in range(len(documents))]
self.collection.add( # ② 入库
embeddings=embeddings,
documents=documents,
ids=ids,
)
def search(self, query, top_k=3):
query_embedding = self.embedding_fn([query]) # ③ 查询向量化
results = self.collection.query( # ④ 相似度检索
query_embeddings=query_embedding,
n_results=top_k,
)
return results['documents'][0] if results['documents'] else []
# 创建实例并灌入前 50 段（演示）
vector_db = VectorDatabase("llama2_vector_db", get_embeddings)
vector_db.add_documents(paragraphs[:50])
```
**注意**：ChromaDB 只负责存与查，**向量化由外部 embedding 函数完成再传入**——这就是“向量库 + Embedding 服务”的标准分工。
|
|
### RAG_Bot：检索与生成整合（第一版·关键词检索）
```python
class RAG_Bot:
"""RAG 智能问答机器人：检索引擎 + 大模型 + 提示词模板"""
def __init__(self, search_func, llm_func, prompt_template, n_results=2):
self.search = search_func # 检索函数（可注入 ES 或向量库）
self.llm = llm_func # 生成函数
self.template = prompt_template # 提示词模板
self.n_results = n_results
def chat(self, user_query):
# 1. 检索相关文档
search_results = self.search(user_query, self.n_results)
if not search_results:
return "抱歉，没有找到相关信息来回答您的问题。"
# 2. 构建提示词
prompt = build_prompt(self.template,
info=search_results, query=user_query)
# 3. 调用 LLM 生成回答
return self.llm(prompt)
rag_bot = RAG_Bot(search_func=search, llm_func=get_completion,
prompt_template=prompt_template, n_results=2)
print(rag_bot.chat("Llama 2有对话版本吗？"))
# 输出：是的，Llama 2有对话版本，名为 Llama 2-Chat。
```
### AdvancedRAG_Bot：向量检索升级版
```python
class AdvancedRAG_Bot:
"""升级版：注入向量数据库，支持语义检索、跨语言查询"""
def __init__(self, vector_database, llm_function, prompt_template, n_results=3):
self.vector_db = vector_database
self.llm = llm_function
self.template = prompt_template
self.n_results = n_results
def chat(self, user_query):
# 1. 向量检索语义相关文档
search_results = self.vector_db.search(user_query, self.n_results)
if not search_results:
return "抱歉，在知识库中未找到相关信息。"
# 2. 构建提示词并生成
prompt = build_prompt(self.template,
info=search_results, query=user_query)
return self.llm(prompt)
def batch_chat(self, questions):
"""批量问答"""
return [{'question': q, 'answer': self.chat(q)} for q in questions]
def debug_search(self, query):
"""调试模式：只看检索结果，不生成（排查检索质量利器）"""
return self.vector_db.search(query, self.n_results)
```
### 课上实测结论
| 测试问题 | 关键词版 | 向量版 | 原因分析 |
|----------|----------|--------|----------|
| “Llama 2有对话版本吗？” | ✅ | ✅ | "chat version" 关键词直接命中 |
| “conversational variant”（同义改写） | ❌ | ✅ | 向量版懂语义，关键词版有词汇鸿沟 |
| “Llama 2有多少个参数？”（中文） | ❌ | ❌ | 中文问题检索不到英文 “7B/13B/70B” 段落 |
| "What is the training data size?" | ✅ | ✅ | 英文问英文文档，无跨语言问题 |
**两条核心经验**：
1. **检索质量决定 RAG 上限**：资料没召回，再强的模型也只能“无法回答”
2. **跨语言是薄弱点**：中文问英文库效果差 → 优化方向：跨语言 Embedding、查询翻译、混合检索（见「RAG优化与评估」）
### 课后作业（课上布置）
1. 独立完成 RAG 流程的构建代码
2. 输入你自己的文档，让 RAG 来回答，看效果如何
|
|
### 朴素方法的问题
想象 1000 万个文档、每个 1536 维向量：
```python
# ❌ 这样会很慢！
for doc_vector in all_documents: # 1000 万次循环
similarity = cos_sim(query_vector, doc_vector)
```
- ⏰ **性能问题**：线性暴力搜索 O(N×D)，无法接受
- 💾 **内存限制**：向量无法全部加载到内存
- 🔧 **缺乏工程能力**：没有索引、元数据过滤、增删改、分布式
### 向量数据库的救赎
向量数据库使用 **近似最近邻（ANN, Approximate Nearest Neighbor）** 算法，用极小的召回损失换取数量级的加速：
- ⚡ **亚秒级检索**：毫秒级响应
- 📈 **可扩展**：支持亿级向量
- 🎯 **高精度**：近似但足够准
- 💪 **生产就绪**：增删改、过滤、监控、分布式
### 核心 ANN 索引算法
| 索引 | 原理 | 特点 |
|------|------|------|
| **Flat 暴力** | 全量逐一比对 | 100% 准确但慢，适合小库或做基准 |
| **IVF 倒排文件** | 向量空间聚类分桶（nlist 个桶），查询只搜最近的几个桶（nprobe） | 大幅剪枝加速，适合大规模；inverted 即“倒排”之意 |
| **PQ 乘积量化** | 向量切分段分别量化压缩存储 | 省 10 倍以上内存，有精度损失 |
| **HNSW 分层小世界图** | 多层“跳表式”近邻图，从顶层粗定位逐层下沉贪心搜索 | 查询快、召回高；内存占用大、建图慢 |
> IVF 适合超大规模、PQ 适合内存受限、HNSW 适合高召回低延迟——很多库支持组合（如 IVF+PQ）。
|
|
### 主流向量数据库对比
| 数据库 | 形态 | 规模 | 语言/部署 | 特点 | 适用场景 |
|--------|------|------|-----------|------|----------|
| **Chroma** | 嵌入式 | 百万级 | Python | 轻量、易上手、pip 即用 | 学习、原型、小规模应用 |
| **FAISS** | 算法库 | 十亿级 | C++/Python | Meta 出品、极致性能、无服务化 | 本地实验、高性能研究（本页上方有专门章节） |
| **Milvus** | 独立/分布式 | 十亿级以上 | Go/C++ | 云原生、生产级、社区最活跃 | 大规模生产环境 |
| **Qdrant** | 独立部署 | 十亿级 | Rust | 性能强、元数据过滤丰富 | 生产环境 |
| **Weaviate** | 独立部署 | 千万~十亿级 | Go | 内置向量化模块、支持混合检索 | 复杂查询需求 |
| **pgvector** | PG 插件 | 千万级 | C（SQL） | 与 PostgreSQL 一体，能 join 业务表 | 已有 PG 技术栈 |
| **Pinecone** | SaaS 云服务 | 任意 | 闭源 | 免运维、开箱即用 | 快速上线、不想管基础设施 |
| **Elasticsearch** | 搜索引擎 | 千万级 | Java | 倒排+向量双索引，混合检索天然优势 | 关键词+语义混合场景 |
| **Zilliz** | Milvus 托管云 | 十亿级 | — | Milvus 商业版 | 省心的 Milvus |
### 课上的选型建议（原话）
- **ChromaDB**：向量规模在 **100 万左右量级** 的时候使用
- **Milvus**：**1000 万甚至上亿、数十亿以上** 适合
- **云厂商 SaaS 平台**：适合省心、数据安全不敏感的场景
### 一句话选型
```text
学习/原型 ──────────▶ Chroma
本地实验/极致性能 ──▶ FAISS
生产·亿级 ──────────▶ Milvus / Qdrant
已有PG·要join ──────▶ pgvector
关键词+语义混合 ────▶ Elasticsearch / Weaviate
不想运维 ───────────▶ Pinecone / Zilliz Cloud
```
|
|
### ChromaDB：最简单的入门之选
官方文档：https://docs.trychroma.com
**① 原生用法（5 行跑通）**
```python
# pip install chromadb
import chromadb
client = chromadb.Client() # 内存模式
# client = chromadb.PersistentClient(path="./chroma_data") # 持久化到磁盘
col = client.get_or_create_collection("docs")
# 不传 embeddings 时，Chroma 自动用内置的默认模型向量化
col.add(
documents=["Llama 2 有 7B、13B、70B 三个版本", "今天天气不错"],
ids=["doc_1", "doc_2"],
)
res = col.query(query_texts=["Llama 2 有多少参数？"], n_results=2)
print(res["documents"]) # 返回最相似的文档列表
```
**② 自定义 Embedding（生产用法）**
```python
emb = get_embeddings(["Llama 2 有 7B、13B、70B 三个版本"]) # 自行调用 embedding API
col.add(
embeddings=emb,
documents=["Llama 2 有 7B、13B、70B 三个版本"],
ids=["doc_1"],
)
res = col.query(
query_embeddings=get_embeddings(["Llama 2 有多少参数？"]),
n_results=3,
where={"source": "llama2_paper"}, # 元数据过滤
)
```
**③ LangChain 集成**
```python
from langchain_chroma import Chroma
db = Chroma.from_documents(chunks, embeddings,
persist_directory="./chroma_db")
retriever = db.as_retriever(search_kwargs={"k": 3})
docs = retriever.invoke("Llama 2 有多少参数？")
```
**优缺点**：✅ pip 即装、API 极简、自带持久化 ❌ 规模百万级、分布式能力弱
|
|
### Milvus：生产级分布式向量库
部署文档：https://milvus.io/docs/zh/install_standalone-docker-compose.md
**① Docker Compose 部署（standalone）**
```bash
# 下载官方 docker-compose.yml 后
docker compose up -d # 默认端口 19530
```
也支持 Zilliz Cloud 免部署托管，或阿里云等云厂商的向量服务。
**② pymilvus 原生用法**
```python
# pip install pymilvus
from pymilvus import MilvusClient
client = MilvusClient(uri="http://localhost:19530")
# 创建 collection（快速模式：只给维度即可）
client.create_collection(
collection_name="docs",
dimension=1536,
)
# 插入：data 是「字典列表」，vector 字段 + 任意标量字段
client.insert(
collection_name="docs",
data=[
{
"id": 1,
"vector": embedding_of_doc1, # 1536 维 list
"text": "Llama 2 有 7B、13B、70B 三个版本",
"source": "llama2.pdf",
},
],
)
# 相似度检索
res = client.search(
collection_name="docs",
data=[query_embedding], # 查询向量（可批量）
limit=3, # Top-K
output_fields=["text", "source"], # 要回传的字段
)
```
**③ LangChain 集成**
```python
from langchain_milvus import Milvus
db = Milvus.from_documents(
chunks, embeddings,
connection_args={"uri": "http://localhost:19530"},
collection_name="docs",
)
```
**优缺点**：✅ 亿级扩展、分布式、索引类型丰富（IVF/HNSW/PQ） ❌ 部署运维较重，小项目杀鸡用牛刀
|
|
### FAISS：Meta 的向量检索算法库
> 本页上方「FAISS / FAISS检索 / FAISS持久化」三个章节已有详细讲解，这里做体系化小结。
**① 本体用法：index 三步曲**
```python
import faiss
import numpy as np
data = np.random.rand(1000, 64).astype('float32') # 1000 个 64 维向量
index = faiss.IndexFlatL2(64) # 建：平面 L2 索引（暴力但精确）
index.add(data) # 增：灌入向量
dis, ind = index.search(data[:1], 5) # 查：最相似的 5 个（返回距离+下标）
```
**② 持久化**
```python
faiss.write_index(index, "my_index.index") # 存
loaded = faiss.read_index("my_index.index") # 读
```
**③ 数据量大时换索引**
| 索引 | 适用 |
|------|------|
| IndexFlatL2 / IndexFlatIP | 小规模基准（精确检索） |
| IndexIVFFlat | 大规模：先聚类分桶再桶内搜，显著提速、召回略降 |
| IndexIVFPQ | 大规模 + 内存受限：向量量化压缩，省内存有精度损失 |
**④ LangChain 封装（from_documents / from_texts / from_embeddings）**
```python
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
db = FAISS.from_documents(texts, OpenAIEmbeddings()) # 文档对象 → 自动切块后入库
db.save_local("faiss_index") # 持久化
db = FAISS.load_local("faiss_index", OpenAIEmbeddings(),
allow_dangerous_deserialization=True)
retriever = db.as_retriever() # 转检索器
docs = retriever.invoke("呼吸") # 统一接口调用
```
**关键认知**：**index 只是 index**——FAISS 存的是向量、搜的输入也是向量，不存原文；原文映射（id → text）需要自己维护（LangChain 的 docstore 帮你做了这事）。
|
|
### pgvector：让 PostgreSQL 秒变向量库
**① 启用扩展与建表**
```sql
CREATE EXTENSION vector;
CREATE TABLE docs (
id bigserial PRIMARY KEY,
content text,
embedding vector(1536) -- pgvector 的向量类型
);
```
**② 插入与检索（余弦距离）**
```sql
-- 插入：向量以字符串字面量写入
INSERT INTO docs (content, embedding)
VALUES ('Llama 2 有 7B、13B、70B 三个版本', '[0.1, 0.2, ...]');
-- 检索：<=> 是余弦距离，<-> 是 L2 距离，<#> 是内积
SELECT content
FROM docs
ORDER BY embedding <=> '[0.1, 0.2, ...]'
LIMIT 3;
-- 优势：可与业务表 join、加 where 条件、事务保障
SELECT d.content, o.order_status
FROM docs d JOIN orders o ON o.doc_id = d.id
WHERE o.status = 'paid'
ORDER BY d.embedding <=> '[0.1, 0.2, ...]'
LIMIT 3;
```
**③ 加速：HNSW 索引**
```sql
CREATE INDEX ON docs USING hnsw (embedding vector_cosine_ops);
```
**④ LangChain 集成**
```python
from langchain_postgres import PGVector
db = PGVector.from_documents(
chunks, embeddings,
connection="postgresql://user:pwd@localhost:5432/mydb",
collection_name="docs",
)
```
### Elasticsearch：关键词 + 向量一站式混合
ES 8.x 起原生支持 dense_vector 字段与 kNN 搜索，同一个库同时有倒排索引和向量索引——**混合检索的最顺手选择**：
```json
PUT /docs
{
"mappings": {
"properties": {
"text": { "type": "text" },
"embedding": { "type": "dense_vector", "dim": 1536,
"index": true, "similarity": "cosine" }
}
}
}
```
```json
GET /docs/_knn_search
{
"knn": {
"field": "embedding",
"query_vector": [0.1, 0.2, "..."],
"k": 3,
"num_candidates": 100
}
}
```
### 选型决策表（总结）
| 你的情况 | 推荐 |
|----------|------|
| 刚入门 / 课程作业 | Chroma |
| 本地研究、追求检索速度 | FAISS |
| 生产环境、亿级向量 | Milvus / Qdrant |
| 已有 PostgreSQL、千万级以内 | pgvector |
| 要关键词+语义混合检索 | Elasticsearch / Weaviate |
| 完全不想运维 | Pinecone / Zilliz Cloud |
|
|
### LangChain 是什么
LangChain 是最流行的大模型应用开发框架，把 RAG 需要的每个环节都抽象成了 **可插拔的标准组件**——你可以像搭积木一样组合它们，随时替换某个环节的实现。
### RAG 全链路组件一览
```text
Document ──▶ Loader ──▶ Splitter ──▶ Embeddings ──▶ VectorStore ──▶ Retriever
(文档) (加载) (切分) (向量化) (向量库) (检索器)
│
▼
Prompt ──▶ LLM ──▶ 回答
(模板) (模型)
```
| 组件 | 职责 | 常用实现 |
|------|------|----------|
| Document | 文本 + 元数据的标准载体 | `page_content` + `metadata` |
| Loader 文档加载器 | 各格式 → Document 列表 | TextLoader / PyPDFLoader / CSVLoader / WebBaseLoader |
| Splitter 文本分割器 | 长文档 → chunk | RecursiveCharacterTextSplitter / CharacterTextSplitter |
| Embeddings 嵌入 | 文本 → 向量 | OpenAIEmbeddings / HuggingFaceEmbeddings |
| VectorStore 向量库 | 存储 + 相似度检索 | FAISS / Chroma / Milvus / PGVector |
| Retriever 检索器 | 统一的“查询→文档”接口 | `db.as_retriever()` / MultiQueryRetriever |
| Prompt 模板 | 带变量的提示词骨架 | ChatPromptTemplate |
| LLM 模型 | 生成回答 | ChatOpenAI（兼容各类国产模型） |
### 10 行串起一条 RAG 链路
```python
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
docs = TextLoader('/tmp/a.txt').load() # ① 加载
chunks = RecursiveCharacterTextSplitter(
chunk_size=500, chunk_overlap=50
).split_documents(docs) # ② 切分
db = FAISS.from_documents(chunks, OpenAIEmbeddings()) # ③④⑤ 向量化+入库
retriever = db.as_retriever() # ⑥ 检索器
results = retriever.invoke("呼吸") # ⑦ 检索
```
这正是本页上方「FAISS」章节的代码骨架——组件化让 ES 版、Chroma 版、FAISS 版可以无缝互换。
|
|
### Document Loaders 家族
| Loader | 来源 |
|--------|------|
| TextLoader / CSVLoader / JSONLoader | 本地文件 |
| PyPDFLoader / Docx2txtLoader | PDF / Word |
| WebBaseLoader / GithubFileLoader | 网页 / 仓库 |
| DirectoryLoader | 整个目录批量加载 |
```python
from langchain_community.document_loaders import PyPDFLoader
docs = PyPDFLoader("llama2.pdf").load() # 每页一个 Document
# docs[0].page_content → 文本内容
# docs[0].metadata → {'source': 'llama2.pdf', 'page': 0}
```
### Text Splitters 家族
| Splitter | 策略 | 适用 |
|----------|------|------|
| CharacterTextSplitter | 按单一分隔符（如换行） | 结构规整的文本 |
| **RecursiveCharacterTextSplitter** | 依次尝试 `\n\n` → `\n` → 空格 → 字符 | **默认首选**，尽量保住段落语义 |
| TokenTextSplitter | 按 token 数切 | 需精确控制 token 时 |
| MarkdownHeaderTextSplitter | 按 md 标题层级 | 文档有清晰标题结构 |
| Language-specific | 按代码语法切 | Python/JS 等代码库问答 |
### 切分实测（中文）
```python
from langchain_text_splitters import RecursiveCharacterTextSplitter
text = "肺主呼吸之气、一身之气。肺司呼吸的功能正常，……（长文本）"
splitter = RecursiveCharacterTextSplitter(
chunk_size=100, chunk_overlap=10, length_function=len
)
texts = splitter.split_text(text)
# ['肺主呼吸之气、一身之气。……（约100字的完整语义块）',
# 'from langchain_community.document_loaders import TextLoader',
# 'from langchain_community.vectorstores import FAISS']
```
**要点**：中文必须传 `length_function=len`，否则默认按“字节数/4”估算导致切块过碎。
### split_documents vs split_text
- `split_text(str)` → 返回 `list[str]`（纯文本）
- `split_documents(list[Document])` → 返回 `list[Document]`，**保留 metadata（来源/页码）**，检索后可溯源
|
|
### LangChain Indexing API：增量索引的正规军
**痛点**：直接 `vectorstore.add_documents()` 反复运行会 **重复入库**——同一文档被向量化多次，浪费 embedding 费用还污染检索；删除源文档后向量库里的旧数据也无人清理。
**Indexing API**（`langchain.indexes`）= 一个 `index()` 函数 + 一个 **RecordManager**（记录哪些文档已入库），自动完成 **去重、增量同步、陈旧数据清理**。
```python
from langchain.indexes import SQLRecordManager, index
# ① 记录管理器：用 SQLite 跟踪入库状态（也可用 Postgres）
record_manager = SQLRecordManager(
"rag/llama2_docs", # 命名空间，不同任务分开
db_url="sqlite:///record_manager.sql"
)
record_manager.create_schema() # 首次建表
# ② 一键增量同步
result = index(
chunks, # 可迭代的 Document 列表
record_manager, # 记录管理器
vectorstore=db, # 目标向量库（FAISS/Chroma/Milvus...）
source_id_key="source", # 用哪个 metadata 字段判定"同一来源"
cleanup="incremental", # 清理模式（见下表）
)
print(result) # IndexingResult(num_added=983, num_updated=0,
# num_deleted=0, num_skipped=0)
```
### 去重原理
- **文档级去重**：按 `source_id_key` 指定的来源 id 判断——同一来源只保留最新一次
- **内容级去重**：内容 hash 没变的 chunk 直接 skip（**省 embedding 费用**）
### cleanup 三种清理模式
| 模式 | 行为 | 适用 |
|------|------|------|
| `None` | 只增不删，不清理任何旧数据 | 一次性灌库、手动管理 |
| `"incremental"` | 内容变了的重灌；**来源已消失的自动删** | 定时增量同步（最常用） |
| `"full"` | 本次批次中没出现的旧数据 **全部删除** | 全量重建、强制对齐 |
### 典型定时同步模式
```python
# 每天定时：源目录变了就增量更新，删掉的文件自动清理
all_docs = DirectoryLoader("./knowledge").load()
chunks = splitter.split_documents(all_docs)
index(
chunks, record_manager, db,
source_id_key="source",
cleanup="incremental",
)
```
> 一句话：**把“向量库与源数据保持一致”这个脏活交给 Indexing API**。
|
|
### LangGraph：LangChain 团队的图编排框架
**为什么 RAG 需要它？** 简单 RAG 是一条直线，但真实系统需要 **分支与循环**：
- 检索质量差 → 改写问题重查（纠错 RAG）
- 问题含糊 → 先拆解成子问题逐个检索
- 需要多跳推理 → 检索 → 生成 → 再检索
LangChain 把这类流程建模为 **状态图（StateGraph）**：**节点 = 处理步骤，边 = 执行流，条件边 = 分支**，状态在节点间共享传递。
### 核心概念
| 概念 | 说明 |
|------|------|
| State | 一个 TypedDict，图中所有节点共享读写的“全局状态” |
| Node 节点 | 一个函数：读状态 → 返回状态增量 |
| Edge 边 | 固定的执行顺序 |
| Conditional Edge 条件边 | 按状态动态决定下一个节点（分支） |
| START / END | 特殊节点：入口与出口 |
### 最小 RAG 图：两节点直线
```python
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class RAGState(TypedDict):
question: str # 输入：用户问题
documents: list # 中间：检索到的文档
answer: str # 输出：最终回答
def retrieve(state: RAGState):
docs = retriever.invoke(state["question"])
return {"documents": docs} # 返回增量，自动合并进状态
def generate(state: RAGState):
prompt = build_prompt(template,
info=state["documents"],
query=state["question"])
return {"answer": llm.invoke(prompt)}
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve) # 加节点
graph.add_node("generate", generate)
graph.add_edge(START, "retrieve") # 连边
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)
app = graph.compile() # 编译成可执行应用
result = app.invoke({"question": "Llama 2 有多少个参数？"})
print(result["answer"])
```
执行流：`START → retrieve → generate → END`
### 加条件边：自纠错 RAG（分支）
```python
def rewrite_query(state: RAGState):
"""检索不相关时：改写问题再查一次"""
better = llm.invoke(f"改写得更利于检索：{state['question']}")
return {"question": better}
def grade_documents(state: RAGState):
"""路由函数：有相关文档走生成，否则走改写"""
return "generate" if state["documents"] else "rewrite"
graph.add_node("rewrite_query", rewrite_query)
graph.add_conditional_edges( # 条件边：动态分支
"retrieve",
grade_documents,
{"generate": "generate", "rewrite": "rewrite_query"},
)
graph.add_edge("rewrite_query", "retrieve") # 循环：改写后重查
```
执行流变成：
```text
START ──▶ retrieve ──▶ 相关？ ──是──▶ generate ──▶ END
│
否
↓
rewrite_query ──▶ 回到 retrieve（循环）
```
### 工程能力
- **持久化**：内置 checkpointer，可中断恢复、时光旅行调试
- **Human-in-the-loop**：敏感节点前暂停等人工确认
- **流式输出**：节点粒度的 streaming
- **多 Agent**：图嵌套图，构建 Agentic RAG / 多智能体系统
|
|
### 同一个 RAG，三种写法
| 维度 | 裸写（课上 notebook） | LangChain LCEL | LangGraph |
|------|----------------------|----------------|-----------|
| 代表 | RAG_Bot 类 | retriever \| prompt \| llm | StateGraph |
| 优点 | 直观、每步可控、易调试 | 声明式、组件可插拔、生态全 | 分支/循环/多Agent、持久化 |
| 缺点 | 组件全要自己粘 | 灵活但黑盒度上升 | 学习成本最高 |
| 适用 | 学习原理、简单脚本 | 标准流水线、快速交付 | 复杂流程（纠错/多跳/Agent） |
### LCEL 写法：用管道符串链
```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
prompt = ChatPromptTemplate.from_template(
"根据以下背景资料回答问题：\n{context}\n\n问题：{question}"
)
def format_docs(docs):
return "\n".join(d.page_content for d in docs)
rag_chain = (
{
"context": retriever | format_docs, # 检索 + 整理
"question": RunnablePassthrough(), # 问题原样透传
}
| prompt # 填模板
| ChatOpenAI(model="deepseek-v3") # 调模型
| StrOutputParser() # 取字符串
)
rag_chain.invoke("Llama 2 有多少个参数？")
```
> `|` 管道符把“检索器、模板、模型、解析器”串成一条链，前一步输出自动作为后一步输入——这就是 LCEL（LangChain Expression Language）。
### 学习建议
1. **先用裸写**理解每一步在干什么（rag_tutorial_improved.ipynb 就是这么做的）
2. **再用 LCEL** 统一组件接口，享受生态（换向量库只改一行）
3. **流程出现分支/循环时上 LangGraph**，别在裸写里堆 if-else
|
|
### Graph RAG 用来解决什么问题？
Graph RAG（基于图的检索增强生成）将 **知识图谱/关系图** 与大模型的检索增强生成结合，用结构化的 **实体-关系-证据** 来约束与引导生成，在复杂知识场景下显著提升 **可追溯性、可解释性和多跳推理能力**。
### 普通向量 RAG 的五大痛点
| 痛点 | 说明 |
|------|------|
| **多源长文档、结构信息缺失** | 文本切块后丢失实体之间的“关系结构”，难以回答跨段落、跨文档的组合问题 |
| **多跳推理与复杂查询** | 仅靠相似度召回难以覆盖链式/路径式问题（A 与 B 的因果链、时间线、供应链关系） |
| **幻觉与不可追踪** | 缺少“谁说的、在哪里说的”，回答难以核验 |
| **实体歧义与别名** | 人名、机构名、产品别名多，普通检索容易混淆 |
| **知识更新与一致性** | 需要增量更新而不破坏既有结构与引用 |
### 核心思路（一句话）
**先构图，后问答**：从原文中抽取“实体、关系、属性”，构建可遍历的知识子图；问答时以图为骨架检索相关证据并驱动生成，保证 **路径清晰、证据可引、结论可验**。
```text
普通向量RAG： 文档 → 切块 → 向量 → 相似度召回 （弱结构）
GraphRAG ： 文档 → 抽取实体/关系 → 知识图谱 → 图检索 （强结构）
```
|
|
### 离线阶段：构图与索引（9 步）
```text
①数据采集 ─▶ ②预处理切分 ─▶ ③信息抽取 ─▶ ④实体归一 ─▶ ⑤构建图谱
│
⑨增量更新 ◀─ ⑧质量控制 ◀─ ⑦多模索引 ◀─ ⑥图增强分析 ◀──┘
```
**① 数据采集与准入**：网页/文档/数据库/日志等多模态来源，统一成结构化清洗后的原文块，保留 `文档ID / 段落ID / 时间戳 / 来源` 等溯源字段。
**② 预处理与切分**：采用避免破坏实体与关系的切分策略（按语义边界、标题、表格/列表识别），保留上下文窗口与锚点。
**③ 信息抽取（IE）——构图的核心**：
| 抽取对象 | 例子 |
|----------|------|
| 实体 | 人/地/组织/产品/术语/指标 |
| 关系 | 隶属、依赖、因果、时间序、地理包含、同义/别名 |
| 属性 | 价格、时间、版本号、指标值 |
可用「规则 + IE 模型（spaCy / HanLP）」或 LLM 提示抽取，输出带置信度。
**④ 实体标准化与去重**：同名/别名归一与冲突合并（字典、实体链接 EL、聚类），建立 `canonical_name` 与 `alias` 映射。
**⑤ 构建知识图谱**：实体作节点、关系作边，存入图数据库（Neo4j / Nebula / ArangoDB），附结构属性（类型、权重、时间、来源、置信度）与文本属性（摘要、原文引用）。
**⑥ 图增强与分析**：计算中心度/介数/社区划分/主题子图；为节点/边生成描述向量，便于语义寻路。
**⑦ 建立多模索引**：
- 图索引：节点/边类型、属性、时间索引
- 文本向量索引：节点描述、边证据、原文块分别建向量库（Milvus / FAISS / pgvector）
- 反向索引：关键词/倒排，辅助可解释检索
**⑧ 质量控制与治理**：冲突检测、冗余去除、证据去噪；每条边/属性记录 `source_citation` 与 `confidence`。
**⑨ 增量更新**：定时/事件驱动的 CDC，采用可幂等的“对齐-合并-标记历史”策略。
|
|
### 在线阶段：检索与生成（7 步）
```text
用户查询
↓
① 查询理解 ──▶ ② 图检索(种子节点+k-hop) ──▶ ③ 图约束下的文本证据检索
↓
⑥ 自检校验 ◀─ ⑤ LLM生成(结论+路径+引用) ◀── ④ 证据打包
│ 不一致 → 回退扩大子图重检
↓ 一致
⑦ 多轮记忆 + 返回答案
```
**① 查询理解**
- 意图分类：事实问答 / 对比 / 解释 / 路径查询 / 汇总
- 实体识别与链接（EL）：把查询中的实体/别名对齐到图节点
- 约束解析：时间、地域、类型、路径长度、元路径（metapath）
**② 图检索（子图构建）**
- 语义检索找 **种子节点**（向量检索节点描述/别名字典）
- 按约束 **k-hop 扩展**、社区/主题过滤、时间窗裁剪
- 路径打分：结构（度、介数）+ 语义（节点/边嵌入）+ 证据置信度
**③ 文本证据检索**：仅在候选子图关联的“原文块/边证据”中做 **二次向量检索**，去重聚合、保留多源视角。
**④ 证据打包**：节点摘要、关键路径、边解释、引用列表（doc_id、段落、URL、时间）；数值/时序问题聚合成可核验表格。
**⑤ 答案生成**：模板化提示（角色、任务、格式、引用规范、**拒答与不臆测约束**）；输出包含 **结论 + 关键路径（谁→关系→谁）+ 证据引用编号**。
**⑥ 自检与事实校验（可选）**：二次 LLM 校验“证据-结论一致性”与“引用覆盖率”，不一致回退扩大子图。
**⑦ 多轮对话记忆**：对话中出现的新实体/别名写入“轻量记忆图”并与主图对齐。
|
|
### 与传统向量 RAG 的对比
| 维度 | 向量 RAG | Graph RAG |
|------|----------|-----------|
| **结构化约束** | 主要靠相似度，弱结构 | 以实体-关系为骨架，路径可解释 |
| **多跳推理** | 难以稳定覆盖链式问题 | 通过 k-hop / 元路径自然表达 |
| **可追溯性** | 引用分散、难组装 | 节点/边即证据锚点，容易对齐来源 |
| **更新治理** | 增量简单，但语义漂移风险 | 更新复杂，但全局一致性更强 |
### 工程落地与选型建议
| 环节 | 推荐 |
|------|------|
| 图数据库 | Neo4j（生态成熟）、Nebula（分布式）、ArangoDB（多模型） |
| 向量库 | FAISS（本地/实验）、Milvus（生产级）、pgvector（PG 一体化） |
| 抽取方案 | 规则 + IE 模型（spaCy/HanLP）或 LLM 提示抽取（配置信度与审计） |
| 编排框架 | LangChain / LlamaIndex + 自定义图检索器 |
| 服务化 | 把“图检索 + 证据打包”做成独立微服务，LLM 只负责生成 |
### 评估指标（侧重“可信”与“覆盖”）
| 指标 | 含义 |
|------|------|
| Faithfulness 事实一致性 | 回答是否完全由证据支持 |
| Coverage 覆盖率 | 是否召回了应出现的关键节点/路径/来源 |
| Precision@K / Recall@K | 子图与证据层面的查准/查全 |
| Path Explainability 路径可解释性 | 读者能否复现推理路径 |
| Citation Rate 引用率 | 回答中包含可核验引用的比例 |
| Latency 时延 | 端到端耗时是否在 SLA 内 |
### 常见陷阱与规避
- **切分破坏关系** → 优先按语义与结构切分，保留跨段落锚点
- **实体归一化困难** → 别名词典 + 向量聚类 + EL 校验闭环
- **子图爆炸** → k-hop 上限、主题/时间窗过滤、路径打分阈值
- **错配与幻觉** → 强制引用与不臆测模板，二次校验不一致则回退
- **更新引发漂移** → 幂等合并策略与版本化溯源
|
|
### 最小可行实现（MVP 七步）
1. 用 LLM 对文本抽取（实体、关系、引用片段）并存入 Neo4j
2. 为“节点描述 + 边说明 + 原文块”分别构建向量库
3. 查询时先做实体链接 → 在图中 1~2 hop 扩展候选
4. 在候选子图关联的原文块中做二次向量检索
5. 将“关键路径 + 证据引用”打包交给 LLM 生成
6. 输出严格包含「结论 / 路径 / 引用」的结构化回答
7. 记录反馈并修正别名词典与抽取提示
### 简化伪代码（示意）
```python
query = user_input()
intent, constraints = parse_intent_and_constraints(query)
linked_entities = entity_linking(query)
seed_nodes = semantic_search_nodes(query, linked_entities)
subgraph = expand_graph(seed_nodes, hop=2, constraints=constraints)
ranked_paths = score_paths(subgraph)
candidate_chunks = retrieve_text_chunks(ranked_paths)
ranked_evidence = rerank_within_graph(candidate_chunks, query)
answer = llm_generate(
instruction="依据证据回答，列出关键路径与引用，禁止臆测",
evidence=ranked_evidence,
graph_paths=topk(ranked_paths, k=5),
constraints=constraints,
)
return answer
```
### 流程图（文字版）
```text
A[用户查询] ──▶ B[查询理解/意图分类] ──▶ C[实体识别/实体链接]
C ──▶ D[图检索: 种子节点 + k-hop 扩展] ──▶ E[路径/子图打分与筛选]
E ──▶ F[图约束下的文本证据检索] ──▶ G[证据打包: 节点摘要/边解释/引用]
G ──▶ H[LLM 生成: 结论+路径+引用] ──▶ I[自检/一致性校验]
I ──不一致──▶ D（回退扩大子图重检）
I ──一致──▶ J[返回答案]
```
### 一句话总结
**Graph RAG = 图结构 + 受约束的检索 + 可溯源的生成**，尤其适合“多跳推理、强可解释、强合规”的知识密集型场景（金融研究、法务合规、供应链分析、技术方案溯源等）。
|
|
### 优化从哪里下手？
RAG = 检索 + 生成。**先优化检索，再优化生成**——资料都不对，生成再好也白搭。
```text
用户问题 → [查询改写] → [混合检索] → [重排序] → Top-K → [提示词] → LLM
查询侧优化 召回侧优化 精排侧优化 生成侧优化
```
### ① 分块策略进阶
| 策略 | 思路 | 适用 |
|------|------|------|
| 固定长度 + 重叠 | chunk_size / chunk_overlap | 通用兜底 |
| 递归分割 | 优先按段落/句子边界切 | 默认首选 |
| 语义切分 | 按 embedding 相似度断句 | 追求语义完整 |
| **父子块（small-to-big）** | 用小块检索、返回其父大块做上下文 | 兼顾精度与上下文 |
| 结构化切分 | 表格/代码整块不拆 | 含表格文档、代码库 |
### ② 混合检索（Hybrid Search）
关键词（BM25）负责精确术语，向量负责语义理解，结果用 **RRF（倒数排名融合）** 合并：
```python
# 思路示意：两路召回 + RRF 融合
kw_docs = es_search(query, top_n=20) # 关键词路（倒排索引）
vec_docs = vector_db.search(query, top_k=20) # 语义路（向量）
def rrf_fusion(list_a, list_b, k=60):
scores = {}
for docs in (list_a, list_b):
for rank, doc in enumerate(docs, 1):
scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + rank)
return sorted(scores, key=scores.get, reverse=True)
final_docs = rrf_fusion(kw_docs, vec_docs)[:5]
```
### ③ 重排序（Rerank）
粗排（向量 ANN）从万里挑一百，**Cross-Encoder 精排**从百里挑五（课上配图 sbert-rerank.png）：
```text
查询 + 100个候选 ──▶ Cross-Encoder(bge-reranker / SBERT) 逐对打分 ──▶ 取 Top-5
```
- 双塔（Bi-Encoder）：query 和 doc 分别编码再算相似度——快，用于粗排
- 交叉编码器（Cross-Encoder）：query 和 doc 拼接后共同编码——准但慢，用于重排
- 常用模型：BGE-reranker、bge-reranker-v2-m3（中文友好）、Cohere Rerank API
### ④ 查询改写与 RAG-Fusion
用户原话往往不适合直接检索（课上配图 rag-fusion.jpeg）：
- **多查询扩展（RAG-Fusion）**：让 LLM 把一个问题改写成多个角度的子问题，各自检索后 RRF 融合
- **查询翻译**：中文问题 → 英文检索（解决课上“中文问题查英文论文失败”的痛点）
- **HyDE**：让 LLM 先生成一个“假想答案”，用假答案的向量去检索（答案与文档的语义空间更接近）
- **多轮问题改写**：把“它支持吗？”这类指代问题，结合历史改写成独立完整的问题
```python
# LangChain 内置多查询检索器（RAG-Fusion 的现成实现）
from langchain.retrievers.multi_query import MultiQueryRetriever
retriever = MultiQueryRetriever.from_llm(
retriever=db.as_retriever(), llm=llm
)
```
### ⑤ Embedding 模型选型
看 **MTEB 榜单**（课上配图 mteb.png，https://huggingface.co/blog/mteb）：
- 中文场景：BGE（bge-large-zh）、M3E、text-embedding-v3
- 多语言/跨语言：bge-m3、multilingual-e5
- 关注点：维度（影响存储）、是否支持指令化检索、最大输入长度
|
|
### RAG 的演进：从朴素到智能
```text
Naive RAG ──▶ Advanced RAG ──▶ Modular RAG ──▶ Agentic RAG
（一把梭） （前后加优化） （模块可插拔） （模型自己决定怎么查）
```
### 主流高级架构一览
| 架构 | 核心思想 | 解决什么 |
|------|----------|----------|
| **Advanced RAG** | 检索前（查询改写）+ 检索后（重排、压缩）优化 | 提升召回与精度 |
| **RAG-Fusion** | 多角度生成子查询 + RRF 融合 | 单一问法召回不全 |
| **Self-RAG** | 模型自反思：判断是否需要检索、内容是否可信 | 过度检索/不可靠内容 |
| **CRAG 纠错式RAG** | 先评估检索质量：好→用；差→改写重查或 web 搜索兜底 | 检索失败仍硬答 |
| **RAPTOR** | 对 chunk 递归聚类摘要，构建多层“摘要树”，不同抽象层检索（课上配图 raptor.jpg） | 细节问题与全局问题要不同粒度 |
| **GraphRAG** | 知识图谱 + 图检索（见上一章） | 多跳推理、可追溯 |
| **Agentic RAG** | Agent 自主决定：查不查、查哪个库、查几轮、何时停止 | 复杂多源任务 |
### Agentic RAG 长什么样
```text
用户问题 ──▶ [规划 Agent]
│ 分析：需要检索吗？查哪些源？
├──▶ 向量库检索 ──▶ 评估相关性 ──不足──▶ 改写重查
├──▶ 知识图谱检索 ─────────────────────┐
└──▶ Web 搜索 ─────────────────────────┤
▼
[综合 Agent] 汇总多源证据 ──▶ LLM 生成带引用回答
```
用 LangGraph 实现最为顺手（条件边做“评估→重查”的循环，见「LangChain索引与图」一章）。
### 选型建议（由简到繁）
1. 先跑通 **Naive RAG**，建立评估基线
2. 检索不准 → 上 **混合检索 + 重排**（性价比最高的一步）
3. 问法多样 → **查询改写 / RAG-Fusion**
4. 全局性/多跳问题 → **RAPTOR / GraphRAG**
5. 流程复杂多变 → **Agentic RAG**
|
|
### 为什么必须评估？
没有评估的优化都是盲改。RAG 评估分 **检索侧** 与 **生成侧** 两段。
### 检索侧指标
| 指标 | 含义 | 说明 |
|------|------|------|
| **Recall@K** | 前K个结果覆盖了多少相关文档 | RAG 最该盯的指标：资料没召回，后面全白搭 |
| **Precision@K** | 前K个结果有多少是相关的 | 噪声多会稀释 LLM 注意力 |
| **MRR** | 第一个相关结果的排名倒数均值 | 关心“第一条准不准” |
| **NDCG@K** | 考虑排名位置加权的综合质量 | 排序质量的标准指标 |
```python
def recall_at_k(retrieved_ids, relevant_ids, k):
topk = set(retrieved_ids[:k])
return len(topk & set(relevant_ids)) / len(relevant_ids)
```
### 生成侧指标
| 指标 | 含义 |
|------|------|
| **Faithfulness 忠实度** | 回答是否完全由检索证据支持（反幻觉核心指标） |
| **Answer Relevancy** | 回答是否切题 |
| **Citation Rate 引用率** | 回答中带可核验引用的比例 |
### RAGAS：自动化评估框架
RAGAS 用 LLM 评估 LLM，无需人工标注即可估测四项核心指标：
```python
# pip install ragas datasets
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
eval_data = Dataset.from_dict({
"question": ["Llama 2 有多少个参数？"],
"answer": ["Llama 2 有 7B、13B、70B 三个版本。"],
"contexts": [["...检索到的 chunk 原文..."]],
"ground_truth": ["7B、13B、70B"], # 参考答案（context_recall 需要）
})
result = evaluate(eval_data, metrics=[
faithfulness, answer_relevancy,
context_precision, context_recall,
])
print(result)
```
### 落地流程
1. 从真实用户问题中抽 50~100 条，人工标注理想答案与相关 chunk → **黄金测试集**
2. 每次改动（换 embedding / 调 chunk_size / 加重排）跑一遍测试集
3. 指标对比 + 人工抽检 bad case，确认再上线
4. 线上收集用户点赞/点踩，回流测试集持续迭代
> 课上实测就是最好的例子：“Llama 2 有多少个参数？”中文提问两版都失败——如果事先有评估集，这个 bad case 在上线前就能暴露。
|
|
### 常见陷阱清单
| # | 陷阱 | 后果 | 规避 |
|---|------|------|------|
| 1 | 切分过碎/过大 | 检索片面 或 语义稀释 | chunk 200~1000 + 10%~20% 重叠，多策略对比 |
| 2 | 中文用默认 length_function | 切块过碎 | 显式传 `length_function=len` |
| 3 | 建库与查询用不同 Embedding 模型 | 相似度完全失效 | 全链路锁定同一模型 |
| 4 | Top-K 盲目调大 | 噪声多、token 贵、超上下文 | K=3~5 起步 + 重排精筛 |
| 5 | 多轮对话直接拿原话检索 | “它支持吗？”检索失败 | 结合历史做问题改写 |
| 6 | 无拒答兜底 | 模型硬编造（幻觉） | 模板强制“信息不足即拒答” |
| 7 | 文档更新后重复灌库 | 内容重复、费用浪费 | LangChain Indexing API 增量同步 |
| 8 | 只用向量检索 | 精确术语（型号/编号）漏召回 | 关键词+向量混合检索 |
| 9 | 不做评估就上线 | 优化靠感觉 | 黄金测试集 + RAGAS |
| 10 | 检索质量差全怪模型 | 方向性错误 | 先用 debug_search 看召回内容再定责 |
### 最佳实践 Checklist
**建库阶段**
- [ ] 文档解析干净（去页眉页脚/水印/乱码）
- [ ] 切分策略与文档结构匹配（表格整块、代码按函数）
- [ ] 保留 metadata（来源/页码/标题）供过滤与溯源
- [ ] 用 Indexing API 做增量与去重
**检索阶段**
- [ ] 混合检索 + RRF 融合
- [ ] 加 Cross-Encoder 重排
- [ ] 多轮场景先改写查询
**生成阶段**
- [ ] 模板含角色/约束/拒答/引用要求
- [ ] context 超长时截断或压缩（如按相关度逐条丢弃）
**工程阶段**
- [ ] 缓存高频 query 的检索结果与回答
- [ ] 监控：检索延迟、召回命中率、拒答率、token 成本
- [ ] 建立评估流水线，每次迭代有数据支撑
### 学习资源
- ElasticSearch：https://www.elastic.co/
- ChromaDB：https://docs.trychroma.com/
- Milvus：https://milvus.io/docs/zh
- MTEB 评测：https://huggingface.co/blog/mteb
- LangChain：https://python.langchain.com/
- LangGraph：https://langchain-ai.github.io/langgraph/
- RAGAS：https://docs.ragas.io/
|
|
### 版本与环境规划（2026-09-05 实机搭建）
| 项 | 选型 | 说明 |
|---|---|---|
| ES 版本 | **7.17.23**（最后的 7.x） | 与课程 `elasticsearch7` python 客户端（7.17.13）版本匹配 |
| 安装目录 | `/opt/app/es7` | tarball 方式，bin/config 直接在其下（ES_HOME 即该目录） |
| JDK | **tarball 自带 bundled JDK 17** | 无需系统 JDK；已有 `/opt/app/jdk-17` 时可用 `ES_JAVA_HOME` 指向 |
| 运行模式 | 单节点、免认证 | `127.0.0.1:9200`，教学/原型友好 |
| 中文分词 | IK 7.17.23 | 插件版本必须与 ES 版本**完全一致** |
| 部署环境 | WSL2（linux 6.6.87） | loopback 模式免 bootstrap 检查 |
### 决策树（选路）
```text
需要 ES？
├─ 教学/原型（单机免认证）→ 本流程（tarball 安装）
└─ 生产集群 → 超出本篇，参考官方 production 文档
网络情况？
├─ 本地已有 tarball → 直接装（最快）
├─ 有远程中转服务器 → ssh 中转下载（实测 13MB/s）
└─ 直连官方 → 国内约 600KB/s，325MB 约 10 分钟
```
### 下载：远程中转（国内提速 20 倍）
elastic 官方源国内直连约 600KB/s；借助一台海外 VPS 中转可达 **13MB/s**：
```bash
# ① 在远程服务器上下载（免密 ssh）
ssh -p 26225 xt@144.34.185.72 \
"cd /tmp && curl -sL -o es7.tar.gz \
https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.23-linux-x86_64.tar.gz"
# ② 回传本地（scp）
scp -P 26225 xt@144.34.185.72:/tmp/es7.tar.gz /tmp/es7.tar.gz
# ③ 清理远程临时文件 + md5 双端校验（一致性：c1d71631ab08fa8c3a8ab36dc5e93c91）
ssh -p 26225 xt@144.34.185.72 "rm -f /tmp/es7.tar.gz"
md5sum /tmp/es7.tar.gz
```
> 💡 下载后务必 `tar -tzf es7.tar.gz > /dev/null` 快速校验完整性——gzip 截断的包要等到解压时才报错。
|
|
### 八步安装流程
```text
①前置检查 ─▶ ②获取tarball ─▶ ③解压 ─▶ ④写配置 ─▶ ⑤JVM堆 ─▶ ⑥可选IK ─▶ ⑦启动 ─▶ ⑧健康检查
```
**① 前置检查**：`/opt/app` 可写（属主为普通用户）；**不能以 root 运行**（tarball 版强制拒绝）；磁盘剩余 2G 以上。
**②③ 解压到 /opt/app/es7**
```bash
mkdir -p /opt/app/es7
tar -xzf /tmp/es7.tar.gz -C /opt/app/es7 --strip-components=1
```
> `--strip-components=1` 去掉 tarball 顶层目录 `elasticsearch-7.17.23/`，使 `$ES_HOME == /opt/app/es7`，bin/config 直接在其下。
**④ 单节点课程配置** `config/elasticsearch.yml`：
```yaml
cluster.name: es7-learn
node.name: node-1
path.data: /opt/app/es7/data
path.logs: /opt/app/es7/logs
network.host: 127.0.0.1 # loopback=开发模式，免 bootstrap 检查
http.port: 9200
discovery.type: single-node # 单节点，不做集群发现
xpack.security.enabled: false # 关闭认证（7.17 首装可能自动开启 basic 认证）
```
**⑤ JVM 堆**：写 `config/jvm.options.d/heap.options`：
```text
-Xms1g
-Xmx1g
```
**JDK 说明**：ES 7.17 tarball **自带 bundled JDK 17**（`$ES_HOME/jdk/`），开箱即用；若想用系统的 `/opt/app/jdk-17`，启动前 `export ES_JAVA_HOME=/opt/app/jdk-17`。
**⑦⑧ 启动与健康检查**
```bash
/opt/app/es7/bin/elasticsearch -d -p /opt/app/es7/es.pid \
< /dev/null > /dev/null 2>&1 # 三个 fd 必须显式重定向（见避坑）
curl -s localhost:9200 # 返回 cluster_name / version 即成功
```
实测输出：
```json
{
"name" : "node-1",
"cluster_name" : "es7-learn",
"version" : { "number" : "7.17.23", "build_type" : "tar", "lucene_version" : "8.11.3" },
"tagline" : "You Know, for Search"
}
```
### ⚠️ 避坑：`-d` 守护进程不脱离 stdout 管道
**现象**：`bash es_install.sh ... | tail -25` 永不退出（tail 等不到 EOF）。
**根因**：`elasticsearch -d` 的 JVM 进程继承了父 shell 的 stdout/stderr 管道 fd 且不关闭；ES 活着 → 管道永远不空 → 下游命令阻塞。
**处置**：启动时显式重定向三个 fd（stdin/stdout/stderr）。此坑适用于一切 nohup/daemon 场景。
|
|
### 为什么需要 IK？
ES 默认的 standard 分析器把中文按**单字**切分——“连续对话模式”变成 [连,续,对,话,模,式]，关键词检索基本退化为单字匹配。IK 提供**词级**中文分词，是中文 RAG 的标配。
### 安装（infini.cloud 国内源）
```bash
/opt/app/es7/bin/elasticsearch-plugin install --batch \
https://get.infini.cloud/elasticsearch/analysis-ik/7.17.23
```
- **版本铁律**：IK 版本必须与 ES 版本**完全一致**（7.17.23 对 7.17.23），否则启动报错
- 装完需**重启 ES** 生效
- 查看已装插件：`/opt/app/es7/bin/elasticsearch-plugin list` → `analysis-ik`
### 验证分词效果
```bash
curl -s 'http://127.0.0.1:9200/_analyze' \
-H 'Content-Type: application/json' \
-d '{"analyzer":"ik_max_word","text":"极客AI音箱支持连续对话模式"}'
```
实测输出（词级切分 ✅）：
```text
极 / 客 / ai / 音箱 / 支持 / 连续 / 对话 / 模式
```
### ik_max_word vs ik_smart 与粒度坑
| 分析器 | 粒度 | "保修期是多久"的切分 |
|---|---|---|
| `ik_max_word` | 细（最细粒度穷尽） | 保修期 / **保修** / 期 / 是 / 多久 |
| `ik_smart` | 粗（智能合并） | **保修期** / 是 / 多久 |
**实测踩坑**：索引侧用 ik_max_word（切成“保修”），查询侧若用 ik_smart（“保修期”整词），**粗粒度查询词错过细粒度索引词 → 漏召回**（问“保修期多久”无命中，但库里明明有“整机保修一年”）。
**结论**：小知识库场景**索引与查询统一用 ik_max_word**（召回优先）；追求精度再在检索后加精排。网传"index=max_word, search=smart"并非铁律，取决于查询词与文档词的粒度关系。
|
|
### 一键脚本（nlp_es_install 智能体）
安装过程已沉淀为可复用脚本（智能体 `nlp_es_install` 的 `scripts/es_install.sh`）：
```bash
# 安装（三种取包方式）
bash es_install.sh install --tarball /tmp/es7.tar.gz # 本地包
bash es_install.sh install --remote "xt@144.34.185.72:26225" # 远程中转下载
bash es_install.sh install --with-ik # 直连官方+装IK
# 生命周期
bash es_install.sh start|stop|status
```
启停实现要点：
```bash
# 启动：-d 守护 + pid 文件 + 三 fd 重定向（防管道挂死）
"$ES_HOME/bin/elasticsearch" -d -p "$PID_FILE" < /dev/null > /dev/null 2>&1
# 停止：按 pid 文件优雅停止，超时再强杀
kill "$(cat "$PID_FILE")"
# 存活判断
[ -f "$PID_FILE" ] && kill -0 "$(cat "$PID_FILE")" 2>/dev/null
```
实测生命周期输出：
```text
[es_install] 已发送停止信号...
[es_install] 已停止
[es_install] 后台启动 ES（首启约 10~30 秒）...
[es_install] 健康检查通过:
cluster: es7-learn version: 7.17.23 lucene: 8.11.3
IK 中文分词: 已安装
```
### 排错手册（实踩坑汇总）
| 症状 | 根因 | 处置 |
|---|---|---|
| `can not run elasticsearch as root` | tarball 版拒绝 root | 切普通用户执行 |
| 启动即退 `bootstrap checks failed` | network.host 非 loopback 触发生产模式 | 改回 127.0.0.1；或 `sudo sysctl -w vm.max_map_count=262144` |
| `curl: Connection refused` | 未起完或崩溃 | `tail -50 $ES_HOME/logs/es7-learn.log`；常见 yml 缩进错误 |
| 客户端 401 | 7.17 首装自动开 basic 认证 | 配 `xpack.security.enabled: false` 后重启 |
| IK 装完分词报错 | 插件版本 ≠ ES 版本 | 卸载重装对应版本 |
| `port 9200 already in use` | 旧实例未停 | 按 pid 文件 kill 或 `--force` 重装 |
| 下载后解压报 gzip 错 | 网络截断 | `tar -tzf` 预检；重下或走远程中转 |
| 外层管道命令永不退出 | `-d` 守护进程持有 stdout | 启动时三 fd 显式重定向 |
|
|
### 0905_rag_es.ipynb 总览
本文 notebook 产出文件：`/mnt/d/wks/jupyter/ai58/0905_rag_es.ipynb`（28 个单元格，py312 环境 + 本地 ES 7.17.23 实测执行通过，0 错误），由 nlp_rag 智能体的一键生成器产出：
```bash
/wks/python/micromamba/envs/py312/bin/python \
.config/agent/nlp_rag/scripts/nlp_rag_nb.py \
--out /mnt/d/wks/jupyter/ai58/0905_rag_es.ipynb \
--execute --env-file /mnt/d/wks/jupyter/env.txt
```
### 变量链（单元格间前后依赖）
```text
kenb_docs(知识库) → chunk_text(切分演示) → es/MAPPINGS(建索引)
→ vectors=get_embeddings(向量化) → helpers.bulk(入库)
→ kw_search(关键词路) / vec_search(向量路) → 词汇鸿沟对比
→ PROMPT_TEMPLATE/build_prompt/get_completion(生成侧)
→ RAG_Bot(整合) → TESTS 测试矩阵 → 清理演示索引
```
### 知识库设计：成败在考点
用虚构产品「极客AI音箱 X1」的 12 条资料当知识库（生产中来自 PDF/网页/数据库）。**关键不是数量，而是埋好三类考点**：
| 考点 | 示例 | 验证目标 |
|---|---|---|
| 直答类 | 售价 399 元 / 整机保修一年 | 两路检索都应命中并正确作答 |
| **同义改写类** | 原文“随时打断” vs 问“能插嘴吗” | 关键词失败、向量成功——词汇鸿沟实证 |
| 知识库外 | “电池续航怎么样”（库里没有） | 拒答兜底生效，不编造 |
数据结构（每条自带粒度合适的 chunk，真实场景需先切分）：
```python
KENB_DOCS = [
{"doc_id": "spec_01", "category": "参数",
"text": "极客AI音箱 X1 搭载 6 麦克风环形阵列，5 米远场唤醒率达到 95%，适合客厅、会议室等大空间使用。"},
{"doc_id": "func_01", "category": "功能",
"text": "X1 支持连续对话模式：唤醒后可以多轮追问、随时打断，无需反复喊唤醒词。"}, # ← 同义改写考点
{"doc_id": "sale_01", "category": "售后",
"text": "X1 整机保修一年，人为损坏除外；音箱主体 15 天内出现质量问题可直接换新。"},
{"doc_id": "price_01", "category": "价格",
"text": "X1 官方售价 399 元，首发期下单立减 50 元，学生认证再享 9 折优惠。"},
# ... 共 12 条：参数2/功能3/售后2/价格2/对比2/隐私1
]
```
### 切分演示（chunk_size / overlap）
```python
def chunk_text(text, chunk_size=120, overlap=20):
"""朴素滑窗切分：按句号边界 + 固定窗口 + 重叠"""
sentences = [s.strip() for s in text.replace("；", "。").split("。") if s.strip()]
chunks, buf = [], ""
for sent in sentences:
piece = sent + "。"
if len(buf) + len(piece) > chunk_size and buf:
chunks.append(buf)
buf = buf[-overlap:] + piece # 保留尾部重叠，防止关键句被切断
else:
buf += piece
if buf:
chunks.append(buf)
return chunks
```
|
|
### 环境准备：加载 API 配置
`env.txt` 含 `OPENAI_API_KEY` / `OPENAI_BASE_URL`（OpenAI 兼容代理），python-dotenv 加载后 OpenAI 客户端自动读取：
```python
import warnings
warnings.simplefilter("ignore") # 屏蔽 elasticsearch7 客户端 body 参数弃用告警
from elasticsearch7 import Elasticsearch, helpers
import numpy as np
from numpy import dot
from numpy.linalg import norm
from dotenv import load_dotenv
ENV_FILE = r"/mnt/d/wks/jupyter/env.txt"
load_dotenv(ENV_FILE, override=True)
from openai import OpenAI
llm_api = OpenAI() # 自动读取上述两个环境变量
```
### 建立双索引：倒排（关键词）+ 向量
一条记录同时写 `text`（IK 分词，供 match）与 `embedding`（dense_vector，供向量检索）——**双路索引 = 混合检索的地基**：
```python
ES_URL = "http://127.0.0.1:9200"
INDEX_NAME = "geeker_kb_v1"
es = Elasticsearch(hosts=[ES_URL])
if es.indices.exists(index=INDEX_NAME):
es.indices.delete(index=INDEX_NAME) # 演示可重复运行
MAPPINGS = {
"mappings": {
"properties": {
"doc_id": {"type": "keyword"}, # 精确匹配/聚合
"category": {"type": "keyword"},
"text": {"type": "text"}, # 关键词检索字段
"embedding": { # 向量检索字段
"type": "dense_vector",
"dims": 1536,
# 注：similarity 是 8.x 参数，7.17 无效——
# 打分由查询侧 script_score 的 cosineSimilarity() 完成
},
}
}
}
# 装了 IK 则升级中文分词；索引与查询统一 ik_max_word 细粒度（防粒度错配漏召回）
try:
MAPPINGS["mappings"]["properties"]["text"] = {
"type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_max_word"}
es.indices.create(index=INDEX_NAME, body=MAPPINGS)
print("✅ 索引创建成功：IK 中文分词（双侧 ik_max_word）")
except Exception:
MAPPINGS["mappings"]["properties"]["text"] = {"type": "text"}
es.indices.create(index=INDEX_NAME, body=MAPPINGS)
print("⚠️ 未装 IK，退化为 standard 单字分词")
```
**两个设计决策的解读**：
1. **双侧都用 ik_max_word**：实测查询侧用 ik_smart 时，“保修期”整词无法命中索引里的“保修”——粗粒度查询 × 细粒度索引 = 漏召回
2. **不加 similarity 参数**：7.17 会告警“no effect”（8.x 特性），7.17 的余弦打分在查询侧 script_score 完成
|
|
### 向量化：get_embeddings + 稳定性自检
```python
EMBED_MODEL = "text-embedding-3-small" # 1536 维
def get_embeddings(texts):
"""调用 OpenAI 兼容 embedding API，返回向量列表"""
resp = llm_api.embeddings.create(input=list(texts), model=EMBED_MODEL)
return [x.embedding for x in resp.data]
texts = [d["text"] for d in KENB_DOCS]
vectors = get_embeddings(texts)
print(f"✅ {len(vectors)} 条文本已向量化，维度 = {len(vectors[0])}")
```
### ⚠️ embedding 后端漂移坑（本次最隐蔽的坑）
实测该代理的 `text-embedding-ada-002` **多后端负载均衡、向量空间不互通**：同一文本两次调用 cos=1.0，第三次 cos≈0（正交）——建库与查询落到不同后端 = 向量检索整体失效（分数普遍 ≈1.0，即 cos≈0）。
**检测方法（任何新代理先跑探针）**：
```python
vecs = [get_embeddings(["稳定探针文本abc"])[0] for _ in range(4)]
stable = all(cos_sim(vecs[0], v) > 0.99 for v in vecs[1:]) # False 即漂移
```
**结论**：换 `text-embedding-3-small`（实测稳定）。通用教训——**向量空间一致性是向量检索的命门**：不只“建库与查询同模型”，还有“同模型的不同后端”。
### 余弦相似度工具 + 自检
```python
def cos_sim(a, b):
"""余弦相似度：-1~1，越大越相似"""
return dot(a, b) / (norm(a) * norm(b))
print("自检:", f"同文档 cos={cos_sim(vectors[0], vectors[0]):.4f},",
f"参数vs售后 cos={cos_sim(vectors[0], vectors[5]):.4f}")
# 实测：同文档 1.0000，参数vs售后 0.4476 —— 语义区分度正常
```
### 批量入库：helpers.bulk
```python
actions = [
{
"_index": INDEX_NAME,
"_id": d["doc_id"], # 用业务 id，可重复执行覆盖
"_source": {
"doc_id": d["doc_id"],
"category": d["category"],
"text": d["text"], # 原文（返回给 LLM）
"embedding": vec, # 向量（检索打分用）
},
}
for d, vec in zip(KENB_DOCS, vectors)
]
helpers.bulk(es, actions)
es.indices.refresh(index=INDEX_NAME) # 立即可查
print(f"✅ 已入库 {es.count(index=INDEX_NAME)['count']} 条")
```
> 解读：**文本与向量一次落库**、`_id` 用 doc_id 保证幂等；refresh 让近实时（NRT）索引立即可搜。
|
|
### 关键词检索：match + 倒排索引
```python
def kw_search(query, top_n=3):
"""关键词检索：match 查询 text 字段（ES 端 IK 分词 → 倒排匹配 → BM25 打分）"""
body = {
"size": top_n,
"_source": ["doc_id", "category", "text"],
"query": {
"match": {
"text": query
}
},
}
res = es.search(index=INDEX_NAME, body=body)
return [(h["_source"], h["_score"]) for h in res["hits"]["hits"]]
def show_results(query, results):
print(f"❓ {query}")
for src, score in results:
print(f" [{score:.2f}][{src['category']}] {src['text'][:40]}...")
```
实测（BM25 分数越高越相关）：
```text
❓ X1 的售价是多少
[2.37][价格] X1 官方售价 399 元，首发期下单立减 50 元...
❓ 保修期多久
[2.17][售后] X1 整机保修一年，人为损坏除外...
```
### 词汇鸿沟实验：关键词检索的天花板
两个**语义相同**的问题，只因用词不同，检索结果天壤之别：
```python
# 原文用的词是"随时打断"；用户问"能插嘴吗"——字面对不上
q1, q2 = "唤醒后能随时打断吗", "说话说到一半能插嘴吗"
```
实测输出：
```text
── 两个问题语义相同，检索结果却不同 ──
❓ 唤醒后能随时打断吗
[7.85][功能] X1 支持连续对话模式：唤醒后可以多轮追问、随时打断... ← "打断"字面命中
[2.54][隐私] ...
❓ 说话说到一半能插嘴吗
[1.34][售后] X1 提供全国联保... ← "插嘴"库里没有，召回的全是弱匹配
💡 观察：'打断' 字面命中 → 召回准确；'插嘴' 知识库里没有 → 召回失败
```
> 这正是课程里 "chat version" vs "conversational variant" 的中文复刻：**倒排索引只认字面，不懂语义**——向量检索的立身之本。
|
|
### 向量检索：script_score + 余弦相似度
```python
def vec_search(query, top_k=3):
"""向量检索：query 向量化 → script_score 余弦相似度打分"""
qvec = get_embeddings([query])[0]
body = {
"size": top_k,
"_source": ["doc_id", "category", "text"],
"query": {
"script_score": {
"query": {"match_all": {}},
"script": {
# cosineSimilarity 取值 -1~1；+1.0 平移为非负的 _score
"source": "cosineSimilarity(params.query_vector, 'embedding') + 1.0",
"params": {"query_vector": qvec},
},
}
},
}
res = es.search(index=INDEX_NAME, body=body)
return [(h["_source"], h["_score"]) for h in res["hits"]["hits"]]
```
**解读**：
- ES 7.17 的向量检索写法是 `script_score` 的 `cosineSimilarity()`——`_knn_search` 是 8.0+ 的 API，7.x 没有
- `+1.0` 把 -1~1 的余弦平移成非负 `_score`（ES 不接受负分数）
- 生产亿级向量请换 HNSW/IVF 索引或专业向量库（见「常见向量数据库」章）
同样的“插嘴”问题，向量检索把它救回来了：
```text
❓ 说话说到一半能插嘴吗
[1.35][功能] X1 支持连续对话模式：唤醒后可以多轮追问、随时打断... ← 语义对齐成功！
[1.29][隐私] ...
[1.24][功能] X1 可通过语音控制智能家居...
💡 观察：'插嘴' 与 '打断' 语义相近 → 向量相似度高 → 正确召回 func_01
```
### 两路检索对比（实测）
```text
问题 关键词Top1 向量Top1 一致?
------------------------------------------------------------
售价多少 price_01 price_01 ✅
保修期多久 sale_01 sale_01 ✅
能插嘴吗 （无命中） priv_01 ❌
能控制智能家居吗 func_02 func_02 ✅
和HomePod比怎么样 cmp_01 cmp_01 ✅
```
**三条结论**：①“能插嘴吗”关键词无命中——词汇鸿沟；②完整句“说话说到一半能插嘴吗”向量能正确召回 func_01，但**短查询向量 Top1 会偏移——查询越具体、语义检索越稳**；③生产做法：关键词+向量**混合检索**互补。
|
|
### 提示词：五要素 + 明确的引用示例
```python
PROMPT_TEMPLATE = """
你是一个专业的产品知识库问答助手。 # ①角色定义
任务：仅根据【参考文档】回答用户问题。 # ②任务说明
约束： # ③约束条件
- 只使用参考文档中的信息，禁止编造
- 若参考文档包含答案（即使措辞与问题不同），必须依据文档作答
- 仅当所有文档都与问题无关时，才回复：根据提供的信息无法回答此问题
- 回答末尾以 [来源:文档编号] 标注引用，编号取自每条文档开头的方括号，例如 [来源:price_01]
- 简洁准确，用中文回答 # ⑤输出要求
【参考文档】
__INFO__ # ④信息注入占位符
【用户问题】
__QUERY__
"""
def build_prompt(template, **kwargs):
"""占位符填充：__INFO__ / __QUERY__（列表按行拼接）"""
prompt = template
for key, value in kwargs.items():
if isinstance(value, list) and all(isinstance(e, str) for e in value):
val = "\n".join(value)
else:
val = str(value)
prompt = prompt.replace("__" + key.upper() + "__", val)
return prompt
def get_completion(prompt, temperature=0):
resp = llm_api.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=temperature, # RAG 要"稳"不要"创"，取 0
)
return resp.choices[0].message.content
```
**实测调优点**：引用约束若只写“以 [来源:doc_id] 标注”，模型会照抄占位符字面量——**必须给具体示例**（例如 [来源:price_01]）且文档行首用 `[price_01]` 格式；拒答约束过硬会导致检索正确也拒答，需补“措辞不同也必须作答”。
### RAG_Bot：检索函数可注入
```python
class RAG_Bot:
"""search_func(查询, top_n) → [(source_dict, score), ...]"""
def __init__(self, search_func, llm_func, prompt_template, n_results=3):
self.search, self.llm = search_func, llm_func
self.template, self.n_results = prompt_template, n_results
def chat(self, user_query):
results = self.search(user_query, self.n_results) # 1. 检索
if not results:
return "抱歉，没有找到相关信息。"
infos = [f"[{src['doc_id']}] {src['text']}" for src, _ in results]
prompt = build_prompt(self.template, info=infos, query=user_query) # 2. 组装
return self.llm(prompt) # 3. 生成
kw_bot = RAG_Bot(kw_search, get_completion, PROMPT_TEMPLATE) # 关键词版
vec_bot = RAG_Bot(vec_search, get_completion, PROMPT_TEMPLATE) # 向量版
```
### 实测验收矩阵（全部达标 ✅）
```text
🧪 [直答·价格] X1 的售价是多少？
── 关键词版: X1 的官方售价是 399 元，首发期下单立减 50 元... [来源:price_01]
── 向量版 : 同上 ✅
🧪 [直答·售后] 保修期是多久？
── 关键词版: X1 整机保修一年，人为损坏除外。 [来源:sale_01]
── 向量版 : 同上（补充了 15 天换新细节）✅
🧪 [同义改写·词汇鸿沟] 说话说到一半能插嘴吗？
── 关键词版: 根据提供的信息无法回答此问题。 ← 检索无命中，正确拒答
── 向量版 : 可以，X1 支持连续对话模式，允许多轮追问和随时打断。 [来源:func_01] ✅
🧪 [知识库外·拒答兜底] X1 的电池续航怎么样？
── 关键词版: 根据提供的信息无法回答此问题。 ✅
── 向量版 : 根据提供的信息无法回答此问题。 ✅ ← 库里没有，绝不编造
```
**一句话总结**：28 个单元格跑完 RAG 全流程——**检索质量决定上限**（插嘴问题关键词版救不回来），**提示词约束守住底线**（库外问题拒答兜底），**向量检索补上语义理解**这最后一块拼图。
|
|
### BM25 是什么
BM25（Best Match 25）是经典的**基于词频与逆文档频率（IDF）的相关性排序算法**，信息检索的基石——**ElasticSearch 的 match 查询默认打分算法就是 BM25**（Lucene 默认相似度）。
因此「用 BM25 替代 ES」本质上是：**把 ES 的打分内核搬进 Python 进程内**，用 jieba（替代 IK 分词）+ rank_bm25（替代 Lucene 打分），**零外部服务**完成同类相似文本检索。
### 打分公式
```text
tf(qi,d)·(k1+1)
score(q,d) = Σ ───────────────────────── · IDF(qi)
tf(qi,d) + k1·(1 − b + b·|d|/avgdl)
```
| 组件 | 含义 | 典型取值 |
|---|---|---|
| IDF(qi) | 词的逆文档频率——越稀有的词越重要 | rank_bm25 变体（带 epsilon 平滑） |
| tf(qi,d) | 词 qi 在文档 d 中的频次 | 分词后统计 |
| \|d\| / avgdl | 文档长度 / 平均长度（长度归一化） | 全语料统计 |
| **k1** | 词频饱和度：越大词频影响越大 | **1.8**（范围 1.2~2.0） |
| **b** | 长度归一化强度：0 不考虑长度，1 完全归一 | **0.75**（范围 0.5~0.8） |
### BM25 分数的三个特性（决定 Top-K 设计）
1. **无固定上界**：分数与查询词数量、IDF 相关，不是 0~1 的相似度
2. **同一查询内可比**：分数越高越相关
3. **跨查询不可比**："售价多少"与"保修期多久"的最高分可能差数倍
> 与余弦相似度（-1~1、跨查询可比）对照：BM25 的量纲特性决定了检索条数控制要用"上限 + 相对阈值"而非绝对分数——详见「Top-K策略设计」页。
|
|
### tpf.BM25Reranker（bm25.ipynb 所用封装）
本地库 `tpf.nlp.text.BM25Reranker` = **jieba 分词 + rank_bm25 的 BM25Okapi** 封装：
```python
import sys
sys.path.insert(0, r"/ai/wks/aitpf/src")
from tpf.nlp.text import BM25Reranker
# ① 建索引：jieba.lcut 分词 + BM25Okapi 统计（词频/IDF/文档长度）
bm25 = BM25Reranker(df=df, use_cols=["text", "label"], k1=1.8, b=0.75)
# 也支持直接传文本列表：BM25Reranker(texts=[...])
# ② 全量打分：新查询与所有文档的 BM25 分数（numpy 数组）
scores = bm25.sim_scores("新查询文本")
# array([25.4, 31.7, 30.8, ..., 18.2], shape=(100000,))
```
### 取 Top-K：argpartition（O(n)）还是 argsort（O(n log n)）
bm25.ipynb 原课演示的两种取法：
```python
import numpy as np
# 方法① 最高效：np.argpartition，O(n)，前 k 个无序但确为最大 k 个
indices = np.argpartition(scores, -3)[-3:]
# 方法② 要有序：先 argsort 再切片反转，O(n log n)
top3 = np.argsort(scores)[-3:][::-1] # 从大到小
```
> RAG 检索的实用组合：**argpartition 先圈定前 K（快），再对这 K 个 argsort（K 小，排序便宜）**。
### label_topk：按标签分组的 Top-K（bm25_scores*.csv 的来源）
```python
bm25.label_topk(new_query, sim_top_k=5, label_top_k=3, outdir="")
# 逻辑：sim_scores 全量打分 → 按 label 分组 → 每组取分数前 5 的均值(mean_score)
# → 按 mean_score 降序取前 3 个标签 → 导出 bm25_scores.csv / bm25_scores_mean5.csv
```
### 性能基准（bm25.ipynb 实测：10 万条语料）
| 指标 | 数值 |
|---|---|
| BM25 初始化（10万条分词+建索引） | 15.8 秒 |
| 单查询全量打分（10万条） | 平均 1.27 秒 |
| 100 个查询总耗时 | 126.8 秒 |
> 结论：进程内 BM25 **十万级语料完全可用**；百万级以上或需实时增删改时上 ES（IVF/HNSW 索引）或专业向量库。
|
|
### 检索结果返回多少条？——参数化的四种策略
`sim_scores` 返回**全量**分数，返回多少条由调用方决定。推荐「**上限 + 过滤**」双保险设计（全部做成参数，可动态调整）：
```python
def bm25_search(query, top_k=100, min_score=None, ratio=None):
"""BM25 检索：全量打分 → O(n) 取前 top_k → 可选阈值过滤
top_k: 返回条数上限（默认 100）
min_score: 绝对分数阈值（低于即丢弃；BM25 分数无上界，需按语料校准）
ratio: 相对阈值（保留 >= 最高分*ratio 的结果；跨语料更稳）
"""
scores = bm25.sim_scores(query)
k = min(top_k, len(scores))
idx = np.argpartition(scores, -k)[-k:] # O(n) 圈定前 k（无序）
top = idx[np.argsort(scores[idx])[::-1]] # 再按分数降序
results = [(df.iloc[i], float(scores[i])) for i in top]
if min_score is not None:
results = [r for r in results if r[1] >= min_score]
elif ratio is not None:
results = [r for r in results if r[1] >= results[0][1] * ratio]
return results
```
### 断崖检测：条数自适应（auto_topk）
相关文档的分数会**聚簇**在高位，断崖之后基本是噪声——找相邻分数差分最大处截断：
```python
def auto_topk(query, max_k=100, min_k=3):
scores = bm25.sim_scores(query)
order = np.argsort(scores)[::-1][:max_k]
s = scores[order]
if len(s) <= min_k:
return order, s
gaps = s[:-1] - s[1:] # 相邻分数差分
cut = max(int(np.argmax(gaps[:len(gaps) - min_k + 1])) + 1, min_k)
return order[:cut], s[:cut]
```
### 策略对比与推荐
| 策略 | 参数 | 适用 | 特点 |
|---|---|---|---|
| ① 固定上限 | `top_k=100`（默认） | 批处理/要稳定条数；检索漏斗第一层 | 简单稳定，但会带噪声 |
| ② 绝对阈值 | `min_score` | 分数低于阈值基本无关 | 不可移植：换语料量纲就变 |
| ③ 相对阈值 | `ratio`（≥最高分×ratio） | **推荐默认** | 分数量纲无关，跨语料稳 |
| ④ 断崖检测 | `auto_topk` | 条数难定、查询质量参差 | 自适应；需 min_k 兜底 |
| ⑤ 送 LLM 再截断 | 最终取 3~5 条 | RAG 上下文有 token 预算 | 100 条是漏斗不是直送 |
**推荐组合**：`top_k=100` 兜底上限 + `ratio≈0.2~0.3` 过滤噪声 → 送 LLM 前再取 **3~5** 条。
实测（同一个查询，三种策略的条数差异）：
```text
① 固定Top-100 → 12 条（分数 0.92 ~ 3.42） # 12条小语料全返回，含 0.92 的噪声
② 相对阈值 ratio=0.3 → 10 条 # 砍掉低分尾部
③ 断崖检测 → 3 条（分数 2.60 ~ 3.42） # 精准聚簇，条数自适应
```
> 💡 小语料差异有限；**百万级语料 + 模糊查询时，阈值/断崖能把上百条噪声砍到个位数**——这正是"返回多少条"不该写死的原因。
|
|
### 0905_rag_bm25.ipynb：零服务依赖的完整 RAG
产出：`/mnt/d/wks/jupyter/ai58/0905_rag_bm25.ipynb`（21 单元格，py312 实测 0 错误），nlp_rag 生成器一键产出：
```bash
python .config/agent/nlp_rag/scripts/nlp_rag_nb.py \
--out 0905_rag_bm25.ipynb --execute --retriever bm25
```
### 流程（与 ES 版同构，仅检索层不同）
```text
知识库 kenb_docs(12条·三类考点)
↓
jieba 粒度修复 → BM25Reranker 建索引（k1=1.8, b=0.75）
↓
bm25_search(top_k=100, ratio) / auto_topk 断崖 ── 检索层（进程内）
↓
PROMPT_TEMPLATE + build_prompt + get_completion ── 生成层（与 ES 版同款）
↓
RAG_Bot(检索函数注入) → TESTS 测试矩阵
```
### ⚠️ 实测踩坑：jieba 分词粒度（与 IK 同款，跨算法通用）
首跑时"保修期多久"BM25 分数**全 0**——库里明明有"整机保修一年"：
```text
jieba.lcut（精确模式） : ['保修期', '是', '多久'] ← "保修期"整词
jieba.lcut_for_search : ['保修', '保修期', '是', '多久'] ← 含子词"保修" ✅
文档 lcut("整机保修一年") : ['整机', '保修', '一年', ...] ← 库里是"保修"
```
**修复**：建索引前 `jieba.lcut = jieba.lcut_for_search`（搜索引擎模式 = jieba 版 ik_max_word，整词+子词），索引与查询同时生效。修复后：
```text
❓ 保修期多久
[2.07][售后] X1 整机保修一年，人为损坏除外... ← Top1 命中 sale_01 ✅
[0.00][隐私] ... ← 未匹配项分数恰为 0
```
> 2.07 vs 0.00 的分数断崖，恰好演示了 ratio 相对阈值的过滤价值。
### 词汇鸿沟：BM25 版同样存在（如实展示）
```text
❓ 唤醒后能随时打断吗 → Top1 分数 7.85（func_01）✅ "打断"词面命中
❓ 说话说到一半能插嘴吗 → Top1 分数 0.00 ❌ "插嘴"库里没有
```
BM25 与 ES 关键词检索同源（都基于词面匹配），**都不懂语义**——同义改写类考点的失败是"符合预期的失败"，正是向量检索的用武之地。
### RAG_Bot 实测验收矩阵
| 测试 | BM25 版结果 |
|---|---|
| 直答·价格 | ✅ 399 元 + [来源:price_01] |
| 直答·售后 | ✅ 保修一年 + [来源:sale_01]（分词修复后） |
| 同义改写（插嘴） | 拒答——词汇鸿沟，BM25 无语义能力（如实预期） |
| 知识库外（电池） | ✅ 拒答兜底，绝不编造 |
RAG_Bot 检索漏斗：`top_k=100 → ratio=0.2 过滤 → 送 LLM 取 3 条`——同一个 RAG_Bot 骨架，ES 版注入 kw_search/vec_search、BM25 版注入 bm25_search，**换引擎只换检索函数**。
|
|
### 三种检索方案对比
| 维度 | BM25（进程内） | ElasticSearch | 向量检索 |
|---|---|---|---|
| 部署 | **零服务**，pip 即用 | 需 JVM 服务（nlp_es_install） | 需向量库，或进程内暴力余弦 |
| 规模 | **十万级实测可用**（10万条查询约1.3s） | 千万~亿级 | 依赖 ANN 索引结构 |
| 打分内核 | BM25（k1/b 可调） | BM25（Lucene 默认） | 余弦相似度（-1~1 跨查询可比） |
| 语义理解 | ❌ 词面匹配（词汇鸿沟） | ❌ 词面匹配（同 BM25） | ✅ 同义/跨语言召回 |
| 精确术语 | ✅ 型号/编号强命中 | ✅ | 可能有偏差 |
| 增量更新 | 重建索引（进程内便宜） | 实时增删改 | 实时增删改 |
| 中文分词 | jieba（注意粒度坑） | IK（注意粒度坑） | 无需分词 |
| API 形态 | sim_scores → numpy 数组 | match / script_score | query → Top-K |
### 选型决策
```text
快速起步 / 教学 / 嵌入现有 Python 流程 ──▶ BM25（nlp_rag 默认引擎）
需要检索服务化 / 大规模 / 实时增删改 ──▶ ElasticSearch
同义改写多 / 跨语言 / 语义问答 ──▶ 向量检索
生产 RAG ──▶ 关键词(BM25/ES) + 向量 混合检索（RRF 融合）
```
### 混合检索：为什么两个都要
| 痛点 | BM25/ES 关键词路 | 向量路 |
|---|---|---|
| "插嘴" vs "打断"（同义改写） | ❌ | ✅ |
| "X1" "7B" 精确型号编号 | ✅ | 可能偏差 |
| 中文问英文库（跨语言） | ❌ | ✅ |
两路各自召回 → RRF（倒数排名融合）合并 → 取融合后 Top-K 送 LLM——词汇鸿沟与精确术语两个痛点同时解决。
### 本机资产（nlp_rag 智能体维护）
- BM25 引擎：tpf 库 `/ai/wks/aitpf/src`（BM25Reranker），py312 已装 rank_bm25/jieba
- ES 引擎：`/opt/app/es7`（7.17.23 + IK），`http://127.0.0.1:9200` 无认证
- 生成命令：`nlp_rag_nb.py --retriever bm25|es`（**默认 bm25**），`--execute` 实测内嵌
- 原始基准实验：`/mnt/d/wks/jupyter/bm25.ipynb`（10 万条语料性能数据）
|
|
### 主流向量库：定位·架构·场景
| 数据库 | 定位与架构 | 适用场景与规模 |
|---|---|---|
| **Milvus** | 专业开源分布式向量库，云原生架构，多索引类型+GPU 加速 | **大规模**（千万级至十亿级以上）、高并发、对性能和可用性有严格要求的企业级生产环境 |
| **Qdrant** | Rust 编写的高性能向量库，向量检索与结构化数据过滤（Payload）为核心设计 | 中等规模（百万~千万级），尤其适合**复杂标量过滤**与**混合检索**的 RAG/实时检索场景，性能与运维复杂度的良好平衡 |
| **Weaviate** | 内置多种模块（如文本向量化），支持 GraphQL，擅长复杂语义与知识图谱类查询 | 需要**复杂查询**、知识图谱或**混合搜索**的企业知识应用（千万级） |
| **Pinecone** | 全托管**云原生 Serverless**，免运维开箱即用 | 快速云上落地、不想维护基础设施、对数据隐私和长期成本不敏感的团队 |
| **pgvector** | **PostgreSQL 扩展插件**，为关系库增加向量存储检索 | 团队已深度用 PG、数据量中等（百万级以内）、**不想引入新组件** |
### 补充：其他可关注的方案
- **轻量级与本地开发：Chroma**——快速原型与本地开发，安装简单零配置，不适合大规模生产
- **融合型/国产方案：电科金仓（KingbaseES）** 等——向量能力内嵌关系库内核，适合信创或 SQL 统一查询复杂业务数据的场景
- **已有搜索/文档生态**：团队已有 **Elasticsearch** 或 **MongoDB** 时，用它们内置的向量检索功能，避免异构系统过多
### 国内使用情况
- **Milvus 是国内企业用得最多的**，可以说是开源向量数据库的"事实标准"
- **pgvector** 凭借与 PostgreSQL 生态的天然融合，在已有 PG 体系的团队中非常受欢迎
### 选型三维度（本质取舍：数据规模 × 运维成本 × 功能特性）
1. **看数据量和并发**：千万~亿级首选 Milvus；百万~千万级 Qdrant 性价比高；百万级以下 pgvector/Chroma 足以应对
2. **看现有技术栈**：PG 重度用户选 pgvector 的"零新增组件"优势无可替代；已有 ES/Mongo 优先复用其向量能力
3. **看运维能力**：团队小不想管服务器选 Pinecone（托管）；有成熟运维团队选 Milvus/Qdrant 开源自托管，性能与数据控制力更强
|
|
### FAISS vs Chroma：算法引擎 vs 产品
两者的对比本质是**"高性能算法引擎"与"开箱即用的产品"**之争：
| 特性维度 | FAISS | Chroma |
|---|---|---|
| 本质定位 | 高效的相似度搜索**算法库** | 完整、开发者友好的**向量数据库** |
| 核心功能 | 仅提供向量检索的算法实现（索引、搜索） | 完整数据管理：存储、元数据过滤、持久化、API |
| 使用体验 | 极客向：自己管数据、做持久化、维护映射 | 新手友好：pip install 后几行代码上手，内置嵌入模型 |
| 性能表现 | **性能王者**：查询延迟低至 **0.19 毫秒**，近 2000 QPS | 够用：10 万级向量查询约 20 毫秒，大规模下降明显 |
| 扩展与规模 | 单机库，无内置分布式，常作更大型系统的底层引擎 | 原型利器：百万级后扩展困难，建议迁移专业库 |
> 形象比喻：**FAISS 是一台 F1 赛车的高性能发动机，Chroma 是一辆可以立刻开上路的家用轿车**。
> 有造车能力和需求选 FAISS；只想快速从 A 点到 B 点选 Chroma。
> （趣闻：Qdrant/Milvus 等不少产品底层索引引擎正是基于 FAISS/HNSW 思想构建）
### 安装复杂度排序（从简到繁，用交通工具比喻）
| 方案 | 复杂度 | 比喻 | 安装成本 | 典型场景 |
|---|---|---|---|---|
| pgvector | ⭐ 极简 | 车库里现成的家用轿车 | `CREATE EXTENSION vector;` 一条 SQL，零新组件 | 已有 PG、百万级以内 |
| Qdrant | ⭐⭐ 简单 | 高性能跑车 | All-in-One Docker 或二进制，一条命令跑起来 | 百万~千万级、高性能+过滤 |
| Weaviate | ⭐⭐⭐ 中等 | 功能齐全的房车 | Docker 便捷，但功能模块多、概念有学习成本 | 混合搜索、知识图谱类查询 |
| Milvus | ⭐⭐⭐⭐⭐ 复杂 | 运载巨大的地铁系统 | 存算分离架构：需连动 etcd + MinIO + Milvus 三组件 | 千万~十亿级、高可用严苛生产 |
### 为什么复杂度差这么多？
- **pgvector**：不是独立数据库，是 PG 扩展——不需要维护任何新服务进程
- **Qdrant / Weaviate**：独立数据库服务，但官方 All-in-One Docker 镜像把所有功能打包，一条 `docker run` 全跑起来
- **Milvus**：天生为大规模集群设计的云原生分布式系统，即便"单机版"也遵循存算分离——etcd（元数据）+ MinIO/S3（数据）+ Milvus（计算）三个容器协同
### 选型一句话
个人开发者/快速验证原型 → pgvector 或 Qdrant；PG 运维成熟且数据量不爆发 → pgvector 性价比最高；企业级千万级+、运维能力强 → 直接上 Milvus 避免未来迁移之痛。
|
|
### Qdrant 非 Docker 安装：复杂度介于"极简"与"复杂"之间
可以理解为**"准备-下载-运行"三步走**——比 pgvector 稍重（独立服务），远轻于 Milvus（无需拉起微服务集群）。
### 方式一：预编译二进制（推荐）
```bash
# Linux 示例
curl -LO https://github.com/qdrant/qdrant/releases/latest/download/qdrant-x86_64-unknown-linux-musl.tar.gz
tar -xzf qdrant-x86_64-unknown-linux-musl.tar.gz
./qdrant # 直接运行（默认端口 6333）
```
macOS 下载对应 darwin 包、Windows 下载 zip 双击 `qdrant.exe`——个人开发/快速测试非常友好。
### 方式二：从源码编译（进阶，一般不需要）
需要自己搭建 Rust 编译环境 + 安装 clang、cmake、protobuf-compiler 等依赖——仅当要启用 GPU 加速或为特定 CPU 架构做深度优化时才有必要。
### 生产化的两个关键动作
1. **变成"服务"**：直接运行 `./qdrant` 关终端即停——Linux 配 **systemd** 服务、macOS 配 **launchd** 服务（官方文档有现成模板，改路径即用）
2. **配置文件**：创建 `config.yaml` 指定数据存储路径等关键信息，别用默认配置裸奔
### 非 Docker 安装横向对比
| 方案 | 复杂度 | 核心工作 | 一句话评价 |
|---|---|---|---|
| pgvector | ⭐ 极简 | `CREATE EXTENSION vector;` | 与数据库深度绑定，几乎零负担 |
| Qdrant | ⭐⭐ 简单 | 下载二进制 → 配置系统服务 | 独立服务，安装直接、生产配置标准清晰 |
| Milvus | ⭐⭐⭐⭐⭐ 复杂 | 分别部署 etcd、MinIO 等组件 | 云原生架构，安装本身就是分布式系统部署 |
> 结论：Qdrant 非 Docker 安装门槛不高——快速试用下载二进制即跑；生产环境配 systemd 也只是一次性标准化工作，难度远低于从零搭建 Milvus 集群。
|
|
### 场景定义：NL2SQL 的 RAG 系统
> 真实案例场景：基于 **1000 张表定义 + 1~2 万行历史 SQL/存储过程 + 优质问答模板 + 映射字典** 的自然语言转 SQL（NL2SQL）系统——本质是一个典型的**结构化企业级 NL2SQL RAG**。
**结论先行**：Qdrant 不仅适合，而且是非常理想的方案（有 QueReyDB、Text2SQL Multi-Agent 等开源项目直接用 Qdrant 做 NL2SQL，路线已验证）。原因从四个维度拆解：
### ① 核心能力匹配：多类型多来源的复杂检索
**组合式文本构建**（重要实践）：
- ❌ 单一字段向量化效果差：只向量化"表名/字段名"，用户问"订单金额"可能匹配不到"订单表.订单金额"完整字段
- ✅ 最佳实践：把 **表描述 + 字段名 + 字段描述** 组合成一个完整文本块再向量化——语义完整、检索精准
- ✅ 多查询并行检索：把"上个月华东地区销售情况"拆解为"销售金额""地区信息""时间字段"等子查询并行检索再合并，显著提高召回率
### ② 查询能力：混合搜索 + 精准过滤
既有语义相似（"销售额" vs "营收"），又有精确约束（业务线、日期范围）：
- **混合搜索（Hybrid Search）**：向量检索理解语义 + 关键词匹配专业术语（SELECT、JOIN、表名），RRF 融合结果——处理历史 SQL 的专业术语特别有效
- **元数据过滤（Metadata Filtering）**：为每个数据点打标签（表名/业务线/创建日期），先过滤限定范围（"只在销售相关的表里搜"）再做向量检索；Qdrant 的**可过滤 HNSW 索引**保证带过滤查询依然低延迟
### ③ 规模与性能：数据量级轻松应对
- 有实际案例：Qdrant 索引**数亿个向量**，典型查询延迟仍控制在 **100 毫秒以内**——本案例的 1000 张表 + 2 万行 SQL 远小于这个量级
- 性能调优手段：HNSW 参数（`ef_construct`、`ef`）调节、启用**量化（Quantization）**压缩内存占用加速检索
### ④ 生态与架构契合
- 与 LLM 集成生态成熟（LangChain/LlamaIndex 原生支持）
- 开源自托管与全托管云服务双轨：数据敏感可自部署（非 Docker 也简单），运维有限用云服务，可对特定集合精细调优
### 实施路径（四步）
| 步骤 | 核心动作 | Qdrant 关键功能 |
|---|---|---|
| 1. 数据建模 | 表定义/历史SQL/问答模板分别结构化，设计元数据（数据源类型、业务标签） | 灵活的 **Payload（元数据）** 存储 |
| 2. 数据向量化 | 表字段用"表描述+字段描述"组合文本；批量嵌入提效 | 批量 Upsert |
| 3. 检索与生成 | 用户提问 → 检索相关上下文 → 拼提示词发大模型 | **混合搜索 + 元数据预过滤** |
| 4. 持续优化 | 收集反馈，"优质问答"增量入库形成正反馈循环 | 实时 **Upsert** 更新 |
> 这个案例的启示可以泛化：**当你的 RAG 检索对象既有“语义型”文本又有“精确型”约束条件时，带强元数据过滤能力的中量级向量库（Qdrant 档）是甜点区**。
|
|
### 一句话区别
- **混合检索（Hybrid Search）**：解决"**怎么算相关**"——多路召回（关键词 BM25 + 向量语义）各自打分，再融合排序（RRF）。作用在**打分/排序层**。
- **复杂过滤（Complex/Metadata Filtering）**：解决"**在谁里面搜**"——先按结构化元数据条件圈定候选集（WHERE 类谓词），再在候选集内做相似度检索。作用在**候选集裁剪层**。
> 口诀：**混合检索管"像不像"，复杂过滤管"属不属于"**。
### 逐项对比
| 维度 | 混合检索 | 复杂过滤 |
|---|---|---|
| 回答的问题 | 哪些文档**像**我的问题 | 哪些文档**符合**我的条件 |
| 作用层 | 相似度打分 / 排序融合 | 检索前的候选集裁剪 |
| 数据对象 | 文本内容、向量（非结构化） | 元数据 payload（结构化：枚举/数值/时间） |
| 组合方式 | 多路召回 + 融合（RRF / 加权分数） | WHERE 谓词（=、范围、in、嵌套 AND/OR） |
| 典型句式 | "像不像" | "属于不属于、在某范围吗" |
| 失败模式 | 词汇鸿沟 / 精确术语失准 | 过滤过严 → 候选集为空、零召回 |
| 对应选型 | Weaviate / ES（混合检索产品化成熟） | Qdrant（可过滤 HNSW + Payload 最强） |
### 两个场景化例子
**例子 A（混合检索的用武之地）**：用户问"能插嘴吗"——关键词路零命中（词汇鸿沟），向量路召回"连续对话模式"。这是**内容像不像**的问题，多路召回 + RRF 融合解决。
**例子 B（复杂过滤的用武之地）**：NL2SQL 场景中问"上个月华东销售额"——必须**先限定** `业务线=销售 AND 时间字段=2024-08`，再在范围内做向量检索找"金额"字段。这是**属不属于范围**的问题，语义再准也不能越界搜到别的业务线。
### 两者的组合关系（生产查询的真实形态）
```text
用户问题
│
├─▶ ① 复杂过滤：WHERE 业务线='销售' AND tenant_id=123 ← 先圈范围（在哪搜）
│ │
│ ▼
│ 候选集（几百~几千条）
│ │
└─▶ ② 混合检索：候选集内 关键词路 + 向量路 → RRF 融合 ← 再算相关（怎么算像）
│
▼
③ 重排 Rerank → Top-3~5 送 LLM ← 最后求精（怎么排准）
```
> 完整生产查询 = **过滤（在哪搜）+ 混合（怎么算相关）+ 重排（怎么排精）**，三层漏斗各司其职。
### 代码示例（Qdrant：过滤 + 向量一体）
```python
from qdrant_client.models import Filter, FieldCondition, MatchValue, Range
res = qdrant.query_points(
collection_name="kb",
query=qvec, # ② 向量检索（语义相关）
query_filter=Filter(must=[ # ① 复杂过滤（先圈范围）
FieldCondition(key="category", match=MatchValue(value="售后")),
FieldCondition(key="created_at", range=Range(gte=1725148800)),
]),
limit=3,
)
```
Qdrant 的**可过滤 HNSW** 索引在图遍历时直接剪枝不满足条件的点——过滤与向量检索**同时完成**，仍保持低延迟；这也是选型表里"复杂过滤 → Qdrant"的原因。
ES 侧对应写法（`bool.filter` 标量条件 + `knn` 向量，或原生 RRF 混合搜索）——同理：过滤用 filter 子句（不参与打分），检索用 match/knn。
### 为什么选型表那样推荐
- **混合检索 → Weaviate / ES**：ES 倒排 + dense_vector 双索引一体、原生 RRF；Weaviate 内置 hybrid 参数（alpha 加权）——混合检索开箱即用
- **复杂过滤 → Qdrant**：Payload 元数据模型 + 可过滤 HNSW，标量谓词丰富（match/range/geo/nested），过滤与向量同链路低延迟
- **两者都要（如 NL2SQL）**：Qdrant 混合搜索（向量 + 稀疏向量/关键词）+ 元数据预过滤 一库搞定——这就是 NL2SQL 案例选它的原因
### 一句话总结
**过滤是"范围正确性"，混合检索是"内容相关性"**——前者保证不越界，后者保证找得准；缺过滤会串数据（权限/租户事故），缺混合会漏召回（词汇鸿沟）。
|
|
### 多角度深入分析：选型的八个视角
| 视角 | 关键问题 | 倾向选择 |
|---|---|---|
| **数据规模** | 百万级？千万级？亿级？ | 百万以下 Chroma/pgvector；百万~千万 Qdrant/Weaviate；千万~亿+ Milvus |
| **团队运维** | 有人运维数据库吗？ | 无→Pinecone/Zilliz 托管；有→自托管开源方案 |
| **现有技术栈** | 已有 PG/ES/Mongo？ | 优先复用：pgvector / ES dense_vector / Atlas Vector Search |
| **数据敏感度** | 能出内网吗？合规要求？ | 敏感→自托管开源（Milvus/Qdrant）；不敏感→SaaS 省心 |
| **查询复杂度** | 要混合检索/复杂过滤/图谱？ | 混合→Weaviate/ES；复杂过滤→Qdrant；图谱→GraphRAG 路线 |
| **成本模型** | 订阅费 vs 自运维人力成本 | 数据量小 SaaS 划算；数据量大长期看自托管更省 |
| **锁定风险** | 换库/迁库成本？ | SaaS（Pinecone）锁定最深；开源+标准 API（Qdrant）迁移成本最低 |
| **性能指标** | 延迟/QPS 硬要求？ | 极致性能→FAISS 自研嵌入；工程平衡→Qdrant；一般够用→Chroma/pgvector |
### 什么场景选什么库（决策速查）
```text
个人学习 / 课程作业 / 快速原型 ──────▶ Chroma（pip 即用）
算法研究 / 极致性能 / 自建系统底座 ──▶ FAISS（引擎级）
初创团队快速上线 / 不想运维 ──▶ Pinecone（SaaS）/ Zilliz Cloud
已有 PostgreSQL / 百万级以内 ──▶ pgvector（零新增组件）
百万~千万级 + 复杂过滤 + 混合检索 ──▶ Qdrant（甜点区）
千万~亿级 + 高并发 + 企业核心生产 ──▶ Milvus（事实标准）
混合检索为主（关键词+向量） ──▶ Elasticsearch / Weaviate
信创 / 国产替代 ──▶ 电科金仓 KingbaseES 等融合方案
NL2SQL / 多源结构化检索 ──▶ Qdrant（元数据过滤 + 混合搜索）
```
### 一页总结
1. **没有最好的向量库，只有最合适的**：选型本质是在数据规模、运维成本、功能特性间取舍
2. **规模定档**：百万以下轻量档（Chroma/pgvector），百万~千万中档（Qdrant/Weaviate），千万以上重型档（Milvus）
3. **生态优先**：已有 PG/ES/Mongo 就先复用，别轻易引入异构系统
4. **复杂度认知**：pgvector 是一条 SQL、Qdrant 是一个容器、Milvus 是一组微服务——运维成本与规模能力成正比
5. **可演进原则**：先用轻量方案跑通业务验证价值，数据量/并发逼近上限再升级重型方案；接口层做抽象（检索函数可注入），换库只换实现——参考 RAG_Bot 的检索函数注入设计
|
|
### 版本与下载（2026-09-05 实机）
| 项 | 值 |
|---|---|
| 版本 | **Qdrant v1.19.1**（最新稳定版） |
| 包 | `qdrant-x86_64-unknown-linux-musl.tar.gz`（musl 静态链接，无 glibc 依赖问题） |
| 大小 | 32MB（解压后单二进制） |
| 安装目录 | `/opt/app/qdrant`（bin/storage/snapshots/config/backups） |
| 端口 | REST **6333** / gRPC **6334** |
### 远程中转下载（国外资源提速 20 倍）
GitHub Releases 国内直连慢，走远程服务器中转（方法已沉淀到 nlp skill 顶层方法论 `resource-download.md`，全 agent 通用）：
```bash
REMOTE="xt@144.34.185.72"; RPORT=26225
# ① 远程查询最新版本号（GitHub API 在远程可达）
ssh -p $RPORT $REMOTE "curl -s https://api.github.com/repos/qdrant/qdrant/releases/latest \
| python3 -c \"import json,sys; print(json.load(sys.stdin)['tag_name'])\""
# ② 远程下载 + md5
ssh -p $RPORT $REMOTE "cd /tmp && curl -sL -o qdrant.tar.gz \
https://github.com/qdrant/qdrant/releases/download/v1.19.1/qdrant-x86_64-unknown-linux-musl.tar.gz \
&& md5sum qdrant.tar.gz"
# ③ 回传本地 + 双端 md5 校验 + 清理远程
scp -P $RPORT $REMOTE:/tmp/qdrant.tar.gz /tmp/qdrant.tar.gz
md5sum /tmp/qdrant.tar.gz # 须与远程一致：2747d070...
ssh -p $RPORT $REMOTE "rm -f /tmp/qdrant.tar.gz"
```
### 安装（解压即用，无需编译）
```bash
mkdir -p /opt/app/qdrant/bin /opt/app/qdrant/storage \
/opt/app/qdrant/snapshots /opt/app/qdrant/config /opt/app/qdrant/backups
tar -xzf /tmp/qdrant.tar.gz -C /opt/app/qdrant/bin
chmod +x /opt/app/qdrant/bin/qdrant
/opt/app/qdrant/bin/qdrant --version # qdrant 1.19.1
```
**目录职责**：
| 目录 | 职责 |
|---|---|
| `bin/` | 二进制 |
| `storage/` | 数据目录（collection 数据、WAL、索引） |
| `snapshots/` | 快照输出目录 |
| `config/` | config.yaml |
| `backups/` | 备份归档（快照 + storage + config 的 tar.gz） |
> 💡 分发包已存档到 nlp_rag 智能体 `resources/qdrant/v1.19.1/`，后续直接复用免重复下载。
|
|
### 配置：config.yaml（显式数据目录）
不指定配置时 Qdrant 用内置默认值（数据落在启动目录的 `./storage`）——**生产必须显式指定**：
```yaml
# /opt/app/qdrant/config/config.yaml
storage:
storage_path: /opt/app/qdrant/storage # 数据目录
snapshots_path: /opt/app/qdrant/snapshots # 快照目录
service:
host: 127.0.0.1
http_port: 6333
grpc_port: 6334
```
### 启动（与 ES 同款的三 fd 重定向坑）
```bash
nohup /opt/app/qdrant/bin/qdrant --config-path /opt/app/qdrant/config/config.yaml \
< /dev/null > /opt/app/qdrant/logs.out 2>&1 &
echo $! > /opt/app/qdrant/qdrant.pid
```
> ⚠️ 与 ES `-d` 守护进程同款坑：不重定向三个 fd 时，外层 `... | tail` 等管道命令会因子进程持有 stdout 而永久挂起。
### 健康检查与客户端连接
```bash
curl -s http://127.0.0.1:6333 # 返回 {"title":"qdrant - vector search engine","version":"1.19.1"}
curl -s http://127.0.0.1:6333/collections # 集合列表
```
Python 客户端（`pip install qdrant-client`）：
```python
from qdrant_client import QdrantClient
qdrant = QdrantClient(url="http://127.0.0.1:6333")
print([c.name for c in qdrant.get_collections().collections])
```
> 版本匹配：qdrant-client **1.19.0** ↔ server **1.19.1** 实测兼容；检索 API 用 `query_points`（旧 `search` 已弃用）。
|
|
### 数据目录结构
```text
/opt/app/qdrant/
├── storage/ # 数据主目录（collection 数据、WAL、HNSW 索引）
│ └── collections/<name>/
├── snapshots/ # 快照输出（集合级 + 全量）
│ ├── <collection>/xxx.snapshot
│ └── full-snapshot-xxx.snapshot
├── config/config.yaml
└── backups/ # 备份归档（tar.gz）
```
### 备份的正确姿势：先快照、再归档
直接 tar `storage/` 有风险——备份过程中服务可能正在写入。**快照是 Qdrant 提供的一致性保证机制**：
```bash
# ① 集合级快照（单集合）
curl -s -X POST http://127.0.0.1:6333/collections/geeker_kb_vec/snapshots
# ② 全量快照（所有集合 + 配置）
curl -s -X POST http://127.0.0.1:6333/snapshots
# ③ 归档：快照目录 + 存储目录 + 配置
tar -czf /opt/app/qdrant/backups/qdrant_backup_all_$(date +%Y%m%d_%H%M%S).tar.gz \
-C /opt/app/qdrant snapshots storage config
```
### 恢复：上传快照重建集合
```bash
curl -s -X POST http://127.0.0.1:6333/collections/geeker_kb_vec/snapshots/upload \
-H 'Content-Type: multipart/form-data' \
-F "snapshot=@/opt/app/qdrant/snapshots/geeker_kb_vec/xxx.snapshot"
```
### 实测链路（qdrant_run.sh 验证记录）
```text
建集合 backup_test + 写入 1 个点
→ snapshot（产出 .snapshot 文件）
→ backup（归档 384K tar.gz）
→ 删除集合
→ restore（上传快照）
→ points_count: 1 ✅ 数据完整恢复
```
### 备份策略建议
| 层级 | 手段 | 频率 |
|---|---|---|
| 日常增量 | 集合快照（快照是原子的，开销小） | 每日/每次入库后 |
| 周期归档 | 快照 + storage + config 打包到 backups/ | 每周 |
| 异地容灾 | 备份归档 rsync 到远程 | 每周/每月 |
|
|
### 一键运维脚本（nlp_rag 智能体沉淀）
`.config/agent/nlp_rag/scripts/qdrant_run.sh`：
```bash
bash qdrant_run.sh start|stop|status|info # 生命周期
bash qdrant_run.sh snapshot [collection] # 快照（不传名=全量）
bash qdrant_run.sh backup [collection] # 备份（快照+归档，带时间戳）
bash qdrant_run.sh restore <快照文件> <集合名> # 恢复
```
实测输出：
```text
[qdrant] 后台启动 Qdrant（config: /opt/app/qdrant/config/config.yaml）...
[qdrant] 健康检查通过:
version: 1.19.1
pid: 271904
collections: []
```
### 与 ES 部署对比（两个本地服务的经验共性）
| 维度 | ES 7.17.23 | Qdrant 1.19.1 |
|---|---|---|
| 包形态 | tarball（需解压目录树） | **单二进制**（musl 静态链接） |
| 依赖 | 自带 JDK 17 | **零依赖** |
| 配置 | elasticsearch.yml | config.yaml |
| 守护坑 | `-d` 持 stdout 挂管道 | nohup 同款坑，都要三 fd 重定向 |
| 备份 | 快照（需注册 repo） | 快照 API（**零配置**，直接打） |
| 教学结论 | 检索服务化标杆 | 向量库服务化标杆，**轻得多** |
### 排错速查
| 症状 | 根因 | 处置 |
|---|---|---|
| `Address already in use` | 旧实例未停 | `kill $(cat qdrant.pid)` 或 stop 子命令 |
| 启动后 curl 拒绝连接 | 配置路径错误/端口占用 | `tail -50 /opt/app/qdrant/logs.out` |
| `search` 方法报错 | 客户端新版弃用 | 改用 `query_points` |
| 备份文件损坏 | 备份时正在写入 | 务必**先快照再归档** |
| PointStruct id 报错 | id 必须是 int 或 uuid | 别用字符串 id，把业务 id 放 payload |
|
|
### Qdrant 开源吗？免费吗？
**完全开源，免费可商用**——Apache License 2.0（宽松协议，允许商用、修改、再分发，只需保留版权声明）。
- 源码仓库：https://github.com/qdrant/qdrant
- 二进制下载：https://github.com/qdrant/qdrant/releases （assets 里选对应平台的 tar.gz）
- 文档：https://qdrant.tech/documentation/
> 商用云托管（Qdrant Cloud）是付费增值服务；**自托管（self-hosted）永远免费**。
### 源码编译全流程（2026-09-06 实机通过，WSL2）
**① 获取源码**（远程中转，删 .git）：
```bash
# 远程服务器操作
git clone --depth 1 --branch v1.19.1 https://github.com/qdrant/qdrant.git qdrant-src
rm -rf qdrant-src/.git # 按约定不传 .git
tar -czf qdrant-src.tar.gz qdrant-src
# 传回本地 /home/xt/wks/rust/ 并解压
scp -P 26225 xt@144.34.185.72:/tmp/qdrant-src.tar.gz /tmp/
tar -xzf /tmp/qdrant-src.tar.gz -C /home/xt/wks/rust/
```
**② 编译依赖**：cargo（rustup 安装）、protoc、cmake、**clang + libclang-dev**（rocksdb 的 bindgen 需要）：
```bash
sudo apt-get install -y clang libclang-dev pkg-config build-essential protobuf-compiler cmake
```
**③ 实踩坑 1：git 依赖 GitHub 直连超时**
Qdrant 有两个 GitHub git 依赖：`tikv/raft-rs`（rev 锁定）与 `qdrant/tonic`（v0.14.6-qdrant fork）。`[patch."git-url"]` 对 git 依赖仍会触发 git 索引更新，**不解决问题**——直接把 Cargo.toml 里的 git 依赖改写为 **path 依赖**（源码同样经远程中转下载，保留 features）：
```toml
# Cargo.toml 改写
raft-proto = { path = "/home/xt/wks/rust/vendor-git/raft-rs/proto", features = ["prost-codec"], default-features = false }
raft = { path = "/home/xt/wks/rust/vendor-git/raft-rs", features = ["prost-codec"], default-features = false }
# [patch.crates-io] 段内（tonic 原本就是 patch 进来的 fork）
tonic = { path = "/home/xt/wks/rust/vendor-git/tonic/tonic" }
```
**④ 实踩坑 2：aws-lc-sys 拒绝系统 GCC**
build script 检测 GCC 的 memcmp bug（gcc.gnu.org/bugzilla #95189）直接 panic——改用 clang：
```bash
CC=clang CXX=clang++ cargo build --release --bin qdrant
```
**⑤ 实踩坑 3：crates.io 直连 78KB/s**——换 USTC sparse 镜像（`~/.cargo/config.toml`）：
```toml
[source.crates-io]
replace-with = 'ustc'
[source.ustc]
registry = "sparse+https://mirrors.ustc.edu.cn/crates.io-index/"
```
**⑥ 编译与产物**：
```bash
cd /home/xt/wks/rust/qdrant
CC=clang CXX=clang++ cargo build --release --bin qdrant
# 实测：912 个依赖、475+ 编译单元、13 分 22 秒、产物 89.7MB
target/release/qdrant --version # qdrant 1.19.1
```
### 编译版接管服务 + 实测
`qdrant_run.sh` 支持 `QD_BIN` 环境变量覆盖二进制路径：
```bash
QD_BIN=/home/xt/wks/rust/qdrant/target/release/qdrant bash qdrant_run.sh start
# 健康检查通过：version 1.19.1
```
配套教学 notebook：`/mnt/d/wks/jupyter/ai58/0905_qdrant_usage.ipynb`（21 单元格，在**本地编译版**上实测 0 错误）——连接/集合管理/写入/向量检索/元数据过滤/更新删除/滚动浏览/快照恢复全链路。
|
|
```
mkdir -p /opt/app/qdrant/bin /opt/app/qdrant/storage \
/opt/app/qdrant/snapshots /opt/app/qdrant/config /opt/app/qdrant/backups
scp -P 26225 qdrant-v1.19.1-x86_64-musl-compiled xt@144.34.185.72:/tmp/
rsync -rltDv /tmp/qdrant-v1.19.1-x86_64-musl-compiled /opt/app/qdrant/bin/qdrant
tar -xzf /tmp/qdrant.tar.gz -C /opt/app/qdrant/bin
chmod +x /opt/app/qdrant/bin/qdrant
/opt/app/qdrant/bin/qdrant --version
vim /opt/app/qdrant/config/config.yaml
storage:
storage_path: /opt/app/qdrant/storage # 数据目录
snapshots_path: /opt/app/qdrant/snapshots # 快照目录
service:
host: 127.0.0.1
http_port: 6333
grpc_port: 6334
nohup /opt/app/qdrant/bin/qdrant --config-path /opt/app/qdrant/config/config.yaml \
< /dev/null > /opt/app/qdrant/logs.out 2>&1 &
echo $! > /opt/app/qdrant/qdrant.pid
curl -s http://127.0.0.1:6333 # 返回 {"title":"qdrant - vector search engine","version":"1.19.1"}
curl -s http://127.0.0.1:6333/collections # 集合列表
```
|
|
### 远程安装步骤（服务器 144.34.185.72，SSH 端口 26225 / 用户 xt）
```bash
# 1. 建目录（数据/快照/配置/备份分离）
mkdir -p /opt/app/qdrant/{bin,storage,snapshots,config,backups}
# 2. 上传 musl 静态编译版二进制（Linux 通用，免依赖）
scp -P 26225 qdrant-v1.19.1-x86_64-musl-compiled xt@144.34.185.72:/tmp/
rsync -rltDv /tmp/qdrant-v1.19.1-x86_64-musl-compiled /opt/app/qdrant/bin/qdrant
chmod +x /opt/app/qdrant/bin/qdrant
/opt/app/qdrant/bin/qdrant --version
```
```yaml
# 3. /opt/app/qdrant/config/config.yaml
storage:
storage_path: /opt/app/qdrant/storage # 数据目录
snapshots_path: /opt/app/qdrant/snapshots # 快照目录
service:
host: 127.0.0.1 # loopback 绑定（安全默认：qdrant 默认无鉴权，裸绑 0.0.0.0 = 裸奔）
http_port: 6333 # REST
grpc_port: 6334 # gRPC
```
```bash
# 4. 启动（守护进程必须三 fd 重定向，否则外层管道挂死）
nohup /opt/app/qdrant/bin/qdrant --config-path /opt/app/qdrant/config/config.yaml < /dev/null > /opt/app/qdrant/logs.out 2>&1 &
```
### 本地访问：SSH 本地端口转发（推荐）
`host: 127.0.0.1` 只允许服务器本机访问 → 本地不可直连。建 SSH 隧道把本地端口映射到远程 loopback，**远程零改动**：
```bash
ssh -L 6333:127.0.0.1:6333 -L 6334:127.0.0.1:6334 -p 26225 -N -o ServerAliveInterval=30 -o ExitOnForwardFailure=yes xt@144.34.185.72
```
- `-N` 纯转发不执行远程命令；`ServerAliveInterval` 保活；`ExitOnForwardFailure` 本地端口被占时立即报错
- 之后本地代码直接连 `http://localhost:6333`（REST）/ `localhost:6334`（gRPC，两个端口都要转发）
- 隧道进程退出即断连，远程数据无损，重跑命令即恢复；需常驻用 autossh / Windows 任务计划 / NSSM
### 实测验证（2026-09-08）
```bash
curl http://localhost:6333/collections # => 200 {"result":{"collections":[]},"status":"ok"}
curl http://localhost:6333/healthz # => 200
# Python 客户端（uv run --with qdrant-client）：
# QdrantClient(url="http://localhost:6333").get_collections() => OK
```
备选（不推荐）：config.yaml 改 `host: 0.0.0.0` 可直连，但必须同时设 `api_key` + 防火墙/安全组白名单。
|
|
## 前端执行
ssh -L 6333:127.0.0.1:6333 -L 6334:127.0.0.1:6334 -p 26225 -N -o ServerAliveInterval=30 -o ExitOnForwardFailure=yes xt@144.34.185.72
## 后端执行
要将这个 SSH 端口转发命令改为后台执行，有几种常用方法：
## 方法一：使用 `-f` 参数（推荐）
SSH 自带后台运行参数 `-f`，配合 `-N` 使用最简洁：
```bash
ssh -f -N \
-L 6333:127.0.0.1:6333 \
-L 6334:127.0.0.1:6334 \
-p 26225 \
-o ServerAliveInterval=30 \
-o ExitOnForwardFailure=yes \
xt@144.34.185.72
```
- `-f`：SSH 在认证成功后转入后台运行
- `-N`：不执行远程命令，只做端口转发
> 注意：`-f` 与需要交互输入密码的场景不兼容。如果使用密钥登录（推荐），则完全没问题。如果必须输入密码，请先用方法二或三。
## 方法二：使用 `nohup` + `&`
```bash
nohup ssh -N \
-L 6333:127.0.0.1:6333 \
-L 6334:127.0.0.1:6334 \
-p 26225 \
-o ServerAliveInterval=30 \
-o ExitOnForwardFailure=yes \
xt@144.34.185.72 > /var/log/ssh-tunnel.log 2>&1 &
```
## 方法三：使用 `screen` 或 `tmux`（便于管理）
```bash
# 使用 screen
screen -dmS ssl-tunnel ssh -N \
-L 6333:127.0.0.1:6333 \
-L 6334:127.0.0.1:6334 \
-p 26225 \
-o ServerAliveInterval=30 \
-o ExitOnForwardFailure=yes \
xt@144.34.185.72
```
之后可用 `screen -r ssl-tunnel` 重新进入查看。
## 补充建议
1. **自动重连**：为防止断线，建议加上 `autossh`：
```bash
autossh -M 0 -f -N \
-L 6333:127.0.0.1:6333 \
-L 6334:127.0.0.1:6334 \
-p 26225 \
-o ServerAliveInterval=30 \
-o ServerAliveCountMax=3 \
-o ExitOnForwardFailure=yes \
xt@144.34.185.72
```
2. **查看进程**：
```bash
ps aux | grep "6333:127.0.0.1:6333"
```
3. **关闭隧道**：
```bash
pkill -f "6333:127.0.0.1:6333"
```
如果只是想快速后台跑起来，**方法一（加 `-f`）** 是最简单直接的方案，前提是你已配置好 SSH 免密登录。
|
|
|
|
|
|
|
|
|
|
|
|
### 回顾前两版的盲区
| 检索路 | 强项 | 盲区（实测） |
|---|---|---|
| **BM25/ES 关键词路** | 精确术语、型号编号强命中 | **词汇鸿沟**："插嘴" vs "打断" 词面不同 → 零命中 |
| **向量路（embedding）** | 语义相近即召回、跨语言 | 精确术语可能偏差；**短查询漂移**（5 字查询向量 Top1 会漂） |
**混合检索（Hybrid Search）**：两路各自召回 → 融合排序 → 用一条路的强补另一条路的弱。
### 0905_rag_hybrid.ipynb 总览
产出：`/mnt/d/wks/jupyter/ai58/0905_rag_hybrid.ipynb`（21 单元格，py312 实测 0 错误），由 nlp_rag 生成器一键产出：
```bash
python .config/agent/nlp_rag/scripts/nlp_rag_nb.py \
--out 0905_rag_hybrid.ipynb --execute --retriever hybrid
```
### 流程全景
```text
━━━━ 阶段一：建立索引（离线） ━━━━
知识库 ──┬─ jieba 分词 → BM25Reranker 索引（进程内）
└─ embedding → Qdrant collection（本地向量库服务）
━━━━ 阶段二：检索生成（在线） ━━━━
用户提问 ──┬─ bm25_search（关键词，Top-100） ┐
└─ vec_search（Qdrant 向量，Top-100）┴─▶ RRF 融合 ─▶ Top-3 ─▶ 提示词 ─▶ LLM
```
**架构认知**：与 ES 版"embedding 塞 dense_vector"的分工不同——**Qdrant 专司向量检索，原文放 payload 随结果带回**；BM25 在进程内零依赖。两个检索器互补、互不可见，融合层才碰头。
|
|
### 第一路：BM25 关键词路（进程内）
```python
import sys
sys.path.insert(0, "/ai/wks/aitpf/src")
from tpf.nlp.text import BM25Reranker
import jieba
# ⚠️ 粒度修复：lcut 精确模式"保修期"整词错过文档"保修"（与 IK 同款坑）
jieba.lcut = jieba.lcut_for_search
bm25 = BM25Reranker(df=kb_df, use_cols=["text", "category"], k1=1.8, b=0.75)
def bm25_search(query, top_k=100):
"""BM25 关键词路：全量打分 → O(n) 取前 top_k"""
scores = bm25.sim_scores(query)
k = min(top_k, len(scores))
idx = np.argpartition(scores, -k)[-k:]
top = idx[np.argsort(scores[idx])[::-1]]
return [(kb_df.iloc[i], float(scores[i])) for i in top]
```
### 第二路：Qdrant 向量路（本地向量库服务）
```python
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
qdrant = QdrantClient(url="http://127.0.0.1:6333")
COLL = "geeker_kb_vec"
# 重建演示集合（余弦距离，1536 维）
if qdrant.collection_exists(COLL):
qdrant.delete_collection(COLL)
qdrant.create_collection(
collection_name=COLL,
vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)
# 向量化并写入：payload 携带原文与元数据（检索结果随点带回，免二次查库）
texts = [d["text"] for d in KENB_DOCS]
vectors = get_embeddings(texts)
points = [
PointStruct(
id=i, vector=vec, # id 必须 int/uuid
payload={"doc_id": d["doc_id"], "category": d["category"], "text": d["text"]},
)
for i, (d, vec) in enumerate(zip(KENB_DOCS, vectors))
]
qdrant.upsert(collection_name=COLL, points=points)
def vec_search(query, top_k=100):
"""Qdrant 向量路：query 向量化 → 余弦 Top-K（分数 0~1 越大越相似）"""
qvec = get_embeddings([query])[0]
res = qdrant.query_points(collection_name=COLL, query=qvec, limit=top_k)
return [(pt.payload, float(pt.score)) for pt in res.points]
```
> 注意：`PointStruct.id` 必须是 int 或 uuid（业务 id 放 payload）；`query_points` 是 qdrant-client 1.10+ 的检索 API（旧 `search` 已弃用）。
|
|
### RRF：倒数排名融合（Reciprocal Rank Fusion）
两路分数**量纲不同**——BM25 无上界（如 5.38）、cosine 在 0~1（如 0.57）——**不能直接相加**。RRF 不看分数只看**排名**：
```text
score(d) = Σ 各路 1/(k + rank(d)) # k=60 为平滑常数
例：文档在 BM25 路排第 2、向量路排第 1
→ RRF 分 = 1/(60+2) + 1/(60+1) ≈ 0.0325
```
```python
def rrf_fusion(list_a, list_b, k=60):
"""两路召回 RRF 融合：按排名倒数累加，跨量纲可比"""
scores, rows = {}, {}
for lst in (list_a, list_b):
for rank, (src, _score) in enumerate(lst, 1):
did = src["doc_id"]
scores[did] = scores.get(did, 0) + 1 / (k + rank)
rows[did] = src
merged = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [(rows[did], s) for did, s in merged]
def hybrid_search(query, top_k=100):
"""混合检索：双路各自 Top-K → RRF 融合
⚠️ BM25 零分=词面零命中，其排名是噪声（argpartition 对并列值顺序不定），
参与融合会污染 RRF 排名——融合前先过滤零分结果"""
a = bm25_search(query, top_k=top_k)
a = [(src, s) for src, s in a if s > 0]
b = vec_search(query, top_k=top_k)
return rrf_fusion(a, b)
```
### ⚠️ 实测踩坑：BM25 零分污染 RRF
**现象**：查询"插嘴"BM25 路全 0 分，融合后混合路 Top1 竟是弱相关文档（priv_01）而非向量路正确命中的 func_01。
**根因**：BM25 全 0 分时 `argpartition` 对并列值的顺序**不确定**——零分排名是噪声，却在 RRF 中白得 rank 1 的 1/61 分贡献。
**处置**：融合前过滤零分（零分=零信息）。修复后"插嘴"混合路由向量路主导，func_01 正确居首（RRF 0.0164）。
|
|
### 直答类：三路全部命中
```text
❓ X1 的售价是多少（BM25 路）
[3.4183][价格] X1 官方售价 399 元... ← 词面强命中
❓ X1 的售价是多少（向量路）
[0.7074][价格] X1 官方售价 399 元... ← 语义命中
❓ X1 的售价是多少（混合 RRF）
[0.0328][价格] X1 官方售价 399 元... ← 双路同投一票，稳居第一
```
### 词汇鸿沟实验：三路分野（混合的价值）
```text
── BM25 路 ──
唤醒后能随时打断吗 → Top1 分数 5.38（func_01） ✅ "打断"词面命中
说话说到一半能插嘴吗 → Top1 分数 0.00 ❌ "插嘴"词面零命中
── 向量路 ──
唤醒后能随时打断吗 → Top1 cos 0.5716（func_01） ✅
说话说到一半能插嘴吗 → Top1 cos 0.3488（func_01） ✅ 语义救回
── 混合（RRF 融合）──
唤醒后能随时打断吗 → Top1 RRF 0.0328（func_01） ✅ 双路同投
说话说到一半能插嘴吗 → Top1 RRF 0.0164（func_01） ✅ 向量路主导（BM25 零分已过滤）
```
**解读**：
- "打断"问题 BM25 与向量**双路都命中** func_01 → 融合后双份贡献，最稳
- "插嘴"问题 BM25 零贡献、向量独扛 → 融合后仍是 func_01，**一路倒下不影响整体召回**
- RRF 分数 0.0328（双路）vs 0.0164（单路）——**分数高低直接反映几路证据背书**
|
|
### RAG_Bot 实测验收矩阵（全部达标 ✅）
```text
🧪 [直答·价格] X1 的售价是多少？
── 混合版: X1 的官方售价是 399 元，首发期下单立减 50 元... [来源:price_01] ✅
🧪 [直答·售后] 保修期是多久？
── 混合版: X1 整机保修期为一年，人为损坏除外... [来源:sale_01] ✅
🧪 [同义改写·词汇鸿沟] 说话说到一半能插嘴吗？
── 混合版: 可以，X1 支持连续对话模式，允许多轮追问和随时打断... [来源:func_01] ✅
（BM25 版此处只能拒答；ES 向量版能答——混合版达到向量版能力）
🧪 [知识库外·拒答兜底] X1 的电池续航怎么样？
── 混合版: 根据提供的信息无法回答此问题。 ✅
```
### 三种检索方案终局对比
| 对比维度 | 纯 BM25 | 纯向量 | **混合（BM25+Qdrant）** |
|---|---|---|---|
| 精确术语（型号/编号） | ✅ | 可能偏差 | ✅ 双保险 |
| 同义改写（词汇鸿沟） | ❌ | ✅ | ✅ 向量路补位 |
| 部署成本 | 零依赖 | 需向量库服务 | 进程内 BM25 + 轻量 Qdrant |
| 检索质量 | 词面 | 语义 | **词面+语义互补** |
| 适用 | 术语密集小库/教学 | 语义问答 | **生产 RAG 默认推荐** |
### 总结三句话
1. **混合检索用一条路的强补另一条路的弱**，是生产 RAG 的默认形态——关键词路守精确术语，向量路救词汇鸿沟
2. **融合层要小心量纲与噪声**：RRF 按排名不按分数（天然跨量纲）；零分/无效结果融合前过滤，防噪声污染排名
3. **架构可演进**：RAG_Bot 检索函数注入的设计让 bm25 / es / hybrid 三引擎即插即换——换引擎只换检索函数，提示词与生成层零改动
|