Clickhouse Agent 知识

设置

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() 即可实现非阻塞操作。

开发者资源

• 查看 Cookbook (Sync)
• 查看 Cookbook (Async)

Assistant

Responses are generated using AI and may contain mistakes.