Skip to content
ClickHouse Docs
ClickHouse DocsClickHouse Docs
49.6kGet Started

Как создать ИИ-агента на базе ClickHouse с помощью Streamlit

View as MarkdownEdit this pageReport an issueSet up docs MCP

В этом руководстве вы узнаете, как создать веб-ИИ-агента с помощью Streamlit, который может взаимодействовать с Песочницей ClickHouse через MCP-сервер ClickHouse и Agno.

Необходимые условия

  • В системе должен быть установлен Python. Также должен быть установлен uv
  • Вам потребуется API-ключ Anthropic или API-ключ другого провайдера LLM

Чтобы создать приложение Streamlit, выполните следующие шаги.

Установка библиотек

Установите необходимые библиотеки, выполнив следующие команды:

pip install streamlit agno ipywidgets

Создайте файл утилит

Создайте файл utils.py с двумя вспомогательными функциями. Первая — это асинхронная функция-генератор для обработки потоковых ответов от агента Agno. Вторая — функция для применения стилей к приложению Streamlit:

utils.pypython
import streamlit as st
from agno.run.response import RunEvent, RunResponse

async def as_stream(response):
    async for chunk in response:
        if isinstance(chunk, RunResponse) and isinstance(chunk.content, str):
            if chunk.event == RunEvent.run_response:
                yield chunk.content

def apply_styles():
    st.markdown("""
  <style>
  hr.divider {
  background-color: white;
  margin: 0;
  }
  </style>
  <hr class='divider' />""", unsafe_allow_html=True)

Настройка учетных данных

Задайте API-ключ Anthropic в переменной окружения:

export ANTHROPIC_API_KEY="your_api_key_here"

Импортируйте необходимые библиотеки

Начните с создания основного файла приложения Streamlit (например, app.py) и добавьте необходимые импорты:

from utils import apply_styles

import streamlit as st
from textwrap import dedent

from agno.models.anthropic import Claude
from agno.agent import Agent
from agno.tools.mcp import MCPTools
from agno.storage.json import JsonStorage
from agno.run.response import RunEvent, RunResponse
from mcp.client.stdio import stdio_client, StdioServerParameters

from mcp import ClientSession

import asyncio
import threading
from queue import Queue

Определите функцию потоковой передачи агента

Добавьте основную функцию агента, которая подключается к Песочнице ClickHouse и передаёт ответы в потоковом режиме:

async def stream_clickhouse_agent(message):
    env = {
            "CLICKHOUSE_HOST": "sql-clickhouse.clickhouse.com",
            "CLICKHOUSE_PORT": "8443",
            "CLICKHOUSE_USER": "demo",
            "CLICKHOUSE_PASSWORD": "",
            "CLICKHOUSE_SECURE": "true"
        }
    
    server_params = StdioServerParameters(
        command="uv",
        args=[
        'run',
        '--with', 'mcp-clickhouse',
        '--python', '3.13',
        'mcp-clickhouse'
        ],
        env=env
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            mcp_tools = MCPTools(timeout_seconds=60, session=session)
            await mcp_tools.initialize()
            agent = Agent(
                model=Claude(id="claude-3-5-sonnet-20240620"),
                tools=[mcp_tools],
                instructions=dedent("""\
                    You are a ClickHouse assistant. Help users query and understand data using ClickHouse.
                    - Run SQL queries using the ClickHouse MCP tool
                    - Present results in markdown tables when relevant
                    - Keep output concise, useful, and well-formatted
                """),
                markdown=True,
                show_tool_calls=True,
                storage=JsonStorage(dir_path="tmp/team_sessions_json"),
                add_datetime_to_instructions=True, 
                add_history_to_messages=True,
            )
            chunks = await agent.arun(message, stream=True)
            async for chunk in chunks:
                if isinstance(chunk, RunResponse) and chunk.event == RunEvent.run_response:
                    yield chunk.content

Добавьте синхронные функции-обёртки

Добавьте вспомогательные функции для работы с асинхронным стримингом в Streamlit:

def run_agent_query_sync(message):
    queue = Queue()
    def run():
        asyncio.run(_agent_stream_to_queue(message, queue))
        queue.put(None)  # Sentinel to end stream
    threading.Thread(target=run, daemon=True).start()
    while True:
        chunk = queue.get()
        if chunk is None:
            break
        yield chunk

async def _agent_stream_to_queue(message, queue):
    async for chunk in stream_clickhouse_agent(message):
        queue.put(chunk)

Создайте интерфейс Streamlit

Добавьте компоненты интерфейса Streamlit и функцию чата:

st.title("A ClickHouse-backed AI agent")

if st.button("💬 New Chat"):
  st.session_state.messages = []
  st.rerun()

apply_styles()

if "messages" not in st.session_state:
  st.session_state.messages = []

for message in st.session_state.messages:
  with st.chat_message(message["role"]):
    st.markdown(message["content"])

if prompt := st.chat_input("What is up?"):
  st.session_state.messages.append({"role": "user", "content": prompt})
  with st.chat_message("user"):
    st.markdown(prompt)
  with st.chat_message("assistant"):
    response = st.write_stream(run_agent_query_sync(prompt))
  st.session_state.messages.append({"role": "assistant", "content": response})

Запустите приложение

Чтобы запустить веб-приложение вашего ClickHouse AI-агента, выполните в терминале следующую команду:

uv run \
  --with streamlit \
  --with agno \
  --with anthropic \
  --with mcp \
  streamlit run app.py --server.headless true

Это откроет ваш веб-браузер и перейдет на страницу http://localhost:8501, где вы сможете взаимодействовать со своим AI-агентом и задавать ему вопросы о демонстрационных наборах данных, доступных в Песочнице ClickHouse.

Navigation