LangChain 设计与实现

第10章向量存储与检索器

作者杨艺韬 · 14,775 字 · 发布于 2026-04-05 · 更新于 2026-09-25

在 RAG 系统的架构中，向量存储（VectorStore）和检索器（Retriever）是连接知识库与语言模型的关键桥梁。前者负责将文档嵌入为向量并高效存储，后者负责根据查询从知识库中检索最相关的文档。LangChain 为这两个层次分别构建了精心设计的抽象，使得开发者可以在不改变应用代码的前提下，轻松切换底层的向量数据库实现。

本章将从 Embeddings 接口开始，逐步展开 VectorStore 的核心抽象和 InMemoryVectorStore 的参考实现，深入分析 BaseRetriever 的 Runnable 集成设计，以及 VectorStoreRetriever 的适配器模式。最后，我们将探讨多种高级检索策略的实现思路。

本章要点

理解 Embeddings 接口的 embed_documents / embed_query 双方法设计
掌握 VectorStore 抽象的核心方法族：similarity_search / MMR / relevance_scores
深入理解 InMemoryVectorStore 的完整实现，包括余弦相似度和 MMR 算法
理解 BaseRetriever 的 Runnable 协议集成与回调机制
掌握 VectorStoreRetriever 的适配器模式与搜索策略选择
了解高级检索器（ParentDocumentRetriever / MultiVectorRetriever / ContextualCompressionRetriever / MergerRetriever）的设计思路

10.1 Embeddings：向量化接口

Embeddings 是向量存储系统的前置依赖 -- 它负责将文本转换为高维向量表示。

# langchain_core/embeddings/embeddings.py
class Embeddings(ABC):
    @abstractmethod
    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        """Embed search docs."""

    @abstractmethod
    def embed_query(self, text: str) -> list[float]:
        """Embed query text."""

    async def aembed_documents(self, texts: list[str]) -> list[list[float]]:
        return await run_in_executor(None, self.embed_documents, texts)

    async def aembed_query(self, text: str) -> list[float]:
        return await run_in_executor(None, self.embed_query, text)

10.1.1 为什么区分 embed_documents 和 embed_query？

这个双方法设计看似冗余，实际上有深刻的技术原因。某些嵌入模型（如 Google 的 embedding-001、BGE 系列模型）会根据文本的角色使用不同的编码策略：

文档嵌入：优化为捕捉文档的全局语义，可能使用更长的上下文窗口
查询嵌入：优化为捕捉用户意图，可能添加特殊的前缀指令

大多数模型（如 OpenAI 的 text-embedding-3-small）的两个方法实现相同，但这种接口分离确保了在需要区分时有扩展空间。

flowchart LR
    subgraph 文档索引阶段
        A[文档列表] --> B[embed_documents]
        B --> C[向量列表]
        C --> D[存入 VectorStore]
    end
    subgraph 查询检索阶段
        E[用户查询] --> F[embed_query]
        F --> G[查询向量]
        G --> H[VectorStore 相似搜索]
        H --> I[最相关文档]
    end

10.1.2 异步接口的默认实现

aembed_documents 和 aembed_query 的默认实现使用 run_in_executor 将同步方法包装到线程池中。这遵循了 LangChain 的一贯模式：提供同步的基础实现，通过线程池桥接异步接口，同时允许子类覆盖为原生异步实现。

10.2 VectorStore：向量存储抽象

VectorStore 是 LangChain 中体量最大的抽象类之一，它定义了向量存储系统的完整接口。

# langchain_core/vectorstores/base.py
class VectorStore(ABC):
    # 写入接口
    def add_texts(self, texts, metadatas=None, *, ids=None, **kwargs) -> list[str]: ...
    def add_documents(self, documents, **kwargs) -> list[str]: ...
    def delete(self, ids=None, **kwargs) -> bool | None: ...
    def get_by_ids(self, ids: Sequence[str], /) -> list[Document]: ...

    # 搜索接口
    @abstractmethod
    def similarity_search(self, query, k=4, **kwargs) -> list[Document]: ...
    def similarity_search_with_score(self, *args, **kwargs) -> list[tuple[Document, float]]: ...
    def similarity_search_with_relevance_scores(self, query, k=4, **kwargs) -> list[tuple[Document, float]]: ...
    def max_marginal_relevance_search(self, query, k=4, fetch_k=20, lambda_mult=0.5, **kwargs) -> list[Document]: ...

    # 构造接口
    @classmethod
    @abstractmethod
    def from_texts(cls: type[VST], texts, embedding, metadatas=None, *, ids=None, **kwargs) -> VST: ...
    def as_retriever(self, **kwargs) -> VectorStoreRetriever: ...

classDiagram
    class VectorStore {
        <<abstract>>
        +embeddings: Embeddings
        +add_texts(texts, metadatas, ids) list~str~
        +add_documents(documents) list~str~
        +delete(ids) bool
        +get_by_ids(ids) list~Document~
        +similarity_search(query, k) list~Document~*
        +similarity_search_with_score() list~tuple~
        +similarity_search_with_relevance_scores(query, k) list~tuple~
        +max_marginal_relevance_search(query, k, fetch_k, lambda_mult) list~Document~
        +search(query, search_type) list~Document~
        +from_texts(texts, embedding) Self*
        +from_documents(documents, embedding) Self
        +as_retriever() VectorStoreRetriever
        #_select_relevance_score_fn() Callable
        #_euclidean_relevance_score_fn(distance) float$
        #_cosine_relevance_score_fn(distance) float$
        #_max_inner_product_relevance_score_fn(distance) float$
    }
    class InMemoryVectorStore {
        +store: dict
        +embedding: Embeddings
        +similarity_search(query, k) list~Document~
        +max_marginal_relevance_search(query, k) list~Document~
        +dump(path) void
        +load(path, embedding) InMemoryVectorStore$
    }
    VectorStore <|-- InMemoryVectorStore

10.2.1 写入接口的双路径设计

VectorStore 的写入接口存在 add_texts 和 add_documents 两个方法，它们之间有一个精巧的互相委托关系：

# 简化：省略了 metadatas 长度校验、metadatas_/ids_ 迭代器的构造，以及把 ids 回传给 kwargs 等细节
def add_texts(self, texts, metadatas=None, *, ids=None, **kwargs):
    if type(self).add_documents != VectorStore.add_documents:
        # 子类实现了 add_documents，委托给它
        docs = [Document(id=id_, page_content=text, metadata=metadata_)
                for text, metadata_, id_ in zip(texts, metadatas_, ids_, strict=False)]
        return self.add_documents(docs, **kwargs)
    raise NotImplementedError(...)

def add_documents(self, documents, **kwargs):
    if type(self).add_texts != VectorStore.add_texts:
        # 子类实现了 add_texts，委托给它（kwargs 里没有 ids 时，源码还会把 doc.id 收集进 kwargs["ids"]）
        texts = [doc.page_content for doc in documents]
        metadatas = [doc.metadata for doc in documents]
        return self.add_texts(texts, metadatas, **kwargs)
    raise NotImplementedError(...)

这种设计允许子类只实现其中一个方法，另一个会自动适配。通过 type(self).method != VectorStore.method 检测子类是否覆盖了特定方法，避免了无限递归。

10.2.2 搜索方法族

VectorStore 提供了三种搜索策略：

方法	返回类型	特点
`similarity_search`	`list[Document]`	最基础的相似度搜索
`similarity_search_with_relevance_scores`	`list[tuple[Document, float]]`	返回归一化到 [0,1] 的相关性分数
`max_marginal_relevance_search`	`list[Document]`	MMR 算法，平衡相关性和多样性

分数阈值过滤

similarity_search_with_relevance_scores 支持 score_threshold 参数：

# libs/core/langchain_core/vectorstores/base.py:506（签名在源码里是多行带类型）
def similarity_search_with_relevance_scores(self, query, k=4, **kwargs):
    score_threshold = kwargs.pop("score_threshold", None)
    docs_and_similarities = self._similarity_search_with_relevance_scores(query, k=k, **kwargs)
    # （省略：分数超出 [0, 1] 时 warnings.warn；阈值过滤后为空时 logger.warning）

    if score_threshold is not None:
        docs_and_similarities = [
            (doc, similarity)
            for doc, similarity in docs_and_similarities
            if similarity >= score_threshold
        ]
    return docs_and_similarities

这个过滤机制使得检索系统可以动态控制结果质量 -- 当没有足够相关的文档时，返回空列表优于返回不相关的结果。

10.2.3 as_retriever：从存储到检索的桥梁

def as_retriever(self, **kwargs) -> VectorStoreRetriever:
    tags = kwargs.pop("tags", None) or [*self._get_retriever_tags()]
    return VectorStoreRetriever(vectorstore=self, tags=tags, **kwargs)

这个方法是适配器模式的经典应用 -- 它将 VectorStore 包装为符合 BaseRetriever 接口的对象，使其能够直接用于 LCEL 链。_get_retriever_tags 自动生成包含 VectorStore 类名和 Embeddings 类名的标签，用于追踪和调试。

10.3 InMemoryVectorStore：参考实现

InMemoryVectorStore 是 LangChain 内置的基于内存的向量存储实现。虽然不适用于生产环境的大规模数据，但它作为参考实现完整展示了 VectorStore 接口的正确用法。

10.3.1 数据存储结构

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

每条记录存储为一个字典：

self.store[doc_id] = {
    "id": doc_id,
    "vector": vector,      # list[float] - 嵌入向量
    "text": doc.page_content,
    "metadata": doc.metadata,
}

10.3.2 余弦相似度搜索

核心搜索方法使用 numpy 实现的余弦相似度：

def _similarity_search_with_score_by_vector(self, embedding, k=4, filter=None):
    docs = list(self.store.values())
    if filter is not None:
        docs = [doc for doc in docs
                if filter(Document(id=doc["id"], page_content=doc["text"], metadata=doc["metadata"]))]

    if not docs:
        return []

    similarity = cosine_similarity([embedding], [doc["vector"] for doc in docs])[0]
    top_k_idx = similarity.argsort()[::-1][:k]

    return [
        (Document(id=doc_dict["id"], page_content=doc_dict["text"], metadata=doc_dict["metadata"]),
         float(similarity[idx].item()),
         doc_dict["vector"])
        for idx in top_k_idx
        if (doc_dict := docs[idx])
    ]

这段代码的执行流程：

如果提供了 filter 函数，先过滤文档
计算查询向量与所有文档向量的余弦相似度
按相似度降序排序，取 top-k

filter 参数接受一个 Callable[[Document], bool]，允许在向量搜索的基础上进行元数据过滤。这比在搜索后再过滤更高效，因为它减少了需要排序的候选集大小。

flowchart TD
    A[查询向量] --> C{有 filter?}
    C -->|是| D[先按元数据过滤]
    D --> E[对过滤后的文档计算相似度]
    C -->|否| E[对所有文档计算相似度]
    E --> F[argsort 降序排序]
    F --> G[取 top-k 个索引]
    G --> H[构造 Document + score 返回]

10.3.3 余弦相似度的实现

# langchain_core/vectorstores/utils.py（节选：省略了未装 numpy 抛 ImportError、空输入返回、
# NaN/Inf 告警与列数不一致抛 ValueError 等前置检查）
def _cosine_similarity(x: Matrix, y: Matrix) -> np.ndarray:
    x = np.array(x)
    y = np.array(y)
    if not _HAS_SIMSIMD:
        x_norm = np.linalg.norm(x, axis=1)
        y_norm = np.linalg.norm(y, axis=1)
        with np.errstate(divide="ignore", invalid="ignore"):
            similarity = np.dot(x, y.T) / np.outer(x_norm, y_norm)
        if np.isnan(similarity).all():
            raise ValueError("NaN values found, please remove the NaN values and try again")
        similarity[np.isnan(similarity) | np.isinf(similarity)] = 0.0
        return similarity
    # 使用 simsimd 加速
    x = np.array(x, dtype=np.float32)
    y = np.array(y, dtype=np.float32)
    return 1 - np.array(simd.cdist(x, y, metric="cosine"))

实现中有几个值得注意的细节：

simsimd 加速：只要装了 simsimd 就优先走它——一个 SIMD 优化的距离计算库；没装时 logger.debug 提示一句、回落到纯 numpy 路径（具体快多少取决于向量维度、批量大小和 CPU 指令集，需自行实测）
NaN/Inf 处理：对于零向量等边界情况，相似度计算会产生 NaN/Inf，统一替换为 0.0；但若结果矩阵全是 NaN，则直接抛 ValueError
安全检查：提前检测输入中的 NaN 和 Inf，发出警告

10.3.4 MMR 算法

最大边际相关性（Maximal Marginal Relevance）算法在保证相关性的同时增加结果的多样性：

# libs/core/langchain_core/vectorstores/utils.py:106（签名在源码里是多行带类型）
def maximal_marginal_relevance(query_embedding, embedding_list, lambda_mult=0.5, k=4):
    # （省略：未装 numpy 时抛 ImportError）
    if min(k, len(embedding_list)) <= 0:
        return []
    if query_embedding.ndim == 1:
        query_embedding = np.expand_dims(query_embedding, axis=0)

    similarity_to_query = _cosine_similarity(query_embedding, embedding_list)[0]
    most_similar = int(np.argmax(similarity_to_query))
    idxs = [most_similar]
    selected = np.array([embedding_list[most_similar]])

    while len(idxs) < min(k, len(embedding_list)):
        best_score = -np.inf
        idx_to_add = -1
        similarity_to_selected = _cosine_similarity(embedding_list, selected)

        for i, query_score in enumerate(similarity_to_query):
            if i in idxs:
                continue
            redundant_score = max(similarity_to_selected[i])
            equation_score = lambda_mult * query_score - (1 - lambda_mult) * redundant_score
            if equation_score > best_score:
                best_score = equation_score
                idx_to_add = i

        idxs.append(idx_to_add)
        selected = np.append(selected, [embedding_list[idx_to_add]], axis=0)
    return idxs

MMR 公式：score = lambda * sim(doc, query) - (1 - lambda) * max(sim(doc, selected_docs))

flowchart TD
    A[计算所有文档与查询的相似度] --> B[选择最相似的文档加入结果集]
    B --> C{结果集达到 k?}
    C -->|否| D[对每个未选文档计算 MMR 分数]
    D --> E["MMR = lambda * 查询相似度 - (1-lambda) * 与已选最大相似度"]
    E --> F[选择 MMR 分数最高的文档]
    F --> G[加入结果集]
    G --> C
    C -->|是| H[返回结果索引]

lambda_mult 参数控制相关性和多样性的平衡：

lambda_mult = 1.0：等同于普通相似度搜索
lambda_mult = 0.0：最大化多样性
lambda_mult = 0.5：默认值，相关性和多样性各占一半

10.3.5 持久化支持

InMemoryVectorStore 提供了简单的持久化方法：

# libs/core/langchain_core/vectorstores/in_memory.py:537
def dump(self, path: str) -> None:
    path_ = Path(path)
    path_.parent.mkdir(exist_ok=True, parents=True)
    with path_.open("w", encoding="utf-8") as f:
        json.dump(dumpd(self.store), f, indent=2)

@classmethod
def load(cls, path, embedding, **kwargs):
    with Path(path).open("r", encoding="utf-8") as f:
        store = load(json.load(f), allowed_objects=[Document])
    vectorstore = cls(embedding=embedding, **kwargs)
    vectorstore.store = store
    return vectorstore

使用 LangChain 的序列化系统（dumpd/load）确保了 Document 对象的正确序列化和反序列化。

10.4 BaseRetriever：检索器的 Runnable 抽象

BaseRetriever 是 LangChain 检索系统的核心抽象。它继承自 RunnableSerializable[str, list[Document]]，这意味着检索器接收字符串查询，返回文档列表，并且完全兼容 LCEL。

# langchain_core/retrievers.py
RetrieverInput = str
RetrieverOutput = list[Document]

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

    @abstractmethod
    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> list[Document]:
        """Get documents relevant to a query."""

    async def _aget_relevant_documents(
        self, query: str, *, run_manager: AsyncCallbackManagerForRetrieverRun
    ) -> list[Document]:
        return await run_in_executor(
            None, self._get_relevant_documents, query, run_manager=run_manager.get_sync(),
        )

10.4.1 invoke 方法的完整回调集成

BaseRetriever.invoke 的实现展示了 LangChain 回调系统的完整工作流：

# libs/core/langchain_core/retrievers.py:179（签名在源码里是多行）
def invoke(self, input: str, config: RunnableConfig | None = None, **kwargs) -> list[Document]:
    config = ensure_config(config)
    inheritable_metadata = {
        **(config.get("metadata") or {}),
        **self._get_ls_params(**kwargs),
    }
    callback_manager = CallbackManager.configure(
        config.get("callbacks"), None,
        verbose=kwargs.get("verbose", False),
        inheritable_tags=config.get("tags"),
        local_tags=self.tags,
        inheritable_metadata=inheritable_metadata,
        local_metadata=self.metadata,
    )
    run_manager = callback_manager.on_retriever_start(
        None, input, name=config.get("run_name") or self.get_name(), ...
    )
    try:
        result = self._get_relevant_documents(input, run_manager=run_manager, ...)
    except Exception as e:
        run_manager.on_retriever_error(e)
        raise
    else:
        run_manager.on_retriever_end(result)
        return result

sequenceDiagram
    participant C as Caller
    participant R as BaseRetriever
    participant CM as CallbackManager
    participant I as Implementation
    C->>R: invoke(query, config)
    R->>CM: configure(callbacks, tags, metadata)
    CM->>CM: on_retriever_start(query)
    R->>I: _get_relevant_documents(query, run_manager)
    alt 成功
        I-->>R: documents
        R->>CM: on_retriever_end(documents)
        R-->>C: documents
    else 异常
        I-->>R: Exception
        R->>CM: on_retriever_error(exception)
        R-->>C: raise Exception
    end

这段代码有几个关键设计点：

LangSmith 集成：_get_ls_params 返回 LangSmithRetrieverParams，基类只填检索器名称，VectorStoreRetriever 再补上向量存储和嵌入类的类名（§10.25），用于追踪
标签继承：通过 inheritable_tags 和 local_tags 的区分，支持标签在链式调用中的传播
错误追踪：无论成功还是失败，都会通知回调管理器，确保可观测性

10.4.2 init_subclass 的兼容性处理

def __init_subclass__(cls, **kwargs):
    super().__init_subclass__(**kwargs)
    parameters = signature(cls._get_relevant_documents).parameters
    cls._new_arg_supported = parameters.get("run_manager") is not None
    # 如果子类未实现 run_manager 参数，自动适配
    if not cls._new_arg_supported and cls._aget_relevant_documents == BaseRetriever._aget_relevant_documents:
        async def _aget_relevant_documents(self, query):
            return await run_in_executor(None, self._get_relevant_documents, query)
        cls._aget_relevant_documents = _aget_relevant_documents

这段代码处理了一个向后兼容问题：早期版本的 _get_relevant_documents 不接受 run_manager 参数。__init_subclass__ 在子类定义时自动检测其签名，如果发现旧式签名（没有 run_manager），就会自动生成一个适配版本的 _aget_relevant_documents。

10.4.3 `_expects_other_args` 与 LangSmith 名字归一化：另外两处容易漏看的子类检测

__init_subclass__ 还多做了一件前面没提的事——检测签名里有没有第三种参数（libs/core/langchain_core/retrievers.py:162-165）：

# retrievers.py:162
# If a V1 retriever broke the interface and expects additional arguments
cls._expects_other_args = (
    len(set(parameters.keys()) - {"self", "query", "run_manager"}) > 0
)

集合减法 parameters - {self, query, run_manager}、非空即 _expects_other_args = True。这个 flag 在 invoke 里有关键作用（libs/core/langchain_core/retrievers.py:220）：

kwargs_ = kwargs if self._expects_other_args else {}

"用户通过 invoke 传的 kwargs 要不要转发给 _get_relevant_documents"——取决于它自己的签名接不接。这样设计的原因：如果一个 V1 时代的 retriever 只写了 def _get_relevant_documents(self, query, run_manager)，你 invoke(query, config, foo="bar") 的 foo 不能强塞——会报 unexpected keyword argument。_expects_other_args=False 时把 kwargs 丢掉是对老 retriever 的保护。反之新式 retriever 声明 def _get_relevant_documents(self, query, run_manager, kw) 或带具体参数的、_expects_other_args=True、kwargs 透传。这是 Python 动态类型下实现"新旧接口共存"的一种手法——不需要版本号、不需要 decorator、靠 inspect.signature 在子类定义时做一次静态分析就完成。

另一处隐蔽细节——_get_ls_params 里的名字归一化（libs/core/langchain_core/retrievers.py:169-174）：

default_retriever_name = self.get_name()
if default_retriever_name.startswith("Retriever"):
    default_retriever_name = default_retriever_name[9:]
elif default_retriever_name.endswith("Retriever"):
    default_retriever_name = default_retriever_name[:-9]
default_retriever_name = default_retriever_name.lower()

如果你的 class 叫 CustomRetriever、LangSmith trace 里显示的名字是 "custom"（去 "Retriever" 后缀 + 小写）；叫 RetrieverMCP 显示成 "mcp"（去前缀）。这个处理让 LangSmith dashboard 里各种 retriever 的名字统一成类似 tag 的短名——容易聚合、容易按名字比较不同实现。

这两处和 __init_subclass__ 的 _new_arg_supported 检测合起来——都属于"在子类定义时做一次轻量静态分析、运行时零开销"的范式。Python 不像 Rust 那样能用类型系统在编译期表达接口演化，但 __init_subclass__ + inspect.signature 这组工具在 runtime 早期（类定义那一瞬间）做等价的事——让新老 retriever 在同一个进程里共存、调用方不必区分。

10.5 VectorStoreRetriever：适配器模式

VectorStoreRetriever 将 VectorStore 适配为 BaseRetriever 接口：

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",
    )

    @model_validator(mode="before")
    @classmethod
    def validate_search_type(cls, values):
        search_type = values.get("search_type", "similarity")
        if search_type not in cls.allowed_search_types:
            raise ValueError(...)
        if search_type == "similarity_score_threshold":
            score_threshold = values.get("search_kwargs", {}).get("score_threshold")
            if (score_threshold is None) or (not isinstance(score_threshold, float)):
                raise ValueError(...)  # 注意：必须是 float，写成整数 1 也会被拒
        return values

10.5.1 搜索策略路由

# VectorStoreRetriever._get_relevant_documents，libs/core/langchain_core/vectorstores/base.py:1040
def _get_relevant_documents(self, query, *, run_manager, **kwargs):
    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:
        raise ValueError(f"search_type of {self.search_type} not allowed.")
    return docs

三种搜索策略的适用场景：

flowchart TD
    A[VectorStoreRetriever] --> B{search_type?}
    B -->|similarity| C["similarity_search(query, k=4)"]
    C --> D[返回 top-k 最相似文档]
    B -->|similarity_score_threshold| E["similarity_search_with_relevance_scores(query, score_threshold=0.8)"]
    E --> F[返回分数高于阈值的文档]
    B -->|mmr| G["max_marginal_relevance_search(query, k=4, fetch_k=20, lambda_mult=0.5)"]
    G --> H[返回兼顾相关性和多样性的文档]

10.5.2 使用示例

# 基本相似度搜索
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# MMR 搜索
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5}
)

# 分数阈值过滤
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.8}
)

# 在 LCEL 链中使用（format_docs 为自定义的「文档列表 → 字符串」函数）
chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt | model | StrOutputParser()
)

注意 similarity_score_threshold 依赖后端实现 _select_relevance_score_fn（或覆写 _similarity_search_with_relevance_scores）。core 自带的 InMemoryVectorStore 两者都没实现，在它上面这样配置，构造时能通过校验，一调用 invoke 就抛 NotImplementedError（pin 版本实跑）。

10.6 高级检索策略

LangChain 在基础检索器之上构建了多种高级检索策略，每种策略都针对特定的检索质量问题提供解决方案。注意：BaseRetriever / VectorStoreRetriever 住在 langchain-core，而下面这几个高级检索器都住在 langchain-classic（libs/langchain/langchain_classic/retrievers/）——它们是 core 抽象的下游组合，不属于 core 的最小内核。

10.6.1 MultiVectorRetriever：多向量检索

MultiVectorRetriever 的核心思想是：用于搜索的向量和最终返回的文档可以是不同的。

# langchain_classic/retrievers/multi_vector.py
class MultiVectorRetriever(BaseRetriever):
    vectorstore: VectorStore        # 存储子文档向量
    docstore: BaseStore[str, Document]  # 存储父文档
    id_key: str = "doc_id"          # 子文档 metadata 中关联父文档的键
    search_type: SearchType = SearchType.similarity

    def _get_relevant_documents(self, query, *, run_manager):
        if self.search_type == SearchType.mmr:
            sub_docs = self.vectorstore.max_marginal_relevance_search(query, **self.search_kwargs)
        elif self.search_type == SearchType.similarity_score_threshold:
            sub_docs_and_similarities = self.vectorstore.similarity_search_with_relevance_scores(
                query, **self.search_kwargs)
            sub_docs = [sub_doc for sub_doc, _ in sub_docs_and_similarities]
        else:
            sub_docs = self.vectorstore.similarity_search(query, **self.search_kwargs)

        # 从子文档的 metadata 中提取父文档 ID
        ids = []
        for d in sub_docs:
            if self.id_key in d.metadata and d.metadata[self.id_key] not in ids:
                ids.append(d.metadata[self.id_key])

        # 从 docstore 中获取父文档
        docs = self.docstore.mget(ids)
        return [d for d in docs if d is not None]

flowchart TD
    subgraph 索引阶段
        A[原始文档] --> B[分割为子文档]
        B --> C[为每个子文档生成嵌入]
        C --> D[存入 VectorStore]
        A --> E[存入 DocStore]
        B --> F["子文档 metadata 记录 doc_id"]
    end
    subgraph 检索阶段
        G[查询] --> H[在 VectorStore 中搜索子文档]
        H --> I[提取唯一的 doc_id 集合]
        I --> J[从 DocStore 获取父文档]
        J --> K[返回完整父文档]
    end

10.6.2 ParentDocumentRetriever：父文档检索

ParentDocumentRetriever 继承自 MultiVectorRetriever，自动化了"小块搜索、大块返回"的过程：

# langchain_classic/retrievers/parent_document_retriever.py
class ParentDocumentRetriever(MultiVectorRetriever):
    child_splitter: TextSplitter      # 创建子文档的分割器
    parent_splitter: TextSplitter | None = None  # 创建父文档的分割器
    child_metadata_fields: Sequence[str] | None = None

    def _split_docs_for_adding(self, documents, ids=None, *, add_to_docstore=True):
        # 可选：先用 parent_splitter 分割为中等大小的父文档
        if self.parent_splitter is not None:
            documents = self.parent_splitter.split_documents(documents)
        if ids is None:
            doc_ids = [str(uuid.uuid4()) for _ in documents]   # 没传 ID 就生成 UUID
            if not add_to_docstore:
                raise ValueError("If IDs are not passed in, `add_to_docstore` MUST be True")
        else:
            if len(documents) != len(ids):
                raise ValueError("Got uneven list of documents and ids. ...")
            doc_ids = ids

        docs = []
        full_docs = []
        for i, doc in enumerate(documents):
            _id = doc_ids[i]
            # 用 child_splitter 进一步分割为子文档
            sub_docs = self.child_splitter.split_documents([doc])
            if self.child_metadata_fields is not None:
                for _doc in sub_docs:
                    _doc.metadata = {k: _doc.metadata[k] for k in self.child_metadata_fields}
            for _doc in sub_docs:
                _doc.metadata[self.id_key] = _id  # 关联父文档 ID
            docs.extend(sub_docs)
            full_docs.append((_id, doc))
        return docs, full_docs

    def add_documents(self, documents, ids=None, add_to_docstore=True, **kwargs) -> None:
        docs, full_docs = self._split_docs_for_adding(
            documents, ids, add_to_docstore=add_to_docstore)
        # 子文档存入向量库，父文档存入文档库
        self.vectorstore.add_documents(docs, **kwargs)
        if add_to_docstore:
            self.docstore.mset(full_docs)

拆分逻辑单独抽成 _split_docs_for_adding，是为了让同步的 add_documents 和异步的 aadd_documents 共用同一份切分代码——两者只在最后写入 vectorstore / docstore 的那一步分叉。

这种设计解决了 RAG 中的经典矛盾：

小块嵌入更精确地捕捉语义，提高检索准确性
大块（父文档）提供更完整的上下文，提高生成质量

10.6.3 ContextualCompressionRetriever：压缩检索

ContextualCompressionRetriever 在基础检索器的结果上应用后处理压缩：

# langchain_classic/retrievers/contextual_compression.py
class ContextualCompressionRetriever(BaseRetriever):
    base_compressor: BaseDocumentCompressor
    base_retriever: RetrieverLike

    def _get_relevant_documents(self, query, *, run_manager, **kwargs):
        docs = self.base_retriever.invoke(
            query, config={"callbacks": run_manager.get_child()}, **kwargs,
        )
        if docs:
            compressed_docs = self.base_compressor.compress_documents(
                docs, query, callbacks=run_manager.get_child(),
            )
            return list(compressed_docs)
        return []

这是装饰器模式的典型应用。base_compressor 可以是：

LLM 抽取压缩器：用 LLM 提取文档中与查询最相关的段落
重排序器（Reranker）：使用交叉编码器模型对结果重新排序
冗余过滤器：去除语义重复的文档（EmbeddingsRedundantFilter 是文档转换器而非压缩器，要放进 DocumentCompressorPipeline 里使用）

10.6.4 MergerRetriever：融合检索

MergerRetriever 将多个检索器的结果合并：

# langchain_classic/retrievers/merger_retriever.py
class MergerRetriever(BaseRetriever):
    retrievers: list[BaseRetriever]

    def merge_documents(self, query, run_manager):
        retriever_docs = [
            retriever.invoke(query, config={"callbacks": run_manager.get_child(f"retriever_{i + 1}")})
            for i, retriever in enumerate(self.retrievers)
        ]
        # 交叉合并：轮流从每个检索器取文档
        merged_documents = []
        max_docs = max(map(len, retriever_docs), default=0)
        for i in range(max_docs):
            for _retriever, doc in zip(self.retrievers, retriever_docs, strict=False):
                if i < len(doc):
                    merged_documents.append(doc[i])
        return merged_documents

flowchart TD
    Q[查询] --> R1[检索器 1: 稠密向量]
    Q --> R2[检索器 2: 稀疏 BM25]
    Q --> R3[检索器 3: 关键词]
    R1 --> M[MergerRetriever]
    R2 --> M
    R3 --> M
    M --> D["交叉合并排序"]
    D --> F["融合结果"]

交叉合并策略（round-robin）确保了每个检索器的高优先级结果都能靠前出现。异步版本 amerge_documents 用 asyncio.gather 并发调用所有检索器，总耗时从"各检索器串行相加"变成"取最慢的那个"。

10.7 LangSmith 检索追踪

BaseRetriever 集成了 LangSmith 追踪参数：

class LangSmithRetrieverParams(TypedDict, total=False):
    ls_retriever_name: str
    ls_vector_store_provider: str | None
    ls_embedding_provider: str | None
    ls_embedding_model: str | None

VectorStoreRetriever 覆盖了 _get_ls_params 方法，自动填充这些追踪参数：

# 简化：省略了 kwargs 与 search_kwargs 的合并，以及从 vectorstore.embedding 属性取类名的 elif 分支（完整代码见 §10.25）
def _get_ls_params(self, **kwargs):
    ls_params = super()._get_ls_params(**kwargs)
    ls_params["ls_vector_store_provider"] = self.vectorstore.__class__.__name__
    if self.vectorstore.embeddings:
        ls_params["ls_embedding_provider"] = self.vectorstore.embeddings.__class__.__name__
    return ls_params

这使得在 LangSmith 中可以按向量存储类和嵌入类进行过滤和分析检索性能。注意 ls_embedding_model 虽然在 TypedDict 里声明了，pin 版本的 core 与 classic 里没有任何代码填写它。

10.8 设计决策分析

VectorStore 为什么不是 Runnable？——与 BaseRetriever 不同，VectorStore 不继承 Runnable。这是因为 VectorStore 的操作语义更复杂 -- 它不仅仅是"输入查询、输出文档"，还涉及写入、删除、ID 查询等操作。as_retriever() 方法将 VectorStore 的搜索能力投射为 Runnable 接口，保持了 VectorStore 自身接口的完整性。

为什么 similarity_search 是唯一的抽象方法？——在 VectorStore 的众多搜索方法中，只有 similarity_search 和 from_texts 是抽象方法。但"非抽象"不等于"有可用的默认实现"：similarity_search_with_score、similarity_search_by_vector、max_marginal_relevance_search 在基类里都直接抛 NotImplementedError；similarity_search_with_relevance_scores 的默认实现要借助 similarity_search_with_score 和 _select_relevance_score_fn，两者都得由子类提供；真正白送的主要是异步版本（run_in_executor 包一层同步方法）、from_documents 和 as_retriever。这降低了实现新后端的门槛 -- 只需实现最基本的相似度搜索就能接入检索器，但要支持 MMR、分数阈值等策略，仍得逐个实现对应方法。

MMR 的 fetch_k 参数——MMR 搜索有一个 fetch_k 参数（默认 20），它决定了先获取多少候选文档，再从中通过 MMR 算法选取 k 个。fetch_k > k 确保了 MMR 有足够的候选集来优化多样性。fetch_k 过小会限制多样性的空间，过大会增加计算成本。

BaseRetriever 的 _new_arg_supported 机制——这是 LangChain 处理 API 演进的一个典型模式。当框架需要为已有的抽象方法添加新参数时，不能简单地修改签名（会破坏所有现有实现）。通过 __init_subclass__ 动态检测子类签名并自动适配，既保持了新 API 的功能，又不破坏旧代码。

10.9 主线小结

LangChain 的向量存储与检索器系统构建了从嵌入到检索的完整链路。Embeddings 接口通过 embed_documents/embed_query 的分离，为不同角色的文本嵌入留下了优化空间。VectorStore 抽象通过精心设计的方法族，支持从简单相似度搜索到 MMR 多样性搜索的全方位检索能力。InMemoryVectorStore 作为参考实现，完整展示了余弦相似度计算、MMR 算法、持久化等核心功能。

BaseRetriever 作为 Runnable 的子类，使得检索器天然融入 LCEL 链。其 invoke 方法中精心编排的回调管理、LangSmith 追踪参数、以及 __init_subclass__ 的兼容性处理，体现了 LangChain 在可观测性和向后兼容方面的深思熟虑。VectorStoreRetriever 通过适配器模式优雅地连接了存储层和检索层。

高级检索策略 -- MultiVectorRetriever 的"小块搜索、大块返回"、ParentDocumentRetriever 的自动化父子文档管理、ContextualCompressionRetriever 的后处理压缩、MergerRetriever 的多源融合 -- 展示了 LangChain 检索系统的丰富生态。这些组件通过统一的 BaseRetriever 接口可以自由组合，构建出适应各种复杂检索需求的解决方案。

检索质量是 RAG 系统的生命线。理解这些抽象的设计意图和实现细节，是构建高质量 RAG 应用的基础。

以下为源码实证附录。 到上一节为止是本章主线，读完已可动手。从这里开始的各节是针对langchain-core 检索器与向量存储的具体实现的独立深挖，彼此不构成阅读顺序，按需查阅即可。

10.10 源码实证：`retrievers.py` 328 行 + `libs/core/langchain_core/vectorstores/base.py` 1111 行

本章前面讲了抽象——真实代码就这么多：

文件	行	内容
`libs/core/langchain_core/retrievers.py`	328	BaseRetriever + `LangSmithRetrieverParams` + 几个类型别名
`libs/core/langchain_core/vectorstores/base.py`	1111	VectorStore + VectorStoreRetriever
`libs/core/langchain_core/vectorstores/in_memory.py`	546	参考实现 InMemoryVectorStore
`libs/core/langchain_core/vectorstores/utils.py`	157	MMR 算法 + 相似度 helpers

总计 2142 行——撑起整个 LangChain 的检索抽象。

10.11 `__init_subclass__` 的"向后兼容"黑魔法

§10.8 提了 _new_arg_supported 机制——源码 libs/core/langchain_core/retrievers.py:145-160 的实现值得逐行拆：

    @override
    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        parameters = signature(cls._get_relevant_documents).parameters
        cls._new_arg_supported = parameters.get("run_manager") is not None
        if (
            not cls._new_arg_supported
            and cls._aget_relevant_documents == BaseRetriever._aget_relevant_documents
        ):
            # we need to tolerate no run_manager in _aget_relevant_documents signature
            async def _aget_relevant_documents(
                self: Self, query: str
            ) -> list[Document]:
                return await run_in_executor(None, self._get_relevant_documents, query)  # type: ignore[call-arg]

            cls._aget_relevant_documents = _aget_relevant_documents  # type: ignore[assignment]

四段魔法——

1——__init_subclass__ 是 Python 3.6+ 的 hook——子类"被定义时" 触发——不是实例化时——所以是类属性初始化。

2——signature(cls._get_relevant_documents).parameters 用 inspect 模块反射子类方法签名——判断是否有 run_manager 参数——新 API 的标记。

3——cls._new_arg_supported 记在类上——运行 invoke 时用这个 flag 选择调用路径。

4——没新 API 的老子类——自动 monkey-patch 一个 async 版本——用 run_in_executor 在线程池里跑同步方法——老代码不改也能跑 async。

这十几行代码（retrievers.py:145-160）——承载了 LangChain 近三年的 API 演进（这套检测在 2023 年 7 月的 v0.0.230 里已经出现）——旧代码零迁移。

10.12 `VectorStoreRetriever.allowed_search_types` 的三档

libs/core/langchain_core/vectorstores/base.py:976-981——三种 search_type：

similarity——标准相似度 top-k（度量由后端决定，InMemoryVectorStore 用余弦）
similarity_score_threshold——只返回 score ≥ threshold 的（后端须支持相关度分数，见 §10.5.2）
mmr——Maximum Marginal Relevance——兼顾相关性 + 多样性

对应的 validate_search_type 校验器（base.py:986-1016，一个 @model_validator(mode="before")）——选 similarity_score_threshold 时强制 search_kwargs 里带一个 float 类型的 score_threshold，防止"忘配 threshold"的典型踩坑。注意它只校验类型：错误信息里虽然写着 0~1，代码并不检查取值范围。

10.13 MMR 算法的"lambda"参数——多样性 vs 相关性

MMR（Carbonell & Goldstein, 1998）的公式——

MMR = argmax [λ * Sim(q, d) - (1-λ) * max_j Sim(d, d_j)]

λ 取值——

λ = 1.0：纯相关性（等同于标准 similarity_search）
λ = 0.5：平衡（LangChain 默认）
λ = 0.0：纯多样性

取值方向（下面是经验性的调参起点、不是实测结论，具体要按自己的评测集扫一遍 λ）——

摘要类召回——偏大的 λ（相关性优先、顺带压掉信息重复）
代码片段检索——偏小的 λ（多样性优先、覆盖不同语言/库）
一般问答召回——从默认的 0.5 起步

fetch_k 应当 > k——fetch_k == k 时 MMR 没有多余候选可挑，结果就是普通 top-k 那几篇（只是顺序可能被重排）——源码默认是 fetch_k=20、k=4，但代码本身不做这个校验、传反了不会报错：以 InMemoryVectorStore 为例，k=5, fetch_k=3 只返回 3 篇（pin 版本实跑）。

10.14 InMemoryVectorStore 546 行：最小参考实现

它的算法就是暴力扫描——每次检索都对 store 里全部向量算一遍余弦相似度再 argsort，没有任何索引结构。所以相似度计算的开销随文档数和向量维度线性增长，再加一次全量排序——数据量一上来就得换 FAISS / Chroma / Pinecone 这类带 ANN 索引的实现（具体拐点在哪，取决于机器和维度，自己压一遍才算数）。

何时值得用 InMemory——

原型验证——不想装 vector db
小数据集 RAG——个人知识库、量级不大
单元测试——mock 掉真实 vector db
CI 集成测试——快速、无外部依赖

生产规模——必须换真实 vector db——接口一致，但构造参数和 filter 等后端特有参数的写法各不相同（InMemoryVectorStore 的 filter 是 Callable[[Document], bool]，传字典会抛 TypeError: 'dict' object is not callable；多数数据库后端用字典），迁移时要一并改。

10.15 `from_texts` 的工厂方法——为什么它是 @classmethod

libs/core/langchain_core/vectorstores/base.py:846-868 的 from_texts——abstract class method：

@classmethod
@abstractmethod
def from_texts(cls, texts, embedding, metadatas=None, ...) -> VST: ...

为什么要 classmethod 工厂——

VectorStore 的构造过程复杂——embedding 模型初始化 + 底层客户端连接 + schema 创建
不同 VectorStore 的 __init__ 参数完全不同（Pinecone 要 API key、QdrantVectorStore 要 client 与 collection_name、Chroma 要 persist_directory）
from_texts 把复杂构造标准化——用户写 Chroma.from_texts([...], embedding)（langchain_chroma）就建好 store 并插入了初始文档

10.16 ContextualCompressionRetriever 的"后处理"流水线

§10.6.3 提到 ContextualCompressionRetriever——它是一个"retriever + compressor" 组合：

流水线——

base_retriever 召回 k=20 文档（原文可能很长）
compressor 逐个 doc 压缩——去掉无关段落——只留精华
最终塞给 LLM 的至多 20 个 doc（抽取结果为空的会被丢掉）、每个都被压缩过——总 token 数显著下降（降幅完全取决于原文冗余度和 compressor 策略）

三类 compressor（均在 langchain_classic/retrievers/document_compressors/）——

LLMChainExtractor（chain_extract.py）——LLM 抽取相关句
LLMChainFilter（chain_filter.py）——LLM 判断"这个 doc 整体相关吗"（过滤级）
EmbeddingsFilter（embeddings_filter.py）——向量相似度二次过滤（不用 LLM，便宜；pin 版本里它的相似度函数从 langchain_community 导入，没装该包时构造即抛 ImportError）

组合——先 EmbeddingsFilter 粗过滤、再 LLMChainExtractor 精抽取——成本和质量平衡。

这一模式和《Harness Engineering》专栏第 11 章 §11.4讨论的"工具结果截断 / 摘要"同源。

10.17 多源检索的两种合并：round-robin 与 Reciprocal Rank Fusion

把多个 retriever 的结果合并成一个，langchain-classic 里有两个不同的实现、两种不同的策略，别搞混：

MergerRetriever（merger_retriever.py）——round-robin 交叉合并，就是 §10.6.4 那段代码：按名次一轮一轮地从每个 retriever 各取一篇拼起来，retriever_1[0], retriever_2[0], ..., retriever_1[1], retriever_2[1], ...。它完全不看分数、只看名次顺序，所以也就不存在分数标准化的问题；代价是没法表达"某个 retriever 更可信"，而且不去重——同一篇文档被两路都召回，就会在结果里出现两次（pin 版本实跑）。EnsembleRetriever 则会按 page_content（或 id_key 指定的 metadata 字段）去重。

EnsembleRetriever（ensemble.py）——加权 Reciprocal Rank Fusion (RRF)：

score(d) = sum over i: weights[i] * 1 / (c + rank_i(d))

源码里的常数是 c: int = 60（libs/langchain/langchain_classic/retrievers/ensemble.py:70）
weights 默认平均分配（1 / n_retrievers），可以给某路召回加权
每个 retriever 给 d 一个 rank（第几名）
RRF 的优点——不需要分数标准化——每个 retriever 的分数尺度可以完全不同（原始论文见源码 docstring 里引的 Cormack et al., SIGIR 2009，https://plg.uwaterloo.ca/~gvcormac/cormacksigir09-rrf.pdf）

"hybrid retrieval"用的是后者——dense vector + sparse keyword (BM25) 两路召回、RRF 合并。它常常优于单一 retriever，但提升幅度取决于语料、查询分布、评测集和合并参数，不能脱离具体 benchmark 给固定百分比。

10.18 `ParentDocumentRetriever`：大 index 小、大 retrieve 大

经典场景——你的文档是 100 页 PDF、chunk 成 500 token/chunk 存向量库——检索后只返回 500 token 太短、LLM 没上下文——但如果存 5000 token/chunk、embedding 又"太糊"、精度下降。

ParentDocumentRetriever 的解法——

"子 chunk" 入 vector store——小、精（embedding 准）
"父 chunk" 入 docstore（key-value 存储）——大、含上下文
检索时——先用子 chunk 找匹配、再用它 metadata 里的 doc_id（id_key 的默认值）把父 chunk 拉回来——精度 + 上下文兼得

这是 RAG 里处理"chunk size 权衡"的常用手法——生产级 RAG 值得试这个 retriever。

10.19 Async 支持的陷阱：同步阻塞破坏 event loop

retrievers.py 里的 _aget_relevant_documents——如果用户懒得实现、默认会调 run_in_executor 把同步版塞进线程池。

优点——老同步代码自动 async 可用

陷阱——如果 _get_relevant_documents 里有 CPU-bound 操作（比如跑 embedding 模型）——会占用 executor 的线程池——高并发时 executor 耗尽、新请求排队

最佳实践——

I/O-bound retriever（访问 DB / API）——必须实现 _aget_relevant_documents、用 async I/O 原语（aiohttp / asyncpg）
CPU-bound retriever（本地 embedding）——可以用 executor fallback、但 pool size 要调大或用 multiprocessing pool

10.20 本章的四层升华

Retriever 抽象的分层：

最底层：Embeddings (embed_query / embed_documents)——把文本变向量
第二层：VectorStore (similarity_search / MMR)——向量相似度检索
第三层：BaseRetriever (Runnable)——把 VectorStore 的能力通过 Runnable 暴露
第四层：高级 Retriever (MultiVector / Parent / Compression / Merger)——组合多个 Retriever 构建复杂检索流程

四层关系——底层换 provider（OpenAI → Cohere embedding）不影响上层、换 vector store（FAISS → Pinecone）不影响 Retriever 接口、换 Retriever 不影响 chain 的下游消费。

这就是"分层抽象"的工程红利——每层可替换、总体不崩。

本专栏第 3 章讲的 Runnable + 本章讲的 Retriever——在第四层自然汇合——Retriever 是 Runnable、可以{"context": retriever, "question": RunnablePassthrough()} | prompt | llm | parser 拼出完整 RAG chain（直接 retriever | llm 不行：聊天模型不认 Document 列表，pin 版本实跑抛 NotImplementedError: Unsupported message type）。

10.21 实战：十来行代码搭起最小 RAG

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

store = InMemoryVectorStore(embedding=OpenAIEmbeddings())
store.add_texts(["RAG stands for ...", "LangChain is ...", "..."])
prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer based on docs:\n{context}"),
    ("human", "{question}")
])
chain = {"context": store.as_retriever(), "question": RunnablePassthrough()} | prompt | ChatOpenAI() | StrOutputParser()
answer = chain.invoke("What is RAG?")

十来行代码、一条完整 RAG——涵盖本章讨论的核心抽象——Embeddings + VectorStore + as_retriever() + Runnable | prompt | llm | parser。

可替换性——

换 DB：InMemoryVectorStore → PineconeVectorStore (一行)
换 embedding：OpenAIEmbeddings → HuggingFaceEmbeddings (一行)
换 llm：ChatOpenAI → ChatAnthropic (一行)

一套架构、多种组合——这就是 LangChain 的"乐高玩法"。

10.22 RAG 质量评估的"四个维度"

Retriever 效果如何测量？—— 四个维度：

维度 1：召回率 (Recall@k)——相关文档是否在 top-k 结果里——召回率低 = 信息漏掉、LLM 答不对

维度 2：准确率 (Precision@k)——top-k 里有多少真相关——准确率低 = 噪声多、LLM 被干扰

维度 3：MRR (Mean Reciprocal Rank)——第一个相关结果在第几名——MRR 高 = 相关文档排前面、LLM 用得到

维度 4：NDCG——考虑相关性等级（不只 0/1、还有 0.5、0.3 等）——更细粒度

业界实践——

快速迭代期——Recall@10 + MRR
深度调优期——NDCG@10 + 人工评估
LLM 端到端评测——《LLM 评测体系》专栏第 6 章 LLM-as-Judge

RAGAS / TruLens / DeepEval 是三个常用的 RAG 评测框架——分别有不同侧重——读者选一个内化进 CI 就够。

10.23 Embedding 模型的"三大流派"

本章主要讲 VectorStore 和 Retriever——但 embedding 模型本身值得一节：

流派 1：OpenAI / 商用 API——

text-embedding-3-small 1536 维、约 $0.02/M tokens
text-embedding-3-large 3072 维、约 $0.13/M tokens
（价格随时会变，完整对照表与免责说明见 §10.35，用之前请以官方价目页为准）
优点：质量高、无需自建
缺点：按量付费、数据出境

流派 2：开源 sentence-transformers——

BAAI/bge-large-en-v1.5 1024 维——英文场景常用的开源模型
BAAI/bge-m3 1024 维——多语言 + dense/sparse 混合
优点：本地跑、免费、可 fine-tune
缺点：GPU 推理成本、维护负担

流派 3：Cohere / Voyage AI——

embed-english-v3.0 / embed-multilingual-v3.0 —— 支持 input_type 区分 query/document（和 LangChain Embeddings 的双方法接口天然契合）
voyage-3-lite —— 性价比高
优点：质量顶尖
缺点：API 调用、国内访问麻烦

选型三原则——

质量优先 → OpenAI / Cohere
成本优先 → 本地 bge / e5
合规优先 → 本地 open-source

一个典型踩坑——把索引侧的 embedding 模型换掉（比如 text-embedding-ada-002 → text-embedding-3-small）、却没重建索引。这类事故最阴险的地方在于它未必报错：ada-002 和 3-small 都是 1536 维，向量维度对得上、similarity_search 照常返回结果，只是两套向量根本不在同一个语义空间里，召回质量退化成"看起来像随机"。如果换的是维度不同的模型（如 3-large 的 3072 维），反而会直接抛维度不匹配的错——早暴露反而是好事。

教训——embedding 模型的升级 = 整个向量库的重建——不可滚动升级——维度相同不代表兼容——提前规划重建窗口。

10.24 `BaseRetriever.invoke` 调用栈的全貌

libs/core/langchain_core/retrievers.py:179-234 的 invoke 方法——一行简单签名、背后一串编排：

    def invoke(
        self, input: str, config: RunnableConfig | None = None, **kwargs: Any
    ) -> list[Document]:
        # …（这里是 17 行 docstring，略去）
        config = ensure_config(config)
        inheritable_metadata = {
            **(config.get("metadata") or {}),
            **self._get_ls_params(**kwargs),
        }
        callback_manager = CallbackManager.configure(
            config.get("callbacks"),
            None,
            verbose=kwargs.get("verbose", False),
            inheritable_tags=config.get("tags"),
            local_tags=self.tags,
            inheritable_metadata=inheritable_metadata,
            local_metadata=self.metadata,
        )
        run_manager = callback_manager.on_retriever_start(
            None,
            input,
            name=config.get("run_name") or self.get_name(),
            run_id=kwargs.pop("run_id", None),
        )
        try:
            kwargs_ = kwargs if self._expects_other_args else {}
            if self._new_arg_supported:
                result = self._get_relevant_documents(
                    input, run_manager=run_manager, **kwargs_
                )
            else:
                result = self._get_relevant_documents(input, **kwargs_)
        except Exception as e:
            run_manager.on_retriever_error(e)
            raise
        else:
            run_manager.on_retriever_end(
                result,
            )
            return result

九步——

ensure_config——把 config=None 变成带默认字段的 config
_get_ls_params(**kwargs) 的结果并进 metadata——所以 LangSmith 的 ls_* 字段是以"可继承 metadata"的身份下传的
CallbackManager.configure——区分 inheritable_tags（来自 config、会往子 run 传）和 local_tags（来自 retriever 实例、只贴在本次 run 上），metadata 同理
on_retriever_start——LangSmith 事件起点、生成 run_id（注意 run_id 是从 kwargs 里 pop 出来的、不是从 config 里取）
判断 _expects_other_args 决定 kwargs 转不转发（§10.11 的姊妹 flag）
判断 _new_arg_supported 决定要不要传 run_manager（§10.11）
调用子类的 _get_relevant_documents
失败 → on_retriever_error + 原样 raise
成功 → on_retriever_end 后返回结果

注意——回调事件只有三种：开始时 on_retriever_start，结束时 on_retriever_end 或 on_retriever_error 二选一——一次 invoke 在 trace 里对应一个完整 retriever span（含输入 query、输出文档、耗时、错误）。

这呼应《Harness Engineering》专栏第 19 章讨论的可观测性仪表盘——每次 retriever 调用都是一条可观测数据——RAG 产品的健康度就看这些事件。

10.25 `_get_ls_params` 的 LangSmith 集成

libs/core/langchain_core/retrievers.py:167-176 的 _get_ls_params 只填一个 ls_retriever_name。

真正加"专用参数"的是 VectorStoreRetriever 的覆写（libs/core/langchain_core/vectorstores/base.py:1024-1035）：

        ls_params["ls_vector_store_provider"] = self.vectorstore.__class__.__name__

        if self.vectorstore.embeddings:
            ls_params["ls_embedding_provider"] = (
                self.vectorstore.embeddings.__class__.__name__
            )
        elif hasattr(self.vectorstore, "embedding") and isinstance(
            self.vectorstore.embedding, Embeddings
        ):
            ls_params["ls_embedding_provider"] = (
                self.vectorstore.embedding.__class__.__name__
            )

这两个字段作为 metadata 随 run 上报——用户能按"vector store 类" 和 "embedding 类" 切片分析：

"Pinecone 的检索准确率 vs Weaviate 是多少"
"OpenAI embedding vs Cohere embedding 在我 agent 上的召回差多少"

注意填进去的是类名（如 OpenAIEmbeddings），不是模型名：同一个类下的 text-embedding-3-small 和 text-embedding-3-large 靠这两个字段区分不开；ls_embedding_model 字段 pin 版本里没有代码填写。

没有这两个字段——你只能看到"一次 retriever 调用"——不知道是哪个 provider 做的。

这是 LangChain 和 LangSmith 的深度集成——框架原生支持追踪切片——不需要用户手动埋点。

10.26 `VectorStore.add_texts` vs `add_documents`

两个看起来相似的方法——实际略有语义差别：

add_texts(texts: list[str], metadatas: list[dict])——用户手动拆开 text 和 metadata
add_documents(documents: list[Document])——用户传入 Document 对象、内部自动拆

为什么两个都有——历史兼容——老 API 是 add_texts、新 API 是 add_documents——后者更 type-safe（Document 类验证过 page_content / metadata 字段）。

推荐用法——

新代码用 add_documents（类型安全）
快速脚本用 add_texts（简洁）
Document 是 Runnable 链里的通用类型——用 add_documents 和 chain 输出天然契合

10.27 VectorStore 性能 benchmark 的基本功

读者想比较"Pinecone vs Chroma vs Qdrant"——哪些指标要测：

指标 1——Insert QPS——每秒能插多少文档——build 阶段关键（1M 文档要插几小时）

指标 2——Search P99 latency——检索延迟——交互式 RAG 的关键（延迟预算得留给下游 LLM 生成，检索这一段拖长了整条链路都跟着慢）

指标 3——Recall@k——相同 ground truth 下召回率——质量关键

指标 4——Recall@k vs ef_search 的曲线——近似搜索（HNSW）的参数调优——权衡精度 vs 速度

指标 5——Metadata filter 的性能损失——过滤条件多了变不变慢——很多 vector db 的 metadata filter 是 post-filter、不是 pre-filter、性能差异明显

指标 6——内存/磁盘占用——先把裸向量的账算清楚：1M 条 × 1536 维 × 4 字节（float32）≈ 6 GB，这还没算索引结构本身的开销；自建要为这部分内存买单，托管服务把它折进账单里

6 项测完——才能"数据驱动"选 vector db——别被营销文案骗。

10.28 `libs/core/langchain_core/vectorstores/utils.py` 157 行——MMR 算法真身

libs/core/langchain_core/vectorstores/utils.py:106-157 的 maximal_marginal_relevance 函数——MMR 算法 Python 实现：

def maximal_marginal_relevance(
    query_embedding: np.ndarray,
    embedding_list: list,
    lambda_mult: float = 0.5,
    k: int = 4,
) -> list[int]:
    # （省略：numpy 检查、k<=0 早返回、一维 query 升维）
    # Step 1: 余弦相似度计算（query vs all docs）
    similarity_to_query = _cosine_similarity(query_embedding, embedding_list)[0]
    # Step 2: 先选 similarity 最高的
    most_similar = int(np.argmax(similarity_to_query))
    idxs = [most_similar]
    selected = np.array([embedding_list[most_similar]])
    # Step 3: 迭代选剩下的 k-1 个
    while len(idxs) < min(k, len(embedding_list)):
        best_score = -np.inf
        idx_to_add = -1
        similarity_to_selected = _cosine_similarity(embedding_list, selected)
        for i, query_score in enumerate(similarity_to_query):
            if i in idxs: continue
            redundant_score = max(similarity_to_selected[i])
            equation_score = (
                lambda_mult * query_score
                - (1 - lambda_mult) * redundant_score
            )
            if equation_score > best_score:
                best_score = equation_score
                idx_to_add = i
        idxs.append(idx_to_add)
        selected = np.append(selected, [embedding_list[idx_to_add]], axis=0)
    return idxs

五十来行 Python——教科书级 MMR 实现——你可以直接抄到自己的项目（utils.py 模块文档声明它属私有 API、可能不经通知改动，所以宜抄不宜直接 import）。

时间复杂度——注意循环体里每一轮都重算一次 cosine_similarity(embedding_list, selected)：第 j 轮的选中集有 j 个向量，于是这一轮是 O(N·j·d)，累加下来约 O(k²·N·d)（N 是候选数、d 是维度）。这也是为什么 MMR 的候选集是 fetch_k 而不是全库——fetch_k 直接进了平方项外的那个线性因子。

所以 fetch_k 不宜设得太大——否则 MMR 这一步自己会变成瓶颈。

10.29 一段额外的工程提醒

RAG 系统上生产前——务必检查三件事：

检查 1：chunk 是否合理——几百到一千 token 是常见起点（经验值，要按自己的语料实测）——太小上下文不够、太大 embedding 糊——用 RecursiveCharacterTextSplitter 按分隔符层级拆。

检查 2：embedding 和检索用同一模型——用 text-embedding-3-small 建索引、检索时也必须用同一模型生成 query 向量——用错会得"接近随机" 的结果——看似检索出来、实际乱搞。

检查 3：metadata filter 能用 index——Pinecone、Qdrant 等支持 metadata index——不建索引的 filter 会全表扫描——性能退化。

这 3 项核对表过一遍——不少 RAG "为什么不工作" 的问题都能在这里被拦下。

10.30 Chunking 策略的"五大流派"

RAG 上游最被忽视的环节是 chunking——5 种常见策略：

策略 1——定长 chunk（CharacterTextSplitter）——按字符数切——最快、最糙——语义边界可能被砍断

策略 2——递归 chunk（RecursiveCharacterTextSplitter）——按分隔符层级（默认 "\n\n" → "\n" → " " → ""，即段落 → 行 → 词 → 字符）——通用场景的常用选择、质量好

策略 3——语义 chunk（SemanticChunker）——用 embedding 相似度找语义边界——切分时要调 embedding、成本高，效果是否更好取决于语料（注意它不在 langchain-text-splitters 包里、住在 langchain-experimental）

策略 4——结构 chunk——Markdown 按 header（MarkdownHeaderTextSplitter）、HTML 按 tag（html.py）、代码按语言语法（RecursiveCharacterTextSplitter.from_language）——场景专用

策略 5——层级 chunk（parent/child 双层）——见§10.18 ParentDocumentRetriever——小 chunk 入索引、大 chunk 做 retrieve

选择原则——

通用文档 → 递归
技术文档 → 结构（按 header）
高质量问答 → 语义
长文档 RAG → 层级

一个具体数字——chunk_size=500 + chunk_overlap=50——是社区实践里最常用的起点值。注意它不是源码默认：TextSplitter 的默认是 chunk_size=4000, chunk_overlap=200，不传参拿到的是后者。500/50 更适合做 RAG 检索粒度，但得你自己传。

10.31 Rerank——RAG 质量的"临门一脚"

Retriever 召回 top-20——再用 reranker 重排挑 top-5 喂 LLM——这一步能让 RAG 效果"再上一个台阶"。

两类 rerank 模型——

Cross-encoder（如 BAAI/bge-reranker-large）——(query, doc) → score 直接打分——质量最高、速度较慢
LLM-based（让 GPT-4o-mini 给每个 doc 打 1-10 分）——质量好、成本高、可定制

LangChain 集成——ContextualCompressionRetriever + CrossEncoderReranker ——一行组合。

典型效果——排序类指标（MRR / NDCG）通常有明显改善、long-tail query 上尤其明显；但具体涨多少必须在你自己的评测集上量——公开 benchmark 的数字换个语料就不成立。

工程实践——生产 RAG 常见配置是"retriever 召回 20 + reranker 取 5"——先放宽召回、再用更贵的模型精排。

10.32 查询改写——"用户 query 不等于好 retrieval query"

用户问 "那个我上次说的 bug 怎么样了"——直接用这句 embedding 去检索——废。

查询改写——在 retrieve 前用 LLM 把 query 重写成"适合检索的形式"：

HyDE（Hypothetical Document Embeddings）——先让 LLM 生成一段"假想答案"、用这段 embedding 去检索——比原 query embedding 更准
Query Decomposition——把复杂问题拆成多个子问题、各自检索、合并结果
Multi-Query——LLM 生成若干 query 变体、分别检索、合并结果

LangChain 集成——MultiQueryRetriever（langchain_classic/retrievers/multi_query.py，默认 prompt 让 LLM 生成 3 个变体，结果按内容去重取并集 unique_union，不做 RRF；同步路径逐个检索，异步路径用 asyncio.gather 并发）；HyDE 在 langchain-classic 里的落地不是一个 retriever 而是一个 Embeddings 包装——HypotheticalDocumentEmbedder（langchain_classic/chains/hyde/base.py，它同时继承 Chain 和 Embeddings）：先让 LLM 生成假想答案、再对答案做嵌入，然后当成普通 embedding 塞进任意 VectorStore。

性能代价——每次检索多一次（甚至多次）LLM 调用——成本 + 延迟都要加——换来的召回增益值不值，取决于你的查询分布，得自己拿评测集比。

10.33 "Document 的 metadata 是第一公民"

Document.metadata 字段常被当附属信息——实际它是生产 RAG 的"钥匙"：

metadata 的生产用途——

过滤——similarity_search(query, filter={"source": "internal_docs"}) —— 只搜内部文档（字典式 filter 的语法因后端而异；InMemoryVectorStore 要传函数，如 filter=lambda d: d.metadata["source"] == "internal_docs"）
权限——filter={"user_id": current_user.id} —— 按用户隔离知识库
多租户——filter={"tenant": "acme_corp"} —— 同一 vector store 服务多客户
时效性——filter={"updated_after": last_week} —— 只搜最近更新
评分加权——metadata.priority 在合并时作为 boost factor

metadata 设计原则——

小——metadata 字段太大会让向量 db 变慢
可索引——常用 filter 字段"建索引"
标准化——枚举值统一（type: "pdf" | "html" | "md"）——方便过滤

一个常见错误——把整个原文 page_content 复制进 metadata——原文存了两份、存储和费用白白增加——保持 metadata 精简。

10.34 最后一张"RAG 系统工程 checklist"

12 项生产 RAG 上线前的自检——

chunking 策略选对（§10.30）
embedding 模型和检索 query 用同一个（§10.29）
chunk_size + overlap 跑过实验（500 + 50 起步）
用 RecursiveCharacterTextSplitter 而不是 CharacterTextSplitter
metadata 设计含权限/租户字段（§10.33）
retrieve 用 MMR 或混合召回（§10.13 / §10.17）
加 reranker（§10.31）
尝试 HyDE / MultiQuery 查询改写（§10.32）
用 RAGAS 跑 eval（§10.22）
LangSmith trace 开启（§10.25）
成本监控：按 tenant 切片 embedding + LLM 成本
有 fallback 策略（retriever 挂了是否有备选）

打勾 ≥ 10 项——你的 RAG 生产级——少于 7 项——上线前再调。

10.35 一段额外的"embedding provider 官方定价对照"

各家 embedding 的官方定价（截至本专栏成稿时）——

Provider	模型	维度	$/M tokens	1M 文档 (×500 tok) 成本
OpenAI	text-embedding-3-small	1536	$0.02	$10
OpenAI	text-embedding-3-large	3072	$0.13	$65
Cohere	embed-english-v3.0	1024	$0.10	$50
Voyage	voyage-3-lite	512	$0.02	$10
HuggingFace	bge-large（本地）	1024	无 API 费	只付 GPU 时间

最后一列是纯算术：1M 文档 × 500 token = 500M token，乘上每 M 单价。定价随时会变，用之前请以各家官方价目页为准。

三个关键观察——

text-embedding-3-small 单价最低——十美元量级就能嵌完百万级文档
text-embedding-3-large 单价是 small 的 6.5 倍（0.13 / 0.02）——质量是否值这个差价，得在你自己的评测集上比，别拿公开榜单的名次直接套
本地 bge 在规模足够大时才划算——省的是 API 费、付的是 GPU 时间和运维

成本优化技巧——只嵌入 changed docs、不重嵌没变的——每次文档更新只产生小增量 embedding 调用。节省比例取决于文档变更率；如果每天只改少量文档，收益很大，如果语料经常全量重建，收益就有限。（langchain-core 的 langchain_core/indexing/ 模块正是为这件事准备的——用文档哈希（page_content 与 metadata 一起算）+ RecordManager 判断哪些文档真的变了，详见第 9 章 §9.12。）

这呼应《Harness Engineering》专栏第 20 章讲的成本-性能取舍——同样是"只为必要的计算付费"的理念。

10.36 自建 vector db 的"一个 weekend 路线"

有读者问——能不能不用任何外部 vector db、自己写一个？—— 能、一个周末：

Day 1——最简

dict[str, np.ndarray] 存 embedding
numpy.dot 算余弦
就是本章§10.14 InMemoryVectorStore 的简化版 < 100 行

Day 2——加 ANN

引入 hnswlib 库（HNSW 算法的 Python 绑定）
用近似最近邻索引换掉暴力扫描——这一步才是量级上的变化
百来行代码

结果——你有了一个能用的"self-hosted vector search"。

代价——持久化、分片、副本、一致性——这些才是商用 vector db 的"真正价值"——自建不划算。

教训——核心算法很简单（HNSW / IVF）、产品化才是价值所在——买商用 vector db 买的是"运维省心"。

10.37 主流 vector db 的"决策矩阵"

生产选型常被问的问题——"我应该用哪个 vector db"——6 个候选 + 5 个维度：

DB	托管	规模	性能	价格	特色
Pinecone	纯托管	10B+	高	贵	零运维、企业友好
Weaviate	自建/托管	1B+	高	中	开源、GraphQL 风
Qdrant	自建/托管	1B+	高	中	Rust 写的、快
Chroma	自建为主	10M	中	免费	SQLite 级简单
Milvus	自建	10B+	高	免费	中国团队主力
pgvector	Postgres 插件	100M	中	免费	不用额外组件

典型选型路径——

原型 → Chroma / InMemoryVectorStore
小规模生产 (< 1M docs) → pgvector（复用现有 PG）
中规模 (1M-100M) → Qdrant / Weaviate 自建
大规模 (100M+) → Pinecone 托管 / Milvus 自建
企业合规 → 自建 Weaviate / Milvus（数据不出境）

上表的"规模 / 性能 / 价格"三列是各家宣传口径下的粗略档位，不是统一 benchmark 的实测结果——真要选型，按 §10.27 的六项指标在自己的语料上压一遍。

成本梯度（从低到高）——pgvector（复用已有 PG 实例、边际成本最低）→ 自建 Qdrant / Weaviate（付服务器和运维）→ 托管 Pinecone（付托管溢价）——多花的钱换的是运维省心。

10.38 附录：三个其他专栏的呼应

本章多次提到其他专栏——一次性集中：

本专栏第 3 章 Runnable 与 LCEL——Retriever 是 Runnable 的子类——学完 Runnable 再看 Retriever 如鱼得水
《LangGraph 设计与实现》专栏第 17 章多智能体模式——RAG 场景里 retriever 通常作为 LangGraph 图里的一个 node——那个专栏的 tool-using agent 模式可以直接拿来用
《MCP 协议设计与实现》专栏第 20 章构建生产级 MCP Server——MCP Server 经常作为 retriever 的"数据源"——本章 retriever 可以通过 MCP 对接外部知识库

三个专栏交叉读——你就能构建"MCP retriever + LangChain wrap + LangGraph 编排"的企业级 RAG 架构。

10.39 "用 RAG 代替 fine-tune"的工程辩证

2023-2024 年业界流行"用 RAG 代替 fine-tune"——2025 开始有更细致的辩证：

RAG 的优势——

冷启动快——不需要训练数据、几天上线
知识可控——每个答案都能 trace 到源文档
更新及时——新增文档立刻生效、不用重训
成本低——只花 retrieval + LLM API 钱、不花训练钱

RAG 的劣势——

推理成本高——每次查询都要 retrieval + LLM，不 scalable 到超高 QPS
上下文窗口限制——再多文档也塞不下全部
复杂推理弱——多步推理时 chunk 间关系丢失
领域适配慢——专业术语、推理模式用 fine-tune 更彻底

常见做法——混合：

事实类查询 → RAG（快、准、可控）
风格 / 推理模式 → Fine-tune（内化、规模化）
长尾 / 罕见场景 → RAG + few-shot prompt

成本曲线的形状不同——RAG 的成本几乎全在每次查询上（检索 + 更长的 prompt），随 QPS 线性增长；fine-tune 是一次性训练成本 + 更短 prompt 的推理成本。所以查询量越大、fine-tune 的一次性投入越容易摊平，早期量小时 RAG 更划算。具体的交叉点在哪，只能拿自己的 QPS、prompt 长度和训练成本算，没有通用数字。

这个决策——和《Harness Engineering》专栏第 20 章讲的成本-性能取舍一致——根据任务类型用最适合的技术。

10.40 "RAG 事故"的三种典型形态

生产 RAG 系统里最容易出现的三种故障——

事故 A：chunk_overlap 设 0——chunk 切在句子中间、上下文被砍断、检索召回了"半句话"、LLM 答得莫名其妙。修复：overlap 设 ~10% 的 chunk_size。

事故 B：embedding 模型升级 embed 全库但忘 reindex query 模型——query 端还用旧模型、向量空间不一致、召回"像是随机"。修复：embed 模型变更必须同时更新 build + query 两端。

事故 C：metadata 不加 user_id 隔离——A 客户的文档被检索给了 B 客户、合规事故。修复：filter={"tenant": current_tenant} 必须成为强制约束。

10.41 延伸资料

LangChain 官方文档的 RAG 教程——https://docs.langchain.com/（文档已从主仓库拆出、不再位于 langchain-ai/langchain 的 docs/ 目录下，本专栏基准的 langchain-core==1.2.26 tag 上只有 libs/）。

libs/langchain/langchain_classic/retrievers/ 目录——包含 EnsembleRetriever（ensemble.py）、ContextualCompressionRetriever（contextual_compression.py）等在 langchain-core 里看不到的实现，是本章接口抽象的具体落地。（其中 bm25.py 只是个带弃用提示的转发层，BM25Retriever 实际从 langchain_community 导入。）

10.42 一段凝练：一句话 + 三组数字

一句话——"Retriever 是 RAG 的"输入输出控制台"——精度决定 LLM 上限、成本决定商业可行。

三组数字——

chunk_size=500, overlap=50（社区常用起点，注意不是源码默认——TextSplitter 的默认是 chunk_size=4000, chunk_overlap=200）
fetch_k=20, k=4, lambda_mult=0.5（MMR 的源码默认，见 libs/core/langchain_core/vectorstores/base.py:659-665 的 max_marginal_relevance_search 签名）
retriever 召回 20 + reranker 取 5（生产 RAG 的常见配置，非源码默认）

三组数字的来源不同，用的时候要分清：第二组是 langchain-core 里写死的默认值，直接不传参就是它；第一、三组是社区实践中被反复验证的起点值，源码不会替你设成这个数。调参时先明确你用的是哪一类，再按业务场景微调。

10.43 "embedding vs rerank 预算分配"

给工程师最常问的成本分配问题一个答案——

先按成本性质把开销分成两类，这个划分是结构性的、不依赖任何实测数字：

一次性 / 随语料规模增长——embedding build、ETL、vector store 的存储费
随查询量增长——每次查询的检索、rerank、telemetry

真实占比在不同业务里差别极大（重建频率高的语料 embedding 占大头，高 QPS 的产品 rerank 占大头），先按这两类把账单拆开、再决定砍哪边，比套用任何固定比例都靠谱。

如果需要降成本——

先砍 rerank（省钱最直接，但会丢掉 §10.31 的精排收益，砍前先在评测集上量）
再砍 embedding（换成 bge 本地）
最后换 vector store（最难、改动大）

如果需要提精度——

先加 rerank（最快）
再加 MultiQuery（中等）
最后换更大 embedding 模型（最慢）

把"钱花在哪里"拆清楚——是 RAG 成熟团队的"财务地图"。