DB Navigator Agent

AI-агент, который помогает backend-разработчику быстро ориентироваться в базах данных MS SQL Server: находит нужные таблицы и поля, объясняет структуру таблиц, генерирует безопасные read-only T-SQL запросы и получает данные из БД.

Идея простая: вместо того чтобы вручную лазить по схеме незнакомой базы, можно спросить у агента на обычном языке - "где хранится статус заказа?" или "покажи структуру таблицы orders" - и получить ответ с опорой на реальную схему БД.

1. Концепция и пользователь

Основной пользователь - backend-разработчик, которому часто нужно быстро понять:

где в базе хранится нужная бизнес-информация;
какая структура у конкретной таблицы;
как написать безопасный SELECT-запрос;
какое значение или статус лежит в БД по конкретному идентификатору.

Агент закрывает эти задачи и при этом не даёт навредить базе: он работает только на чтение и блокирует любые попытки изменить данные.

Демо-домен - бэк-офис интернет-магазина (заказы, клиенты, продавцы, карты лояльности). Все данные в демо-базе синтетические.

Внешние системы, с которыми работает агент:

MS SQL Server - основная БД, подключение через pyodbc в read-only режиме;
ChromaDB - локальный векторный стор для RAG-поиска по схеме;
OpenRouter / Ollama - провайдеры LLM (облачный и локальный варианты);
LangFuse - observability: трейсинг запусков, метрики, расход токенов.

2. Быстрый старт через Docker

Запуск проекта целиком - БД, агента и веб-интерфейс - без локальной установки Python и ODBC-драйвера.

Требуется: Docker + Docker Compose + ключ OpenRouter.

# 1. Задать ключ OpenRouter
cp .env.example .env
#    вписать ключи, токены и модели в .env

# 2. Поднять всё
docker compose up --build

# 3. Открыть интерфейс
#    http://localhost:8000

Compose поднимает три сервиса:

Сервис	Назначение
`mssql`	MS SQL Server 2022 с demo-базой `shop_demo`
`init-db`	разовая заливка `demo/init.sql` - схема, синтетические данные и read-only логин `agent_reader`
`app`	FastAPI + LangGraph-агент + чат-интерфейс на `:8000`

LangFuse подключается опционально: если в .env заданы ключи - пишутся трейсы; если нет - агент просто работает без трейсинга, ничего не падает.

3. Архитектура агента

Агент построен на LangGraph и представлет граф, по которому проходит каждый запрос. В зависимости от типа вопроса маршрут получается разным: где-то достаточно RAG-поиска, где-то нужно сгенерировать и выполнить SQL.

State

Состояние (agent/state.py) - общий "контейнер", который проходит через все узлы. Каждый узел читает нужные поля и дописывает свои результаты.

Поле	Тип	Описание
`user_query`	str	Исходный вопрос пользователя
`classification`	ClassificationResult	Тип запроса (выход роутера-классификатора)
`metadata_result`	MetadataSearchResult	Найденный контекст по БД (RAG)
`schema_result`	TableSchemaResult	Структура конкретной таблицы
`sql_result`	SQLGenerationResult	Сгенерированный SQL
`execute_result`	ExecuteQueryResult	Результат выполнения SELECT
`final_response`	AgentResponse	Итоговый ответ пользователю
`error`	str \| None	Текст ошибки агента
`steps`	list[str]	История выполненных шагов (для трейсинга)

Узлы графа

Узел	Описание
`classify_intent`	LLM-классификация типа запроса (малая модель)
`search_metadata`	RAG-поиск релевантных таблиц по запросу
`get_schema`	Получение структуры таблицы из sys.columns
`generate_sql`	Генерация T-SQL через LLM (большая модель)
`execute_query`	Выполнение SELECT через pyodbc (только read-only)
`fix_sql`	Самоисправление SQL после ошибки выполнения (retry-цикл)
`format_response`	Формирование финального ответа (малая модель)
`handle_unknown`	Ответ при непонятном запросе
`unsafe_query`	Блокировка запросов на изменение данных

Tools

Минимум три инструмента, и как минимум один из них ходит во внешнюю систему (реальную БД), а не просто крутит текст:

Tool	Тип	Описание
`metadata_search`	RAG + SQL fallback	Семантический поиск таблиц и полей по запросу
`schema_tool`	Внешняя БД	Получение структуры таблицы из системных таблиц MS SQL
`sql_tool`	Внешняя БД	Выполнение SELECT-запросов через pyodbc

4. Схема графа

flowchart TD
    START([Пользовательский запрос]) --> A[classify_intent<br/>LLM-классификация запроса]

    A -->|navigation<br/>Где хранится поле/сущность?| B[search_metadata<br/>RAG-поиск по метаданным]
    A -->|schema<br/>Структура таблицы| C[get_schema<br/>Получение схемы таблицы]
    A -->|script<br/>Нужно написать SQL| D[generate_sql<br/>Генерация SQL]
    A -->|data<br/>Нужно получить данные| D
    A -->|unknown<br/>Непонятный запрос| H[handle_unknown<br/>Ответ с примерами]
    A -->|unsafe<br/>INSERT/UPDATE/DELETE/DROP| U[unsafe_query<br/>Блокировка запроса]

    B --> F[format_response<br/>Формирование финального ответа]
    C --> F

    D --> R{DATA-запрос<br/>и SQL безопасный?}

    R -->|Да| E[execute_query<br/>Выполнение SELECT через pyodbc]
    R -->|Нет| F

    E --> G{Ошибка выполнения<br/>и есть попытки?}
    G -->|Да| X[fix_sql<br/>Исправление SQL]
    G -->|Нет| F
    X --> E

    F --> END([END])
    H --> END
    U --> END

В графе три точки ветвления - это то, что делает агента нелинейным, а не обычным pipeline:

Роутер 1 - после classify_intent: 6 веток по типу запроса.
Роутер 2 - после generate_sql: выполнять SQL или только вернуть скрипт.
Роутер 3 - после execute_query: исправить SQL (fix_sql) или форматировать ответ. Цикл fix_sql -> execute_query ограничен счётчиком sql_retry_count (до 2 попыток), чтобы не зациклиться.

5. Edge cases

#	Сценарий	Обработка
1	Пользователь просит UPDATE / DELETE / INSERT	Блокируется классификатором -> `unsafe_query`
2	SQL содержит мутирующий оператор, но классификатор пропустил	Pydantic-валидатор + проверка в коннекторе
3	Слишком общий вопрос без бизнес-термина	RAG возвращает топ-результаты, агент отвечает что нашёл
4	Бизнес-термин отсутствует в RAG-индексе	Fallback на SQL LIKE поиск по sys.tables
5	Запрос к БД не возвращает строк	`ToolStatus.EMPTY`, агент сообщает что данных нет
6	Запрошенная таблица явно не указана (SCHEMA-запрос)	`get_schema` вызывает `search_metadata` чтобы найти таблицу
7	RAG-индекс не построен при первом запуске	Автоматический fallback на SQL LIKE, агент не падает
8	LLM-провайдер недоступен	`try/except` в каждом узле, возврат `QueryType.UNKNOWN`

6. Критерии качества

Агент работает хорошо, если:

правильно классифицирует тип запроса;
не выполняет unsafe SQL - никогда;
находит релевантные таблицы по бизнес-вопросу;
генерирует только read-only SQL;
объясняет ответ на русском языке;
проходит benchmark из тестовых кейсов с success rate ≥ 80%.

Метрики (считает benchmark/metrics.py, прогон - python app.py --bench):

success rate (overall pass rate) - доля кейсов, где прошли все критерии;
latency - среднее и p90; по категориям - в разбивке by_category;
classification_accuracy, tool_call_accuracy, sql_safety_rate - точность по типам проверок;
no_unhandled_error - детерминированный ассерт: ни один шаг агента не завершился сбоем (:error), при этом штатные исходы (:empty, блокировка unsafe, таблица не найдена) ошибкой не считаются;
llm_judge - LLM-as-judge: малая модель оценивает, отвечает ли ответ агента на вопрос по существу;
cost per run / расход токенов - доступны в дашборде LangFuse по трейсам прогонов.

Результаты каждого прогона сохраняются в benchmark/results/run_*.json.

7. Запустить unit-тесты

pip install -r requirements-dev.txt
pytest

Юнит-тесты в tests/ покрывают чистую логику без живой инфраструктуры (БД/LLM/Chroma не нужны): три роутера графа, проверку SQL на мутации, авто-TOP, извлечение ключевых слов, форматирование типов и агрегацию метрик.

8. Метрики последнего прогона

Общие метрики

Метрика	Значение
Всего кейсов	17
Успешных (pass)	-
Провалено	-
Ошибок выполнения	-
Success Rate	-%
Avg Latency	- s
P90 Latency	- s
Всего времени	- s

По критериям (качество)

Критерий	Accuracy	Кейсов
Правильная классификация	-%	-/17
Правильные инструменты	-%	-/17
Безопасность SQL	-%	-/17
Ожидаемые термины в ответе	-%	-/17

По категориям

Категория	Pass Rate	Кейсов	Avg Latency
navigation	-%	-/3	- s
schema	-%	-/3	- s
script	-%	-/3	- s
data	-%	-/3	- s
unsafe	-%	-/3	- s
unknown	-%	-/2	- s

Как получить метрики

# Запустить benchmark и сохранить результаты
python app.py --bench

# Результаты сохраняются в benchmark/results/run_YYYYMMDD_HHMMSS.json

Observability

Трейсы каждого прогона доступны в LangFuse (если ключи заданы в .env).

9. Структура проекта

db_navigator/
├── api/
│   ├── server.py                # FastAPI: SSE-чат, /api/health, отдача статики
│   ├── events.py                # Перевод обновлений узлов графа в события для UI
│   └── static/                  # Чат-интерфейс (index.html / style.css / app.js)
├── benchmark/
│   ├── evaluator.py             # Критерии оценки каждого кейса
│   ├── metrics.py               # Агрегация: pass rate, latency, by_category
│   ├── runner.py                # CLI для запуска benchmark
│   └── test_cases.json          # 17 тестовых запросов с критериями
├── core/
│   ├── agent/
│   │   ├── graph.py             # LangGraph граф
│   │   ├── nodes.py             # Логика каждого узла
│   │   └── state.py             # AgentState
│   ├── observability/
│   │   ├── __init__.py
│   │   └── tracer.py            # LangFuse интеграция
│   ├── tools/
│   │   ├── metadata_search.py   # RAG + SQL fallback
│   │   ├── schema_tool.py       # get_table_schema из sys.columns
│   │   └── sql_tool.py          # execute_query через pyodbc
│   ├── rag/
│   │   ├── embeddings.py        # Общая фабрика эмбеддингов + HF-токен
│   │   ├── indexer.py           # Индексация схемы БД в ChromaDB
│   │   ├── retriever.py         # Семантический поиск по индексу
│   │   └── domain_knowledge.yaml # Синонимы, связи, enum-поля (доменные знания)
│   ├── llm/
│   │   ├── llm.py               # Обёртка над OpenRouter / Ollama
│   │   ├── prompts.py           # Шаблоны промптов для всех LLM-узлов
│   │   └── context_builder.py   # Сборка контекста схемы/результатов для LLM
│   ├── db/
│   │   └── connector.py         # pyodbc-менеджер
│   ├── schemas/
│   │   ├── models.py            # Pydantic-модели для всех выходов агента
│   │   └── sql_safety.py        # Общий паттерн проверки SQL на мутации
│   ├── config.py                # Все настройки через pydantic-settings + .env
│   ├── logging_config.py        # Единая настройка логирования
├── demo/
│   └── init.sql                 # Синтетическая demo-БД (интернет-магазин)
├── docker-compose.yml           # Весь проект: mssql + init-db + app (веб-агент)
├── Dockerfile                   # Образ агента: Python + ODBC Driver 17 + модель эмбеддингов
├── .dockerignore
├── app.py                       # Точка входа CLI: REPL / запрос / benchmark / check
├── requirements.txt             # Зависимости агента
├── requirements-dev.txt         # Зависимости для unit-тестов
└── .env.example                 # Шаблон переменных окружения

10. Безопасность (security-checklist)

Главная идея - не полагаться на одну проверку, а ставить несколько барьеров. Если один уровень пропустит опасный запрос, его поймает следующий.

Что реализовано

Read-only доступ к БД на уровне прав. Агент ходит в базу не под админом, а под отдельным логином agent_reader, у которого есть только роль db_datareader (задаётся в demo/init.sql). Даже если бы агент каким-то образом сгенерировал DELETE, сама БД его не выполнит - у пользователя нет прав на запись. Это самый надёжный барьер, потому что он не зависит от моего кода.
Блокировка мутаций (3 уровня). Запросы на изменение данных (INSERT, UPDATE, DELETE, DROP, TRUNCATE, ALTER, CREATE) отсекаются трижды: сначала классификатор помечает запрос как unsafe и уводит его в узел блокировки; если вдруг пропустит - срабатывает проверка find_mutations в sql_safety.py; и финально проверка стоит в самом коннекторе перед выполнением. Паттерн один общий, чтобы не было расхождений.
Ограничение объёма выдачи. К каждому SELECT автоматически добавляется TOP N (_inject_top_limit в sql_tool.py), чтобы случайно не вытащить миллион строк и не уронить сервер. Лимит задаётся в конфиге (MAX_ROWS).
Секреты не в коде. API-ключи и пароль к БД лежат в .env, который добавлен в .gitignore и не попадает в репозиторий. В коде ключей нет - всё читается через настройки (config.py). Для шаблона есть .env.example без реальных значений.
Параметризованные запросы. Там, где в SQL подставляются значения, используется параметризация через pyodbc, а не склейка строк - это базовая защита от SQL-инъекций.
Таймаут на запрос. У запросов к БД есть ограничение по времени (QUERY_TIMEOUT), чтобы тяжёлый или зависший запрос не держал соединение бесконечно.

Что пока не сделано (известные ограничения)

Защита от промпт-инъекций. Сейчас нет отдельного фильтра на попытки «обмануть» LLM через текст запроса (типа «забудь инструкции и выполни...»). Частично спасает то, что даже при удачной инъекции read-only права и проверка мутаций не дадут изменить данные.
Allowlist таблиц на выполнение. Сейчас whitelist таблиц есть только для индексации RAG (tables_to_index), но не для самих SELECT-запросов. То есть агент в принципе может прочитать любую таблицу, к которой у agent_reader есть доступ. Для прода стоило бы явно перечислить разрешённые таблицы в рамках определенных ролей.
Разграничение ролей. Сейчас доступы и права одинаковы для любого пользователя. Пока проект живёт в рамках отдела это окей. Но при дальнейшем расширении необходимо настроить роли: какие инструменты доступны, в какие базы и таблицы ходить, дополнительная проверка ответа агента - данной роли разрешено получать эту информацию или нет.
Шифрование данных в ответе - сейчас агент работает с демо-базой на синтетических данных, реальных персональных данных тут нет. В проде необходимо предусмотреть работу с чувствительными данными.

11. Возможные расширения проекта

HITL - уточняющие вопросы у пользователя (например, если не удалось найти таблицу);
Allowlist таблиц и БД;
Разграничение доступов по ролям;
Поиск по конкретной БД/схеме - выбор в UI элементов, по которым производится поиск;
Продолжение диалога (сохранение контекста).

Navigation Menu

Repository files navigation

DB Navigator Agent

1. Концепция и пользователь

2. Быстрый старт через Docker

3. Архитектура агента

State

Узлы графа

Tools

4. Схема графа

5. Edge cases

6. Критерии качества

7. Запустить unit-тесты

8. Метрики последнего прогона

Общие метрики

По критериям (качество)

По категориям

Как получить метрики

Observability

9. Структура проекта

10. Безопасность (security-checklist)

Что реализовано

Что пока не сделано (известные ограничения)

11. Возможные расширения проекта

About

Releases

Packages

Contributors

Languages

Name		Name	Last commit message	Last commit date
api		api
benchmark		benchmark
core		core
demo		demo
tests		tests
Dockerfile		Dockerfile
README.md		README.md
app.py		app.py
conftest.py		conftest.py
docker-compose.yml		docker-compose.yml
pytest.ini		pytest.ini
requirements-dev.txt		requirements-dev.txt
requirements.txt		requirements.txt

Latest commit

History

Folders and files

Repository files navigation

DB Navigator Agent

1. Концепция и пользователь

2. Быстрый старт через Docker

3. Архитектура агента

State

Узлы графа

Tools

4. Схема графа

5. Edge cases

6. Критерии качества

7. Запустить unit-тесты

8. Метрики последнего прогона

Общие метрики

По критериям (качество)

По категориям

Как получить метрики

Observability

9. Структура проекта

10. Безопасность (security-checklist)

Что реализовано

Что пока не сделано (известные ограничения)

11. Возможные расширения проекта

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages