--- title: "Агент-аналітик: архітектура" description: Чат-інтерфейс, який сам пише SQL до ClickHouse — ReAct-агент на LangChain зі стрімінгом через SSE. --- Документ описує архітектуру агента-аналітика для маркетологів: чат-інтерфейс, який відповідає на аналітичні питання, сам пише SQL до ClickHouse через власні тули і повертає результати табличками. Стек: FastAPI, **ReAct-агент** (`create_agent` з LangChain 1.0, поверх LangGraph) + `langchain-anthropic` (прямий Anthropic API з prompt caching), `clickhouse-connect` (прямий HTTP-клієнт до ClickHouse), React-фронт зі стрімінгом через SSE. Дані по креативах доїжджають з MongoDB у ClickHouse окремим worker'ом. ## Загальна схема ``` ┌──────────────┐ SSE ┌─────────────────────────┐ │ UI (React) │ ◄────────────────► │ FastAPI │ │ чат + грід │ POST /chat │ /threads, /messages, │ └──────────────┘ GET /results │ /results/{id} (CSV) │ └───────────┬─────────────┘ │ astream ┌───────────▼─────────────┐ │ ReAct (create_agent) │ │ write_todos (план) │ │ explore_schema (subag.)│ │ checkpointer: MongoDB │ └───────────┬─────────────┘ │ run_sql / render_chart / schema-тули ┌───────────▼─────────────┐ │ ClickHouseGateway │ │ guard → preflight → │ │ execute → CSV persist │ └───────────┬─────────────┘ │ readonly user agent_analyst (HTTP 8123) ┌──────────────┐ kumquat-ch-sync ┌───────────▼─────────────┐ │ MongoDB │ ─────────────────► │ ClickHouse cluster ml │ │ creatives │ Airflow, щогодини │ creative_performance │ └──────────────┘ │ + creatives │ └─────────────────────────┘ ``` > **Історія рішення: чому пішли з MCP.** Перша версія ходила в ClickHouse через sidecar > `mcp-clickhouse`. Це давало готові тули, але: (1) `run_select_query` повертає **весь** > результат у контекст моделі — дорого і шумно; (2) зайвий процес у поді і зайвий hop > (HTTP → MCP → HTTP); (3) формат відповіді MCP змінювався між версіями (парсили > content-блоки). Замінили на прямий `clickhouse-connect` + власний `ClickHouseGateway`: > guard, cost preflight (`EXPLAIN ESTIMATE`), ліміти сесії і персист повного результату в > CSV — модель бачить лише прев'ю. ## Агент: ReAct (`create_agent`) + TodoListMiddleware Ядро — `create_agent` з LangChain 1.0 (звичайний ReAct tool-loop на LangGraph) плюс один middleware — `TodoListMiddleware`, який дає тул `write_todos` для планування (з нього UI будує чекліст плану). > **Історія рішення.** Перша версія була на > [`deepagents`](https://docs.langchain.com/oss/python/deepagents/overview) — harness з > плануванням, віртуальною файловою системою і сабагентами. Для нашого профілю задачі > (питання → 1–5 SQL-запитів → висновок) з усього цього реально використовувались лише > `write_todos` і AGENT.md; решта (9 вбудованих тулів, ~2.7k токенів схем у кожному > LLM-виклику, файли у стейті → важкі чекпоінти, subgraph-стрімінг сабагентів) була чистим > оверхедом. Мігрували на голий `create_agent`: планування лишили через `TodoListMiddleware`, > AGENT.md переїхав у system prompt (статичний префікс краще кешується). Якщо сабагенти > знадобляться — `deepagents.middleware.subagents.SubAgentMiddleware` підключається до > `create_agent` як звичайний middleware, без повернення на deepagents. ```python from langchain.agents import create_agent from langchain.agents.middleware import TodoListMiddleware from langchain_anthropic import ChatAnthropic gateway = ClickHouseGateway(settings.clickhouse, settings.analyst_agent) agent = create_agent( model=ChatAnthropic(model_name="claude-sonnet-5", max_tokens_to_sample=4096), system_prompt=build_system_prompt(semantic), # промпт + AGENT.md + семантичний шар tools=build_agent_tools(gateway, semantic, settings), # run_sql, validate_sql, lookup_metric, # render_chart, schema-тули, explore_schema middleware=[ TodoListMiddleware(), # write_todos → SSE-подія plan SummarizationMiddleware( # стискає контекст довгих тредів; model=ChatAnthropic(...), # UI не зачіпає — історія рендериться trigger=("tokens", 60_000), # з agent__messages, не з graph state keep=("messages", 20), ), ToolCallLimitMiddleware(run_limit=20), # м'яка межа: тул блокується, модель # сама завершує відповідь AnthropicPromptCachingMiddleware(ttl="1h"), # кеш-брейкпоінти: system + tools + # хвіст історії; 1h переживає паузи ], # «людина думає між ходами» checkpointer=mongodb_checkpointer, # треди, продовження діалогу ) ``` - **`ClickHouseGateway`** (`gateway.py`) — єдиний шлях від агента до сховища: guard (`sqlglot`) → cost preflight (`EXPLAIN ESTIMATE`, стеля рядків конфігом `estimate_row_ceiling`) → виконання із session-лімітами (`max_execution_time`, `max_result_rows`, ... — сервер дозволяє лише знижувати) → персист повного результату в CSV (`results_dir`, emptyDir у поді; найстаріші файли підчищаються понад `results_max_files`). - **`run_sql(sql, purpose)`** — головний робочий тул: - guard валідує, що це один read-statement (SELECT/WITH/DESCRIBE/SHOW/EXPLAIN), дописує `LIMIT`, якщо його нема — *convenience layer* для швидких, зрозумілих моделі помилок; справжній guardrail — права ClickHouse-юзера (нижче); - у контекст моделі повертає **markdown-прев'ю (20 рядків за замовчуванням, модель може попросити більше через `preview_rows`, кап 200)** + row/scan-статистику + `result_id` CSV-файла з повним результатом — не весь результат; - повні дані UI отримує окремою SSE-подією `table` (до 1000 рядків) з `result_id` для скачування повного CSV через `GET /api/v1/agent/results/{result_id}`. - **`validate_sql`** — перевірка без виконання: guard + план-оцінка вартості. Для запитів по широких вікнах дат — перед `run_sql`. - **`lookup_metric`** — канонічна формула метрики з семантичного шару; модель не вигадує власних визначень CPA/ROAS. - **Schema-тули** (`list_tables`, `describe_table`, `show_create_table`, `profile_column`) — інтроспекція через `system.*` (той самий gateway, без персисту CSV). Основний агент з них отримує лише `show_create_table` (потрібен для дешевих фільтрів перед запитом); решта живе тільки в explorer-сабагенті, щоб списки таблиць і describe-дампи не роздували основний контекст. - **`explore_schema`** — **сабагент** schema-explorer: окремий `create_agent` на дешевій моделі (`explorer_model`, haiku) лише зі schema-тулами. Інтроспекція токеноємна і низько-судженнєва: DDL-дампи і профілі значень не потрапляють в основний контекст — сабагент повертає компактний звіт. Основний агент викликає його як звичайний тул. - **`render_chart`** — той самий патерн для графіків: модель передає SQL + специфікацію (`chart_type` line/bar/area/pie, `title`, `x_key`, `series`), тул сам виконує запит і шле UI подію `chart` з готовими для recharts рядками-словниками. Дані йдуть повз контекст моделі — вона бачить лише прев'ю з `run_sql`, тому передати великий ряд сама не змогла б. Числові рядки (ClickHouse Decimal/UInt64 приходять через JSON стрінгами) коерсяться у float. - **Checkpointer — MongoDB** (`langgraph-checkpoint-mongodb`): Postgres в інфраструктурі немає, а Mongo вже є. Той самий інстанс, що й основна база kumquat, колекції за неймінг-конвенцією (`agent__checkpoints`, `agent__checkpoint_writes`). З версії 0.4 бібліотека лишила тільки синхронний `MongoDBSaver` (async-виклики йдуть через executor базового класу) — саме він у `analyst_agent/checkpoint.py`. In-memory — тільки для локальної розробки. Checkpointer також обов'язковий для interrupt'ів (clarify) у LangGraph. - **Self-correction безкоштовно:** помилка ClickHouse повертається як tool result, цикл сам переписує запит. Ліміти двоярусні: `ToolCallLimitMiddleware(run_limit=20)` — м'яка межа з graceful-завершенням, `recursion_limit=50` — жорсткий бекстоп (має лишати м'якій межі простір спрацювати першою). - **Clarify (фаза 2):** уточнювальні питання — окремий тул `ask_user` через `interrupt` LangGraph (потребує checkpointer, він у нас і так є). UI рендерить interrupt як питання з варіантами. ### Модель LLM — **прямий Anthropic API** (`langchain-anthropic`, ключ `ANTHROPIC_API_KEY`), модель — конфігом `ANALYST_AGENT__MODEL` (дефолт `claude-sonnet-5`; на складних агентних сесіях можна підняти до Opus, але спершу поміряти якість на реальних питаннях). Головний важіль економії — **prompt caching**: system prompt із семантичним шаром + AGENT.md — великий статичний префікс (~13k символів), що повторюється в кожному кроці агентного циклу. Кешуванням займається `AnthropicPromptCachingMiddleware(ttl="1h")`: брейкпоінти на system prompt, tool definitions і хвості історії; кешоване читання коштує 10% від ціни входу, на практиці це −50–80% вартості вхідних токенів. TTL 1h (а не дефолтні 5m), бо людина думає над відповіддю довше 5 хвилин — мертвий кеш дорожчий за дорожчий запис. Звіряти по `cache_read_input_tokens` в usage-звіті. > **Історія: чому пішли з OpenRouter.** Перша версія ходила через OpenRouter (один > ключ/біллінг, перемикання провайдерів конфігом), але він бере ~5% комісії при поповненні > кредитів, а кешування через посередника — додатковий шар, де воно може мовчки не > спрацювати. Після фіксації на Anthropic-only посередник не давав нічого. Відкат назад, > якщо знадобиться, — дзеркальний: залежність `langchain-anthropic` → `langchain-openrouter`, > `ANALYST_AGENT__MODEL`, ключ. ## Шар даних: ClickHouse - **Прямий клієнт** `clickhouse-connect` (HTTP 8123), конфіг — спільна секція `CLICKHOUSE__*`; юзер/пароль — з секрета `analyst-agent-secrets`. - **Окремий read-only юзер `agent_analyst`** — головний guardrail. Python-валідація в `guard.py` — лише для швидких помилок; зупиняє запис саме ClickHouse. - **Doctor-скрипт** (`python -m analyst_api.doctor`) — preflight проти дрейфу семантичного шару: таблиці/колонки існують, метрики виконуються, джойни валідні, freshness партицій, write-проба відхиляється сервером. `--offline` — лише синтаксис, для CI; exit code 1 — deploy gate. За правилами кластера (кореневий `CLAUDE.md`): **всі DDL — `ON CLUSTER 'ml'`**, гранти перевіряти на всіх репліках. Додати в `db/init.sql`: ```sql -- readonly = 2 дозволяє SELECT і дозволяє клієнту ЗНИЖУВАТИ ліміти, -- але забороняє INSERT/ALTER/DDL. readonly = 1 заборонив би клієнту -- передавати будь-які settings взагалі (наш gateway їх шле) — тому 2 -- плюс явні MAX-обмеження, вище яких клієнт піднятись не може. CREATE SETTINGS PROFILE IF NOT EXISTS agent_analyst_profile ON CLUSTER 'ml' SETTINGS readonly = 2, allow_ddl = 0, max_execution_time = 60 MAX 180, max_result_rows = 100000 MAX 1000000, max_result_bytes = 100000000 MAX 1000000000, max_rows_to_read = 2000000000 MAX 20000000000, max_bytes_to_read = 200000000000 MAX 2000000000000, max_memory_usage = 8000000000 MAX 32000000000, max_threads = 8 MAX 16, result_overflow_mode = 'break', timeout_overflow_mode = 'throw', max_concurrent_queries_for_user = 8; CREATE USER IF NOT EXISTS agent_analyst ON CLUSTER 'ml' IDENTIFIED WITH sha256_password BY '<>' SETTINGS PROFILE agent_analyst_profile; GRANT ON CLUSTER 'ml' SELECT ON kumquat_market_analytics.* TO agent_analyst; -- Інтроспекція для schema-тулів. Список тримати вузьким; він дзеркалить -- ALLOWED_SYSTEM_TABLES в app/ch/guard.py. GRANT ON CLUSTER 'ml' SELECT ON system.tables TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.columns TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.databases TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.parts TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.parts_columns TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.data_skipping_indices TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.table_engines TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.functions TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.numbers TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.one TO agent_analyst; GRANT ON CLUSTER 'ml' SELECT ON system.query_log TO agent_analyst; GRANT ON CLUSTER 'ml' SHOW TABLES, SHOW COLUMNS ON kumquat_market_analytics.* TO agent_analyst; -- Явно відрізаємо exfiltration surface: SOURCES гейтить url(), s3(), file(), -- remote(), jdbc(). За замовчуванням їх і так нема, але revoke страхує від -- випадкового успадкування через роль. REVOKE ON CLUSTER 'ml' SOURCES ON *.* FROM agent_analyst; REVOKE ON CLUSTER 'ml' INTROSPECTION ON *.* FROM agent_analyst; -- Обмежити blast radius агентного циклу, що пішов у рознос. CREATE QUOTA IF NOT EXISTS agent_analyst_quota ON CLUSTER 'ml' KEYED BY user_name FOR INTERVAL 1 hour MAX queries = 2000, errors = 200, execution_time = 3600 TO agent_analyst; -- Перевірка (на КОЖНІЙ репліці, не через Service): -- SHOW GRANTS FOR agent_analyst; -- SELECT * FROM system.settings_profile_elements -- WHERE profile_name = 'agent_analyst_profile'; ``` ## Семантичний шар Найважливіша частина для якості відповідей. LLM не вгадає, що «Мета» в цій таблиці записана як `fb` (а не `facebook`), або що CPA рахується по `purch`, а не по `all_purch`. Шар — YAML-файли, які збираються в system prompt. Дві частини: описи таблиць (grain, колонки, граблі, приклади) і канонічні метрики/розрізи (metrics/dimensions/joins/conventions). ### semantic/tables.yaml — опис таблиць Опис таблиці відповідає на чотири питання, на які не відповідає `DESCRIBE TABLE`: що таке один рядок (grain), що означають значення колонок, як правильно рахувати метрики і де граблі. Описаних таблиць три: `creative_performance` (атрибуційний фід — гроші, конверсії, усі джерела), `creatives` / `creatives_current` (вимір картки) і `tiktok_creative_performance` (власний звіт TikTok — старти відео, 6-секундні перегляди, реакції). Остання **не** дублює першу: вона є єдиним місцем, де поміряно саме відео, і водночас має власні `spend`, `impressions` і власну атрибуцію, які з фідом розходяться. Через це в її `gotchas` окремим рядком стоїть заборона складати два набори конверсій, а метрика `spend` отримала явний `alias: total_spend` — інакше `AS spend` збігався б із сирою колонкою нової таблиці й результат був би неоднозначним (валідатор семантичної моделі таку колізію не пропускає). ```yaml version: 1 timezone: Europe/Kyiv default_database: kumquat_market_analytics tables: - name: kumquat_market_analytics.creative_performance description: > Щоденна перформанс-статистика рекламних креативів по всіх продуктах і платформах. Основна таблиця для питань про спенд, закупівлю, ефективність креативів і роботу дизайн-команди. # Найважливіший рядок усього опису. Без нього LLM зробить # SUM(spend_usd) по creo_name і задвоїть цифри, бо не знає, # що креатив живе в кількох адсетах. grain: > Один рядок = (date, campaign_name, adset_name, creo_name). Той самий креатив зустрічається в багатьох кампаніях/адсетах — метрики по креативу завжди агрегувати через SUM/GROUP BY. freshness: "Оновлюється Airflow-DAG щоночі; даних за сьогодні нема, за вчора — можуть бути неповними до ~08:00 UTC." time_column: date columns: - name: date description: "Дата показу (не дата створення креативу)." - name: product description: "Код продукту." values: {plantme: PlantIn, coins: CoinIn, rockin: RockIn, sportdex: SportDex} - name: platform description: "Платформа пристрою: ios / android / web." - name: media_source description: > Рекламне джерело закупівлі. Саме це мають на увазі, коли питають «по каналах». values: {fb: "Meta (FB+IG)", tiktok: TikTok, applovin: AppLovin} # інших значень у таблиці немає - name: creo_name description: > Канонічна назва креативу за неймінг-конвенцією (див. [Workflow](/backend/workflow)); джойниться на creatives.name. - name: spend_usd description: "Витрати в USD (єдина валюта таблиці)." - name: purch description: "Первинні покупки, атрибутовані креативу. Дефолт для CPA/CR." - name: all_purch description: "Усі покупки включно з повторними. Для LTV-питань, не для CPA." - name: base_purch description: "Покупки base-плану без апсейлів." - name: ltv description: "Прогнозний LTV когорти в USD (модельний, перераховується заднім числом ~60 днів)." # ... решта колонок за тим самим шаблоном gotchas: - "Порожній рядок '' у LowCardinality-колонках означає «невідомо» — NULL там не буває. Фільтрувати country != ''." - "spend без покупок — норма (рядок є, purchases=0); ділення тільки через nullIf." - "Тестові кампанії мають префікс 'test_' у campaign_name — виключати за замовчуванням." - name: kumquat_market_analytics.creatives description: > Довідник креативів, синхронізований з MongoDB (kumquat, колекція creatives). Атрибути креативу: тип, формат хука, джерело, статус тестування, VLM-розмітка. Для питань «які хуки працюють», «чиї креативи найкращі», «AI vs дизайнерські». grain: "Один рядок = один креатив (name). Оновлюється щогодини." gotchas: - "Engine ReplacingMergeTree (ключ — name): історичні версії рядка можуть співіснувати до merge. Читати через argMax(col, updated_at) GROUP BY name або FINAL." - "vlm_* колонки порожні, поки креатив не пройшов VLM-аналіз." ``` ### semantic/metrics.yaml — канонічні метрики і розрізи CPA можна порахувати чотирма способами (`purch`, `all_purch`, `base_purch`, по інсталах), і всі виглядають правдоподібно. Канонічні формули прибирають цю варіативність: ```yaml metrics: spend: sql: sum(spend_usd) format: currency tables: [creative_performance] impressions: sql: sum(impressions) tables: [creative_performance] clicks: sql: sum(clicks) tables: [creative_performance] ctr: sql: sum(clicks) / nullIf(sum(impressions), 0) description: Link click-through rate. format: percent higher_is_better: true cpm: sql: 1000 * sum(spend_usd) / nullIf(sum(impressions), 0) description: > Cost per thousand impressions. Зростання CPM при стабільному CTR — зазвичай тиск аукціону, а не деградація креативу. format: currency higher_is_better: false cpc: sql: sum(spend_usd) / nullIf(sum(clicks), 0) format: currency higher_is_better: false cpi: sql: sum(spend_usd) / nullIf(sum(installs), 0) description: Cost per install. format: currency higher_is_better: false cpa: sql: sum(spend_usd) / nullIf(sum(purch), 0) description: > Cost per purchase, за замовчуванням по purch (первинні покупки, платформна атрибуція). format: currency higher_is_better: false roas: sql: sum(ltv) / nullIf(sum(spend_usd), 0) description: > Прогнозний, на базі модельного LTV — озвучувати це в кожній відповіді. format: number higher_is_better: true trial_to_purchase_cr: sql: sum(purch) / nullIf(sum(trials), 0) format: percent higher_is_better: true creative_age_days: sql: dateDiff('day', min(creatives.launched_at), max(creative_performance.date)) description: Скільки днів креатив у ротації. Fatigue-аналіз крутиться навколо цього. tables: [creative_performance, creatives] dimensions: product: sql: product table: creative_performance platform: sql: platform table: creative_performance media_source: sql: media_source table: creative_performance week: sql: toMonday(date) table: creative_performance description: Тижні починаються з понеділка. country: sql: country table: creative_performance hook_format: sql: creatives.vlm_hook_format table: creatives creative_source: sql: creatives.source table: creatives performance_tier: sql: creatives.performance_tier table: creatives joins: - left: kumquat_market_analytics.creative_performance right: kumquat_market_analytics.creatives on: creative_performance.creo_name = creatives.name type: LEFT conventions: - > Ratio-метрики — завжди sum поверх sum. Ніколи не avg() порядкову ставку: це зважує креатив зі 100 показами так само, як креатив з 10 млн. - > При порівнянні креативів — мінімум 1000 показів на креатив, якщо не просили інакше; застосований поріг озвучувати у відповіді. - > creatives — ReplacingMergeTree: у джойнах брати актуальну версію через argMax(..., updated_at) або FINAL. - > Дефолтний період — останні 30 днів, якщо питання не вказує інше. Завжди фільтрувати по date, щоб працював partition pruning. - > campaign_name за замовчуванням НЕ фільтрувати: скоркард рахує всі кампанії, а маркер тесту стоїть усередині назви — positionCaseInsensitive(campaign_name, 'test') > 0, не LIKE 'test_%'. - > media_source має рівно три значення: fb, tiktok, applovin. ``` ### semantic/examples.yaml — few-shot пари Few-shot пари «питання → SQL» дають більший приріст якості, ніж будь-який тюнінг промпта: агент копіює стиль (nullIf, HAVING від малих вибірок, argMax для creatives, періоди по вчора). ```yaml examples: - question: "Топ-10 креативів по спенду за минулий місяць на Meta" sql: | SELECT creo_name, sum(spend_usd) AS spend, sum(spend_usd) / nullIf(sum(purch), 0) AS cpa FROM kumquat_market_analytics.creative_performance WHERE date >= date_trunc('month', today() - INTERVAL 1 MONTH) AND date < date_trunc('month', today()) AND media_source = 'fb' GROUP BY creo_name ORDER BY spend DESC LIMIT 10 - question: "Які формати хуків дають найкращий CPA за останні 30 днів?" sql: | SELECT c.hook_format, sum(p.spend_usd) AS spend, sum(p.spend_usd) / nullIf(sum(p.purch), 0) AS cpa FROM kumquat_market_analytics.creative_performance AS p LEFT JOIN ( SELECT name, argMax(vlm_hook_format, updated_at) AS hook_format FROM kumquat_market_analytics.creatives GROUP BY name ) AS c ON p.creo_name = c.name WHERE p.date >= today() - 30 AND p.date < today() AND c.hook_format != '' GROUP BY c.hook_format HAVING sum(p.impressions) > 1000 ORDER BY cpa ASC ``` Навіщо кожен блок: - **`grain`** — захист від найдорожчої помилки text-to-SQL: подвоєння метрик через неврахований рівень агрегації. - **`values` з розшифровкою** — маркетолог спитає «по Меті», а в даних `fb`; без мапінгу агент поверне нуль рядків. Саме так це й ламалося: у шарі стояли `facebook` / `google` / `apple` / `organic`, яких у таблиці не існує. - **`metrics` з готовим SQL** — одна канонічна формула на метрику. - **`conventions`** — правила, що застосовуються до кожного запиту, а не до однієї таблиці. - **`gotchas`** — те, що знає тільки команда. Блок росте: кожна знайдена помилка агента перетворюється на рядок тут. > **Застереження:** семантика `purch` / `all_purch` / `base_purch` досі частково припущена — > звірити з маркетологами. Помилка в описі гірша за його відсутність: агент буде впевнено > помилятися. Значення `media_source` (`fb` / `tiktok` / `applovin`) і маркер тест-кампанії > в назві заміряні по джерелу 2026-08-02 — це вже не припущення. ## AGENT.md — контекст команди Нотатки про команду (конвенції звітності, відомі дані-граблі), які `build_system_prompt()` вшиває в system prompt секцією «Team context». Тримати коротко — це в контекст-вікні кожного ходу і конкурує з самим питанням. Довгоживуче і структурне належить семантичному шару, не сюди. ```markdown # Team context ## Хто питає Перформанс-маркетологи і growth lead. Вільно володіють CPA, ROAS, hook rate; SQL не пишуть. Пропускай розбір запиту, якщо не просили, але тримай «плівку» чесною — коли цифра здивує, вони підуть дивитись SQL. ## Конвенції звітності - Тиждень — понеділок–неділя. - «Минулий тиждень» = останній повний тиждень пн–нд, не trailing 7 днів. - «Цей місяць» = month-to-date. - Валюта — цілі числа від 1 000, два знаки нижче. Рейти — відсотки з одним знаком. ## Відомі граблі даних - Дані за сьогодні відсутні (нічний DAG); за вчора неповні до ~08:00 UTC. «Обвал у нуль за вчора» вранці — майже завжди це. - ltv модельний і перераховується заднім числом ~60 днів — ROAS свіжих когорт «пливе». Згадувати, коли період < 60 днів. - creatives.vlm_* порожні для непроаналізованих креативів — «немає розмітки» ≠ «немає хука». ## Стандартні преференції - Спочатку цифра, потім драйвер, потім застереження. - Якщо щось схоже на артефакт трекінгу, а не реальну зміну — казати про це першим, до аналізу. - Не рекомендувати зміну бюджетів на менш ніж 7 днях даних без прапорця про розмір вибірки. ``` ## Worker: MongoDB → ClickHouse (`kumquat-ch-sync`) Атрибути креативів живуть у MongoDB (`creatives`, див. `packages/kumquat-creatives/src/kumquat_creatives/models.py`), а агент читає тільки ClickHouse. Новий workspace-пакет `packages/kumquat-ch-sync` з CLI синхронізує колекцію в таблицю `kumquat_market_analytics.creatives`. ``` kumquat-ch-sync --mode=batch інкрементальний догон за watermark (updated_at). Дефолт: Airflow DAG щогодини. Дешево, надійно. MVP. kumquat-ch-sync --mode=stream Change Streams, near-realtime. Фаза 2 — для випадку, коли креатив залили в Kumquat і хочуть бачити його в аналітиці за хвилину. Потребує replica set; resume token комітити з WriteConcern("majority"). ``` ### Таблиця в ClickHouse Mongo-документ пласкається до колонок, потрібних аналітиці (enum'и → `LowCardinality(String)`, масиви → `Array(String)`, вкладені блоки `vlm.hook.*` → `vlm_hook_*`): ```sql CREATE TABLE IF NOT EXISTS kumquat_market_analytics.creatives ON CLUSTER 'ml' ( creative_id String, -- Mongo _id hex name String, task_type LowCardinality(String), work_stream LowCardinality(String), product_id LowCardinality(String), media_type LowCardinality(String), source LowCardinality(String), geo LowCardinality(String), language LowCardinality(String), test_status LowCardinality(String), performance_tier LowCardinality(String), vlm_hook_format LowCardinality(String), vlm_hook_script LowCardinality(String), vlm_body_format LowCardinality(String), vlm_emotional_driver LowCardinality(String), vlm_visual_style LowCardinality(String), launched_at Nullable(DateTime64(3, 'UTC')), created_at DateTime64(3, 'UTC'), updated_at DateTime64(3, 'UTC'), schema_version UInt16 ) ENGINE = ReplicatedReplacingMergeTree(updated_at) ORDER BY name; ``` - **`ReplicatedReplacingMergeTree(updated_at)`** — і replicated (обов'язково для кластера, plain MergeTree дасть розсинхрон реплік), і природні upsert'и: повторна вставка тієї ж версії або новішої безпечна, дедуплікація з перемогою свіжішого `updated_at`. - **`ORDER BY name`, а не `creative_id`** — дедуплікація ReplacingMergeTree йде по ORDER BY-ключу. `name` — бізнес-ключ, по якому все джойниться (`creo_name = creatives.name`, unique в Mongo); з ключем по `_id` сценарій «видалили і залили заново з тим самим name» лишив би два живі рядки на одне ім'я і задвоїв метрики в джойнах. Для JOIN індекс у ClickHouse все одно не використовується (hash join читає праву таблицю повністю) — ключ тут про коректність дедуплікації, а бонусом і первинний індекс для `WHERE name = ...`. - Читання актуальної версії — через `argMax(..., updated_at)` / `FINAL`; це зафіксовано в `conventions` семантичного шару, щоб агент не рахував по історичних версіях. ### Batch-алгоритм 1. Watermark — `SELECT max(updated_at) FROM creatives` у самому ClickHouse: стан ніде окремо не зберігається, рестарти безкоштовні. 2. З Mongo читаються документи `updated_at > watermark - overlap` (overlap ~10 хв — страховка від clock skew і довгих транзакцій запису). 3. Вставка блоками; overlap створює дублікати — їх гасить ReplacingMergeTree, а ретраї таски — дедуплікація ідентичних insert-блоків replicated-таблиць. Задача повністю ідемпотентна. 4. Hard delete у Mongo watermark не ловить; якщо такі з'являться — щоденна повна звірка `creative_id` (окрема легка таска) або soft-delete прапорець у документі. ### Оркестрація DAG у `dags/` за існуючим патерном: `KubernetesPodOperator`, `schedule="@hourly"`, `catchup=False`, `max_active_runs=1`; креденшели Mongo і CH — з `kumquat-analytics-secrets`. Ідемпотентність вище означає, що ретраї і пропущені години не потребують спецобробки. ## FastAPI ``` POST /threads → створити тред POST /threads/{id}/messages → повідомлення, відповідь стрімом (SSE) GET /threads/{id} → історія GET /threads/{id}/export → тред як Markdown-звіт (питання, висновки, таблиці, дані графіків) GET /results/{result_id} → повний результат запиту як CSV (ephemeral, живе до рестарту пода) ``` - Стрім через **SSE** (простіше за WebSocket, для чату достатньо). Джерело — `agent.astream(..., stream_mode=["updates", "messages", "custom"])` (звичайний LangGraph-граф). - **Протокол подій** — контракт з UI, фіксується одразу: ```json {"type": "token", "content": "За останній місяць..."} {"type": "plan", "todos": [{"text": "...", "status": "done"}]} {"type": "tool_start", "name": "run_sql", "sql": "SELECT ..."} {"type": "table", "columns": [...], "rows": [...], "sql": "...", "result_id": "agent-..."} {"type": "chart", "chart_type": "line", "title": "...", "x_key": "day", "series": ["spend"], "data": [...], "sql": "..."} {"type": "clarify", "question": "За який період?", "options": [...]} {"type": "done", "message_id": "..."} {"type": "error", "detail": "..."} ``` Ключові рішення: - **Таблиця — окремий тип події, а не markdown у тексті.** Агент відповідає текстом-висновком; сирі дані UI бере зі структурованого результату `run_sql`. Надійніше (LLM не переписує числа) і дешевше (повний результат не ганяється через модель). `result_id` у події — лінк на повний CSV (`GET /results/{result_id}`). - **`plan`** — прогрес todo-списку агента (`write_todos`); довгий аналіз перестає виглядати як завислий спінер. - **`clarify`** — interrupt з LangGraph; відповідь юзера продовжує граф з checkpointer'а. - SSE тримає з'єднання — потрібні heartbeat-події, інакше проксі в k8s поріже idle-конекшн. ## UI **React + Vite, кастомний чат** поверх SSE-протоколу: - текст як markdown; події `table` — компонентом-гридом (TanStack Table: сортування, копіювання, експорт CSV); - події `chart` — компонент на recharts (`ResultChart`): line/bar/area/pie зі специфікації, кольори з CSS-змінних теми (`--chart-1..5`); - під кожною табличкою і графіком — розгортається «показати SQL» (довіра + дебаг); - блок плану (події `plan`) — компактний чекліст над відповіддю; - кнопка «Експорт .md» над тредом — качає `GET /threads/{id}/export`; - сайдбар з тредами. Альтернатива для швидкого MVP — Chainlit, але кастомізація впирається в стелю дуже швидко. Для production-інструмента — одразу свій фронт. ## Структура проєкту Реалізовано в межах backend uv-workspace (не окремим репо-в-репо), UI — сторінка в наявному Next.js `frontend/`: ``` backend/packages/analyst-agent/src/analyst_agent/ # ядро агента, без fastapi ├── agent.py # create_agent(...) + middleware-стек ├── gateway.py # ClickHouseGateway: guard → preflight → execute → CSV persist ├── guard.py # sqlglot-валідація (guard_sql) + ALLOWED_SYSTEM_TABLES ├── tools/ # query.py (run_sql, validate_sql, lookup_metric, render_chart), │ # schema.py (інтроспекція), explorer.py (сабагент explore_schema) ├── prompts.py # system prompt: інструкції + AGENT.md + рендер семантичного шару ├── semantic/ # typed-модель (pydantic) + tables.yaml, metrics.yaml, examples.yaml ├── doctor.py # preflight-перевірки дрейфу семантичного шару ├── llm.py # фабрика ChatAnthropic ├── streaming.py # astream-чанки → типізовані SSE-події ├── checkpoint.py # MongoDBSaver (agent__checkpoints*) ├── threads.py # метадані тредів (agent__threads) ├── events.py # pydantic-моделі SSE-подій └── AGENT.md # контекст команди (секція system prompt) backend/analyst_api/ # окремий FastAPI-процес (Clerk auth з kumquat_api) ├── app.py # lifespan: mongo, checkpointer, gateway (+ping), агент ├── routes.py # /api/v1/agent/threads*, /results/{id}, SSE ├── doctor.py # python -m analyst_api.doctor (CLI-обгортка з settings) ├── sse.py # heartbeat + серіалізація подій └── dependencies.py # провайдери backend/packages/kumquat-ch-sync/ # worker Mongo → ClickHouse (uv workspace) dags/dag_kumquat_ch_sync_creatives.py # Airflow DAG, щогодини frontend/src/modules/analyst/ # чат: SSE-клієнт, треди, грід, «показати SQL» k8s/analyst-agent/ # deployment (app + emptyDir для CSV), external-secret ``` Деплой — у кластер за існуючим патерном: секрети (`ANTHROPIC_API_KEY`, пароль `agent_analyst`, Mongo URI — і для checkpointer'а, і для worker'а) через External Secrets з AWS Secrets Manager, як зроблено для `kumquat-analytics-secrets`. ## Ризики 1. **Якість = семантичний шар.** Без описів таблиць агент пише синтаксично правильний, але семантично хибний SQL. Основний час закладати сюди, не в граф. 2. **Семантичний шар не звірений.** Формули метрик і значення enum'ів у цьому доці частково припущені — перший крок імплементації, а не деталь. 3. **Дрейф creatives.** Схема Mongo-моделі зміниться — worker має падати голосно (нове обов'язкове поле), а не тихо писати NULL; `schema_version` в таблиці для контролю. 4. **Контекст-бюджет.** AGENT.md + семантичний шар + план + історія — стежити за розміром промпта. Поки семантичний шар статичний, prompt caching робить його майже безкоштовним; якщо він розростеться — виносити описи таблиць у тул «прочитати опис на вимогу» (progressive disclosure). 5. **Довгі запити.** SSE heartbeat обов'язковий; ліміти `agent_analyst_profile` — стеля на runaway-запити, квота — на runaway-цикли.