设置
请按照 Qdrant 设置指南 中的说明在本地安装 Qdrant。这是获取 API 密钥的指南：Qdrant API 密钥。示例：PDF 知识库的自定义检索器
下面是使用agno 库实现自定义检索器函数的详细示例。此示例演示了如何设置包含 PDF 文档的知识库、定义自定义检索器，并将其与代理一起使用。
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()
异步实现
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())
说明
- 嵌入器和向量数据库设置：我们首先定义一个嵌入器并初始化与向量数据库的连接。此设置对于将查询转换为嵌入并将其存储在数据库中至关重要。
- 加载知识库：知识库已加载 PDF 文档。此步骤涉及将文档转换为嵌入并将其存储在向量数据库中。
-
自定义检索器函数：
retriever函数用于处理文档检索。它接受一个查询，将其转换为嵌入，然后在向量数据库中搜索相关文档。 - 代理初始化：使用自定义检索器初始化代理。代理使用此检索器来搜索知识库并检索信息。
-
示例查询：
main函数演示了如何使用代理执行查询并从知识库检索信息。