Setup
Install Agno, the Weaviate client, the OpenAI client, and the PDF reader:uv pip install -U agno weaviate-client openai pypdf
export OPENAI_API_KEY=xxx
docker run -d \
-p 8080:8080 \
-p 50051:50051 \
--name weaviate \
cr.weaviate.io/semitechnologies/weaviate:1.28.4
Example
agent_with_knowledge.py
from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.search import SearchType
from agno.vectordb.weaviate import Distance, VectorIndex, Weaviate
vector_db = Weaviate(
collection="recipes",
search_type=SearchType.hybrid,
vector_index=VectorIndex.HNSW,
distance=Distance.COSINE,
local=True, # Set to False if using Weaviate Cloud and True if using local instance
)
# Create knowledge base
knowledge_base = Knowledge(
vector_db=vector_db,
)
# Create and use the agent
agent = Agent(
knowledge=knowledge_base,
search_knowledge=True,
)
if __name__ == "__main__":
knowledge_base.insert(
url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
)
agent.print_response("How to make Thai curry?", markdown=True)
Async Support
Use asynchronous operations when knowledge loading or search must not block the event loop.
async_weaviate_db.py
import asyncio
from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.search import SearchType
from agno.vectordb.weaviate import Distance, VectorIndex, Weaviate
vector_db = Weaviate(
collection="recipes_async",
search_type=SearchType.hybrid,
vector_index=VectorIndex.HNSW,
distance=Distance.COSINE,
local=True, # Set to False if using Weaviate Cloud and True if using local instance
)
# Create knowledge base
knowledge_base = Knowledge(
vector_db=vector_db,
)
agent = Agent(
knowledge=knowledge_base,
search_knowledge=True,
)
async def main():
await knowledge_base.ainsert(
url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
)
await agent.aprint_response("How to make Tom Kha Gai", markdown=True)
if __name__ == "__main__":
asyncio.run(main())
The async methods use
WeaviateAsyncClient for non-blocking vector operations.Weaviate Params
| Parameter | Type | Description | Default |
|---|---|---|---|
wcd_url | Optional[str] | Weaviate Cloud URL (or use WCD_URL env var) | None |
wcd_api_key | Optional[str] | Weaviate Cloud API key (or use WCD_API_KEY env var) | None |
client | Optional[weaviate.WeaviateClient] | Pre-configured Weaviate client | None |
local | bool | Whether to use a local Weaviate instance | False |
collection | str | Name of the Weaviate collection | "default" |
name | Optional[str] | Vector database name | None |
description | Optional[str] | Vector database description | None |
id | Optional[str] | Vector database ID. Generated when omitted | None |
vector_index | VectorIndex | Type of vector index (HNSW, FLAT, DYNAMIC) | VectorIndex.HNSW |
distance | Distance | Distance metric (COSINE, DOT, etc.) | Distance.COSINE |
embedder | Optional[Embedder] | Embedder to use for generating embeddings | OpenAIEmbedder() |
search_type | SearchType | Search type (vector, keyword, hybrid) | SearchType.vector |
reranker | Optional[Reranker] | Reranker to refine search results | None |
hybrid_search_alpha | float | Weighting factor for hybrid search | 0.5 |