自定义向量数据库实践：从 LangChain VectorStore 到生产适配器

LangChain 已经集成了很多向量数据库，比如 FAISS、Pinecone、TCVectorDB、Weaviate、Milvus、Qdrant、PGVector。多数项目直接选一个成熟集成就够了。

但生产里总会遇到一些“不标准”的情况：

• 公司内部已经有自研向量检索服务。
• 某个官方封装能力不完整，过滤、删除、批量写入或分数解释不符合业务要求。
• 需要把现有搜索系统、权限系统、审计系统和 RAG 链路接起来。
• 需要临时修复社区集成里的 bug，但又不想 fork 整个框架。
• 需要做一个统一适配层，屏蔽底层 FAISS、Pinecone、Weaviate、TCVectorDB 的差异。

这时就要理解 LangChain 的核心抽象：到底应该实现 VectorStore，还是只实现一个 Retriever。

先分清 VectorStore 和 Retriever

这两个概念很容易混。

抽象	负责什么	适合场景
VectorStore	存储向量、写入文档、删除文档、相似度搜索	你要把一个向量数据库完整接入 LangChain
Retriever	根据 query 返回相关 Document	你只需要把某个检索服务接进 RAG，不关心写入和删除

如果你在做一个向量数据库适配器，通常实现 VectorStore。

如果你已有一个内部搜索 API，比如 /search?q=...，它自己完成召回、过滤、排序，你只是想让 LangChain 的 RAG 链路能调用它，那么实现 BaseRetriever 往往更简单。

生产里不要为了“看起来更完整”而硬写 VectorStore。抽象越大，责任越多，后续测试和兼容成本也越高。

LangChain VectorStore 的当前接口

当前 LangChain 对向量库的统一接口主要包括：

方法	作用
add_documents	写入 Document 对象
add_texts	写入原始文本和 metadata
delete	按 id 删除文档
similarity_search	按文本 query 做相似度检索
similarity_search_with_score	返回文档和分数
similarity_search_by_vector	直接按 query vector 检索
as_retriever	包装成 Retriever

最小可用实现不一定要覆盖所有方法，但生产适配器至少要把以下能力考虑清楚：

• 是否支持稳定 id。
• 是否支持删除。
• 是否支持 metadata filter。
• 是否支持分数返回。
• 是否支持按向量搜索。
• 是否支持批量写入。
• 是否支持异步。
• 分数是距离还是相关性。
• 删除和更新是否幂等。

一个最小内存 VectorStore

先用内存实现一个小型向量库，帮助理解接口关系。这个实现不适合生产，只适合理解抽象和写单元测试。

from __future__ import annotations

import uuid
from typing import Any, Iterable

import numpy as np
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.vectorstores import VectorStore


class MemoryVectorStore(VectorStore):
    def __init__(self, embedding: Embeddings) -> None:
        self.embedding = embedding
        self.records: dict[str, dict[str, Any]] = {}

    def add_texts(
        self,
        texts: Iterable[str],
        metadatas: list[dict] | None = None,
        ids: list[str] | None = None,
        **kwargs: Any,
    ) -> list[str]:
        text_list = list(texts)

        if metadatas is not None and len(metadatas) != len(text_list):
            raise ValueError("metadatas length must match texts length")

        if ids is not None and len(ids) != len(text_list):
            raise ValueError("ids length must match texts length")

        vectors = self.embedding.embed_documents(text_list)
        output_ids = ids or [str(uuid.uuid4()) for _ in text_list]

        for index, text in enumerate(text_list):
            record_id = output_ids[index]
            self.records[record_id] = {
                "id": record_id,
                "text": text,
                "vector": vectors[index],
                "metadata": metadatas[index] if metadatas else {},
            }

        return output_ids

    def similarity_search(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[Document]:
        results = self.similarity_search_with_score(query=query, k=k, **kwargs)
        return [doc for doc, _score in results]

    def similarity_search_with_score(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[tuple[Document, float]]:
        query_vector = self.embedding.embed_query(query)
        return self.similarity_search_by_vector_with_score(query_vector, k=k, **kwargs)

    def similarity_search_by_vector(
        self,
        embedding: list[float],
        k: int = 4,
        **kwargs: Any,
    ) -> list[Document]:
        results = self.similarity_search_by_vector_with_score(embedding, k=k, **kwargs)
        return [doc for doc, _score in results]

    def similarity_search_by_vector_with_score(
        self,
        embedding: list[float],
        k: int = 4,
        **kwargs: Any,
    ) -> list[tuple[Document, float]]:
        filters = kwargs.get("filter")
        candidates = []

        for record in self.records.values():
            if filters and not self._match_filter(record["metadata"], filters):
                continue

            distance = self._euclidean_distance(embedding, record["vector"])
            candidates.append((record, distance))

        top_k = sorted(candidates, key=lambda item: item[1])[:k]

        return [
            (
                Document(
                    page_content=record["text"],
                    metadata={**record["metadata"], "id": record["id"]},
                ),
                score,
            )
            for record, score in top_k
        ]

    def delete(self, ids: list[str] | None = None, **kwargs: Any) -> bool | None:
        if ids is None:
            return None

        for record_id in ids:
            self.records.pop(record_id, None)

        return True

    @classmethod
    def from_texts(
        cls,
        texts: list[str],
        embedding: Embeddings,
        metadatas: list[dict] | None = None,
        ids: list[str] | None = None,
        **kwargs: Any,
    ) -> MemoryVectorStore:
        store = cls(embedding=embedding)
        store.add_texts(texts=texts, metadatas=metadatas, ids=ids)
        return store

    @staticmethod
    def _euclidean_distance(left: list[float], right: list[float]) -> float:
        return float(np.linalg.norm(np.array(left) - np.array(right)))

    @staticmethod
    def _match_filter(metadata: dict, filters: dict[str, Any]) -> bool:
        return all(metadata.get(key) == value for key, value in filters.items())

使用：

from langchain_openai import OpenAIEmbeddings


embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

store = MemoryVectorStore.from_texts(
    texts=[
        "企业用户可以在控制台申请发票。",
        "退款申请需要提交订单号和付款凭证。",
        "索引更新后需要通过评估集验证召回质量。",
    ],
    embedding=embeddings,
    metadatas=[
        {"tenant_id": "tenant-a", "status": "published"},
        {"tenant_id": "tenant-a", "status": "published"},
        {"tenant_id": "tenant-a", "status": "draft"},
    ],
    ids=[
        "invoice:v1:chunk-0001",
        "refund:v1:chunk-0001",
        "rag-eval:v1:chunk-0001",
    ],
)

docs = store.similarity_search(
    "发票怎么申请",
    k=2,
    filter={"status": "published"},
)

这个例子能跑通核心接口，但离生产还很远。它没有索引结构、没有持久化、没有并发控制、没有权限隔离，也没有真正的 metadata 查询引擎。

为什么不能直接把内存版上生产

内存版适合理解，不适合上线。

主要问题：

• 进程重启后数据丢失。
• 所有向量全量遍历，数据量大后延迟不可控。
• 多进程和多实例之间数据不一致。
• metadata filter 只是 Python 层过滤，没有索引。
• 没有批量写入限流和重试。
• 没有删除审计。
• 没有向量维度校验。
• 没有索引版本和回滚。

生产里的自定义适配器应该接一个真实后端：PostgreSQL + pgvector、Elasticsearch、OpenSearch、Milvus、自研 ANN 服务、内部搜索平台，或者某个云向量库的原生 SDK。

生产适配器的分层

一个更可靠的结构是三层。

层级	职责
Domain Service	文档解析、权限、租户、知识库、版本、审计
Vector Adapter	对齐 LangChain VectorStore 接口
Backend Client	调用真实向量数据库或内部搜索服务

不要让 VectorStore 承担所有业务逻辑。它应该只负责把 LangChain 的调用翻译成后端检索能力。权限、租户、发布、评估、审计这些业务规则，应放在更上层的服务里。

封装真实后端

生产适配器一般长这样：

from dataclasses import dataclass
from typing import Any

from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.vectorstores import VectorStore


@dataclass(frozen=True)
class SearchRecord:
    id: str
    text: str
    metadata: dict[str, Any]
    score: float


class InternalVectorClient:
    def upsert(self, records: list[dict[str, Any]]) -> None:
        ...

    def delete(self, ids: list[str]) -> None:
        ...

    def search(
        self,
        vector: list[float],
        k: int,
        filters: dict[str, Any] | None,
    ) -> list[SearchRecord]:
        ...


class InternalVectorStore(VectorStore):
    def __init__(self, client: InternalVectorClient, embedding: Embeddings) -> None:
        self.client = client
        self.embedding = embedding

    def add_texts(
        self,
        texts: list[str],
        metadatas: list[dict] | None = None,
        ids: list[str] | None = None,
        **kwargs: Any,
    ) -> list[str]:
        if ids is None:
            raise ValueError("ids are required in production")

        vectors = self.embedding.embed_documents(texts)

        records = [
            {
                "id": ids[index],
                "text": text,
                "vector": vectors[index],
                "metadata": metadatas[index] if metadatas else {},
            }
            for index, text in enumerate(texts)
        ]

        self.client.upsert(records)
        return ids

    def similarity_search_with_score(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[tuple[Document, float]]:
        query_vector = self.embedding.embed_query(query)
        records = self.client.search(
            vector=query_vector,
            k=k,
            filters=kwargs.get("filter"),
        )

        return [
            (
                Document(
                    page_content=record.text,
                    metadata={**record.metadata, "id": record.id},
                ),
                record.score,
            )
            for record in records
        ]

    def similarity_search(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[Document]:
        return [doc for doc, _score in self.similarity_search_with_score(query, k, **kwargs)]

    def delete(self, ids: list[str] | None = None, **kwargs: Any) -> bool | None:
        if not ids:
            return None
        self.client.delete(ids)
        return True

这个结构的关键点是：业务侧必须显式传 ids，适配器负责向量化和接口翻译，真实存储和检索由后端 client 完成。

什么时候只写 Retriever

如果后端已经是一个完整搜索服务，只暴露查询接口，不希望 LangChain 写入或删除，那么实现 BaseRetriever 更轻。

from typing import Any

from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever


class InternalSearchRetriever(BaseRetriever):
    client: Any
    k: int = 5

    def _get_relevant_documents(
        self,
        query: str,
        *,
        run_manager: CallbackManagerForRetrieverRun,
    ) -> list[Document]:
        results = self.client.search(query=query, k=self.k)

        return [
            Document(
                page_content=item["text"],
                metadata=item["metadata"],
            )
            for item in results
        ]

这种方式尤其适合：

• 只读知识库。
• 内部搜索服务已经完成排序和权限。
• 数据写入不由当前应用负责。
• 想快速接入 LCEL、RAG chain 或 Agent 工具。

分数要讲清楚

不同向量数据库返回的分数含义不同。

分数类型	越大越相关	越小越相关	常见来源
cosine similarity	是	否	余弦相似度
distance	否	是	L2、cosine distance
inner product	通常是	否	点积检索
rerank score	通常是	否	重排模型

自定义适配器要做两件事：

第一，文档里写清楚 similarity_search_with_score 返回的 score 是相似度还是距离。

第二，如果要使用 similarity_search_with_relevance_scores，要把分数归一化到清晰的区间，并说明是否可以跨索引比较。

生产里不要拿一个未校准的 score 做全局阈值。阈值应该来自评估集，而不是来自几次人工试问。

metadata filter 的边界

自定义适配器最容易偷懒的地方是 filter。

最简单的写法是在 Python 里先召回一批向量，再对 metadata 做过滤。但生产里这会有两个问题：

• 如果先取 k=5 再过滤，过滤后可能一条都不剩。
• 如果先取 fetch_k=1000 再过滤，延迟和成本不可控。

更稳的做法是把权限字段和高频过滤字段下推到底层数据库：

filters = {
    "tenant_id": "tenant-a",
    "knowledge_base_id": "kb-help-center",
    "status": "published",
}

底层服务应该在 ANN 检索前或检索过程中应用过滤，而不是只在最终结果里做装饰。

稳定 id 与幂等写入

生产里强制稳定 id。

推荐格式：

def build_chunk_id(document_id: str, chunk_version: str, chunk_index: int) -> str:
    return f"{document_id}:{chunk_version}:chunk-{chunk_index:04d}"

写入应该是 upsert，而不是每次生成新 id：

• 同一文档同一 chunk 版本重复写入，不产生重复数据。
• 文档删除时可以按 document_id 或 id 前缀清理。
• 索引重建时可以回放任务。
• 线上问题可以从命中文档 id 追到源文档和构建版本。

没有稳定 id 的向量库，后面做删除、更新、审计和回滚都会很难受。

异步与批量写入

生产入库链路通常不应该同步卡在请求里。

更合理的结构：

1. 用户上传文档。
2. 文档进入解析任务。
3. 切分 chunk。
4. 批量调用 Embedding。
5. 批量 upsert 向量库。
6. 写入索引 manifest。
7. 跑评估集。
8. 通过后切换线上检索版本。

如果要实现异步接口，可以补充：

async def aadd_texts(...):
    ...


async def asimilarity_search(...):
    ...

但不要为了“支持异步”简单把同步代码丢到线程池里。真正的异步收益来自后端 client、HTTP 连接池、批处理和限流策略都支持异步。

测试自定义 VectorStore

自定义适配器必须写测试。

最低测试清单：

• add_texts 能返回传入 ids。
• similarity_search 返回 Document。
• similarity_search_with_score 分数排序符合预期。
• delete 幂等，删除不存在 id 不报错。
• metadata filter 能过滤租户和状态。
• 文本数量、metadata 数量、ids 数量不一致时报错。
• Embedding 维度不一致时报错。
• 后端异常能转换成业务可理解的错误。
• 批量写入失败时不会造成部分不可追踪数据。

测试时可以使用固定向量的 fake embedding，不要在单元测试里依赖真实外部模型。

from langchain_core.embeddings import DeterministicFakeEmbedding


embeddings = DeterministicFakeEmbedding(size=8)

上线前检查清单

• 是否真的需要实现 VectorStore，还是只需要 Retriever。
• 是否强制传入稳定 ids。
• 是否支持 delete 和 upsert。
• metadata filter 是否下推到后端。
• 分数含义是否写清楚。
• Embedding 模型、维度、chunk 策略是否写入 manifest。
• 写入是否有批量、重试、限流和幂等。
• 查询是否记录 query、filter、k、score、latency、命中文档 id。
• 权限字段是否来自服务端认证态。
• 是否有固定评估集验证召回质量。
• 是否有索引版本切换和回滚方案。

问题

自定义向量数据库最常见的问题不是代码写不出来，而是边界没想清楚。

第一，误把 demo 内存库当成生产方案。内存库没有持久化、并发、一致性和索引能力。

第二，接口实现过少。只实现 similarity_search 能跑 RAG，但删除、分数、filter、按向量搜索、批量写入都会在上线后追着你补。

第三，filter 没有下推。权限过滤如果只在召回后做，很容易漏召回或越权。

第四，分数含义不清楚。距离和相似度混着用，会让阈值、排序和评估都失真。

第五，id 不稳定。没有稳定 id，就没有可靠更新、删除、回放和审计。

拓展

自定义适配器可以继续往几个方向扩展。

第一，统一多后端。用一个业务 VectorStoreFactory 封装 FAISS、Pinecone、Weaviate、TCVectorDB 和内部服务。

第二，混合检索。适配器内部同时调用关键词召回和向量召回，再用 reranker 排序。

第三，多向量。标题、正文、摘要、图片分别建向量，查询时按场景选择不同向量字段。

第四，可观测。每次检索都记录 trace，能回放 query、filter、候选文档、分数和最终答案。

第五，标准测试套件。给所有后端适配器跑同一组功能测试，避免切换底层数据库时行为漂移。

实际生产是否使用

会使用，但不是优先选项。

如果已有成熟 LangChain 集成，优先用官方或维护良好的独立包。自定义适配器适合内部搜索服务、特殊权限模型、特殊分数逻辑、旧系统接入、统一多后端和修复封装缺口。

生产里真正常见的是“薄适配器”：LangChain 只负责调用，复杂数据治理放在自己的服务层。不要把所有数据库运维逻辑都塞进一个 VectorStore 类里。

现在是否抛弃

没有抛弃。

LangChain 当前仍然以 VectorStore 和 Retriever 作为重要抽象。官方文档也仍然强调向量库的统一接口：写入、删除、相似度搜索、metadata 过滤。

需要抛弃的是旧认知：只要继承 VectorStore，实现三五个方法就算完成。现在更应该从生产能力看：接口完整性、幂等、删除、过滤、分数、评估、观测和回滚。

最新生产如何实现

当前更推荐这样落地：

第一，先判断目标：完整向量库接入用 VectorStore，只读检索服务接入用 BaseRetriever。

第二，适配器必须强制稳定 ids，不接受生产写入自动生成不可追踪 id。

第三，Embedding 由入库链路统一生成，模型、维度、chunk 策略写入 manifest。

第四，metadata filter 和权限条件必须下推到后端，并由服务端认证态生成。

第五，similarity_search_with_score 必须说明分数含义，必要时提供归一化 relevance score。

第六，批量写入、删除、重建、审计走后台任务，不放在在线问答请求里。

第七，用同一套评估集验证自定义适配器和成熟向量库之间的召回差异。

一个更接近生产的查询入口：

def search_knowledge(vector_store: VectorStore, query: str, auth_context, knowledge_base_id: str):
    if knowledge_base_id not in auth_context.allowed_knowledge_base_ids:
        raise PermissionError("knowledge base not allowed")

    filters = {
        "tenant_id": auth_context.tenant_id,
        "knowledge_base_id": knowledge_base_id,
        "status": "published",
    }

    return vector_store.similarity_search_with_score(
        query=query,
        k=5,
        filter=filters,
    )

这段代码表达的是边界：权限先判断，过滤服务端生成，向量库只接收已经收敛过的检索条件。真实上线时还要记录 trace、做错误降级、接入评估回放。

总结

自定义向量数据库适配器的价值，不是为了重复造一个 FAISS 或 Pinecone，而是把内部检索能力、特殊业务约束和 LangChain 的 RAG 生态接起来。

写得好的适配器很薄：接口清晰、行为稳定、分数明确、过滤可靠、id 可追踪、错误可观测。写得差的适配器会变成所有检索、权限、写入、运维问题的混合大泥团。

参考：

• LangChain Vector Store Integrations (opens new window)
• LangChain VectorStore API Reference (opens new window)
• LangChain BaseRetriever API Reference (opens new window)