dimitriev.gs/llm-agent-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add сaching of files, content of files no longer stored in db, add $ and !$ shoftcuts

Browse Source
This commit is contained in:
dimitrievgs 2025-11-10 03:28:48 +03:00
parent 899b0f3d5e
commit 00f3038d1b
3 changed files with 386 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
app/api.py
View File

@ -202,14 +202,15 @@ def send_user_message():
message = data.get("message")
graph_id = data.get("graph_id")
parent_node_id = data.get("parent_node_id")
attachments = data.get("attachments", []) # НОВОЕ
attachments = data.get("attachments", [])
cache_folder = data.get("cache_folder")
if not message:
return jsonify({"error": "Сообщение не может быть пустым."}, 400)
try:
result = graph_history_manager.create_user_node(
message, graph_id, parent_node_id, attachments # ИЗМЕНЕНО
message, graph_id, parent_node_id, attachments, cache_folder
)
graph_history_manager.title_generator.add_node_to_queue_direct(
result.get("graph_id"), result.get("node_id"))
@ -227,6 +228,7 @@ def chat_stream():
user_node_id = data.get("user_node_id")
system_prompt = data.get("system_prompt")
model = data.get("model")
cache_folder = data.get("cache_folder")
# Необязательный параметр для существующего узла ассистента при регенерации
existing_assistant_node_id = data.get("existing_assistant_node_id")
@ -254,7 +256,8 @@ def chat_stream():
try:
for chunk_data in run_agent_streaming(graph_id, user_node_id,
assistant_node_id,
system_prompt, model):
system_prompt, model,
cache_folder):
if chunk_data.get("type") == "chunk":
accumulated_content += chunk_data.get("content", "")
yield f"data: {json.dumps(chunk_data)}\n\n"
@ -324,6 +327,7 @@ def regenerate_message():
"node_id") # Это ID узла, который нужно регенерировать (llm-ответ)
model = data.get("model")
system_prompt = data.get("system_prompt")
cache_folder = data.get("cache_folder")
if not graph_id or not node_id:
return jsonify({"error": "Не указаны graph_id или node_id."}, 400)
@ -353,14 +357,12 @@ def regenerate_message():
graph_id, parent_node_id)
return jsonify({
"existing_assistant_node_id":
assistant_node_id_to_use, # Возвращаем ID узла, который будет регенерирован/использован
"user_node_id":
parent_node_id, # Это родительский узел (пользовательский)
"existing_assistant_node_id": assistant_node_id_to_use,
"user_node_id": parent_node_id,
"graph_id": graph_id,
"model": model,
"system_prompt":
system_prompt # НОВОЕ: Возвращаем системный промпт, чтобы ChatView мог его передать в streamLLMResponse
"system_prompt": system_prompt,
"cache_folder": cache_folder
})
except Exception as e:
print(f"Ошибка при регенерации сообщения: {e}")

364
app/graph_history_manager.py
View File

@ -43,7 +43,9 @@ class GraphHistoryManager:
"""
Деструктор класса. Останавливает генератор заголовков при уничтожении объекта.
"""
print("🚩 Завершение работы GraphHistoryManager. Останавливаем генератор заголовков...")
print(
"🚩 Завершение работы GraphHistoryManager. Останавливаем генератор заголовков..."
)
self.stop_title_generator()
def stop_title_generator(self):
@ -107,6 +109,23 @@ class GraphHistoryManager:
FOREIGN KEY (target_node_id) REFERENCES graph_nodes_data (node_id) ON DELETE CASCADE
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS graph_attachments (
id INTEGER PRIMARY KEY AUTOINCREMENT,
graph_id TEXT NOT NULL,
node_id TEXT NOT NULL,
type TEXT NOT NULL,
name TEXT NOT NULL,
source_path TEXT NOT NULL,
cached_path TEXT NOT NULL,
cache_folder TEXT NOT NULL,
uuid TEXT NOT NULL,
permanent BOOLEAN NOT NULL,
mime_type TEXT,
FOREIGN KEY (graph_id, node_id) REFERENCES graph_nodes_data (graph_id, node_id) ON DELETE CASCADE
)
""")
conn.commit()
def _ensure_graph_exists(self, graph_id: str) -> bool:
@ -117,7 +136,7 @@ class GraphHistoryManager:
"""
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT 1 FROM graphs WHERE id = ?", (graph_id,))
cursor.execute("SELECT 1 FROM graphs WHERE id = ?", (graph_id, ))
return cursor.fetchone() is not None
# ----------------------------------------------- Graph Node/Edge Access Helpers ------------------------------------------------------------
@ -128,18 +147,15 @@ class GraphHistoryManager:
"SELECT node_id, node_type, node_data, title, title_generated FROM graph_nodes_data WHERE graph_id = ?",
(graph_id, ))
nodes = []
for node_id, node_type, node_data_json, title, title_generated in cursor.fetchall():
for node_id, node_type, node_data_json, title, title_generated in cursor.fetchall(
):
node_data = json.loads(node_data_json)
# Добавляем заголовок в данные узла
if title and title_generated:
node_data["title"] = title
node_data["title_generated"] = bool(title_generated)
nodes.append({
"id": node_id,
"type": node_type,
"data": node_data
})
nodes.append({"id": node_id, "type": node_type, "data": node_data})
return nodes
def _get_graph_edges(self, conn, graph_id: str) -> List[Dict[str, Any]]:
@ -187,7 +203,8 @@ class GraphHistoryManager:
try:
cursor.execute(
"INSERT INTO graphs (id, title, title_generated, custom_system_prompt) VALUES (?, ?, FALSE, NULL)",
(new_graph_id, "Новый граф") # Устанавливаем заголовок по умолчанию
(new_graph_id, "Новый граф"
) # Устанавливаем заголовок по умолчанию
)
conn.commit()
return new_graph_id
@ -209,8 +226,9 @@ class GraphHistoryManager:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT current_node_id, title, title_generated, custom_system_prompt FROM graphs WHERE id = ?", # Извлекаем title и title_generated
(graph_id, ))
cursor.execute(
"SELECT current_node_id, title, title_generated, custom_system_prompt FROM graphs WHERE id = ?", # Извлекаем title и title_generated
(graph_id, ))
result = cursor.fetchone()
if result:
@ -223,12 +241,11 @@ class GraphHistoryManager:
resolved_current_node_id = target_node_id if target_node_id else db_current_node_id
messages = self.get_messages_from_root_to_node(
graph_id,
{
graph_id, {
"graph_nodes": graph_nodes,
"graph_edges": graph_edges
}, resolved_current_node_id)
# Определяем "первое_сообщение" для отображения, если заголовок еще не сгенерирован
first_message_content = None
if messages:
@ -238,7 +255,7 @@ class GraphHistoryManager:
break
if not first_message_content and messages:
first_message_content = messages[0].get("content")
return {
"id": graph_id,
"messages": messages,
@ -274,8 +291,7 @@ class GraphHistoryManager:
# Получаем все сообщения до current_node_id
full_messages = self.get_messages_from_root_to_node(
gid,
{
gid, {
"graph_nodes": graph_nodes,
"graph_edges": graph_edges
}, current_node_id)
@ -306,10 +322,9 @@ class GraphHistoryManager:
self,
graph_id: str,
graph_data: Dict[str, Any],
target_node_id: str) -> List[Dict[str, Any]]: # ИЗМЕНЕНО: Dict[str, str] -> Dict[str, Any]
target_node_id: str) -> List[Dict[str, Any]]:
"""
Получает список сообщений от корневого узла до указанного узла.
Сообщения извлекаются из данных узлов, а не из отдельного поля.
"""
all_nodes = graph_data.get("graph_nodes", [])
edges = graph_data.get("graph_edges", [])
@ -331,24 +346,28 @@ class GraphHistoryManager:
path_to_target = get_path_to_root(target_node_id)
messages_for_path = []
for node_id in path_to_target:
node = node_map.get(node_id)
if node and 'message' in node.get('data', {}):
original_message = node['data']['message']
message_dict = {
"role": original_message.get("role", "unknown"),
"type": node.get("type", "unknown"),
"content": original_message.get("content", ""),
"node_id": original_message.get("node_id", node_id),
"graph_id": graph_id
}
# НОВОЕ: Добавляем attachments, если они есть
if "attachments" in original_message:
message_dict["attachments"] = original_message["attachments"]
messages_for_path.append(message_dict)
with self._get_connection() as conn: # TODO: Это, кажется, не нужно
for node_id in path_to_target:
node = node_map.get(node_id)
if node and 'message' in node.get('data', {}):
original_message = node['data']['message']
message_dict = {
"role": original_message.get("role", "unknown"),
"type": node.get("type", "unknown"),
"content": original_message.get("content", ""),
"node_id": original_message.get("node_id", node_id),
"graph_id": graph_id
}
# ИЗМЕНЕНО: Загружаем attachments из БД
if original_message.get("role") == "user":
attachments = self._get_attachments(conn, graph_id, node_id)
if attachments:
message_dict["attachments"] = attachments
messages_for_path.append(message_dict)
return messages_for_path
@ -368,7 +387,7 @@ class GraphHistoryManager:
print(f"Ошибка при удалении графа: {e}")
return False
def delete_all_graphs(self) -> bool: # ДОБАВЛЕНО
def delete_all_graphs(self) -> bool: # ДОБАВЛЕНО
"""Удаляет все графы из базы данных."""
with self._get_connection() as conn:
cursor = conn.cursor()
@ -381,7 +400,7 @@ class GraphHistoryManager:
print(f"Ошибка при удалении всех графов: {e}")
return False
def rename_graph(self, graph_id: str, new_title: str) -> bool: # ДОБАВЛЕНО
def rename_graph(self, graph_id: str, new_title: str) -> bool: # ДОБАВЛЕНО
"""Переименовывает граф, устанавливая пользовательское название."""
with self._get_connection() as conn:
cursor = conn.cursor()
@ -405,18 +424,23 @@ class GraphHistoryManager:
cursor = conn.cursor()
try:
# 1. Получаем информацию о текущем узле графа
cursor.execute("SELECT current_node_id FROM graphs WHERE id = ?", (graph_id,))
current_graph_node_id = cursor.fetchone()[0] if cursor.rowcount > 0 else None
cursor.execute(
"SELECT current_node_id FROM graphs WHERE id = ?",
(graph_id, ))
current_graph_node_id = cursor.fetchone(
)[0] if cursor.rowcount > 0 else None
# 2. Находим родителей и детей удаляемого узла
# Родители
cursor.execute("SELECT source_node_id FROM graph_edges_data WHERE graph_id = ? AND target_node_id = ?",
(graph_id, node_id))
cursor.execute(
"SELECT source_node_id FROM graph_edges_data WHERE graph_id = ? AND target_node_id = ?",
(graph_id, node_id))
parents = [row[0] for row in cursor.fetchall()]
# Дети
cursor.execute("SELECT target_node_id FROM graph_edges_data WHERE graph_id = ? AND source_node_id = ?",
(graph_id, node_id))
cursor.execute(
"SELECT target_node_id FROM graph_edges_data WHERE graph_id = ? AND source_node_id = ?",
(graph_id, node_id))
children = [row[0] for row in cursor.fetchall()]
# 3. Переподключаем родителей к детям
@ -427,14 +451,16 @@ class GraphHistoryManager:
cursor.execute(
"INSERT OR IGNORE INTO graph_edges_data (graph_id, edge_id, source_node_id, target_node_id) VALUES (?, ?, ?, ?)",
(graph_id, new_edge_id, parent_id, child_id))
# 4. Удаляем все ребра, связанные с удаляемым узлом (входящие и исходящие)
cursor.execute("DELETE FROM graph_edges_data WHERE graph_id = ? AND (source_node_id = ? OR target_node_id = ?)",
(graph_id, node_id, node_id))
cursor.execute(
"DELETE FROM graph_edges_data WHERE graph_id = ? AND (source_node_id = ? OR target_node_id = ?)",
(graph_id, node_id, node_id))
# 5. Удаляем сам узел
cursor.execute("DELETE FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
cursor.execute(
"DELETE FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
node_deleted = cursor.rowcount > 0
# TODO: Здесь не полная логика, т.к. если родителей нет, то current_node_id=None останется
@ -449,16 +475,21 @@ class GraphHistoryManager:
new_current_node_id = children[0]
# Если нет ни родителей, ни детей, current_node_id останется None.
cursor.execute("UPDATE graphs SET current_node_id = ? WHERE id = ?",
(new_current_node_id, graph_id))
print(f"current_node_id для графа {graph_id} обновлен на {new_current_node_id}")
cursor.execute(
"UPDATE graphs SET current_node_id = ? WHERE id = ?",
(new_current_node_id, graph_id))
print(
f"current_node_id для графа {graph_id} обновлен на {new_current_node_id}"
)
conn.commit()
return node_deleted
except Exception as e:
conn.rollback()
print(f"Ошибка при удалении узла {node_id} из графа {graph_id}: {e}")
raise # Пробрасываем исключение, чтобы API мог вернуть 500
print(
f"Ошибка при удалении узла {node_id} из графа {graph_id}: {e}"
)
raise # Пробрасываем исключение, чтобы API мог вернуть 500
# ----------------------------------------------- Title Update Methods ------------------------------------------------------------
@ -481,7 +512,7 @@ class GraphHistoryManager:
(graph_id, node_id))
count = cursor.fetchone()[0]
print(f"Найдено узлов для обновления: {count}")
cursor.execute(
"UPDATE graph_nodes_data SET title = ?, title_generated = TRUE WHERE graph_id = ? AND node_id = ?",
(title, graph_id, node_id))
@ -489,9 +520,14 @@ class GraphHistoryManager:
print(f"Обновлено строк: {updated_rows}")
conn.commit()
def create_user_node(self, message: str, graph_id: Optional[str] = None,
parent_node_id: Optional[str] = None,
attachments: List[Dict[str, Any]] = None) -> Dict[str, Any]: # ИЗМЕНЕНО
def create_user_node(
self,
message: str,
graph_id: Optional[str] = None,
parent_node_id: Optional[str] = None,
attachments: List[Dict[str, Any]] = None,
cache_folder: Optional[str] = None) -> Dict[str, Any]:
"""Создаёт узел пользователя с вложениями."""
if not graph_id:
graph_id = str(uuid.uuid4())
@ -499,39 +535,46 @@ class GraphHistoryManager:
raise Exception(f"Граф с ID {graph_id} не найден.")
user_node_id = str(uuid.uuid4())
with self._get_connection() as conn:
try:
node_data = {
"label": message[:30] + "..." if len(message) > 30 else message,
"label":
message[:30] + "..." if len(message) > 30 else message,
"message": {
"role": "user",
"content": message,
"node_id": user_node_id,
"attachments": attachments or [] # НОВОЕ
"role": "user",
"content": message,
"node_id": user_node_id
# УДАЛЕНО: "attachments": attachments or []
}
}
self._add_graph_node(conn, graph_id, {
"id": user_node_id,
"type": "user",
"data": node_data
})
# ДОБАВЛЕНО: Сохраняем attachments отдельно
if attachments:
self._save_attachments(conn, graph_id, user_node_id,
attachments, cache_folder)
if parent_node_id:
edge_id = str(uuid.uuid4())
self._add_graph_edge(conn, graph_id, {
"id": edge_id,
"source": parent_node_id,
"target": user_node_id
})
self._add_graph_edge(
conn, graph_id, {
"id": edge_id,
"source": parent_node_id,
"target": user_node_id
})
self._update_graph_current_node_id(conn, graph_id, user_node_id)
conn.commit()
if not self.get_graph_title_generation_status(graph_id):
self.title_generator.add_graph_to_queue_direct(graph_id)
return {
"graph_id": graph_id,
"node_id": user_node_id,
@ -542,44 +585,52 @@ class GraphHistoryManager:
print(f"Ошибка при создании узла пользователя: {e}")
raise
def create_assistant_placeholder_node(self, graph_id: str, parent_node_id: str) -> str:
def create_assistant_placeholder_node(self, graph_id: str,
parent_node_id: str) -> str:
"""
Создаёт узел-заглушку для ответа ассистента.
"""
assistant_node_id = str(uuid.uuid4())
with self._get_connection() as conn:
try:
node_data = {
"label": "Получение ответа...",
"message": {"role": "assistant", "content": "", "node_id": assistant_node_id},
"message": {
"role": "assistant",
"content": "",
"node_id": assistant_node_id
},
"is_placeholder": True
}
self._add_graph_node(conn, graph_id, {
"id": assistant_node_id,
"type": "llm",
"data": node_data
})
edge_id = str(uuid.uuid4())
self._add_graph_edge(conn, graph_id, {
"id": edge_id,
"source": parent_node_id,
"target": assistant_node_id
})
self._update_graph_current_node_id(conn, graph_id, assistant_node_id)
self._add_graph_edge(
conn, graph_id, {
"id": edge_id,
"source": parent_node_id,
"target": assistant_node_id
})
self._update_graph_current_node_id(conn, graph_id,
assistant_node_id)
conn.commit()
return assistant_node_id
except Exception as e:
conn.rollback()
print(f"Ошибка при создании узла-заглушки: {e}")
raise
def mark_node_as_placeholder_and_clear_content(self, graph_id: str, node_id: str, parent_node_id: str):
def mark_node_as_placeholder_and_clear_content(self, graph_id: str,
node_id: str,
parent_node_id: str):
"""
Очищает содержимое существующего узла LLM, помечает его как заглушку,
и обновляет current_node_id графа.
@ -591,8 +642,9 @@ class GraphHistoryManager:
cursor = conn.cursor()
try:
# 1. Получаем текущие данные узла
cursor.execute("SELECT node_data FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
cursor.execute(
"SELECT node_data FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
result = cursor.fetchone()
if result:
@ -604,66 +656,78 @@ class GraphHistoryManager:
# Сбрасываем флаги генерации заголовка для перегенерации
node_data["title"] = None
node_data["title_generated"] = False
# Обновляем узел
cursor.execute("UPDATE graph_nodes_data SET node_type = ?, node_data = ?, title = NULL, title_generated = FALSE WHERE graph_id = ? AND node_id = ?",
("llm", json.dumps(node_data), graph_id, node_id))
cursor.execute(
"UPDATE graph_nodes_data SET node_type = ?, node_data = ?, title = NULL, title_generated = FALSE WHERE graph_id = ? AND node_id = ?",
("llm", json.dumps(node_data), graph_id, node_id))
# 2. Обновляем current_node_id графа на этот узел
self._update_graph_current_node_id(conn, graph_id, node_id)
conn.commit()
else:
raise ValueError(f"Узел с ID {node_id} не найден в графе {graph_id} для обновления.")
raise ValueError(
f"Узел с ID {node_id} не найден в графе {graph_id} для обновления."
)
except Exception as e:
conn.rollback()
print(f"Ошибка при очистке и маркировке узла {node_id} как заглушки: {e}")
print(
f"Ошибка при очистке и маркировке узла {node_id} как заглушки: {e}"
)
raise
def update_assistant_node_content(self, graph_id: str, node_id: str, content: str):
def update_assistant_node_content(self, graph_id: str, node_id: str,
content: str):
"""
Обновляет содержимое узла ассистента после завершения стриминга.
"""
with self._get_connection() as conn:
cursor = conn.cursor()
try:
cursor.execute("SELECT node_data FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
cursor.execute(
"SELECT node_data FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
result = cursor.fetchone()
if result:
node_data = json.loads(result[0])
node_data["message"]["content"] = content
node_data["label"] = content[:30] + "..." if len(content) > 30 else content
node_data["label"] = content[:30] + "..." if len(
content) > 30 else content
node_data["is_placeholder"] = False
cursor.execute("UPDATE graph_nodes_data SET node_data = ? WHERE graph_id = ? AND node_id = ?",
(json.dumps(node_data), graph_id, node_id))
cursor.execute(
"UPDATE graph_nodes_data SET node_data = ? WHERE graph_id = ? AND node_id = ?",
(json.dumps(node_data), graph_id, node_id))
conn.commit()
except Exception as e:
conn.rollback()
print(f"Ошибка при обновлении узла: {e}")
raise
def mark_node_as_error(self, graph_id: str, node_id: str, error_message: str):
def mark_node_as_error(self, graph_id: str, node_id: str,
error_message: str):
"""
Помечает узел как ошибочный.
"""
with self._get_connection() as conn:
cursor = conn.cursor()
try:
cursor.execute("SELECT node_data FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
cursor.execute(
"SELECT node_data FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
result = cursor.fetchone()
if result:
node_data = json.loads(result[0])
node_data["message"]["content"] = error_message
node_data["label"] = "Ошибка"
node_data["is_placeholder"] = False
cursor.execute("UPDATE graph_nodes_data SET node_type = ?, node_data = ? WHERE graph_id = ? AND node_id = ?",
("error", json.dumps(node_data), graph_id, node_id))
cursor.execute(
"UPDATE graph_nodes_data SET node_type = ?, node_data = ? WHERE graph_id = ? AND node_id = ?",
("error", json.dumps(node_data), graph_id, node_id))
conn.commit()
except Exception as e:
conn.rollback()
@ -676,8 +740,9 @@ class GraphHistoryManager:
"""
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT source_node_id FROM graph_edges_data WHERE graph_id = ? AND target_node_id = ?",
(graph_id, node_id))
cursor.execute(
"SELECT source_node_id FROM graph_edges_data WHERE graph_id = ? AND target_node_id = ?",
(graph_id, node_id))
result = cursor.fetchone()
return result[0] if result else None
@ -690,8 +755,9 @@ class GraphHistoryManager:
"""
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT node_type FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
cursor.execute(
"SELECT node_type FROM graph_nodes_data WHERE graph_id = ? AND node_id = ?",
(graph_id, node_id))
result = cursor.fetchone()
return result[0] if result else None
@ -703,11 +769,13 @@ class GraphHistoryManager:
"""
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT title_generated FROM graphs WHERE id = ?", (graph_id,))
cursor.execute("SELECT title_generated FROM graphs WHERE id = ?",
(graph_id, ))
result = cursor.fetchone()
return bool(result[0]) if result else False
def update_graph_settings(self, graph_id: str, custom_system_prompt: Optional[str]) -> bool:
def update_graph_settings(self, graph_id: str,
custom_system_prompt: Optional[str]) -> bool:
"""
Обновляет настройки графа, включая пользовательский системный промпт.
Если граф не существует, он будет создан.
@ -717,7 +785,9 @@ class GraphHistoryManager:
"""
if not self._ensure_graph_exists(graph_id):
print(f"Граф с ID {graph_id} не найден. Невозможно обновить настройки.")
print(
f"Граф с ID {graph_id} не найден. Невозможно обновить настройки."
)
return False
with self._get_connection() as conn:
@ -725,11 +795,55 @@ class GraphHistoryManager:
try:
cursor.execute(
"UPDATE graphs SET custom_system_prompt = ? WHERE id = ?",
(custom_system_prompt, graph_id)
)
(custom_system_prompt, graph_id))
conn.commit()
return cursor.rowcount > 0
except Exception as e:
conn.rollback()
print(f"Ошибка при обновлении системного промпта для графа {graph_id}: {e}")
print(
f"Ошибка при обновлении системного промпта для графа {graph_id}: {e}"
)
return False
def _save_attachments(self, conn, graph_id: str, node_id: str,
attachments: List[Dict[str, Any]],
cache_folder: Optional[str] = None):
"""Сохраняет attachments в базу данных."""
cursor = conn.cursor()
for att in attachments:
cursor.execute(
"""
INSERT INTO graph_attachments
(graph_id, node_id, type, name, source_path, cached_path, cache_folder, uuid, permanent, mime_type)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(graph_id, node_id, att.get('type'), att.get('name'),
att.get('sourcePath'), att.get('cachedPath'), cache_folder or '',
att.get('uuid'), att.get('permanent', False), att.get('mimeType')))
def _get_attachments(self, conn, graph_id: str,
node_id: str) -> List[Dict[str, Any]]:
"""Получает attachments для узла."""
cursor = conn.cursor()
cursor.execute(
"""
SELECT type, name, source_path, cached_path, cache_folder, uuid, permanent, mime_type
FROM graph_attachments
WHERE graph_id = ? AND node_id = ?
""", (graph_id, node_id))
attachments = []
for row in cursor.fetchall():
attachments.append({
'type': row[0],
'name': row[1],
'sourcePath': row[2],
'cachedPath': row[3],
'cacheFolder': row[4],
'uuid': row[5],
'permanent': bool(row[6]),
'mimeType': row[7]
})
return attachments

156
app/workflows.py
View File

@ -11,6 +11,10 @@ from models import AgentState
from nodes import parse_command_node, execute_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
import uuid # Добавлено для генерации UUID
from llm_client import get_llm
import os
import base64
from pathlib import Path
# --- LangGraph: Построение графа ---
graph_history_manager = GraphHistoryManager()
@ -92,61 +96,173 @@ app = workflow.compile()
def run_agent_streaming(graph_id: str,
user_node_id: str,
assistant_node_id: str,
system_prompt: Optional[str] = None,
model: Optional[str] = None):
user_node_id: str,
assistant_node_id: str,
system_prompt: Optional[str] = None,
model: Optional[str] = None,
cache_folder: Optional[str] = None):
"""Генератор для стримингового ответа LLM с поддержкой вложений."""
try:
loaded_graph_data = graph_history_manager.get_graph(
graph_id, target_node_id=user_node_id)
if not loaded_graph_data:
yield {"type": "error", "error": "Граф не найден."}
return
messages = loaded_graph_data.get("messages", [])
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
messages_for_llm = []
if system_prompt:
messages_for_llm.append(SystemMessage(content=system_prompt))
# ИЗМЕНЕНО: Обрабатываем вложения
# Обрабатываем вложения
for i, msg in enumerate(messages):
is_last = (i == len(messages) - 1)
if msg["role"] == "user":
content = msg["content"]
text_content = msg["content"]
attachments = msg.get("attachments", [])
# Раскрываем постоянные вложения и вложения последнего сообщения
# Проверяем, есть ли мультимодальный контент
has_multimodal = False
content_parts = [{"type": "text", "text": text_content}]
# Обрабатываем вложения
if attachments:
for att in attachments:
# Раскрываем только permanent или вложения последнего сообщения
if att.get("permanent") or is_last:
formatted_content = format_attachment_for_llm(att)
content += "\n" + formatted_content
messages_for_llm.append(HumanMessage(content=content))
attachment_content = prepare_attachment_for_llm(
att, cache_folder)
# Если это изображение, то это мультимодальный контент
if attachment_content.get("type") == "image_url":
has_multimodal = True
content_parts.append(attachment_content)
# Формируем финальный контент
if has_multimodal:
# Мультимодальное сообщение
messages_for_llm.append(
HumanMessage(content=content_parts))
else:
# Текстовое сообщение (объединяем все текстовые части)
full_text = text_content
for part in content_parts[
1:]: # Пропускаем первую часть (основной текст)
if part.get("type") == "text":
full_text += part.get("text", "")
messages_for_llm.append(HumanMessage(content=full_text))
elif msg["role"] == "assistant":
messages_for_llm.append(AIMessage(content=msg["content"]))
# Стриминг ответа
llm = get_llm(model or "gemini-2.5-flash")
for chunk in llm.stream2(messages_for_llm):
content = chunk.content if hasattr(chunk, 'content') else str(chunk)
content = chunk.content if hasattr(chunk,
'content') else str(chunk)
yield {"type": "chunk", "content": content}
except Exception as e:
print(f"Ошибка при стриминге: {e}")
import traceback
traceback.print_exc()
yield {"type": "error", "error": str(e)}
def prepare_attachment_for_llm(attachment: Dict[str, Any],
cache_folder: str) -> Dict[str, Any]:
"""Подготавливает вложение для отправки в LLM."""
mime_type = attachment.get("mimeType", "application/octet-stream")
cached_path = attachment.get("cachedPath")
att_cache_folder = attachment.get("cacheFolder", cache_folder) # ИЗМЕНЕНО
source_path = attachment.get("sourcePath")
att_type = attachment.get("type")
name = attachment.get("name")
# Получаем полный путь к кешированному файлу
full_cached_path = os.path.join(att_cache_folder, cached_path)
# Проверяем существование кеша
if not os.path.exists(full_cached_path):
print(f"❌ Кеш не найден: {full_cached_path}")
return {"type": "text", "text": f"\n[Файл {name} недоступен]\n"}
# Проверяем тип файла
if mime_type.startswith('image/'):
# Для изображений отправляем как base64
with open(full_cached_path, 'rb') as f:
image_data = base64.b64encode(f.read()).decode('utf-8')
return {
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{image_data}"
}
}
elif mime_type.startswith('text/') or mime_type in [
'application/json', 'application/xml'
]:
# Для текстовых файлов вставляем содержимое
try:
with open(full_cached_path, 'r', encoding='utf-8') as f:
content = f.read()
except UnicodeDecodeError:
# Если не получилось прочитать как текст
return {
"type": "text",
"text": f"\n[Файл {name} не может быть прочитан как текст]\n"
}
att_type_ru = "промпта" if att_type == "prompt" else "файла"
header = f"\n\n--- Содержимое {att_type_ru}: {name} ---\n"
footer = f"\n--- Конец {att_type_ru}: {name} ---\n\n"
return {"type": "text", "text": header + content.strip() + footer}
else:
# Для других типов файлов возвращаем информацию
return {
"type":
"text",
"text":
f"\n[Файл {name} ({mime_type}) не может быть отображен в чате]\n"
}
def restore_cache_from_source(source_path: str, att_type: str,
cache_folder: str, cached_path: str) -> str:
"""Восстанавливает кеш из исходного файла."""
full_cached_path = os.path.join(cache_folder, cached_path)
# Создаем папку кеша, если не существует
os.makedirs(cache_folder, exist_ok=True)
if att_type == 'external':
# Для внешних файлов source_path - абсолютный
if not os.path.exists(source_path):
raise FileNotFoundError(f"Исходный файл не найден: {source_path}")
import shutil
shutil.copy2(source_path, full_cached_path)
else:
# Для внутренних файлов и промптов нужно получить полный путь
# Это требует доступа к vault, что на бэкенде недоступно
# Поэтому это нужно делать на фронтенде
raise NotImplementedError(
"Восстановление кеша для внутренних файлов должно происходить на фронтенде"
)
return full_cached_path
def format_attachment_for_llm(attachment: Dict[str, Any]) -> str:
"""Форматирует вложение для вставки в сообщение LLM."""
att_type = "промпта" if attachment["type"] == "prompt" else "файла"
header = f"\n\n--- Содержимое {att_type}: {attachment['name']} ---\n"
footer = f"\n--- Конец {att_type}: {attachment['name']} ---\n\n"
return header + attachment.get("content", "").strip() + footer
return header + attachment.get("content", "").strip() + footer