langchain_core.retrievers.BaseRetriever 是 LangChain 框架中的一个核心抽象基类，用于定义文档检索系统的标准接口，广泛应用于检索增强生成（RAG）系统和代理（Agent）系统中。它通过接受非结构化查询（如字符串）并返回相关文档列表，为开发者提供了灵活的检索功能。

1. 概述

1.1 什么是 BaseRetriever？

BaseRetriever 是 langchain_core.retrievers 模块中的抽象基类，继承自 RunnableSerializable[str, List[Document]] 和 Python 的 ABC（抽象基类）。它定义了检索器的核心接口，允许开发者通过子类化实现从各种数据源（如向量存储、数据库或 Web API）检索文档的逻辑。BaseRetriever 的主要任务是将输入查询（如文本字符串）映射到相关的 Document 对象列表，支持同步和异步操作。

该类在 langchain-core 0.2.14 版本中引入，是 LangChain 检索系统的基石，适用于构建 RAG 管道、代理工具或自定义检索逻辑。相关信息可参考 LangChain 自定义检索器指南 和 BaseRetriever API 参考。

1.2 核心功能

• 标准化检索接口：定义从任意数据源检索文档的统一接口，接受字符串查询并返回 Document 列表。
• Runnable 支持：继承 RunnableSerializable，支持 LangChain 的标准操作（如 invoke、ainvoke、batch、abatch），便于链式组合。
• 同步与异步操作：提供同步方法（_get_relevant_documents）和可选异步方法（_aget_relevant_documents），满足不同性能需求。
• 自定义扩展：通过子类化实现自定义检索逻辑，适用于向量存储、数据库、Web API 或其他数据源。
• 元数据与标签：支持配置 metadata 和 tags，传递给回调用于日志和跟踪。
• 实验性功能：提供 Beta 版 API（如 as_tool 和 astream_events），扩展功能。
• 版本要求：需 langchain-core>=0.2.14，推荐使用最新版本。

2. 定义与结构

2.1 类定义

BaseRetriever 是抽象基类，定义如下（简化表示）：

from abc import ABC
from typing import Any, Dict, List, Optional
from langchain_core.documents import Document
from langchain_core.runnables import RunnableConfig, RunnableSerializable
from langchain_core.callbacks import AsyncCallbackManagerForRetrieverRun, CallbackManagerForRetrieverRun

class BaseRetriever(RunnableSerializable[str, List[Document]], ABC):
    metadata: Optional[Dict[str, Any]]
    tags: Optional[List[str]]

    def _get_relevant_documents(self, query: str, *, run_manager: Optional[CallbackManagerForRetrieverRun] = None) -> List[Document]:
        raise NotImplementedError

    async def _aget_relevant_documents(self, query: str, *, run_manager: Optional[AsyncCallbackManagerForRetrieverRun] = None) -> List[Document]:
        ...

2.2 初始化参数

参数	类型	默认值	描述
metadata	Optional[Dict[str, Any]]	None	可选，检索器的元数据，传递给回调用于跟踪。
tags	Optional[List[str]]	None	可选，检索器的标签，传递给回调用于日志记录。

2.3 核心方法

以下是 BaseRetriever 的主要方法，基于 API 参考：

方法	描述	参数	返回值	异步
invoke	同步调用检索器，返回相关文档列表。	input: str, config: Optional[RunnableConfig], **kwargs: Any	List[Document]	No
ainvoke	异步调用检索器，返回相关文档列表。	同 invoke	List[Document]	Yes
batch	同步批量处理多个查询，返回文档列表。	inputs: List[str], config: Optional[Union[RunnableConfig, List[RunnableConfig]]], return_exceptions: bool, **kwargs: Any	List[List[Document]]	No
abatch	异步批量处理多个查询，返回文档列表。	同 batch	List[List[Document]]	Yes
stream	同步流式输出检索结果。	input: str, config: Optional[RunnableConfig], **kwargs: Any	Iterator[List[Document]]	No
astream	异步流式输出检索结果。	同 stream	AsyncIterator[List[Document]]	Yes
astream_events	异步流式输出事件（Beta 功能）。	input: Any, config: Optional[RunnableConfig], version: Literal['v1', 'v2'], **kwargs: Any	AsyncIterator[StreamEvent]	Yes
_get_relevant_documents	抽象方法，定义文档检索逻辑，子类必须实现。	query: str, run_manager: Optional[CallbackManagerForRetrieverRun]	List[Document]	No
_aget_relevant_documents	可选异步方法，定义异步检索逻辑，默认调用同步方法，子类可覆盖。	query: str, run_manager: Optional[AsyncCallbackManagerForRetrieverRun]	List[Document]	Yes
as_tool	将检索器转换为 BaseTool（Beta 功能）。	name: str, description: Optional[str], args_schema: Optional[Type[BaseModel]]	BaseTool	No

已废弃方法：

• get_relevant_documents（自 langchain-core==0.1.46 起废弃，推荐使用 invoke）。
• aget_relevant_documents（自 langchain-core==0.1.46 起废弃，推荐使用 ainvoke）。

3. 使用方法

使用 BaseRetriever 通常需要创建子类并实现 _get_relevant_documents 方法，以下是具体使用方式和示例：

3.1 创建自定义检索器

实现一个简单的字符串匹配检索器：

from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from typing import List
import asyncio

class ToyRetriever(BaseRetriever):
    documents: List[Document] = []
    k: int = 3

    def _get_relevant_documents(self, query: str, *, run_manager=None) -> List[Document]:
        query_lower = query.lower()
        relevant_docs = [
            doc for doc in self.documents
            if query_lower in doc.page_content.lower()
        ]
        return relevant_docs[:self.k]

    async def _aget_relevant_documents(self, query: str, *, run_manager=None) -> List[Document]:
        return await asyncio.get_event_loop().run_in_executor(None, lambda: self._get_relevant_documents(query, run_manager=run_manager))

# 初始化文档
docs = [
    Document(page_content="我有一只狗"),
    Document(page_content="我有一只猫"),
    Document(page_content="我有一条金鱼"),
]

# 创建检索器
retriever = ToyRetriever(documents=docs, k=2)

# 同步调用
results = retriever.invoke("狗")
print(results)  # 输出: [Document(page_content="我有一只狗", ...)]

# 异步调用
results_async = await retriever.ainvoke("狗")
print(results_async)  # 输出: [Document(page_content="我有一只狗", ...)]

3.2 批量处理

处理多个查询：

queries = ["狗", "猫"]
results = retriever.batch(queries)
print(results)  # 输出: [[Document(page_content="我有一只狗", ...)], [Document(page_content="我有一只猫", ...)]]

3.3 结合 RAG 链

将自定义检索器集成到 RAG 链：

from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 定义提示模板
prompt = ChatPromptTemplate.from_template("根据以下文档回答问题：\n{docs}\n问题：{question}")

# 创建 RAG 链
chain = (
    {"docs": retriever | (lambda docs: "\n".join(doc.page_content for doc in docs)), "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-4o-mini")
)

# 调用链
result = chain.invoke("我有什么宠物？")
print(result)  # 输出模型生成答案，如“我有狗、猫和金鱼。”

4. 实际应用

BaseRetriever 在以下场景中广泛应用：

• 检索增强生成（RAG）：检索相关文档后传递给语言模型生成答案，适用于问答系统、知识库查询。
• 代理系统：为代理提供检索功能，代理根据用户输入调用检索工具获取上下文，如智能助手。
• 自定义检索逻辑：通过子类化实现从特定数据源（如数据库、Web API、文件系统）检索文档，满足定制需求。
• 批量与流式处理：处理大规模查询或实时输出，适用于高并发或交互式应用。
• 测试与原型设计：结合虚假嵌入模型（如 FakeEmbeddings）或简单检索逻辑，快速验证管道。

5. 最佳实践

• 实现异步方法：若数据源支持异步（如数据库或 API），实现 _aget_relevant_documents，优化性能。
• 定义清晰逻辑：在 _get_relevant_documents 中实现明确的检索逻辑，避免复杂性。
• 文档与注释：为自定义检索器提供详细文档字符串，说明初始化参数和使用场景，例如：

  class ToyRetriever(BaseRetriever):
      """一个简单的字符串匹配检索器。
  
      Args:
          documents: 要搜索的文档列表。
          k: 返回的最大文档数，默认为 3。
      """
  

• 测试覆盖：测试 invoke、ainvoke、batch 和 abatch，确保同步和异步行为一致。
• 元数据利用：使用 metadata 和 tags 记录上下文，方便 LangSmith 跟踪。
• 版本检查：通过 pip install -qU langchain 确保安装 langchain-core>=0.2.14。
• 调试工具：使用 LangSmith 跟踪检索操作，调试复杂链或代理系统。

6. 注意事项与限制

• 抽象基类：BaseRetriever 不可直接实例化，必须子类化并实现 _get_relevant_documents。
• 废弃方法：避免使用 get_relevant_documents 和 aget_relevant_documents（自 langchain-core==0.1.46 起废弃），优先使用 invoke 和 ainvoke。
• 异步默认实现：若未实现 _aget_relevant_documents，默认调用同步方法，可能影响性能。
• Pydantic 兼容性：自定义检索器的字段需符合 Pydantic 要求，避免验证错误。
• 模块路径：确保正确导入 langchain_core.retrievers.BaseRetriever。
• 版本依赖：功能可能受版本限制，需确保兼容最新版本。
• Beta 功能：as_tool 和 astream_events 是实验性 API，未来可能调整。

7. 结论

langchain_core.retrievers.BaseRetriever 是 LangChain 框架中定义文档检索系统标准接口的抽象基类，通过支持同步和异步操作、批量处理和流式输出，为 RAG 系统、代理系统和自定义检索逻辑提供了强大支持。其灵活的 Runnable 接口和元数据管理使其易于集成到复杂工作流中。通过子类化实现自定义检索逻辑，结合 Embeddings、VectorStoreRetriever 和 LangChain 生态系统，开发者可以构建高效、精准的检索系统。遵循最佳实践并注意版本要求，将有助于充分发挥其功能。

8. 参考资料

• LangChain 自定义检索器指南
• BaseRetriever API 参考
• LangChain 检索器文档
• LangChain 检索博客
• 如何创建自定义检索器
• 在 LangChain 中使用自定义检索器
• 自定义检索器设置问题
• 检索器与 typing.Protocol 的兼容性问题