Setup
uv pip install -U clickhouse-connect pypdf openai sqlalchemy agno
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
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,
)
if __name__ == "__main__":
# Comment out after first run
asyncio.run(agent.knowledge.ainsert(
url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
)
)
# Create and use the agent
asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True))
Use
ainsert() and aprint_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.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 |