196 lines
19 KiB
Markdown
196 lines
19 KiB
Markdown
Отличная идея! Гарантировать уникальность `id` узлов и ребер является критически важным для целостности графа. SQLite с `PRIMARY KEY` и `INSERT OR IGNORE` прекрасно с этим справляется, автоматически пропуская попытки вставить уже существующие записи. Если нам не нужна строгая последовательность `id`, а только их уникальность, то это работает очень хорошо.
|
||
|
||
Что касается `_get_max_graph_id`, ты прав. Если мы создаём `id` как `graph_1`, `graph_2` и так далее, то `_get_max_graph_id` работает как способ получения **следующего свободного числового суффикса** для нового графа. При этом совершенно неважно, в каком порядке графы были *добавлены* или *изменены* в БД. Важно, что каждый новый граф получит уникальный ID.
|
||
|
||
### Детальное ТЗ для изменений
|
||
|
||
**Цель:** Обеспечить целостность и конкурентную работу с графами в многопоточной среде Flask, используя SQLite. Предотвратить состояние гонки, гарантировать уникальность ID узлов/ребер и атомарность операций.
|
||
|
||
**Основные изменения:**
|
||
|
||
1. **Изменение модели хранения графа в БД:**
|
||
* Вместо хранения `graph_nodes` и `graph_edges` как JSON-строк в одной строке таблицы `graphs`, будет создано 3 таблицы:
|
||
* `graphs`: Основная таблица для информации о графе (`id`, `current_node_id`).
|
||
* `graph_nodes_data`: Для хранения каждого узла графа как отдельной записи (`graph_id`, `node_id`, `node_type`, `node_data_json`).
|
||
* `graph_edges_data`: Для хранения каждого ребра графа как отдельной записи (`graph_id`, `edge_id`, `source_node_id`, `target_node_id`).
|
||
* Это позволит инкрементально добавлять узлы и ребра без перезаписи всего графа целиком.
|
||
|
||
2. **Обработка конкуренции:**
|
||
* **Удаление `threading.Lock()`:** Мы не будем использовать мьютекс Python для блокировки всего `GraphHistoryManager`. Вместо этого мы будем полагаться на встроенные механизмы конкурентной обработки SQLite и атомарности транзакций.
|
||
* **Транзакции:** Все операции по модификации графа (добавление узлов, ребер, обновление `current_node_id`) будут обернуты в одну SQLite транзакцию. Это гарантирует, что либо все изменения будут применены, либо ни одно.
|
||
* **`INSERT OR IGNORE`:** В операциях добавления узлов и ребер будет использоваться `INSERT OR IGNORE`. Это важно: если два конкурирующих запроса попытаются добавить один и тот же новый узел/ребро (с одинаковым `id`), первый запрос успешно его добавит, а второй будет проигнорирован без ошибки. Это предотвращает "Invalid ID" ошибки при инкрементальном добавлении.
|
||
* **Timeout для подключения SQLite:** Увеличение таймаута при подключении к SQLite (`timeout=5.0`) даст другим потокам больше времени на освобождение блокировки файлов БД при пиковых нагрузках, избегая `sqlite3.OperationalError: database is locked`.
|
||
|
||
3. **Генерация ID:**
|
||
* `_get_max_graph_id` будет использоваться только для генерации нового уникального `graph_id`, если он не предоставлен вызывающим кодом (т.е. создание нового графа).
|
||
* В `nodes.py` генерация `node_id` будет опираться на `UUID`, чтобы гарантировать уникальность без необходимости глобального отслеживания. Это критично, поскольку разные потоки могут одновременно генерировать узлы для разных веток.
|
||
|
||
4. **`run_agent` логика:**
|
||
* `run_agent` будет загружать полный граф (все узлы и ребра) из `GraphHistoryManager` в начале.
|
||
* После выполнения LangGraph, `run_agent` будет сравнивать `final_state.graph_nodes` и `final_state.graph_edges` с исходными загруженными данными, чтобы определить *только новые* узлы и ребра, которые нужно сохранить.
|
||
* Затем `graph_history_manager.save_graph_changes` будет вызван с этим списком новых узлов/ребер и новым `current_node_id`.
|
||
|
||
**Файлы для изменения:**
|
||
|
||
1. `app/graph_history_manager.py`
|
||
2. `app/workflows.py`
|
||
3. `app/nodes.py`
|
||
|
||
---
|
||
|
||
Хорошо, полностью согласен с переходом на UUID для генерации ID графов, узлов и ребер. Это упрощает логику, исключает необходимость в `_get_max_graph_id` и гарантирует глобальную уникальность ID без дополнительных сложностей с нумерацией.
|
||
|
||
### Детальное ТЗ для изменений (Обновленное)
|
||
|
||
**Цель:** Обеспечить целостность и конкурентную работу с графами в многопоточной среде Flask, используя SQLite. Предотвратить состояние гонки, гарантировать уникальность ID узлов/ребер через UUID и атомарность операций.
|
||
|
||
**Основные изменения:**
|
||
|
||
1. **Генерация ID с использованием UUID:**
|
||
* Все ID для графов (`graph_id`), узлов (`node_id`) и ребер (`edge_id`) будут генерироваться с использованием `uuid.uuid4().hex`.
|
||
* Это исключает необходимость в методе `_get_max_graph_id`.
|
||
|
||
2. **Изменение модели хранения графа в БД:** (Остается как в предыдущем ТЗ)
|
||
* 3 таблицы: `graphs`, `graph_nodes_data`, `graph_edges_data`.
|
||
* Инкрементальное добавление узлов и ребер без перезаписи всего графа.
|
||
|
||
3. **Обработка конкуренции:** (Остается как в предыдущем ТЗ)
|
||
* Удаление `threading.Lock()` из `GraphHistoryManager`.
|
||
* Все операции по модификации графа в `save_graph_changes` будут обернуты в одну SQLite транзакцию.
|
||
* Использование `INSERT OR IGNORE` для узлов и ребер.
|
||
* Timeout при подключении к SQLite (`timeout=5.0`).
|
||
|
||
4. **`run_agent` логика:** (Остается как в предыдущем ТЗ)
|
||
* `run_agent` будет загружать полный граф (все узлы и ребра) из `GraphHistoryManager` в начале.
|
||
* После выполнения LangGraph, `run_agent` будет сравнивать `final_state.graph_nodes` и `final_state.graph_edges` с исходными загруженными данными, чтобы определить *только новые* узлы и ребра, которые нужно сохранить.
|
||
* Затем `graph_history_manager.save_graph_changes` будет вызван с этим списком новых узлов/ребер и новым `current_node_id`.
|
||
|
||
**Файлы для изменения:**
|
||
|
||
1. `app/graph_history_manager.py`
|
||
2. `app/workflows.py`
|
||
3. `app/nodes.py`
|
||
|
||
---
|
||
|
||
### ТЗ для `app/graph_history_manager.py`
|
||
|
||
**Цель:** Адаптировать менеджер истории графов для работы с UUID и обеспечить надежное инкрементальное хранение и конкурентный доступ.
|
||
|
||
**Изменения:**
|
||
|
||
1. **Импорты:**
|
||
* Удалить `threading`.
|
||
|
||
2. **Удаление `_get_max_graph_id`:**
|
||
* Полностью удалить этот метод, так как ID будут генерироваться с помощью UUID.
|
||
|
||
3. **Метод `save_graph_changes`:**
|
||
* **Генерация `graph_id`**: Если `existing_graph_id` равен `None`, новый `graph_id` генерируется внутри метода с использованием `uuid.uuid4().hex`.
|
||
* Пример: `graph_id = uuid.uuid4().hex`
|
||
* Логика сохранения остальных изменений остаётся прежней (использование `INSERT OR IGNORE`, транзакции).
|
||
* Важно: В `INSERT OR IGNORE INTO graphs (id) VALUES (?)` теперь просто используем сгенерированный или переданный `graph_id`.
|
||
|
||
4. **Методы `_add_graph_edge` и `_add_graph_node`:**
|
||
* Эти методы должны использовать `INSERT OR IGNORE` для добавления узлов и ребер. Это гарантирует, что если узел или ребро с таким `(graph_id, node_id)` или `(graph_id, edge_id)` уже существует из-за конкурентной операции, попытка вставки будет проигнорирована без ошибки.
|
||
* Предполагается, что `node["id"]` и `edge["id"]` уже должны быть UUID, сгенерированными на этапе создания узла/ребра.
|
||
* **Внимание для `_add_graph_edge`**: SQL-запрос `FOREIGN KEY` требует, чтобы узлы `source_node_id` и `target_node_id` уже существовали. Если для `INSERT OR IGNORE` возникнет нарушение `FOREIGN KEY`, он все равно выдаст ошибку `IntegrityError`. Нужно убедиться, что узлы всегда добавляются *до* ребер, которые на них ссылаются, что обеспечивается логикой в `workflows.py`. Если мы будем использовать `executescript` как в предыдущем примере, то нужно убедиться, что синтаксис верен для `INSERT OR IGNORE`. Простой `cursor.execute` с `INSERT OR IGNORE` также должен работать, если узлы гарантированно существуют.
|
||
|
||
Пример `_add_graph_edge` с `cursor.execute`:
|
||
```python
|
||
def _add_graph_edge(self, conn, graph_id: str, edge: Dict[str, Any]):
|
||
cursor = conn.cursor()
|
||
cursor.execute(
|
||
"INSERT OR IGNORE INTO graph_edges_data (graph_id, edge_id, source_node_id, target_node_id) VALUES (?, ?, ?, ?)",
|
||
(graph_id, edge["id"], edge["source"], edge["target"])
|
||
)
|
||
```
|
||
Этот вариант предпочтительнее `executescript`, так как он более читаем и безопасен от SQL-инъекций.
|
||
|
||
---
|
||
|
||
### ТЗ для `app/workflows.py`
|
||
|
||
**Цель:** Адаптировать логику запуска агента для работы с UUID и инкрементального сохранения изменений в графе.
|
||
|
||
**Изменения:**
|
||
|
||
1. **Импорты:**
|
||
* Добавить `import uuid`.
|
||
* Удалить `GraphUpdateConflictError` из импортов, так как этот exception больше не используется прямой логикой `run_agent`.
|
||
|
||
2. **Метод `run_agent`:**
|
||
* **Загрузка графа:** `loaded_graph_data` будет содержать `graph_nodes` и `graph_edges`.
|
||
* **Отслеживание исходного состояния:** После загрузки графа, перед вызовом `app.invoke`, необходимо сохранить текущие `graph_nodes` и `graph_edges` из `initial_state` (которые были загружены из БД). Это нужно, чтобы потом определить, какие узлы/ребра являются *новыми*.
|
||
* Пример:
|
||
```python
|
||
original_nodes_ids = {node['id'] for node in initial_state.graph_nodes}
|
||
original_edges_ids = {edge['id'] for edge in initial_state.graph_edges}
|
||
```
|
||
* **Генерация `graph_id` при создании нового графа:** Если `existing_graph_id` равен `None`, то `graph_id` должен быть сгенерирован здесь, до `app.invoke`, и передан в `initial_state`.
|
||
* Пример:
|
||
```python
|
||
if existing_graph_id is None:
|
||
new_graph_id = uuid.uuid4().hex
|
||
# Также может быть полезно добавить его в initial_state, если AgentState это поддерживает
|
||
# initial_state.graph_id = new_graph_id
|
||
```
|
||
(Или просто передавать `None` в `save_graph_changes` и пусть генерация там происходит).
|
||
* **Определение новых узлов и ребер для сохранения:** После получения `final_state` из `app.invoke`:
|
||
* Проитерировать `final_state.graph_nodes` и собрать те, `id` которых нет в `original_nodes_ids`. Это будут `newly_added_nodes`.
|
||
* Аналогично для `final_state.graph_edges` и `original_edges_ids`, чтобы получить `newly_added_edges`.
|
||
* **Вызов `graph_history_manager.save_graph_changes`:**
|
||
* Вызвать `save_graph_changes` с `existing_graph_id` (или сгенерированным `new_graph_id` если он новый), `newly_added_nodes`, `newly_added_edges` и `final_state.current_node_id`.
|
||
* Пример:
|
||
```python
|
||
final_graph_id = existing_graph_id or new_graph_id # если new_graph_id был сгенерирован ранее
|
||
graph_id_after_save = graph_history_manager.save_graph_changes(
|
||
final_graph_id, newly_added_nodes, newly_added_edges, final_state.current_node_id
|
||
)
|
||
```
|
||
* **Обновление ID в `response_data`**: Убедиться, что `response_data["graph_id"]` содержит актуальный `graph_id_after_save`.
|
||
|
||
---
|
||
|
||
### ТЗ для `app/nodes.py`
|
||
|
||
**Цель:** Модифицировать генерацию ID узлов и ребер на UUID.
|
||
|
||
**Изменения:**
|
||
|
||
1. **Импорты:**
|
||
* Добавить `import uuid`.
|
||
|
||
2. **Генерация `node_id`:**
|
||
* Во всех функциях узлов (`parse_command_node`, `call_llm_node`, `generate_images_node`, `analyze_image_node`, `get_meet_subtitles_node`, `get_teams_subtitles_node`, `summarize_history_node`, `handle_error_node`, `help_node`) заменить строки типа `f"llm_{len(state.graph_nodes) + 1}"` на `uuid.uuid4().hex`.
|
||
* Это обеспечит уникальность `node_id` каждого нового узла.
|
||
|
||
3. **Генерация `edge_id`:**
|
||
* При создании ребер (например, `f"e{state.parent_node_id}-{user_node_id}"`) также заменять на `uuid.uuid4().hex`.
|
||
* Пример:
|
||
```python
|
||
edge_id = uuid.uuid4().hex
|
||
state.graph_edges.append({
|
||
"id": edge_id,
|
||
"source": state.parent_node_id,
|
||
"target": user_node_id
|
||
})
|
||
```
|
||
|
||
---
|
||
|
||
**По поводу фронтенда:**
|
||
|
||
**Да, фронтенд, скорее всего, придется менять.**
|
||
|
||
Причина в том, что теперь бэкенд ожидает, что вы можете отправить `parent_node_id` в запросах к `/api/chat`. Если ваш фронтенд не был разработан с учетом "продолжения" чата с определенного узла (т.е. просто отправлял `message` и `graph_id`), то теперь ему нужно будет:
|
||
|
||
1. **Отслеживать `current_node_id`:** После каждого ответа от бэкенда (`response.graph_visualization_data.current_node_id` или `response.messages[-1].node_id`), фронтенду нужно будет сохранять этот `node_id`.
|
||
2. **Отправлять `parent_node_id`:** Когда пользователь хочет продолжить чат с какого-либо узла (например, нажав на него или просто отправляя новый запрос после получения ответа), фронтенд должен будет включать этот `current_node_id` (или `node_id` того узла, с которого он хочет продолжить) в качестве `parent_node_id` в Payload POST-запроса к `/api/chat`.
|
||
|
||
Если фронтенд отправляет только `message` и `graph_id`, и вы хотите простой линейный диалог, где новый ответ всегда продолжается с последнего сгенерированного узла, то `parent_node_id` можно не отправлять, и бэкенд будет продолжать с `current_node_id`, который он хранит для `graph_id`. Однако, если вы хотите реализовать возможность "ответить на конкретное сообщение/узел", то `parent_node_id` необходим.
|
||
|
||
**Резюме:**
|
||
|
||
* **`api.py`:** Нужны изменения, как показано выше.
|
||
* **Фронтенд:** Потенциально требует изменений для использования новой функциональности `parent_node_id` и более гибкого управления историей чата/графом. |