设置
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
示例
agent_with_knowledge.py
from agno.agent import Agent
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.storage.sqlite import SqliteStorage
from agno.vectordb.clickhouse import Clickhouse
agent = Agent(
storage=SqliteStorage(table_name="recipe_agent"),
knowledge=PDFUrlKnowledgeBase(
urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=Clickhouse(
table_name="recipe_documents",
host="localhost",
port=8123,
username="ai",
password="ai",
),
),
# Show tool calls in the response
show_tool_calls=True,
# 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)
Async Support ⚡
Clickhouse 也支持异步操作，能够实现并发并带来更好的性能。
async_clickhouse.py
import asyncio
from agno.agent import Agent
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.storage.agent.sqlite import SqliteAgentStorage
from agno.vectordb.clickhouse import Clickhouse
agent = Agent(
storage=SqliteAgentStorage(table_name="recipe_agent"),
knowledge=PDFUrlKnowledgeBase(
urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=Clickhouse(
table_name="recipe_documents",
host="localhost",
port=8123,
username="ai",
password="ai",
),
),
# Show tool calls in the response
show_tool_calls=True,
# 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.aload(recreate=False))
# Create and use the agent
asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True))
在吞吐量高的应用中使用
aload() 和 aprint_response() 方法配合 asyncio.run() 即可实现非阻塞操作。