Documentation Index
Fetch the complete documentation index at: /llms.txt
Use this file to discover all available pages before exploring further.
Use Weaviate as a vector database for your Knowledge Base.
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
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)
These methods await Weaviate client I/O. In Agno v3.0.4, PDF parsing and embedding still run synchronously and can block the event loop. Offload large ingestion jobs or slow embedding calls to a worker.
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())
WeaviateAsyncClient for non-blocking vector operations.| 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 |
Was this page helpful?