向量存储是RAG解决方案的核心，目前市面上由很多向量存储产品，由免费开源的，也有商业闭源的；有本地部署的，也有完全云托管的；有传统数据库产品推出的针对向量存储的扩展，也有新势力专门针对向量存储设计的新产品。基于所有的VectorStore都提供了对应的检索器，后者可以以一个Runnable对象的形式作为LCEL链上的一环。

1. InMemoryVectorStore

InMemoryVectorStore是一个轻量级、完全基于内存的向量存储实现。它主要用于快速原型开发、测试环境或处理不需要持久化且数据量较小的临时检索任务。它的核心特性包括：

• 非持久化：数据存储在当前运行进程的内存中。一旦程序停止或重启，所有存储的向量和文档都会丢失;
• 无需配置：不像Pinecone或Milvus需要设置API密钥或启动Docker容器，它是开箱即用的，依赖极少;
• 速度极快：由于不涉及网络请求和磁盘I/O，对于小规模数据集（如几百到几千个文档）的检索速度非常快;
• 易于调试：是学习RAG (检索增强生成) 流程的最理想工具;

适合如下的典型场景：

• 快速原型设计：在决定使用哪种大型向量数据库之前，先验证RAG管道的逻辑;
• 单元测试：在自动化测试中模拟向量搜索行为，无需搭建复杂的基础设施;
• 临时上下文检索：例如处理单个长文档或用户上传的临时文件，任务完成后即可释放内存;

InMemoryVectorStore的创建很方便，只需要指定Embeddings调用构造函数就行。InMemoryVectorStore采用余弦相似度算法，而且similarity_search_with_score/asimilarity_search_with_score方法返回的不是两个向量的余弦距离，而是归一化的余弦相似度。由于_select_relevance_score_fn方法并未重写，所以similarity_search_with_relevance_scores/asimilarity_search_with_relevance_scores方法在被调用时会抛出异常。

class InMemoryVectorStore(VectorStore):
    def __init__(self, embedding: Embeddings) -> None:

如下的代码演示了针对InMemoryVectorStore的简单使用：

from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore

load_dotenv()

embedding=OpenAIEmbeddings(model="text-embedding-3-small")
texts=[
        "I love eating spicy Sichuan food.",
        "My favorite sport is swimming."]
query = "What are my food preferences?"

store = InMemoryVectorStore.from_texts(
    texts=texts,
    embedding=embedding,
    metadatas=[{"category": "food"}, {"category": "sport"}],
    similarity = "euclidean"
)
assert store.similarity_search(query=query,k=1)[0].page_content == texts[0]

2. 本地轻量级方案（适合中小型项目）

如果您希望数据能持久化到本地文件，而不仅仅存在于内存中，以下是首选：

• Chroma:

  • 特点：目前最受LangChain开发者欢迎的本地数据库。它开源且对 AI 极其友好，支持多模态数据；
  • 优势：安装简单（pip install chromadb），支持将数据持久化到本地磁盘，且能够方便地切换到分布式云端模式；

• FAISS (Facebook AI Similarity Search):

  • 特点：由Meta开发的高性能相似度搜索库，严格来说它是一个库而非数据库；
  • 优势：搜索速度极快，特别是在处理数千万级向量时表现卓越。它支持多种索引类型（如HNSW,IVF），并可利用GPU加速；

3. 全托管云端方案（适合生产环境/免运维）

对于不想管理服务器或需要跨地域扩展的团队：

• Pinecone:

  • 特点：目前市场上最成熟的商业全托管向量数据库。
  • 优势：Serverless 架构，完全不需要配置基础设施。其毫秒级的检索性能和高度的稳定性使其成为企业级 RAG 应用的首选。

• Zilliz Cloud (Milvus 托管版):

  • 特点：基于开源 Milvus 提供的云服务，针对大规模向量检索做了极致优化。

4. 开源企业级/分布式方案（适合大规模自研）

如果您需要处理十亿级数据或有严格的私有化部署需求：

• Milvus:

  • 特点：国产开源之光，专为云原生设计的分布式向量数据库。
  • 优势：支持百亿级向量规模，具备极强的水平扩展能力，并提供丰富的索引策略和混合搜索（向量 + 标量过滤）功能。

• Qdrant:

  • 特点：使用 Rust 编写，以高性能和低延迟著称。
  • 优势：在元数据过滤（Metadata Filtering）方面表现极其出色，适合需要复杂业务逻辑筛选的检索场景。

• Weaviate:

  • 特点：支持混合搜索（Dense + Sparse）和 GraphQL 查询，能很好地处理结构化与非结构化数据的关联。

5. 传统数据库扩展

如果您现有的架构中已经在使用关系型数据库：

• pgvector (PostgreSQL): 这是目前开发者最常用的方案，让最强大的开源关系型数据库具备了向量检索能力

  • 特点：通过插件让PostgreSQL具备向量存储能力。支持精确搜索（L2、余弦、点积）和模糊搜索（IVFFlat、HNSW 索引）；
  • 优势：无需引入新的组件，可以直接在现有的SQL环境中进行向量检索，极大地降低了系统复杂度；

• RediSearch（Redis）：Redis通过Stack 模块支持向量存储和检索

  • 特点：数据存储在内存中，检索延迟极低；
  • 优势：速度极快，支持 HNSW 算法。适合需要高并发、低延迟的场景（如实时推荐、毫秒级语义匹配）；

• OpenSearch (Elasticsearch)：作为传统的全文搜索引擎，它们很早就引入了dense_vector类型

  • 特点：将传统关键词搜索 (BM25) 与向量搜索 (k-NN) 完美融合（即“混合搜索”）；
  • 优势：处理大规模文本数据的经验丰富，支持复杂的过滤逻辑和分布式集群扩展；

• Atlas Vector Search（MongoDB）：MongoDB在其云服务Atlas中原生集成了向量搜索

  • 特点：基于 Lucene 分级索引实现；
  • 优势：适合 JSON 文档型数据结构。如果你的数据本身就在 MongoDB 里，不需要为了向量检索再迁移数据；

• ClickHouse：作为高性能的列式数据库（OLAP），ClickHouse 引入了向量索引支持

  • 特点：极其擅长大规模数据的批量插入和高吞吐查询；
  • 优势：适合在处理海量日志或分析型数据时，顺便进行向量比对；

6. 检索器

被定义成Runnable的检索器使数据检索也成为了LCEL链上的一环。绝大部分VectorStore类型都提供了as_rereiver方法生分成基于自身的检索器对象。

6.1 BaseRetriever

所有的检索器类型都继承自如下这个BaseRetriever抽象类，它继承自RunnableSerializable，其输入和输出类型为RetrieverInput和RetrieverOutput，分别表示作为输入的查询文本和作为输出的Document列表。BaseRetriever将具体的检索操作利用抽象方法_get_relevant_documents下放给子类，_aget_relevant_documents默认会以Crorutine的形式调用此方法。invoke和ainvoke方法最终会调用这两个方法返回检索结果。

class BaseRetriever(RunnableSerializable[RetrieverInput, RetrieverOutput], ABC):
    tags: list[str] | None = None
    metadata: dict[str, Any] | None = None

    @override
    def invoke(
        self, input: str, config: RunnableConfig | None = None, **kwargs: Any
    ) -> list[Document]

    @override
    async def ainvoke(
        self,
        input: str,
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> list[Document]

    @abstractmethod
    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> list[Document]

    async def _aget_relevant_documents(
        self, query: str, *, run_manager: AsyncCallbackManagerForRetrieverRun
    ) -> list[Document]

RetrieverInput = str
RetrieverOutput = list[Document]

_get_relevant_documents/_aget_relevant_documents方法的run_manager返回一个CallbackManagerForRetrieverRun/AsyncCallbackManagerForRetrieverRun, 这两个对象是LangChain框架中专门用于追踪、监控）和调试的“事件调度器”。它们是LangChain可观测性的基石。有了它们，我们才能在LangSmith仪表盘上清晰地看到每一条检索到的Document及其背后的分数和耗时。

6.2 VectorStoreRetriever

VectorStore一般会利用as_retriever方法返回一个VectorStoreRetriever对象。VectorStoreRetriever是对一个VectorStore对象的封装，并针对此向量存储实施检索。

class VectorStoreRetriever(BaseRetriever):
    vectorstore: VectorStore
    search_type: str = "similarity"
    search_kwargs: dict = Field(default_factory=dict)

    allowed_search_types: ClassVar[Collection[str]] = (
        "similarity",
        "similarity_score_threshold",
        "mmr",
    )

    @override
    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun, **kwargs: Any
    ) -> list[Document]:
        kwargs_ = self.search_kwargs | kwargs
        if self.search_type == "similarity":
            docs = self.vectorstore.similarity_search(query, **kwargs_)
        elif self.search_type == "similarity_score_threshold":
            docs_and_similarities = (
                self.vectorstore.similarity_search_with_relevance_scores(
                    query, **kwargs_
                )
            )
            docs = [doc for doc, _ in docs_and_similarities]
        elif self.search_type == "mmr":
            docs = self.vectorstore.max_marginal_relevance_search(query, **kwargs_)
        else:
            msg = f"search_type of {self.search_type} not allowed."
            raise ValueError(msg)
        return docs

    @override
    async def _aget_relevant_documents(
        self,
        query: str,
        *,
        run_manager: AsyncCallbackManagerForRetrieverRun,
        **kwargs: Any,
    ) -> list[Document]

    def add_documents(self, documents: list[Document], **kwargs: Any) -> list[str]:
        return self.vectorstore.add_documents(documents, **kwargs)

    async def aadd_documents(
        self, documents: list[Document], **kwargs: Any
    ) -> list[str]:
        return await self.vectorstore.aadd_documents(documents, **kwargs)

VectorStoreRetriever提供了三个实例字段和一个静态字段：

• vectorstore：实施检索任务的向量存储；
• search_type：采用的相似度算法；
• search_kwargs：为具体查询方法提供的关键字参数；
• allowed_search_types：可用相似度算法白名单，作为一个Pydantic模型，会验证search_type字段的值在此名单中；

VectorStoreRetriever实现了_get_relevant_documents方法，并根据search_type字段针对相似度算法的设置调用对应的方法返回检索结果的Document列表。除此之外，它还提供了add_documents/aadd_documents方法向存储中添加数据。

在如下的程序中， 我们调用InMemoryVectorStore对象的as_retriever方法，并利用它返回的VectorStoreRetriever对象以Runnable对象的形式实施检索。

from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore

load_dotenv()

embedding=OpenAIEmbeddings(model="text-embedding-3-small")
texts=[
        "I love eating spicy Sichuan food.",
        "My favorite sport is swimming."]
query = "What are my food preferences?"

store = InMemoryVectorStore.from_texts(
    texts=texts,
    embedding=embedding,
    metadatas=[{"category": "food"}, {"category": "sport"}],
    similarity = "euclidean"
)

retriever = store.as_retriever()
documents = retriever.invoke(query,k=2)
assert len(documents) == 2
assert documents[0].page_content == texts[0]