101 lines
5.7 KiB
Python
101 lines
5.7 KiB
Python
from app.db import database
|
|
|
|
def get_system_capabilities_text() -> str:
|
|
"""Формирует список возможностей и примеров команд для текущей стадии проекта."""
|
|
return (
|
|
"🛠 Доступные команды и возможности системы SCUD Orion AI:\n\n"
|
|
"1. Управление задачами (Бэклог):\n"
|
|
" • «Покажи последние задачи» — вывод списка задач проекта.\n"
|
|
" • «Добавь задачу: Настроить интеграцию 1С с REST API» — создание новой задачи.\n"
|
|
" • «Измени статус задачи TASK-01 на COMPLETED» — обновление состояния.\n"
|
|
" • «Удали задачу TASK-02» — удаление задачи из реестра.\n\n"
|
|
"2. База знаний и правила (ai_knowledge_base):\n"
|
|
" • «Покажи базу знаний» (или «покажи правила») — вывод текущих правил проекта.\n"
|
|
" • «Добавь в базу знаний: Офисные сотрудники отдела разработки работают по гибкому графику с 09:00 до 11:00» — запись нового правила.\n\n"
|
|
"3. Системные промпты (system_prompts):\n"
|
|
" • «Покажи системный промпт» — просмотр активных инструкций ИИ.\n"
|
|
" • «Обнови системный промпт: Ты — ассистент SCUD Orion AI. Отвечай кратко и строго в формате Plain Text.» — смена роли ИИ на лету.\n\n"
|
|
"Вы можете написать любой из этих запросов в чат в свободной форме."
|
|
)
|
|
|
|
def handle_tool_call(fn_name: str, fn_args: dict) -> str:
|
|
"""Диспетчеризация вызова функций и форматирование текстового ответа."""
|
|
|
|
if fn_name == "db_get_tasks":
|
|
tasks = database.db_get_tasks(status=fn_args.get("status"))
|
|
if not tasks:
|
|
return "Список задач пуст."
|
|
tasks_sorted = sorted(tasks, key=lambda x: x.get('task_id', ''), reverse=True)
|
|
formatted_text = "📋 Последние задачи проекта:\n\n"
|
|
for t in tasks_sorted:
|
|
due = f" (до {t['due_date']})" if t.get('due_date') else ""
|
|
priority = t.get('priority') or 'MEDIUM'
|
|
status_str = t.get('status') or 'BACKLOG'
|
|
module_str = t.get('module') or 'general'
|
|
formatted_text += f"{t.get('task_id')} — {t.get('title')} [{status_str}]\n"
|
|
formatted_text += f" • Приоритет: {priority} | Модуль: {module_str}{due}\n\n"
|
|
return formatted_text
|
|
|
|
elif fn_name == "db_get_rules":
|
|
rules = database.db_get_rules()
|
|
if not rules:
|
|
return "База знаний пуста."
|
|
formatted_text = "🧠 База знаний ИИ (ai_knowledge_base):\n\n"
|
|
for r in rules:
|
|
formatted_text += f"{r.get('id')}. {r.get('rule_text')}\n\n"
|
|
return formatted_text
|
|
|
|
elif fn_name == "db_add_rule":
|
|
res = database.db_add_rule(
|
|
rule_text=fn_args.get("rule_text", ""),
|
|
added_by=fn_args.get("added_by", "Operator")
|
|
)
|
|
return res.get("message", "Правило добавлено в базу знаний.")
|
|
|
|
elif fn_name == "db_get_system_prompt":
|
|
prompt_text = database.db_get_active_system_prompt()
|
|
return f"⚙️ Актуальный системный промпт ИИ (из базы SQLite):\n\n{prompt_text}"
|
|
|
|
elif fn_name == "db_add_system_prompt":
|
|
res = database.db_add_system_prompt(
|
|
name=fn_args.get("name", "main_agent"),
|
|
prompt_text=fn_args.get("prompt_text", "")
|
|
)
|
|
return res.get("message", "Промпт обновлен.")
|
|
|
|
elif fn_name == "db_add_task":
|
|
res = database.db_add_task(
|
|
module=fn_args.get("module", "general"),
|
|
title=fn_args.get("title"),
|
|
priority=fn_args.get("priority", "MEDIUM"),
|
|
due_date=fn_args.get("due_date")
|
|
)
|
|
if "error" in res:
|
|
return f"Ошибка добавления задачи: {res['error']}"
|
|
assigned_id = res.get("task_id", "TASK")
|
|
due_str = fn_args.get('due_date') or 'Без срока'
|
|
return (
|
|
f"Задача {assigned_id} успешно создана!\n\n"
|
|
f"• Название: {fn_args.get('title')}\n"
|
|
f"• Приоритет: {fn_args.get('priority', 'MEDIUM')}\n"
|
|
f"• Срок: {due_str}\n"
|
|
f"• Модуль: {fn_args.get('module', 'general')}"
|
|
)
|
|
|
|
elif fn_name == "db_update_task_status":
|
|
res = database.db_update_task_status(
|
|
task_id=fn_args.get("task_id"),
|
|
status=fn_args.get("status"),
|
|
due_date=fn_args.get("due_date")
|
|
)
|
|
if "error" in res:
|
|
return f"Ошибка обновления задачи: {res['error']}"
|
|
return f"Статус задачи {fn_args.get('task_id')} успешно обновлен!"
|
|
|
|
elif fn_name == "db_delete_task":
|
|
res = database.db_delete_task(task_id=fn_args.get("task_id"))
|
|
if "error" in res:
|
|
return f"Ошибка удаления задачи: {res['error']}"
|
|
return f"Задача {fn_args.get('task_id')} успешно удалена из базы!"
|
|
|
|
return f"Функция {fn_name} выполнена." |