ClickHouse Vector Database

Use ClickHouse as a vector database for your Knowledge Base.

Setup

uv pip install -U clickhouse-connect pypdf openai sqlalchemy agno

Run ClickHouse with Docker:

docker run -d \
  -e CLICKHOUSE_DB=ai \
  -e CLICKHOUSE_USER=ai \
  -e CLICKHOUSE_PASSWORD=ai \
  -e CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1 \
  -v clickhouse_data:/var/lib/clickhouse/ \
  -v clickhouse_log:/var/log/clickhouse-server/ \
  -p 8123:8123 \
  -p 9000:9000 \
  --ulimit nofile=262144:262144 \
  --name clickhouse-server \
  clickhouse/clickhouse-server

Set your OpenAI API key. The examples below use the default OpenAI embedder and model.

export OPENAI_API_KEY=xxx

Example

agent_with_knowledge.py

from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.db.sqlite import SqliteDb
from agno.vectordb.clickhouse import Clickhouse

knowledge=Knowledge(
    vector_db=Clickhouse(
        table_name="recipe_documents",
        host="localhost",
        port=8123,
        username="ai",
        password="ai",
    ),
)

knowledge.insert(
  url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
)

agent = Agent(
    db=SqliteDb(db_file="agno.db"),
    knowledge=knowledge,
    # Enable the agent to search the knowledge base
    search_knowledge=True,
    # Enable the agent to read the chat history
    read_chat_history=True,
)

agent.print_response("How do I make pad thai?", markdown=True)
agent.print_response("What was my last question?", stream=True)

Async Support ⚡

ClickHouse also supports asynchronous operations, enabling concurrency and leading to better performance.

async_clickhouse.py

import asyncio

from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.db.sqlite import SqliteDb
from agno.vectordb.clickhouse import Clickhouse

agent = Agent(
    db=SqliteDb(db_file="agno.db"),
    knowledge=Knowledge(
        vector_db=Clickhouse(
            table_name="recipe_documents",
            host="localhost",
            port=8123,
            username="ai",
            password="ai",
        ),
    ),
    # Enable the agent to search the knowledge base
    search_knowledge=True,
    # Enable the agent to read the chat history
    read_chat_history=True,
)

async def main():
    # Comment out after first run
    await agent.knowledge.ainsert(
        url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
    )

    # Create and use the agent
    await agent.aprint_response("How to make Tom Kha Gai", markdown=True)

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

Use ainsert() and aprint_response() to integrate these operations into an async application. Await both in the same event loop.

ClickHouse Params

Parameter	Type	Default	Description
table_name	str	-	Name of the table to store vectors and metadata in Clickhouse
host	str	-	Hostname of the Clickhouse server
username	Optional[str]	None	Username for Clickhouse authentication
password	str	""	Password for Clickhouse authentication
port	int	0	Port number for Clickhouse connection
database_name	str	"ai"	Name of the database to use in Clickhouse
dsn	Optional[str]	None	DSN string for Clickhouse connection
compress	str	"lz4"	Compression algorithm to use
client	Optional[Client]	None	Optional pre-configured Clickhouse client
asyncclient	Optional[AsyncClient]	None	Optional pre-configured async Clickhouse client
embedder	Optional[Embedder]	OpenAIEmbedder()	Embedder instance to generate embeddings
distance	Distance	Distance.cosine	Distance metric to use for similarity search
index	Optional[HNSW]	HNSW()	HNSW index configuration for vector similarity search