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
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.add_content(
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,
)
# Comment out after first run
agent.knowledge.load(recreate=False) # type: ignore
agent.print_response("How do I make pad thai?", markdown=True)
agent.print_response("What was my last question?", stream=True)
Clickhouse also supports asynchronous operations, enabling concurrency and leading to better performance.
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.add_content_async(
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))
aload()
and aprint_response()
methods with asyncio.run()
for non-blocking operations in high-throughput applications.