吾食吾味1. 为什么需要自定义向量数据库
在LangChain生态中，向量数据库扮演着核心角色，负责存储文本嵌入向量并支持高效的相似性搜索。当前主流方案如Pinecone、Weaviate等虽然功能完善，但实际业务中常遇到三类典型问题：
- 技术栈适配问题：某些企业内部已自建向量检索服务，采用私有协议或特殊存储格式
- 性能调优需求：业务场景可能需要特殊的索引结构或距离算法（如医疗领域需要Jaccard相似度）
- 轻量化部署：生产环境有时需要极简的内存数据库，避免引入重型依赖
重要提示：选择继承VectorStore基类而非现有实现时，意味着你需要完全控制底层存储逻辑。这种方案适合需要深度定制或对接全新存储引擎的场景。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 核心接口设计解析
2.1 必须实现的三个基础方法
LangChain通过抽象基类强制约定接口规范，这三个方法是向量数据库的"最小可行实现"：
python复制from typing import Iterable, List, Optional, Any, Dict
from abc import ABC, abstractmethod
class VectorStore(ABC):
@abstractmethod
def add_texts(
self,
texts: Iterable[str],
metadatas: Optional[List[dict]] = None,
**kwargs: Any
) -> List[str]:
"""批量添加文本数据及其元数据"""
@abstractmethod
def similarity_search(
self,
query: str,
k: int = 4,
**kwargs: Any
) -> List[Document]:
"""基于文本查询的相似性搜索"""
@classmethod
@abstractmethod
def from_texts(
cls,
texts: List[str],
embedding: Embeddings,
metadatas: Optional[List[dict]] = None,
**kwargs: Any,
) -> "VectorStore":
"""类方法：从文本集合构建向量库"""
参数设计要点：
metadatas需要与texts严格对齐（长度一致）**kwargs为未来扩展保留参数通道- 返回的ID列表应保证全局唯一性
2.2 推荐实现的增强方法
虽然以下方法非强制，但实现它们能获得更完整的LangChain集成体验：
| 方法名 | 作用域 | 典型应用场景 |
|---|---|---|
| delete | 实例方法 | 按ID删除特定文档 |
| _select_relevance_score_fn | 保护方法 | 自定义相似度评分策略 |
| similarity_search_with_score | 实例方法 | 需要显示相似度得分的检索 |
| max_marginal_relevance_search | 实例方法 | 兼顾相关性与多样性的检索 |
3. 内存型向量数据库完整实现
3.1 存储引擎设计
我们采用Python字典作为底层存储，结构设计如下：
python复制{
"doc_id_1": {
"vector": [0.1, 0.2, ...], # 嵌入向量
"text": "原始文本内容",
"metadata": {"source": "web"}, # 关联元数据
"timestamp": 1620000000.0 # 可选扩展字段
},
...
}
这种结构提供O(1)的单文档访问效率，但全量扫描时为O(N)。适合文档量小于10万的场景。
3.2 距离计算优化
原始示例使用欧氏距离，实际生产更推荐余弦相似度：
python复制def _cosine_similarity(self, vec1, vec2):
"""计算余弦相似度（范围[-1,1]，越大越相似）"""
dot = np.dot(vec1, vec2)
norm = np.linalg.norm(vec1) * np.linalg.norm(vec2)
return dot / (norm + 1e-10) # 防止除零
距离算法选型建议：
- 文本相似度：优先余弦相似度
- 图像检索：可尝试曼哈顿距离
- 推荐系统：常用内积相似度
3.3 批处理优化
原始add_texts逐个处理文本，改进后的批处理版本：
python复制def add_texts(self, texts: Iterable[str], metadatas: Optional[List[dict]] = None, **kwargs):
if metadatas and len(metadatas) != len(texts):
raise ValueError("Metadatas长度必须与texts一致")
# 批量生成嵌入（减少API调用次数）
embeddings = self._embedding.embed_documents(list(texts))
# 批量生成UUID
ids = [str(uuid.uuid4()) for _ in texts]
# 原子性写入
batch = {
id_: {
"id": id_,
"vector": emb,
"text": text,
"metadata": meta if metadatas else {}
}
for id_, emb, text, meta in zip(
ids, embeddings, texts, metadatas or [{}]*len(texts)
)
}
self.store.update(batch)
return ids
4. 生产环境实践要点
4.1 性能优化策略
当文档量超过1万时，建议：
-
索引优化：
python复制# 使用FAISS构建索引 import faiss index = faiss.IndexFlatIP(embedding_dim) index.add(np.array([doc["vector"] for doc in self.store.values()])) -
缓存机制：
python复制from functools import lru_cache @lru_cache(maxsize=1000) def get_embedding(text: str) -> List[float]: return self._embedding.embed_query(text)
4.2 常见问题排查
问题1：相似度分数范围异常
- 检查距离函数返回值范围
- 确认
_select_relevance_score_fn转换逻辑
问题2：内存泄漏
- 定期调用
gc.collect() - 使用弱引用存储大对象：
python复制import weakref self.store = weakref.WeakValueDictionary()
4.3 监控指标建议
在similarity_search中添加性能埋点：
python复制import time
from prometheus_client import Summary
SEARCH_TIME = Summary('vector_search_latency', '相似搜索耗时')
def similarity_search(self, query: str, k: int = 4, **kwargs):
start = time.time()
try:
# ...原有逻辑...
finally:
SEARCH_TIME.observe(time.time() - start)
5. 扩展应用场景
5.1 多模态向量存储
改造存储结构支持图像向量：
python复制def add_multimodal(
self,
texts: Optional[List[str]] = None,
image_vectors: Optional[List[List[float]]] = None,
**kwargs
):
# 合并文本和图像特征向量
combined_vec = np.concatenate([
text_embedding,
image_embedding
])
# ...存储逻辑...
5.2 混合检索方案
结合关键词与向量搜索：
python复制def hybrid_search(
self,
query: str,
k: int = 4,
keyword_weight: float = 0.3
) -> List[Document]:
# 1. 关键词搜索（基于倒排索引）
keyword_results = self.keyword_index.search(query)
# 2. 向量搜索
vector_results = self.similarity_search(query, k*2)
# 3. 混合排序
combined = self._merge_results(
keyword_results,
vector_results,
keyword_weight
)
return combined[:k]
这种实现方式在电商商品搜索等场景能提升20%+的召回率。关键在于调整keyword_weight参数，可通过A/B测试确定最佳值。