向量数据库与AI融合的年中总结：RAG场景下的存储架构演进
2026上半年，RAG（检索增强生成）从实验性技术变成了企业级AI应用的标准架构。而作为RAG的核心基础设施，向量数据库经历了从"专业的向量检索工具"到"AI应用的基础存储层"的定位升级。本文基于团队在三个RAG项目中的实践经验，复盘向量数据库的架构演进和关键决策。
一、从一个失败的RAG项目说起：为什么基础版向量检索不够用
今年2月，团队接到一个内部知识库的RAG项目，需求听起来很简单：将公司内部10万份技术文档向量化，支持自然语言检索。技术方案选择了当时最成熟的方案：OpenAI Embedding + Milvus。
第一个版本在上线两周后暴露了严重问题。用户的反馈集中在两点：一是检索结果不准确，明明有相关的文档却检索不到；二是检索延迟高，在文档数量增长到5万后，单次检索从50ms增长到了500ms。
深入分析后发现三个核心问题：第一，单一的向量相似度检索忽略了关键词匹配的精确性；第二，所有文档使用同一套参数进行向量化和检索，忽略了不同文档类型的特征差异；第三，embedding模型的维度固定为1536维，无法根据文档的语义复杂度动态调整。
二、RAG存储架构的三层演进模型
第一代架构适合概念验证阶段，简单的向量检索+Top-K就可以跑通。但当文档量超过1万、查询复杂度提升后，单一向量检索的召回率会从90%快速下降到60%以下。
第二代架构引入了混合检索和重排序，这是目前生产环境的主流方案。它将BM25的关键词匹配与向量相似度搜索融合，通过重排序模型（如BGE-Reranker）对候选结果做精排。在我们的测试中，这种方法将召回率从60%提升到了85%以上。
第三代架构代表了前沿方向：智能切分替代固定窗口切分、多模态Embedding支持图文混合检索、知识图谱增强语义理解。但工程复杂度显著增加。
三、实战：构建混合检索RAG存储层
import numpy as np
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field
import json
import hashlib
# 假设使用Milvus作为向量数据库
try:
from pymilvus import Collection, connections, FieldSchema, CollectionSchema, DataType
MILVUS_AVAILABLE = True
except ImportError:
MILVUS_AVAILABLE = False
print("[WARNING] pymilvus not installed, using in-memory fallback")
@dataclass
class Document:
doc_id: str
content: str
metadata: Dict = field(default_factory=dict)
embedding: Optional[List[float]] = None
keywords: List[str] = field(default_factory=list)
class HybridRetrievalStore:
"""混合检索存储层：向量检索 + BM25关键词检索"""
def __init__(self, collection_name: str, embedding_dim: int = 1536):
self.collection_name = collection_name
self.embedding_dim = embedding_dim
# 内存中的文档存储（用于BM25和元数据）
self.documents: Dict[str, Document] = {}
# BM25需要的词频统计
self.word_doc_freq: Dict[str, int] = {}
self.total_docs = 0
# Milvus连接
if MILVUS_AVAILABLE:
try:
connections.connect("default", host="localhost", port="19530")
self._init_collection()
except Exception as e:
print(f"[ERROR] Milvus connection failed: {e}")
def _init_collection(self):
"""初始化Milvus集合"""
if not MILVUS_AVAILABLE:
return
try:
# 检查集合是否已存在
if self.collection_name in [c.name for c in Collection.list()]:
self.collection = Collection(self.collection_name)
self.collection.load()
return
fields = [
FieldSchema(name="id", dtype=DataType.VARCHAR,
max_length=128, is_primary=True),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR,
dim=self.embedding_dim),
FieldSchema(name="content_hash", dtype=DataType.VARCHAR,
max_length=64),
]
schema = CollectionSchema(fields, "Hybrid retrieval store")
self.collection = Collection(self.collection_name, schema)
# 创建索引
index_params = {
"metric_type": "IP",
"index_type": "IVF_FLAT",
"params": {"nlist": 1024}
}
self.collection.create_index("embedding", index_params)
self.collection.load()
except Exception as e:
print(f"[ERROR] Collection init failed: {e}")
def add_document(self, doc: Document) -> bool:
"""添加文档到混合存储"""
try:
doc_id = doc.doc_id or hashlib.md5(
doc.content.encode()
).hexdigest()[:32]
doc.doc_id = doc_id
# 内存中存储（用于BM25）
self.documents[doc_id] = doc
self.total_docs += 1
# 更新词频统计
for keyword in doc.keywords or []:
self.word_doc_freq[keyword] = (
self.word_doc_freq.get(keyword, 0) + 1
)
# 向量数据库存储
if MILVUS_AVAILABLE and doc.embedding:
self.collection.insert([{
"id": doc_id,
"embedding": doc.embedding,
"content_hash": hashlib.md5(doc.content.encode()).hexdigest()
}])
return True
except Exception as e:
print(f"[ERROR] Add document failed: {e}")
return False
def vector_search(self, query_embedding: List[float],
top_k: int = 20) -> List[Tuple[str, float]]:
"""向量相似度检索"""
if not MILVUS_AVAILABLE:
# 内存中的余弦相似度检索（fallback）
results = []
for doc_id, doc in self.documents.items():
if doc.embedding:
similarity = np.dot(query_embedding, doc.embedding) / (
np.linalg.norm(query_embedding) *
np.linalg.norm(doc.embedding)
)
results.append((doc_id, float(similarity)))
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
try:
search_params = {"metric_type": "IP", "params": {"nprobe": 10}}
results = self.collection.search(
data=[query_embedding],
anns_field="embedding",
param=search_params,
limit=top_k,
output_fields=["id"]
)
return [(hit.id, hit.score) for hit in results[0]]
except Exception as e:
print(f"[ERROR] Vector search failed: {e}")
return []
def bm25_search(self, query_keywords: List[str],
top_k: int = 20) -> List[Tuple[str, float]]:
"""BM25关键词检索"""
scores = {}
avg_doc_len = sum(len(d.content) for d in self.documents.values()) / max(self.total_docs, 1)
k1, b = 1.5, 0.75
for doc_id, doc in self.documents.items():
score = 0.0
doc_len = len(doc.content)
for keyword in query_keywords:
if keyword in doc.content.lower():
tf = doc.content.lower().count(keyword)
df = self.word_doc_freq.get(keyword, 0)
idf = np.log((self.total_docs – df + 0.5) / (df + 0.5) + 1)
numerator = tf * (k1 + 1)
denominator = tf + k1 * (1 – b + b * doc_len / avg_doc_len)
score += idf * numerator / max(denominator, 1e-8)
if score > 0:
scores[doc_id] = score
return sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
def hybrid_search(self, query_embedding: List[float],
query_keywords: List[str],
top_k: int = 10,
vector_weight: float = 0.7) -> List[str]:
"""混合检索：向量 + BM25 加权融合"""
try:
vector_results = self.vector_search(query_embedding, top_k * 3)
bm25_results = self.bm25_search(query_keywords, top_k * 3)
# 归一化分数
max_vec_score = max(s for _, s in vector_results) if vector_results else 1.0
max_bm25_score = max(s for _, s in bm25_results) if bm25_results else 1.0
# 加权融合
merged_scores: Dict[str, float] = {}
for doc_id, score in vector_results:
merged_scores[doc_id] = vector_weight * (score / max(max_vec_score, 1e-8))
for doc_id, score in bm25_results:
normalized = (1 – vector_weight) * (score / max(max_bm25_score, 1e-8))
merged_scores[doc_id] = merged_scores.get(doc_id, 0) + normalized
# 排序返回
ranked = sorted(merged_scores.items(), key=lambda x: x[1], reverse=True)
return [doc_id for doc_id, _ in ranked[:top_k]]
except Exception as e:
print(f"[ERROR] Hybrid search failed: {e}")
return [doc_id for doc_id, _ in vector_results[:top_k]]
# 使用示例
if __name__ == "__main__":
store = HybridRetrievalStore("knowledge_base")
# 添加文档
docs = [
Document(
content="MySQL 8.0的InnoDB引擎支持原子DDL操作",
keywords=["mysql", "innodb", "ddl", "原子操作"],
embedding=[0.1] * 1536 # 实际应使用embedding模型生成
),
Document(
content="ClickHouse MergeTree引擎使用LSM树结构进行数据组织",
keywords=["clickhouse", "mergetree", "lsm tree"],
embedding=[0.2] * 1536
),
]
for doc in docs:
store.add_document(doc)
# 混合检索
query_emb = [0.15] * 1536
query_kw = ["mysql", "innodb"]
results = store.hybrid_search(query_emb, query_kw)
print(f"检索结果: {results}")
四、向量存储选型的五个决策维度
维度一：数据规模。 100万以下向量，pgvector足够；100万-1亿，Milvus或Qdrant是成熟选择；1亿以上需要分布式部署和GPU加速。
维度二：检索精度要求。 纯向量检索的召回率天花板约85%，混合检索可到92%，加上重排序可到95%。
维度三：实时性需求。 毫秒级检索需要GPU索引（如RAFT），百毫秒级可用IVF类索引。
维度四：运维复杂度。 pgvector运维最简单（就和管PostgreSQL一样），Milvus需要专门的向量索引管理，Qdrant介于两者之间。
维度五：成本。 自建向量数据库的硬件成本约是传统数据库的3-5倍（GPU服务器），云服务按调用量计费在小规模时更经济。
五、总结
过去半年，向量数据库在RAG场景下的角色经历了三个阶段的演进：从单一向量检索到混合检索，再到智能分层检索。核心经验是：不要试图让向量检索做所有事情——关键词匹配的精确性和向量的语义理解是互补的，而非替代关系。下半年计划在多层索引架构（粗筛+精排）和局部敏感哈希加速检索两个方向做深入探索。
