Upstash Vector Database

Use Upstash Vector as a serverless vector database for your Knowledge Base.

Setup

uv pip install -U upstash-vector pypdf openai agno

Create an index in the Upstash Console, then copy its REST URL and token.

export UPSTASH_VECTOR_REST_URL="your-index-url"
export UPSTASH_VECTOR_REST_TOKEN="your-index-token"

The example uses OpenAI for the agent model, so set your API key:

export OPENAI_API_KEY=xxx

Example

agent_with_knowledge.py
import os

from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.upstashdb import UpstashVectorDb

vector_db = UpstashVectorDb(
    url=os.getenv("UPSTASH_VECTOR_REST_URL"),
    token=os.getenv("UPSTASH_VECTOR_REST_TOKEN"),
)

knowledge = Knowledge(
    name="Upstash Knowledge Base",
    vector_db=vector_db,
)

knowledge.insert(
    name="Recipes",
    url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
    metadata={"doc_type": "recipe_book"},
)

agent = Agent(knowledge=knowledge)
agent.print_response("How to make Pad Thai?", markdown=True)

Without an embedder, UpstashVectorDb uses Upstash's hosted embedding models. Create your index with an embedding model in the Upstash Console. To embed locally instead, pass an embedder and make sure its dimension matches the index.

Async Support ⚡

Upstash can be used through Agno's async ingestion and agent APIs. Its adapter still performs synchronous index writes; Knowledge async search falls back to synchronous search on the event-loop thread.

async_upstash_db.py
import asyncio
import os

from agno.agent import Agent
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.upstashdb import UpstashVectorDb

vector_db = UpstashVectorDb(
    url=os.getenv("UPSTASH_VECTOR_REST_URL"),
    token=os.getenv("UPSTASH_VECTOR_REST_TOKEN"),
    dimension=1536,
    embedder=OpenAIEmbedder(enable_batch=True),
)

knowledge = Knowledge(
    vector_db=vector_db,
)

agent = Agent(knowledge=knowledge)

if __name__ == "__main__":
    asyncio.run(
        knowledge.ainsert(
            url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
        )
    )

    asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True))

The async API preserves the application calling style, but does not make the Upstash adapter nonblocking. Move blocking work to workers when required.

UpstashVectorDb Params

ParameterTypeDefaultDescription
urlstr-The Upstash Vector database URL.
tokenstr-The Upstash Vector API token.
retriesint3Number of retry attempts for operations.
retry_intervalfloat1.0Time between retries, in seconds.
dimensionintNoneThe dimension of the embeddings. Validated against the index dimension on first use.
embedderEmbedderNoneThe embedder to use. When None, the index must use an Upstash hosted embedding model.
namespacestr""The namespace to use.
rerankerRerankerNoneThe reranker to use.
namestrNoneName of the vector database.
descriptionstrNoneDescription of the vector database.