如何创建自定义检索器

概览

许多大型语言模型应用都涉及使用检索器从外部数据源获取信息。

检索器负责根据给定的用户 query 检索相关文档列表。

检索到的文档通常会被格式化为提示词，输入到大型语言模型中，使大型语言模型能够利用这些信息生成适当的回应（例如，基于知识库回答用户问题）。

界面

要创建自己的检索器，您需要扩展 BaseRetriever 类并实现以下方法：

方法	描述	Required/Optional
`_get_relevant_documents`	Get documents relevant to a query.	Required
`_aget_relevant_documents`	Implement to provide async native support.	Optional

_get_relevant_documents 内部的逻辑可以包含对数据库或通过 requests 调用网络的任意操作。

提示

通过继承 BaseRetriever，您的检索器将自动成为 LangChain 可运行对象，并且会自带标准的 Runnable 功能！

信息

您可以使用 RunnableLambda 或 RunnableGenerator 来实现检索器。

将检索器实现为 BaseRetriever 而不是 RunnableLambda（自定义可运行函数）的主要优势在于，BaseRetriever 是 LangChain 中一个众所周知的实体，因此某些监控工具可能会针对检索器实现专门的行为。另一个区别是，在某些 API 中，BaseRetriever 的行为会与 RunnableLambda 略有不同；例如，在 astream_events API 中的 start 事件将变为 on_retriever_start，而不是 on_chain_start。

示例

让我们实现一个简单的检索器，该检索器返回文本中包含用户查询内容的所有文档。

from typing import List

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


class ToyRetriever(BaseRetriever):
    """A toy retriever that contains the top k documents that contain the user query.

    This retriever only implements the sync method _get_relevant_documents.

    If the retriever were to involve file access or network access, it could benefit
    from a native async implementation of `_aget_relevant_documents`.

    As usual, with Runnables, there's a default async implementation that's provided
    that delegates to the sync implementation running on another thread.
    """

    documents: List[Document]
    """List of documents to retrieve from."""
    k: int
    """Number of top results to return"""

    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
        """Sync implementations for retriever."""
        matching_documents = []
        for document in documents:
            if len(matching_documents) > self.k:
                return matching_documents

            if query.lower() in document.page_content.lower():
                matching_documents.append(document)
        return matching_documents

    # Optional: Provide a more efficient native implementation by overriding
    # _aget_relevant_documents
    # async def _aget_relevant_documents(
    #     self, query: str, *, run_manager: AsyncCallbackManagerForRetrieverRun
    # ) -> List[Document]:
    #     """Asynchronously get documents relevant to a query.

    #     Args:
    #         query: String to find relevant documents for
    #         run_manager: The callbacks handler to use

    #     Returns:
    #         List of relevant documents
    #     """

API 参考：CallbackManagerForRetrieverRun | 文档 |BaseRetriever

测试一下 🧪

documents = [
    Document(
        page_content="Dogs are great companions, known for their loyalty and friendliness.",
        metadata={"type": "dog", "trait": "loyalty"},
    ),
    Document(
        page_content="Cats are independent pets that often enjoy their own space.",
        metadata={"type": "cat", "trait": "independence"},
    ),
    Document(
        page_content="Goldfish are popular pets for beginners, requiring relatively simple care.",
        metadata={"type": "fish", "trait": "low maintenance"},
    ),
    Document(
        page_content="Parrots are intelligent birds capable of mimicking human speech.",
        metadata={"type": "bird", "trait": "intelligence"},
    ),
    Document(
        page_content="Rabbits are social animals that need plenty of space to hop around.",
        metadata={"type": "rabbit", "trait": "social"},
    ),
]
retriever = ToyRetriever(documents=documents, k=3)

retriever.invoke("that")

[Document(page_content='Cats are independent pets that often enjoy their own space.', metadata={'type': 'cat', 'trait': 'independence'}),
 Document(page_content='Rabbits are social animals that need plenty of space to hop around.', metadata={'type': 'rabbit', 'trait': 'social'})]

这是一个 可运行的 对象，因此它将受益于标准的 Runnable 接口！ 🤩

await retriever.ainvoke("that")

[Document(page_content='Cats are independent pets that often enjoy their own space.', metadata={'type': 'cat', 'trait': 'independence'}),
 Document(page_content='Rabbits are social animals that need plenty of space to hop around.', metadata={'type': 'rabbit', 'trait': 'social'})]

retriever.batch(["dog", "cat"])

[[Document(page_content='Dogs are great companions, known for their loyalty and friendliness.', metadata={'type': 'dog', 'trait': 'loyalty'})],
 [Document(page_content='Cats are independent pets that often enjoy their own space.', metadata={'type': 'cat', 'trait': 'independence'})]]

async for event in retriever.astream_events("bar", version="v1"):
    print(event)

{'event': 'on_retriever_start', 'run_id': 'f96f268d-8383-4921-b175-ca583924d9ff', 'name': 'ToyRetriever', 'tags': [], 'metadata': {}, 'data': {'input': 'bar'}}
{'event': 'on_retriever_stream', 'run_id': 'f96f268d-8383-4921-b175-ca583924d9ff', 'tags': [], 'metadata': {}, 'name': 'ToyRetriever', 'data': {'chunk': []}}
{'event': 'on_retriever_end', 'name': 'ToyRetriever', 'run_id': 'f96f268d-8383-4921-b175-ca583924d9ff', 'tags': [], 'metadata': {}, 'data': {'output': []}}

贡献

我们感谢对有趣检索器的贡献！

以下是帮助确保您的贡献被添加到 LangChain 的检查清单：

Documentation:

检索器包含所有初始化参数的文档字符串，因为这些内容将在 API 参考中显示。
该模型的类文档字符串中包含指向用于检索器的相关API链接（例如，如果检索器是从维基百科获取信息，则应链接到维基百科API！）

Tests:

添加单元测试或集成测试以验证invoke和ainvoke work.

Optimizations:

如果检索器连接到外部数据源（例如 API 或文件），几乎肯定能从异步原生优化中获益！

提供原生的异步实现_aget_relevant_documents（被用于ainvoke)

概览​

界面​

示例​

测试一下 🧪​

贡献​

概览

界面

示例

测试一下 🧪

贡献