如何在LangChain中构建并使用自定义向量数据库

1. 自定义向量数据库对接

向量数据库的发展非常迅速，几乎每隔几天就会出现新的向量数据库产品。LangChain 不可能集成所有的向量数据库，此外，一些封装好的数据库可能存在 bug 或者其他问题。这种情况下，我们需要考虑创建自定义向量数据库，以实现特定需求。

在 LangChain 中，实现自定义向量数据库的类主要有两种模式：

• 继承已有封装的数据库类 ：通常用于扩展现有类的功能或者修复已知问题。
• 继承基类 VectorStore ：通常用于对接新的向量数据库。

在 LangChain 中，继承 VectorStore 后，只需要实现以下三个基础方法即可正常使用：

• add_texts：将对应的数据添加到向量数据库中。
• similarity_search：实现最基础的相似性搜索功能。
• from_texts：从特定的文本列表、元数据列表中构建向量数据库。

此外，还有一些使用频率较低的方法，LangChain 并未将其定义为虚方法。如果在未实现的情况下直接调用这些方法，会引发异常。具体包括：

• delete()：删除向量数据库中的数据。
• _select_relevance_score_fn()：根据距离计算相似性得分的函数。
• similarity_search_with_score()：附带得分的相似性搜索功能。
• similarity_search_by_vector()：传递向量进行相似性搜索。
• max_marginal_relevance_search()：最大边际相关性搜索功能。
• max_marginal_relevance_search_by_vector()：传递向量进行最大边际相关性搜索功能。

---

资料推荐

• 💡大模型中转API推荐
• ✨中转使用教程

---

2. 自定义 VectorStore 示例

要在 LangChain 中对接自定义向量数据，本质上就是将向量数据库提供的方法集成到 add_texts、similarity_search、from_texts 方法下，例如自建一个基于内存+欧几里得距离的“向量数据库”，示例如下：

import uuid
from typing import List, Optional, Any, Iterable, Typeimport dotenv
import numpy as np
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.vectorstores import VectorStore
from langchain_openai import OpenAIEmbeddingsclass MemoryVectorStore(VectorStore):"""自定义向量数据库"""store: dict = {}  # 在内存中开辟位置存储向量def __init__(self, embedding: Embeddings, **kwargs):self._embedding = embeddingdef add_texts(self, texts: Iterable[str], metadatas: Optional[List[dict]] = None, **kwargs: Any) -> List[str]:"""将数据添加到内存向量数据库中"""# 1.判断metadatas和texts的长度是否保持一致if metadatas is not None and len(metadatas) != len(texts):raise ValueError("元数据格式必须和文本数据保持一致")# 2.将文本转换为向量embeddings = self._embedding.embed_documents(texts)# 3.生成uuidids = [str(uuid.uuid4()) for text in texts]# 4.将原始文本、向量、元数据、id构建字典并存储for idx, text in enumerate(texts):self.store[ids[idx]] = {"id": ids[idx],"vector": embeddings[idx],"text": text,"metadata": metadatas[idx] if metadatas is not None else {}}return idsdef similarity_search(self, query: str, k: int = 4, **kwargs: Any) -> List[Document]:"""执行相似性搜索"""# 1.将query转换成向量embedding = self._embedding.embed_query(query)# 2.循环遍历记忆存储，计算欧几里得距离result: list = []for key, record in self.store.items():distance = self._euclidean_distance(embedding, record["vector"])result.append({"distance": distance,**record,})# 3.找到欧几里得距离最小的k条记录sorted_result = sorted(result, key=lambda x: x["distance"])result_k = sorted_result[:k]# 4.循环构建文档列表并返回documents = [Document(page_content=item["text"], metadata={**item["metadata"], "score": item["distance"]})for item in result_k]return documents@classmethoddef from_texts(cls: Type["MemoryVectorStore"], texts: List[str], embedding: Embeddings,metadatas: Optional[List[dict]] = None,**kwargs: Any) -> "MemoryVectorStore":"""通过文本、嵌入模型、元数据构建向量数据库"""memory_vector_store = cls(embedding=embedding, **kwargs)memory_vector_store.add_texts(texts, metadatas)return memory_vector_store@classmethoddef _euclidean_distance(cls, vec1, vec2) -> float:"""计算两个向量的欧几里得距离"""return np.linalg.norm(np.array(vec1) - np.array(vec2))dotenv.load_dotenv()# 1.创建初始数据与嵌入模型
texts = ["笨笨是一只很喜欢睡觉的猫咪","我喜欢在夜晚听音乐，这让我感到放松。","猫咪在窗台上打盹，看起来非常可爱。","学习新技能是每个人都应该追求的目标。","我最喜欢的食物是意大利面，尤其是番茄酱的那种。","昨晚我做了一个奇怪的梦，梦见自己在太空飞行。","我的手机突然关机了，让我有些焦虑。","阅读是我每天都会做的事情，我觉得很充实。","他们一起计划了一次周末的野餐，希望天气能好。","我的狗喜欢追逐球，看起来非常开心。",
]
metadatas = [{"page": 1},{"page": 2},{"page": 3},{"page": 4},{"page": 5},{"page": 6, "account_id": 1},{"page": 7},{"page": 8},{"page": 9},{"page": 10},
]
embedding = OpenAIEmbeddings(model="text-embedding-3-small")# 2.构建自定义向量数据库
db = MemoryVectorStore.from_texts(texts, embedding, metadatas)# 3.执行检索
print(db.similarity_search("我养了一只猫，叫笨笨"))