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

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

Пример приложения

В этом примере создаётся полнофункциональное веб-приложение, предоставляющее чат-интерфейс для выполнения запросов к данным ClickHouse. Исходный код этого примера вы можете найти в репозитории examples.

Предварительные требования

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

Выполните следующие шаги для создания приложения Streamlit.

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

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

pip install streamlit agno ipywidgets

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

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

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="ваш_ключ_api"

Использование другого провайдера LLM

Если у вас нет ключа API Anthropic и вы хотите использовать другого провайдера LLM, вы можете найти инструкции по настройке учетных данных в документации Agno «Integrations»

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

Начните с создания основного файла приложения 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

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

Добавьте основную функцию агента, которая подключается к SQL-песочнице 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("""\
                    Вы — ассистент ClickHouse. Помогайте пользователям выполнять запросы и работать с данными в ClickHouse.
                    - Выполняйте SQL-запросы с помощью инструмента ClickHouse MCP
                    - Представляйте результаты в виде таблиц markdown, когда это уместно
                    - Выводите информацию кратко, полезно и в удобном формате
                """),
                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)  # Сигнальное значение для завершения потока
    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("ИИ-агент на базе ClickHouse")

if st.button("💬 Новый чат"):
  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("Чем могу помочь?"):
  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})

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

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

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

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