在深度学习兴起之前，几乎所有搜索引擎和知识库系统都依赖关键词匹配（BM25、TF-IDF）来查找内容。其核心思想非常直接：把用户查询拆成几个关键词，然后在文档库中找同时包含这些词的文档，按出现次数排序。
举例来说，当你在企业知识库搜索"怎么处理客户投诉"时：
局限一：同义词问题——意思一样却找不到
用户想搜"如何退款"，但文档写的是"退货流程""钱款返还"，传统检索无法理解这些词背后的语义关联。这是最常见的痛点，尤其在中文语境中，一个意思可以用数十种表达方式。
局限二：长尾查询——关键词太少搜不到
当用户输入一个较长的自然语言问题时，比如"员工出差报销需要提供哪些票据"，这句话里并没有明确的黑体关键词，BM25打分机制很难匹配到真正相关的文档，即使文档内容完全回答了这个问题。
局限三：语义理解缺失——字面匹配≠语义相关
"苹果多少钱一斤"和"苹果股价今天涨了多少"，两个查询共享"苹果"这个词，但含义完全不同。传统检索无法区分实体类型，更无法理解上下文含义。
局限四：排序质量有限——相关性信号单一
BM25只考虑词频和文档长度，没有办法综合考虑查询与文档之间的语义相似度、实体关系、意图类别等深层特征。
向量化检索（Vector Search / Semantic Search）的核心思路是：将文本映射到一个高维稠密向量空间中，让语义相似的文本在向量空间中距离相近。检索时，用户的查询会被转换为向量，然后在向量空间中找最近的邻居（Nearest Neighbor），这就是语义检索的本质。
假设"怎么处理客户投诉"和"客户投诉处理方案"这两个文本的向量在空间中位置接近，那么无论用户用哪种表述问，系统都能准确返回相关文档——这就是语义搜索相比关键词搜索的核心优势。
Embedding，本质上是将离散的非结构化数据（文字、图片、声音）转换为连续稠密向量表示的过程。以文本为例：一段文字经过Embedding模型处理后，变成一个固定维度（如1536维）的实数向量。这个向量包含了文本的语义信息，相似的文本在向量空间中距离更近。
Embedding的本质是特征提取和降维压缩——把一段文字的语义"压缩"进一个N维向量里，同时保留最重要的语义特征。
现代Embedding模型大多基于Transformer架构（如BERT、GPT系列）。其核心原理是：利用自注意力机制（Self-Attention），让文本中每个字词在上下文中动态调整与其他字词的关联强度，从而得到一个上下文相关的向量表示。
以OpenAI的text-embedding-3-small模型为例，它输出的向量维度为1536维，每个维度是一个浮点数。这1536个数字共同刻画了这段文本的语义特征。维度越高，模型能表达的信息越丰富，但存储和计算成本也越高。
向量检索的核心距离度量方式之一是余弦相似度。它的取值范围是[-1, 1]，值越接近1表示两个向量越相似。
给定两个向量 A = [a₁, a₂, ..., aₙ] 和 B = [b₁, b₂, ..., bₙ]，余弦相似度的计算公式为：
余弦相似度 = (向量A · 向量B) / (||向量A|| × ||向量B||)
= Σ(aᵢ × bᵢ) / (√Σaᵢ² × √Σbᵢ²)
其中：
· 表示向量点积（对应元素相乘后求和）
||A|| 表示向量A的L2范数（模长）
Σ 表示求和符号公式解读：分子是两个向量的点积，反映了它们的"方向一致性"；分母是两个向量模长的乘积，起到归一化作用，使得不同长度的向量也能公平比较。
# ============================================================
# 纯Python版余弦相似度计算
# 不依赖任何外部库，仅使用Python内置函数实现
# 适用于没有NumPy环境或学习理解原理的场景
# ============================================================
def cosine_similarity_pure_python(vec_a, vec_b):
"""
纯Python实现余弦相似度算法。
参数:
vec_a: list[float] - 第一个向量，如 [0.1, 0.5, 0.3, ...]
vec_b: list[float] - 第二个向量，与vec_a维度相同
返回:
float - 余弦相似度值，范围 [-1.0, 1.0]
值接近1.0表示两向量高度相似
值接近0.0表示两向量不相关
值接近-1.0表示两向量方向相反
"""
# ----- 第1步：校验输入 -----
# 确保两个向量的维度一致，如果不一致则抛出异常
if len(vec_a) != len(vec_b):
raise ValueError(
f"向量维度不匹配：vec_a是{len(vec_a)}维，vec_b是{len(vec_b)}维"
)
# ----- 第2步：计算向量点积（Dot Product）-----
# 点积 = Σ(aᵢ × bᵢ)，即对应位置的元素相乘后累加
# 例如 vec_a=[1,2,3]，vec_b=[4,5,6]
# 点积 = 1×4 + 2×5 + 3×6 = 32
dot_product = 0.0 # 初始化点积为0
for i in range(len(vec_a)):
dot_product += vec_a[i] * vec_b[i] # 累加每个维度的乘积
# ----- 第3步：计算向量A的L2范数（模长）-----
# ||A|| = √(a₁² + a₂² + ... + aₙ²)
# 即每个元素平方后求和，再开平方根
norm_a = 0.0 # 初始化模长为0
for val in vec_a:
norm_a += val * val # 累加每个元素的平方
norm_a = norm_a ** 0.5 # 开平方得到模长
# ----- 第4步：计算向量B的L2范数（模长）-----
norm_b = 0.0
for val in vec_b:
norm_b += val * val
norm_b = norm_b ** 0.5
# ----- 第5步：计算余弦相似度 -----
# 如果某个向量模长为0，说明是零向量，无法计算相似度
if norm_a == 0.0 or norm_b == 0.0:
raise ValueError("零向量无法计算余弦相似度")
# 余弦相似度 = 点积 / (模长A × 模长B)
similarity = dot_product / (norm_a * norm_b)
return similarity
# ===== 测试代码 =====
if __name__ == "__main__":
# 定义两组测试向量
# 场景1：语义相似的两个句子（"我想退订服务" 和 "如何取消我的订阅"）
vec1 = [0.2, 0.8, 0.3, 0.1] # 句子1的向量表示
vec2 = [0.25, 0.75, 0.35, 0.12] # 句子2的向量表示（方向接近）
# 场景2：一个句子与完全不相关的句子
vec3 = [0.9, 0.1, 0.8, 0.2] # 与vec1方向差异很大的向量
# 计算并打印相似度结果
sim_1_2 = cosine_similarity_pure_python(vec1, vec2)
sim_1_3 = cosine_similarity_pure_python(vec1, vec3)
print(f"向量1 vs 向量2（语义相似）：{sim_1_2:.4f}")
print(f"向量1 vs 向量3（语义不相关）：{sim_1_3:.4f}")
# ============================================================
# NumPy版余弦相似度计算
# 在生产环境中推荐使用NumPy实现，性能远超纯Python循环
# ============================================================
import numpy as np # 导入NumPy库，用于高效数值计算
def cosine_similarity_numpy(vec_a, vec_b):
"""
使用NumPy计算余弦相似度。
参数:
vec_a: list[float] 或 np.ndarray - 第一个向量
vec_b: list[float] 或 np.ndarray - 第二个向量
返回:
float - 余弦相似度值，范围 [-1.0, 1.0]
"""
# ----- 第1步：转换为NumPy数组 -----
# np.array() 将Python列表转换为NumPy的ndarray对象
# dtype=np.float64 指定数据类型为64位浮点数，保证计算精度
vec_a = np.array(vec_a, dtype=np.float64)
vec_b = np.array(vec_b, dtype=np.float64)
# ----- 第2步：校验维度 -----
if vec_a.shape[0] != vec_b.shape[0]:
raise ValueError(
f"向量维度不匹配：{vec_a.shape[0]}维 vs {vec_b.shape[0]}维"
)
# ----- 第3步：计算点积 -----
# np.dot() 是高度优化的矩阵乘法实现，比Python循环快100-1000倍
# 当输入是1维向量时，np.dot(a, b) 等价于 a · b
dot_product = np.dot(vec_a, vec_b)
# ----- 第4步：计算两个向量的L2范数 -----
# np.linalg.norm() 是NumPy内置的向量范数计算函数
# 默认ord=2表示L2范数，即欧氏距离中的模长
norm_a = np.linalg.norm(vec_a)
norm_b = np.linalg.norm(vec_b)
# ----- 第5步：处理零向量情况 -----
if norm_a == 0 or norm_b == 0:
# 如果任一向量为零向量，直接返回0（避免除零错误）
return 0.0
# ----- 第6步：计算归一化余弦相似度 -----
similarity = dot_product / (norm_a * norm_b)
return float(similarity) # 转换为Python原生float便于后续使用
# ===== 批量计算函数：一次计算一个向量与多个向量的相似度 =====
def cosine_similarity_batch(query_vec, doc_vectors):
"""
批量计算查询向量与多个文档向量的余弦相似度。
这是向量检索系统的核心计算模块，在实际应用中会一次处理数千乃至数百万文档。
参数:
query_vec: list[float] - 用户查询的向量（1个向量）
doc_vectors: list[list[float]] - 文档向量库（多个向量组成的二维列表）
返回:
list[float] - query与每个文档向量的相似度分数列表
"""
# 转换为NumPy二维数组，行=文档数，列=向量维度
# shape[0] = 文档数量，shape[1] = 向量维度
doc_matrix = np.array(doc_vectors, dtype=np.float64)
query = np.array(query_vec, dtype=np.float64)
# ----- 批量点积计算 -----
# 矩阵乘法：query (1,D) · doc_matrix (N,D)^T = (1,N) 相似度向量
# 等价于对doc_vectors中每个向量分别做点积，但只需一次矩阵乘法
# np.dot(query, doc_matrix.T) 利用NumPy的广播机制一次性算出所有相似度
dot_products = np.dot(query, doc_matrix.T) # shape: (N,)
# ----- 批量范数计算 -----
# 计算每个文档向量的L2范数，形成一个长度为N的数组
norms = np.linalg.norm(doc_matrix, axis=1) # axis=1表示按行计算范数
# ----- 批量归一化 -----
# 将点积除以对应范数乘积，完成所有文档的余弦相似度计算
# 利用NumPy的逐元素除法（broadcasting），一行代码完成全部N个计算
similarities = dot_products / norms
# query自身的范数，用于归一化
query_norm = np.linalg.norm(query)
# 最终归一化：每个相似度除以query的模长
result = similarities / query_norm
return result.tolist() # 转换为Python列表返回
# ===== 测试代码 =====
if __name__ == "__main__":
# 示例：3个文档向量（实际应用中通常是1536维或768维）
documents = [
[0.2, 0.8, 0.3, 0.1], # 文档1向量
[0.9, 0.1, 0.8, 0.2], # 文档2向量
[0.1, 0.3, 0.1, 0.9], # 文档3向量
]
# 用户查询向量（与文档1最语义相近）
query = [0.2, 0.8, 0.3, 0.12]
# 批量计算查询向量与所有文档的相似度
scores = cosine_similarity_batch(query, documents)
print("各文档相似度分数：")
for i, score in enumerate(scores):
print(f" 文档{i+1}: {score:.4f}")
# ============================================================
# 使用OpenAI官方Embedding API将文本转换为向量
# 推荐模型：text-embedding-3-small（高性能，1536维）
# text-embedding-3-large（高精度，3072维）
# ============================================================
import os
import math
def get_openai_embedding(text, model="text-embedding-3-small"):
"""
调用OpenAI Embedding API，将文本转换为高维向量。
参数:
text: str - 输入文本（如用户查询或文档内容）
model: str - 使用的Embedding模型名称
text-embedding-3-small: 1536维，速度快，成本低（$0.02/1M tokens）
text-embedding-3-large: 3072维，精度高（$0.13/1M tokens）
返回:
list[float] - 文本对应的embedding向量
注意：
- API会按token计费，建议将长文本预先分段
- 文本最大长度约8,191 tokens（模型context window）
- 返回的向量已经过归一化（模长=1），可直接用于余弦相似度计算
"""
try:
# 导入OpenAI官方客户端库
from openai import OpenAI
except ImportError:
# 如果未安装openai库，提示用户安装
raise ImportError(
"请先安装OpenAI库：pip install openai\n"
"然后设置环境变量：export OPENAI_API_KEY='your-api-key'"
)
# 从环境变量读取API密钥（安全做法，避免硬编码密钥）
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
raise ValueError(
"未找到OPENAI_API_KEY环境变量。\n"
"请在终端执行：export OPENAI_API_KEY='sk-xxxxxxx'"
)
# 创建OpenAI客户端实例
# 客户端会自动读取OPENAI_API_KEY环境变量
client = OpenAI(api_key=api_key)
# 调用Embedding API
# response是一个Embedding对象，包含data列表
# 每个data对象包含embedding（向量列表）和index（序号）
response = client.embeddings.create(
model=model, # 指定使用的Embedding模型
input=text, # 输入文本（单文本或文本列表均可）
encoding_format="float" # 返回float格式的向量（非base64）
)
# 提取向量：response.data[0].embedding 就是我们需要的向量
# response.data是一个列表，当input是单个文本时只有一项
embedding_vector = response.data[0].embedding
return embedding_vector
# ===== 使用本地模型（Sentence-Transformers）作为替代方案 =====
def get_local_embedding(text, model_name="shibing624/text2vec-base-chinese"):
"""
使用开源本地Embedding模型将文本转换为向量。
适合：没有OpenAI API Key、隐私敏感数据、需要离线部署的场景。
推荐中文模型：shibing624/text2vec-base-chinese
备选英文模型：sentence-transformers/all-MiniLM-L6-v2
参数:
text: str - 输入文本
model_name: str - HuggingFace上的模型名称
返回:
list[float] - 文本对应的embedding向量
"""
try:
from sentence_transformers import SentenceTransformer
except ImportError:
raise ImportError(
"请先安装sentence-transformers：pip install sentence-transformers"
)
# 加载预训练模型（首次调用会从HuggingFace下载模型权重）
# device="cpu" 表示使用CPU计算（也可设为"cuda"使用GPU，大幅加速）
model = SentenceTransformer(model_name, device="cpu")
# encode()方法将文本转为向量
# normalize_embeddings=True 会自动将输出向量归一化（模长=1）
# 这样后续余弦相似度计算就只需要点积，不需要再除以模长
embedding_vector = model.encode(
text,
normalize_embeddings=True, # 归一化，向量模长=1
convert_to_numpy=True # 转换为NumPy数组
)
# 转换为Python原生list，便于后续处理
return embedding_vector.tolist()
# ===== 测试代码 =====
if __name__ == "__main__":
test_texts = [
"怎么处理客户投诉",
"如何办理退款手续",
"苹果公司今天的股价"
]
print("=" * 60)
print("测试文本向量生成（本地模型）")
print("=" * 60)
for text in test_texts:
vec = get_local_embedding(text)
# 打印向量的前5个维度+总维度数
print(f"\n文本: {text}")
print(f"向量维度: {len(vec)}")
print(f"向量前5维: {[round(v, 4) for v in vec[:5]]}")
有了向量之后，需要解决的核心问题是：如何在海量向量中快速找到与查询向量最相似的K个。如果直接暴力遍历所有向量（Brute Force），时间复杂度是O(N)，在百万级文档规模下会非常慢（每次查询可能需要几十秒）。
向量数据库就是专门解决这个问题的系统，它通过构建索引结构，将查询复杂度从O(N)降低到O(log N)甚至O(1)，同时召回率（Accuracy）损失控制在可接受范围内。
什么是HNSW？
HNSW（Hierarchical Navigable Small World）是一种基于图的近似最近邻搜索算法，也是目前最流行的向量索引算法，PostgreSQL的pgvector扩展、Chroma、Qdrant等主流向量数据库均支持HNSW。
HNSW的工作原理：
第一层：构建多层图结构
HNSW构建一个多层次的图，类似于一座电梯大厦：顶层是高速公路（跳层连接），底层是逐层深入的小区道路。顶层节点数量少，连接稀疏，可以快速跳过大部分不相关的区域；底层节点数量多，连接密集，保证最终的搜索精度。
第二层：贪婪式图遍历搜索
查询时，从顶层开始找到当前层的最近邻，然后跳到下一层，继续在该层寻找更近的邻居。如此逐层向下，直到最底层（原始数据层）。这个过程就像：你先坐电梯到大楼高层俯瞰全局，找到大概方向后，一层一层下楼，最终找到目标办公室。
核心参数：
HNSW的优势在于：构建后查询速度极快（毫秒级），且支持动态插入新向量（无需全量重建）。劣势是内存占用较大（需要存储完整的图结构），并且构建时间较长。
IVF（Inverted File Index）是一种基于聚类的向量索引方法，其核心思想类似图书馆的分类卡片系统。
构建过程：
使用K-Means聚类算法（如k=1024）将N个向量分成K个簇（cluster），每个簇有一个中心点。系统预先记录"哪些向量属于哪个簇"，形成倒排索引——从簇心快速定位到簇内的所有向量。
查询过程：
给定查询向量，先找到距离最近的M个簇心（如M=3），然后只在选中的这3个簇内做精确搜索（可以用Brute Force），跳过其余的簇。由于只搜索了3/1024≈0.3%的数据，查询速度大幅提升。
核心参数：
IVF的优势是内存占用相对可控，适合内存受限的场景。劣势是精度受聚类效果影响，查询时需要额外做聚类中心匹配。
PQ（Product Quantization，乘积量化）是一种有损向量压缩算法，用于在内存受限情况下存储海量向量。其核心思想是"分而治之"：将高维向量拆分为多个子空间，分别做量化编码。
具体原理：
假设一个向量是1536维（每维float32，占12字节），对1百万个向量来说，原始存储需要 1,000,000 × 1536 × 4字节 ≈ 6GB。而通过PQ压缩，可以将每个向量压缩到几十字节。
PQ将向量拆分为多段（如16段×96维），对每段单独做K-Means聚类（如每段聚类中心数k=256）。每个子向量不再存储原始浮点数，而是存储其所属聚类中心的ID（1个字节即可），从而实现10-100倍的压缩比。
查询时的距离计算：
PQ使用查表法计算近似距离：预计算查询向量各子段与各聚类中心的距离表（Lookup Table），查询时用ID查表累加即可，时间复杂度与原始维度无关。
PQ通常与IVF结合使用（IVF-PQ），先用IVF快速定位到相关簇，再用PQ压缩方式存储簇内向量，大幅减少内存占用的同时保持较高召回率。这是工业界处理十亿级向量数据的主流方案。
# ============================================================
# Chroma向量数据库完整使用示例
# Chroma是一个轻量级本地向量数据库，开源免费，易于上手
# 支持增删改查、过滤、元数据管理，是搭建AI知识库的入门首选
# ============================================================
import chromadb
from chromadb.config import Settings
# ============================================================
# 第1步：初始化Chroma客户端
# ============================================================
# 创建持久化客户端，数据会保存在本地磁盘
# 这样重启服务后数据不会丢失
# 若设置persist_directory=None，则为内存模式（重启后数据丢失）
client = chromadb.PersistentClient(
path="./chroma_data", # 数据持久化存储路径
settings=Settings(
anonymized_telemetry=False # 关闭匿名遥测（保护隐私）
)
)
# ============================================================
# 第2步：创建Collection（向量集合）
# ============================================================
# collection类似关系数据库中的"表"的概念
# 每个collection管理一批向量及其元数据
# name参数：集合名称（同一客户端下不能重复）
# metadata：可存储集合的描述信息
collection = client.get_or_create_collection(
name="knowledge_base_articles", # 知识库文章集合
metadata={
"description": "投肯智能知识库技术文章集合",
"version": "1.0"
}
)
# ============================================================
# 第3步：添加向量数据（增）
# ============================================================
# 每个向量条目需要：id、embedding向量、document文本内容、metadata元数据
# add()支持批量添加，大幅减少API调用次数
# 示例数据：3篇知识库文章
documents = [
"本文介绍如何申请报销，需要提供发票抬头和明细清单",
"员工出差交通费用报销标准：机票不超过2000元，高铁不超过800元",
"公司笔记本电脑采购流程：需先提交IT需求申请，经审批后统一采购"
]
# 对应的embedding向量（实际应用中应调用API或本地模型生成）
# 这里用随机向量模拟，实际系统请替换为真实模型生成的向量
embeddings = [
[0.12, 0.45, 0.78, 0.23, 0.91], # 文章1的向量（1536维，实际场景需要更多维度）
[0.34, 0.67, 0.12, 0.89, 0.56], # 文章2的向量
[0.78, 0.23, 0.45, 0.91, 0.34], # 文章3的向量
]
# 元数据：可用来过滤、分类
# 支持字符串、数字、布尔类型，但不能存list
metadatas = [
{"category": "财务制度", "author": "小云", "publish_date": "2026-05-01"},
{"category": "差旅报销", "author": "小虾", "publish_date": "2026-04-15"},
{"category": "IT采购", "author": "小刚", "publish_date": "2026-03-20"}
]
# 唯一ID：每个向量条目的标识符
ids = ["doc_001", "doc_002", "doc_003"]
# 执行添加操作
collection.add(
ids=ids, # 唯一ID列表（必填）
embeddings=embeddings, # 向量列表（必填）
documents=documents, # 原始文本列表（必填）
metadatas=metadatas # 元数据字典列表（可选）
)
print(f"✅ 成功添加 {len(documents)} 条文档到向量数据库")
print(f" Collection名称: {collection.name}")
print(f" 当前文档总数: {collection.count()}")
# ============================================================
# 第4步：查询向量（查）
# ============================================================
# 使用query()方法检索与查询向量最相似的文档
# n_results: 返回最相似的几条结果（默认10条）
# where: 元数据过滤条件（可选，用于精确筛选）
# include: 返回哪些字段（["documents", "metadatas", "distances"]）
query_embedding = [0.15, 0.50, 0.80, 0.25, 0.88] # 查询向量
results = collection.query(
query_embeddings=[query_embedding], # 支持批量查询（一次查多个向量）
n_results=3, # 返回最相似的3条
where={"category": "财务制度"}, # 只在"财务制度"分类中搜索（可选）
include=["documents", "metadatas", "distances"] # 返回内容+元数据+距离
)
print("\n📊 查询结果：")
print(f" 查询向量: {query_embedding}")
print(f" 找到 {len(results['ids'][0])} 条相关文档：")
for i, doc_id in enumerate(results['ids'][0]):
# distance表示向量间的距离（Chroma使用L2距离）
# distance=0表示完全匹配，值越大相似度越低
print(f"\n 第{i+1}名 [{doc_id}]")
print(f" 相似度距离: {results['distances'][0][i]:.4f}")
print(f" 内容: {results['documents'][0][i]}")
print(f" 分类: {results['metadatas'][0][i]['category']}")
# ============================================================
# 第5步：更新向量数据（改）
# ============================================================
# update()方法：根据ID更新已有文档的向量、文本或元数据
# 如果ID不存在则报错
# 适合内容修改场景（如文章编辑后重新计算向量）
collection.update(
ids=["doc_001"], # 要更新的文档ID
documents=["【更新】报销申请流程已调整：新增加急审批通道，最长3个工作日完成"],
metadatas=[{"category": "财务制度", "author": "小云", "publish_date": "2026-05-20"}]
)
print("\n✅ 文档doc_001已更新")
# 验证更新结果
updated_doc = collection.get(ids=["doc_001"])
print(f" 更新后内容: {updated_doc['documents'][0]}")
# ============================================================
# 第6步：删除向量数据（删）
# ============================================================
# delete()方法：根据ID删除指定文档
# 谨慎使用，删除后不可恢复！
# 删除单条
# collection.delete(ids=["doc_003"])
# 删除满足条件的多条（删除所有"IT采购"分类的文章）
# collection.delete(where={"category": "IT采购"})
print("\n⚠️ 删除操作已就绪（实际未执行，请根据需要取消注释）")
# ============================================================
# 第7步：使用Peek查看数据（不执行查询）
# ============================================================
# peek()方法：查看数据库中的前N条数据，不做向量搜索
# 适合调试、数据检查场景
all_data = collection.peek(limit=10)
print("\n📋 数据库中的全部文档（peek预览）：")
print(f" IDs: {all_data['ids']}")
print(f" 文档数: {len(all_data['documents'])}")
# ============================================================
# 第8步：重置Collection（清空所有数据）
# ============================================================
# ⚠️ 危险操作：删除并重建Collection，所有数据将不可恢复
# 请在生产环境中谨慎使用，务必提前备份数据
# client.delete_collection(name="knowledge_base_articles")
# print("🗑️ Collection已清空")
| 对比维度 | Chroma | Milvus | Qdrant | Pinecone |
|---|---|---|---|---|
| 架构类型 | 单机/轻量客户端 | 分布式集群 | 分布式集群+混合检索 | 云原生托管服务 |
| 开源协议 | Apache 2.0 | Apache 2.0 | Apache 2.0 | 专有（闭源） |
| 支持语言 | Python/JS/Go/Java | Python/JS/Go/Java/Rust | Python/JS/Go/Rust | Python/JS/Go/Java/Node |
| 索引算法 | HNSW（默认） | HNSW / IVF / PQ / 混合 | HNSW + 过滤混合 | HNSW（闭源自研） |
| 向量维度上限 | 4096 | 32768 | 4096 | 3072（标准） |
| 数据规模 | 百万级（单节点） | 十亿级（分布式） | 亿级（分布式） | 十亿级（云端） |
| 元数据过滤 | 支持（有限） | 支持（强大） | 支持（强大） | 支持（强大） |
| 部署难度 | ⭐ 极简（pip安装即用） | ⭐⭐⭐⭐ 复杂（需K8s） | ⭐⭐⭐ 中等（Docker单节点） | ⭐ 无需部署（API直连） |
| 云端托管 | 不支持 | Zilliz Cloud（商业版） | Qdrant Cloud | Pinecone Cloud（原生） |
| 适用场景 | 本地开发/原型验证 | 企业级大规模部署 | 生产级+混合检索 | 快速接入+免运维 |
| 推荐指数 | ⭐⭐⭐⭐ 学习入门首选 | ⭐⭐⭐⭐ 大规模生产首选 | ⭐⭐⭐⭐ 混合检索首选 | ⭐⭐⭐ 云端免运维首选 |
我们用实际代码模拟一个知识库的检索场景，对比关键词检索（基于TF-IDF）和向量检索（基于Embedding）的实际效果差异。
# ============================================================
# 关键词检索 vs 向量检索对比测试
# 测试场景：企业知识库问答系统
# 评估指标：准确率（Precision）、召回率（Recall）、MRR
# ============================================================
import math
# ===== 准备测试数据集 =====
# 模拟一个企业知识库，包含10条文档（实际场景可达百万级）
# 每条文档包含：ID、标题、内容、正确答案的文档ID列表
knowledge_base = [
{"id": "d001", "title": "发票报销流程", "content": "员工报销需要提供发票、费用明细单、部门负责人签字，经财务审核后打款。"},
{"id": "d002", "title": "年假计算规则", "content": "员工入职满一年享受5天年假，工作满3年享10天，满5年享15天，按自然年度计算。"},
{"id": "d003", "title": "笔记本电脑采购申请", "content": "员工需填写IT采购申请单，经部门主管和IT部门审批后，统一由行政部集中采购。"},
{"id": "d004", "title": "客户投诉处理规范", "content": "客服收到投诉后需在2小时内响应，24小时内给出处理方案，重要投诉上报部门负责人。"},
{"id": "d005", "title": "合同审批流程", "content": "合同金额超过5万元需经法务部门审核，超过50万元需经总经理审批，所有合同需用印前存档。"},
{"id": "d006", "title": "员工考勤制度", "content": "上班时间9:00-18:00，迟到一次扣50元，旷工一天扣当日三倍工资，请假需提前一天申请。"},
{"id": "d007", "title": "技术方案评审流程", "content": "技术方案完成后需经过内部评审会议，参与者包括产品、技术、测试，评审通过后才能进入开发阶段。"},
{"id": "d008", "title": "差旅费用报销标准", "content": "国内出差机票最高报销2000元，高铁最高800元，住宿费每天最高500元，餐补每天100元。"},
{"id": "d009", "title": "离职手续办理", "content": "员工离职需提前30天提交申请，完成工作交接后退还公司资产，人事部门办理社保减员和离职证明。"},
{"id": "d010", "title": "会议室预约规则", "content": "员工可通过企业微信预约会议室，提前最多7天预约，单次最长使用2小时，超时将被自动释放。"},
]
# 定义测试查询及对应的正确答案（relevant doc IDs）
test_queries = [
{
"query": "我想报销差旅费，需要准备什么材料",
"relevant_ids": ["d001", "d008"] # 正确答案：发票报销 + 差旅报销
},
{
"query": "年假是怎么算的，工作多久有年假",
"relevant_ids": ["d002"] # 正确答案：年假规则
},
{
"query": "要买一台新电脑，流程是什么",
"relevant_ids": ["d003"] # 正确答案：采购申请
},
{
"query": "被客户投诉了怎么办",
"relevant_ids": ["d004"] # 正确答案：投诉处理
},
{
"query": "合同盖章需要审批吗",
"relevant_ids": ["d005"] # 正确答案：合同审批
},
]
# ============================================================
# 关键词检索：简化版TF-IDF实现
# ============================================================
import re
from collections import Counter, defaultdict
def tokenize(text):
"""
中文分词：将文本拆分为单词列表
这里使用简单的字符级切分（实际应用应使用jieba等分词库）
"""
# 去除标点符号，按空格和标点分割
text = re.sub(r"[，。！？、；：''\""]", " ", text)
# 简单的字符bigram分词（两个相邻字符为一个词）
chars = list(text.replace(" ", ""))
words = []
for i in range(len(chars) - 1):
words.append(chars[i] + chars[i + 1])
return words
def compute_tf_idf(documents, query):
"""
简化TF-IDF检索：
1. 计算每个文档的词频（TF）
2. 计算每个词的逆文档频率（IDF）
3. 计算query中每个词在每个文档中的TF-IDF得分
4. 返回按得分排序的文档ID列表
返回: list[tuple(doc_id, score)]，按得分降序排列
"""
# ----- 第1步：构建词频矩阵 -----
doc_words = [] # 每个文档的分词结果
all_words = set() # 所有出现过的词
for doc in documents:
words = tokenize(doc["content"])
doc_words.append(set(words))
all_words.update(words)
# ----- 第2步：计算每个词的IDF（简化版：log(N / df)）-----
N = len(documents) # 文档总数
word_doc_freq = defaultdict(int) # 词出现在多少个文档中
for words in doc_words:
for word in words:
word_doc_freq[word] += 1
# 计算IDF：log(N / df)，如果词不在任何文档中，IDF=0
idf = {}
for word in all_words:
df = word_doc_freq[word]
idf[word] = math.log(N / (df + 1))
# ----- 第3步：计算查询与每个文档的相似度 -----
query_words = set(tokenize(query))
scores = []
for i, doc in enumerate(documents):
score = 0.0
doc_word_set = doc_words[i]
# 对query中的每个词，计算其在当前文档中的TF-IDF贡献
for word in query_words:
if word in doc_word_set:
# 该词在文档中的TF = 1（简化：出现就为1）
tf = 1
score += tf * idf[word] # TF × IDF
scores.append((doc["id"], score))
# 按得分降序排列
scores.sort(key=lambda x: x[1], reverse=True)
return scores
# ============================================================
# 向量检索：基于余弦相似度实现
# ============================================================
def cosine_similarity_numpy(vec_a, vec_b):
"""使用NumPy计算余弦相似度（参考模块二实现）"""
import numpy as np
vec_a = np.array(vec_a, dtype=np.float64)
vec_b = np.array(vec_b, dtype=np.float64)
dot = np.dot(vec_a, vec_b)
norm_a = np.linalg.norm(vec_a)
norm_b = np.linalg.norm(vec_b)
if norm_a == 0 or norm_b == 0:
return 0.0
return float(dot / (norm_a * norm_b))
def simple_embedding(text, seed=42):
"""
简化Embedding：用字符哈希生成固定维度的"伪向量"
实际应用中请使用真实的Embedding模型（OpenAI/text2vec等）
原理：用字符的Unicode码生成固定维度的向量
同样的文本产生同样的向量，语义相似的文本产生方向接近的向量
"""
import numpy as np
dim = 32 # 简化维度（实际模型通常1536或768维）
np.random.seed(seed + len(text)) # 根据文本内容生成确定性随机种子
base = np.random.randn(dim) * 0.1 # 基础随机向量
# 将文本中每个字符的信息注入向量
for i, char in enumerate(text):
char_vec = np.random.randn(dim) * 0.01 * (ord(char) % 100)
base += char_vec
# 归一化：向量模长=1
norm = np.linalg.norm(base)
return (base / norm).tolist()
def vector_search(query, documents, top_k=5):
"""
向量检索：
1. 将查询转换为向量
2. 将每个文档转换为向量
3. 计算查询与所有文档的余弦相似度
4. 返回得分最高的top_k个文档
"""
query_vec = simple_embedding(query) # 查询向量
doc_scores = []
for doc in documents:
doc_vec = simple_embedding(doc["content"])
similarity = cosine_similarity_numpy(query_vec, doc_vec)
doc_scores.append((doc["id"], similarity))
# 按相似度降序排列
doc_scores.sort(key=lambda x: x[1], reverse=True)
return doc_scores[:top_k]
# ============================================================
# 评估指标计算函数
# ============================================================
def compute_precision_recall(ranked_ids, relevant_ids, k):
"""
计算Precision@K和Recall@K
参数:
ranked_ids: list[str] - 检索系统返回的排序结果（从第1名到第K名）
relevant_ids: list[str] - 标准答案中的相关文档ID列表
k: int - 只看前k个结果
返回:
precision@k, recall@k
"""
top_k_ids = ranked_ids[:k]
# 在top_k结果中有多少命中了正确答案
hits = len([doc_id for doc_id in top_k_ids if doc_id in relevant_ids])
precision = hits / k
recall = hits / len(relevant_ids) if len(relevant_ids) > 0 else 0.0
return precision, recall
def compute_mrr(ranked_ids, relevant_ids):
"""
计算MRR（Mean Reciprocal Rank）
MRR衡量正确答案的平均排名位置
MRR=1.0表示所有查询的第一名就是正确答案
MRR=0.5表示正确答案平均排第2名
"""
for i, doc_id in enumerate(ranked_ids):
if doc_id in relevant_ids:
return 1.0 / (i + 1) # 命中了，返回1/排名（排名从1开始）
return 0.0 # 未命中任何正确答案
# ============================================================
# 执行对比测试
# ============================================================
print("=" * 70)
print("企业知识库检索系统：关键词检索 vs 向量检索 对比测试")
print("=" * 70)
print(f"知识库文档数: {len(knowledge_base)}")
print(f"测试查询数: {len(test_queries)}")
print("-" * 70)
# 存储每个查询的评估结果
results_history = []
for q_data in test_queries:
query = q_data["query"]
relevant = set(q_data["relevant_ids"])
# ----- 关键词检索 -----
tfidf_ranking = compute_tf_idf(knowledge_base, query)
tfidf_ranked_ids = [doc_id for doc_id, score in tfidf_ranking]
# ----- 向量检索 -----
vector_ranking = vector_search(query, knowledge_base, top_k=5)
vector_ranked_ids = [doc_id for doc_id, score in vector_ranking]
# ----- 计算评估指标 -----
k = 3 # 查看前3名结果
# 关键词检索指标
tfidf_prec, tfidf_rec = compute_precision_recall(tfidf_ranked_ids, relevant, k)
tfidf_mrr = compute_mrr(tfidf_ranked_ids, relevant)
# 向量检索指标
vec_prec, vec_rec = compute_precision_recall(vector_ranked_ids, relevant, k)
vec_mrr = compute_mrr(vector_ranked_ids, relevant)
print(f"\n查询: {query}")
print(f"正确答案: {relevant}")
print(f"\n关键词检索 top-{k}:")
print(f" 排序结果: {tfidf_ranked_ids[:k]}")
print(f" Precision@{k}: {tfidf_prec:.2f} | Recall@{k}: {tfidf_rec:.2f} | MRR: {tfidf_mrr:.2f}")
print(f"向量检索 top-{k}:")
print(f" 排序结果: {vector_ranked_ids[:k]}")
print(f" Precision@{k}: {vec_prec:.2f} | Recall@{k}: {vec_rec:.2f} | MRR: {vec_mrr:.2f}")
results_history.append({
"query": query,
"tfidf": {"prec": tfidf_prec, "rec": tfidf_rec, "mrr": tfidf_mrr},
"vector": {"prec": vec_prec, "rec": vec_rec, "mrr": vec_mrr}
})
# ----- 汇总统计 -----
print("\n" + "=" * 70)
print("汇总统计（所有查询的平均值）")
print("=" * 70)
avg = lambda lst: sum(lst) / len(lst)
tfidf_prec_avg = avg([r["tfidf"]["prec"] for r in results_history])
tfidf_rec_avg = avg([r["tfidf"]["rec"] for r in results_history])
tfidf_mrr_avg = avg([r["tfidf"]["mrr"] for r in results_history])
vec_prec_avg = avg([r["vector"]["prec"] for r in results_history])
vec_rec_avg = avg([r["vector"]["rec"] for r in results_history])
vec_mrr_avg = avg([r["vector"]["mrr"] for r in results_history])
print(f"\n{'指标':<15} {'关键词检索':<15} {'向量检索':<15} {'胜出方':<10}")
print("-" * 55)
print(f"{'Precision@3':<15} {tfidf_prec_avg:<15.3f} {vec_prec_avg:<15.3f} {'向量检索' if vec_prec_avg > tfidf_prec_avg else '关键词检索':<10}")
print(f"{'Recall@3':<15} {tfidf_rec_avg:<15.3f} {vec_rec_avg:<15.3f} {'向量检索' if vec_rec_avg > tfidf_rec_avg else '关键词检索':<10}")
print(f"{'MRR':<15} {tfidf_mrr_avg:<15.3f} {vec_mrr_avg:<15.3f} {'向量检索' if vec_mrr_avg > tfidf_mrr_avg else '关键词检索':<10}")
print("\n✅ 测试完成。注意：本测试使用简化伪向量，实际生产环境使用真实Embedding模型效果会更好。")
| 对比维度 | 关键词检索（TF-IDF/BM25） | 向量检索（Embedding） |
|---|---|---|
| 核心原理 | 统计词频+逆文档频率 | 深度学习语义向量空间 |
| 理解能力 | 字面匹配，不理解语义 | 理解上下文和语义关系 |
| 同义词处理 | ❌ 无法处理（"退款"≠"退货"） | ✅ 语义相近即可匹配 |
| 长查询处理 | ⭐⭐ 一般（匹配度分散） | ⭐⭐⭐⭐⭐ 优秀（提取整体语义） |
| 查询速度 | ⭐⭐⭐⭐⭐ 极快（倒排索引） | ⭐⭐⭐⭐ 需要索引加速（可接受） |
| 计算资源 | ⭐ 低（CPU即可） | ⭐⭐⭐ 较高（Embedding模型需要GPU） |
| 适用场景 | 精确匹配、专有名词检索 | 语义理解、问答系统、推荐 |
问题描述：Embedding向量的维度从128维到3072维不等，很多人认为维度越高表示的信息越丰富，效果越好。但这个认知忽略了"维度灾难"（Curse of Dimensionality）问题。
维度灾难的原理：
当向量维度升高时，高维空间中的数据分布会变得极其稀疏。以余弦相似度为例，在低维空间中方向差异明显的两个向量，在高维空间中可能"看起来都一样远"——即所有向量之间的相似度都趋近于0。这是因为在高维空间中，单位超球面的体积增长极快，数据点分布在越来越薄的球壳上，点与点之间的区分度反而降低。
实际影响：
解决方案：
三种主流相似度度量方式：
① 余弦相似度（Cosine Similarity）
衡量两个向量方向的相似性，取值[-1, 1]。OpenAI等主流Embedding模型输出的是归一化向量（模长=1），此时余弦相似度等价于点积，是最常用的选择。
② 点积（Dot Product / 内积）
当两个向量已归一化时，点积与余弦相似度等价。但若向量未归一化，点积会同时考虑方向和模长，可能导致长度偏差。建议在所有检索场景中优先使用归一化向量。
③ 欧氏距离（Euclidean Distance / L2）
衡量向量在空间中的直线距离，取值[0, +∞)。当向量已归一化时，L2距离与余弦相似度是单调关系（距离越小相似度越高），Chroma数据库默认使用L2距离。
选择建议：
语言特性差异：
中文和英文在语言结构上有根本性差异：英文以空格分词，词边界清晰；中文以字为单位，词语边界需要算法判断（分词）。这直接影响了Embedding模型的设计。
模型选择建议：
分词器差异：
中文Embedding模型需要使用对应的中文分词器（如jieba、pkuseg），英文模型使用空格分词。分词质量直接影响Embedding效果——如果分词粒度过粗或过细，都会导致向量表示不准确。
评估一个Embedding模型是否适合你的知识库，常用方法是计算召回率（Recall@K）：给定一组标准查询-答案对，用模型检索，观察正确答案出现在前K名的比例。
具体操作流程：
索引参数调优：
数据更新策略：
备份与恢复：