Valkey Async

Add and query knowledge asynchronously with ValkeyDB as the vector database.

Code

import asyncio

from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.valkey import ValkeyDB
from agno.vectordb.search import SearchType

# Initialize Valkey Vector DB
vector_db = ValkeyDB(
    index_name="agno_docs",
    host="localhost",
    port=6379,
    search_type=SearchType.vector,
)

# Build a Knowledge base backed by Valkey
knowledge = Knowledge(
    name="My Valkey Knowledge Base",
    description="Knowledge base using Valkey as the vector store",
    vector_db=vector_db,
)


async def main():
    # Add content (async)
    await knowledge.ainsert(
        name="Recipes",
        url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
        metadata={"category": "recipe_book"},
        skip_if_exists=True,
    )

    # Query with an Agent (async)
    agent = Agent(knowledge=knowledge)
    await agent.aprint_response("List down the ingredients to make Massaman Gai", markdown=True)


if __name__ == "__main__":
    asyncio.run(main())

Usage

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Install dependencies

uv pip install -U valkey-glide-sync pypdf openai agno

Set your OpenAI API key

export OPENAI_API_KEY="your-api-key"

Run Valkey

docker run -d --name my-valkey -p 6379:6379 valkey/valkey-bundle

Run Agent

python async_valkey_db.py