Weaviate Vector Database
Use Weaviate as a vector database for your Knowledge Base.
Setup
Install Agno, the Weaviate client, the OpenAI client, and the PDF reader:
uv pip install -U agno weaviate-client openai pypdfSet your OpenAI API key for the default embedder and agent model:
export OPENAI_API_KEY=xxxStart Weaviate locally:
docker run -d \
-p 8080:8080 \
-p 50051:50051 \
--name weaviate \
cr.weaviate.io/semitechnologies/weaviate:1.28.4See Weaviate's local Docker quickstart for other deployment options.
Example
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
These methods await Weaviate client I/O and async embeddings during insertion. Query embeddings and PDF parsing still run synchronously and can block the event loop. Offload that work when needed for your workload.
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 |