Custom Retriever
Implement custom retrieval logic for full control over how agents search knowledge.
Custom retrievers let you implement your own search logic instead of using the default knowledge search. This is useful when you need to:
- Query external APIs or databases directly
- Implement custom ranking or filtering
- Reformulate queries before searching
- Combine multiple data sources
from typing import Optional
from agno.agent import Agent
def knowledge_retriever(query: str, num_documents: Optional[int] = None, **kwargs) -> list[dict]:
# Your custom retrieval logic here
return [{"content": "..."}]
agent = Agent(
knowledge_retriever=knowledge_retriever,
search_knowledge=True,
)How It Works
When the agent decides to search for information:
- The agent calls your
knowledge_retrieverfunction with the query - Your function retrieves documents however you want
- Results are returned to the agent as a list of dictionaries
- The agent uses the retrieved content to generate a response
Retriever Function Signature
from typing import Optional
from agno.agent import Agent
def knowledge_retriever(
query: str,
agent: Optional[Agent] = None,
num_documents: Optional[int] = None,
**kwargs
) -> Optional[list[dict]]:
"""
Args:
query: The search query from the agent
agent: The agent instance (optional, for accessing agent state)
num_documents: Number of documents to retrieve. The agent passes the
knowledge base's max_results, or None if no Knowledge is attached.
**kwargs: Additional arguments passed from the agent
Returns:
List of documents as dictionaries, or None if search fails
"""
# Your logic here
return [{"content": "..."}]Example: Direct Vector Database Query
This adaptation bypasses the Knowledge abstraction and queries an existing Qdrant service directly. Before running custom_retriever.py:
- Start Qdrant at
http://localhost:6333and populate arecipescollection with unnamed 1,536-dimensional vectors generated bytext-embedding-3-smalland document text in the payload. The Qdrant guide covers ingestion; ensure your collection matches this direct-query schema. - Create and activate a Python virtual environment, then install
agno openai qdrant-clientwithuv pip install. - Set
OPENAI_API_KEYfor both the query embedder and the agent’s default model.
The example does not create or populate the collection. Its exception handler returns None on retrieval failure, so an agent answer alone does not prove the search succeeded.
from typing import Optional
from agno.agent import Agent
from agno.knowledge.embedder.openai import OpenAIEmbedder
from qdrant_client import QdrantClient
embedder = OpenAIEmbedder(id="text-embedding-3-small")
qdrant_client = QdrantClient(url="http://localhost:6333")
def knowledge_retriever(
query: str, num_documents: Optional[int] = None, **kwargs
) -> Optional[list[dict]]:
try:
# Generate embedding for the query
query_embedding = embedder.get_embedding(query)
# Search Qdrant directly
results = qdrant_client.query_points(
collection_name="recipes",
query=query_embedding,
limit=num_documents or 5,
)
return results.model_dump().get("points")
except Exception as e:
print(f"Search error: {e}")
return None
agent = Agent(
knowledge_retriever=knowledge_retriever,
search_knowledge=True,
)
agent.print_response("What ingredients do I need for Massaman Gai?")Example: Query Reformulation
Expand or modify queries before searching:
from typing import Optional
from agno.knowledge.knowledge import Knowledge
knowledge = Knowledge(vector_db=vector_db)
def knowledge_retriever(query: str, num_documents: Optional[int] = None, **kwargs) -> list[dict]:
# Expand common terms
expanded_query = query.replace("vacation", "vacation PTO paid time off")
expanded_query = expanded_query.replace("WFH", "work from home remote")
# Search with expanded query
results = knowledge.search(expanded_query, max_results=num_documents)
return [doc.to_dict() for doc in results]Example: Multi-Source Retrieval
Combine results from multiple knowledge bases:
def knowledge_retriever(query: str, num_documents: Optional[int] = None, **kwargs) -> list[dict]:
# Search multiple sources
policy_results = policy_knowledge.search(query, max_results=3)
faq_results = faq_knowledge.search(query, max_results=3)
# Combine and deduplicate
all_results = []
seen_ids = set()
for doc in policy_results + faq_results:
if doc.id not in seen_ids:
all_results.append(doc.to_dict())
seen_ids.add(doc.id)
return all_results[:num_documents]When to Use Custom Retrievers
| Use Case | Why Custom Retriever |
|---|---|
| Direct database access | Skip the Knowledge abstraction for performance |
| Query expansion | Add synonyms or related terms before searching |
| Multi-source search | Combine results from multiple knowledge bases |
| External APIs | Search third-party services (Elasticsearch, Algolia, etc.) |
| Custom ranking | Implement domain-specific relevance scoring |
| Conditional logic | Apply different search strategies based on query type |
For most use cases, the built-in Knowledge search is sufficient. Use custom retrievers when you need full control over the retrieval process.