Async PgVector Usage

import asyncio

from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.pgvector import PgVector

knowledge = Knowledge(
    vector_db=PgVector(
        table_name="async_recipes",
        db_url="postgresql+psycopg://ai:ai@localhost:5532/ai",
    )
)
agent = Agent(knowledge=knowledge, search_knowledge=True)


async def main() -> None:
    await knowledge.ainsert(
        url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
    )
    await agent.aprint_response("How do I make pad thai?", markdown=True)


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

Knowledge.ainsert() awaits reader and embedding work. PgVector uses a synchronous SQLAlchemy engine for database writes. PgVector.async_search() runs the synchronous search method in a worker thread.

Run the Example

Set up your virtual environment

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

Install dependencies

uv pip install -U agno openai pgvector psycopg pypdf sqlalchemy

Run PgVector

docker run -d \
  -e POSTGRES_DB=ai \
  -e POSTGRES_USER=ai \
  -e POSTGRES_PASSWORD=ai \
  -e PGDATA=/var/lib/postgresql \
  -v pgvolume:/var/lib/postgresql \
  -p 5532:5432 \
  --name pgvector \
  agnohq/pgvector:18

Export the API key

export OPENAI_API_KEY=your_openai_api_key_here

Run the example

python async_pgvector_db.py

Next Steps

Task	Guide
Use the synchronous API	PgVector usage
Configure search behavior	PgVector overview