> ## Documentation Index
> Fetch the complete documentation index at: https://ikun.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 实现自定义检索器

> 了解如何实现自定义检索器，以精确控制知识库中文档的检索。

在某些情况下，您可能需要完全控制代理如何从知识库中检索信息。这可以通过实现自定义检索器函数来实现。自定义检索器允许您定义从向量数据库搜索和检索文档的逻辑。

## 设置

请按照 [Qdrant 设置指南](https://qdrant.tech/documentation/guides/installation/) 中的说明在本地安装 Qdrant。这是获取 API 密钥的指南：[Qdrant API 密钥](https://qdrant.tech/documentation/cloud/authentication/)。

### 示例：PDF 知识库的自定义检索器

下面是使用 `agno` 库实现自定义检索器函数的详细示例。此示例演示了如何设置包含 PDF 文档的知识库、定义自定义检索器，并将其与代理一起使用。

```python theme={null}
from typing import Optional
from agno.agent import Agent
from agno.embedder.openai import OpenAIEmbedder
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.vectordb.qdrant import Qdrant
from qdrant_client import QdrantClient

# ---------------------------------------------------------
# 此部分加载知识库。如果您的知识库 elsewhere 已填充，请跳过。
# 定义嵌入器
embedder = OpenAIEmbedder(id="text-embedding-3-small")
# 初始化向量数据库连接
vector_db = Qdrant(collection="thai-recipes", url="http://localhost:6333", embedder=embedder)
# 加载知识库
knowledge_base = PDFUrlKnowledgeBase(
    urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
    vector_db=vector_db,
)

# 加载知识库
knowledge_base.load(recreate=True)  # 首次运行后注释掉
# 知识库现已加载
# ---------------------------------------------------------

# 定义自定义检索器
# 这是代理将用于检索文档的函数
def retriever(
    query: str, agent: Optional[Agent] = None, num_documents: int = 5, **kwargs
) -> Optional[list[dict]]:
    """
    自定义检索器函数，用于在向量数据库中搜索相关文档。

    Args:
        query (str): 搜索查询字符串
        agent (Agent): 发出查询的代理实例
        num_documents (int): 要检索的文档数量（默认为 5）
        **kwargs: 其他关键字参数

    Returns:
        Optional[list[dict]]: 检索到的文档列表，如果搜索失败则返回 None
    """
    try:
        qdrant_client = QdrantClient(url="http://localhost:6333")
        query_embedding = embedder.get_embedding(query)
        results = qdrant_client.query_points(
            collection_name="thai-recipes",
            query=query_embedding,
            limit=num_documents,
        )
        results_dict = results.model_dump()
        if "points" in results_dict:
            return results_dict["points"]
        else:
            return None
    except Exception as e:
        print(f"向量数据库搜索错误: {str(e)}")
        return None

def main():
    """演示代理用法的函数。"""
    # 使用自定义检索器初始化代理
    # 记住将 search_knowledge=True 设置为使用 agentic_rag 或将 add_reference=True 添加到传统 RAG
    # 当您添加知识库时，search_knowledge=True 是默认值，但此处需要它
    agent = Agent(
        retriever=retriever,
        search_knowledge=True,
        instructions="在知识库中搜索信息",
        show_tool_calls=True,
    )

    # 示例查询
    query = "列出制作 Massaman Gai 所需的配料"
    agent.print_response(query, markdown=True)

if __name__ == "__main__":
    main()
```

#### 异步实现

```python theme={null}
import asyncio
from typing import Optional
from agno.agent import Agent
from agno.embedder.openai import OpenAIEmbedder
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.vectordb.qdrant import Qdrant
from qdrant_client import AsyncQdrantClient

# ---------------------------------------------------------
# 知识库设置（与同步示例相同）
embedder = OpenAIEmbedder(id="text-embedding-3-small")
vector_db = Qdrant(collection="thai-recipes", url="http://localhost:6333", embedder=embedder)
knowledge_base = PDFUrlKnowledgeBase(
    urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
    vector_db=vector_db,
)
# ---------------------------------------------------------

# 定义自定义异步检索器
async def retriever(
    query: str, agent: Optional[Agent] = None, num_documents: int = 5, **kwargs
) -> Optional[list[dict]]:
    """
    自定义异步检索器函数，用于在向量数据库中搜索相关文档。
    """
    try:
        qdrant_client = AsyncQdrantClient(path="tmp/qdrant")
        query_embedding = embedder.get_embedding(query)
        results = await qdrant_client.query_points(
            collection_name="thai-recipes",
            query=query_embedding,
            limit=num_documents,
        )
        results_dict = results.model_dump()
        if "points" in results_dict:
            return results_dict["points"]
        else:
            return None
    except Exception as e:
        print(f"向量数据库搜索错误: {str(e)}")
        return None

async def main():
    """演示代理用法的异步主函数。"""
    agent = Agent(
        retriever=retriever,
        search_knowledge=True,
        instructions="在知识库中搜索信息",
        show_tool_calls=True,
    )

    # 加载知识库（首次运行时取消注释）
    await knowledge_base.aload(recreate=True)

    # 示例查询
    query = "列出制作 Massaman Gai 所需的配料"
    await agent.aprint_response(query, markdown=True)

if __name__ == "__main__":
    asyncio.run(main())
```

### 说明

1. **嵌入器和向量数据库设置**：我们首先定义一个嵌入器并初始化与向量数据库的连接。此设置对于将查询转换为嵌入并将其存储在数据库中至关重要。

2. **加载知识库**：知识库已加载 PDF 文档。此步骤涉及将文档转换为嵌入并将其存储在向量数据库中。

3. **自定义检索器函数**：`retriever` 函数用于处理文档检索。它接受一个查询，将其转换为嵌入，然后在向量数据库中搜索相关文档。

4. **代理初始化**：使用自定义检索器初始化代理。代理使用此检索器来搜索知识库并检索信息。

5. **示例查询**：`main` 函数演示了如何使用代理执行查询并从知识库检索信息。

## 开发资源

* 查看 [同步检索器](https://github.com/agno-agi/agno/blob/main/cookbook/agent_concepts/knowledge/custom/retriever.py)
* 查看 [异步检索器](https://github.com/agno-agi/agno/blob/main/cookbook/agent_concepts/knowledge/custom/async_retriever.py)