mirror of
https://gitflic.ru/project/alt-gnome/karkas.git
synced 2024-12-24 00:33:06 +03:00
Delete external modules for OCAB Lite
This commit is contained in:
Armatik
parent
39500b77c2
commit
b185acd871
@ -1 +0,0 @@
|
||||
from . import yandexgpt
|
||||
@ -1,19 +0,0 @@
|
||||
# Модуль Create Report Apps
|
||||
|
||||
Модуль `create_report_apps` предназначен для помощи пользователям в создании отчетов об ошибках в приложениях.
|
||||
|
||||
## Функциональность
|
||||
|
||||
- Задает пользователю ряд вопросов, необходимых для составления отчета.
|
||||
- Собирает информацию о системе пользователя.
|
||||
- Формирует отчет в текстовом формате.
|
||||
|
||||
## Команды
|
||||
|
||||
- `/create_report_apps` - запустить процесс создания отчета.
|
||||
|
||||
## Использование
|
||||
|
||||
1. Отправьте команду `/create_report_apps` боту в личных сообщениях или в групповом чате.
|
||||
2. Ответьте на вопросы бота.
|
||||
3. Бот сформирует отчет и отправит его вам.
|
||||
@ -1 +0,0 @@
|
||||
from .main import module_init
|
||||
@ -1,136 +0,0 @@
|
||||
from aiogram import Bot, Router
|
||||
from aiogram.enums import ParseMode
|
||||
from aiogram.fsm.context import FSMContext
|
||||
from aiogram.fsm.state import State, StatesGroup
|
||||
from aiogram.types import (
|
||||
BufferedInputFile,
|
||||
KeyboardButton,
|
||||
Message,
|
||||
ReplyKeyboardMarkup,
|
||||
ReplyKeyboardRemove,
|
||||
)
|
||||
|
||||
from ocab_core.modules_system.public_api import Utils, get_fsm_context
|
||||
|
||||
from .report import Report
|
||||
|
||||
router = Router()
|
||||
|
||||
|
||||
class ReportState(StatesGroup):
|
||||
input_system_info = State()
|
||||
input_app_name = State()
|
||||
input_problem_step_by_step = State()
|
||||
input_actual_result = State()
|
||||
input_expected_result = State()
|
||||
input_additional_info = State()
|
||||
|
||||
|
||||
system_info_code = """echo "SESSION_TYPE: ${XDG_SESSION_TYPE:-Unknown}"
|
||||
[ -f /etc/os-release ] && grep "^PRETTY_NAME=" /etc/os-release | cut -d= -f2 \
|
||||
| tr -d '"' | xargs echo "OS: "
|
||||
echo "Kernel: $(uname -r)"
|
||||
echo "DE: ${XDG_CURRENT_DESKTOP:-Unknown}"
|
||||
grep "^model name" /proc/cpuinfo | head -n1 | cut -d: -f2 \
|
||||
| xargs echo "CPU: "
|
||||
lspci | grep "VGA compatible controller" | cut -d: -f3 \
|
||||
| xargs -I{} echo "GPU: {}"
|
||||
"""
|
||||
|
||||
|
||||
system_info_message = """Укажите параметры свой системы.
|
||||
Собрать информацию о системе можно с помощью данного скрипта:
|
||||
""" + Utils.code_format(
|
||||
system_info_code,
|
||||
"shell",
|
||||
)
|
||||
|
||||
|
||||
async def start_report(chat_id: int, bot: Bot):
|
||||
await bot.send_message(
|
||||
chat_id=chat_id,
|
||||
text=system_info_message,
|
||||
parse_mode=ParseMode.HTML,
|
||||
reply_markup=ReplyKeyboardRemove(),
|
||||
)
|
||||
state = await get_fsm_context(chat_id, chat_id)
|
||||
|
||||
await state.set_state(ReportState.input_system_info)
|
||||
|
||||
|
||||
app_info_message = """Укажите название и версию приложения.
|
||||
Узнать можно с помощью данной команды:""" + Utils.code_format(
|
||||
"rpm -qa | grep -i НАЗВАНИЕ_ПРИЛОЖЕНИЯ", "shell"
|
||||
)
|
||||
|
||||
|
||||
@router.message(ReportState.input_system_info)
|
||||
async def system_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(system=message.text)
|
||||
await message.answer(
|
||||
text=app_info_message,
|
||||
parse_mode=ParseMode.HTML,
|
||||
)
|
||||
await state.set_state(ReportState.input_app_name)
|
||||
|
||||
|
||||
step_by_step_message = (
|
||||
"""Опиши проблему пошагово, что ты делал, что происходило, что не так."""
|
||||
)
|
||||
|
||||
|
||||
@router.message(ReportState.input_app_name)
|
||||
async def app_name_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(app=message.text)
|
||||
await message.answer(text=step_by_step_message)
|
||||
await state.set_state(ReportState.input_problem_step_by_step)
|
||||
|
||||
|
||||
@router.message(ReportState.input_problem_step_by_step)
|
||||
async def problem_step_by_step_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(problem_step_by_step=message.text)
|
||||
await message.answer(text="Опиши, что произошло (фактический результат).")
|
||||
await state.set_state(ReportState.input_actual_result)
|
||||
|
||||
|
||||
@router.message(ReportState.input_actual_result)
|
||||
async def actual_result_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(actual=message.text)
|
||||
await message.answer(text="Опиши ожидаемый результат.")
|
||||
await state.set_state(ReportState.input_expected_result)
|
||||
|
||||
|
||||
@router.message(ReportState.input_expected_result)
|
||||
async def expected_result_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(expected=message.text)
|
||||
await message.answer(
|
||||
text="Если есть дополнительная информация, то напиши ее.",
|
||||
reply_markup=ReplyKeyboardMarkup(
|
||||
resize_keyboard=True,
|
||||
keyboard=[
|
||||
[KeyboardButton(text="Дополнительной информации нет")],
|
||||
],
|
||||
),
|
||||
)
|
||||
await state.set_state(ReportState.input_additional_info)
|
||||
|
||||
|
||||
@router.message(ReportState.input_additional_info)
|
||||
async def additional_info_entered(message: Message, state: FSMContext):
|
||||
if message.text == "Дополнительной информации нет":
|
||||
additional_info = ""
|
||||
else:
|
||||
additional_info = message.text
|
||||
await state.update_data(additional=additional_info)
|
||||
await message.answer(
|
||||
text="Вот твой отчет сообщением, а также файлом:",
|
||||
reply_markup=ReplyKeyboardRemove(),
|
||||
)
|
||||
data = await state.get_data()
|
||||
|
||||
report = Report(data)
|
||||
file_report = report.export().encode()
|
||||
|
||||
await message.answer(text=report.export())
|
||||
await message.answer_document(document=BufferedInputFile(file_report, "report.txt"))
|
||||
await state.clear()
|
||||
@ -1,14 +0,0 @@
|
||||
{
|
||||
"id": "external.create_report_apps",
|
||||
"name": "Create Report Apps",
|
||||
"description": "Модуль для создания отчетов о ошибках в приложениях",
|
||||
"author": [
|
||||
"OCAB Team",
|
||||
"Maxim Slipenko"
|
||||
],
|
||||
"version": "1.0.0",
|
||||
"privileged": false,
|
||||
"dependencies": {
|
||||
"standard.command_helper": "^1.0.0"
|
||||
}
|
||||
}
|
||||
@ -1,115 +0,0 @@
|
||||
from typing import Union
|
||||
|
||||
from aiogram import Bot, F, Router
|
||||
from aiogram.exceptions import TelegramForbiddenError
|
||||
from aiogram.filters import BaseFilter, Command, CommandStart
|
||||
from aiogram.types import (
|
||||
CallbackQuery,
|
||||
InlineKeyboardButton,
|
||||
InlineKeyboardMarkup,
|
||||
Message,
|
||||
)
|
||||
|
||||
from ocab_core.modules_system.public_api import get_module, register_router
|
||||
|
||||
from .create_report import router as create_report_router
|
||||
from .create_report import start_report
|
||||
|
||||
register_command = get_module("standard.command_helper", "register_command")
|
||||
|
||||
router = Router()
|
||||
|
||||
|
||||
class ChatTypeFilter(BaseFilter):
|
||||
def __init__(self, chat_type: Union[str, list]):
|
||||
self.chat_type = chat_type
|
||||
|
||||
async def __call__(self, message: Message) -> bool:
|
||||
if isinstance(self.chat_type, str):
|
||||
return message.chat.type == self.chat_type
|
||||
return message.chat.type in self.chat_type
|
||||
|
||||
|
||||
@router.message(
|
||||
ChatTypeFilter(chat_type=["group", "supergroup"]), Command("create_report_apps")
|
||||
)
|
||||
async def create_report_apps_command_group(message: Message):
|
||||
keyboard = InlineKeyboardMarkup(
|
||||
inline_keyboard=[
|
||||
[
|
||||
InlineKeyboardButton(
|
||||
text="Да", callback_data=f"create_report:{message.from_user.id}"
|
||||
),
|
||||
InlineKeyboardButton(
|
||||
text="Нет", callback_data=f"cancel_report:{message.from_user.id}"
|
||||
),
|
||||
]
|
||||
]
|
||||
)
|
||||
await message.answer(
|
||||
"Я могу отправить тебе пару вопросов "
|
||||
"для помощи в составлении репорта личными "
|
||||
"сообщениями.",
|
||||
reply_markup=keyboard,
|
||||
)
|
||||
|
||||
|
||||
@router.message(
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
CommandStart(deep_link=True, magic=F.args == "create_report_apps"),
|
||||
)
|
||||
@router.message(ChatTypeFilter(chat_type=["private"]), Command("create_report_apps"))
|
||||
async def create_report_apps_command(message: Message, bot: Bot):
|
||||
await start_report(message.from_user.id, bot)
|
||||
|
||||
|
||||
@router.callback_query(F.data.startswith("cancel_report"))
|
||||
async def cancel_report_callback(callback_query: CallbackQuery):
|
||||
callback_user_id = int(callback_query.data.split(":")[1])
|
||||
if callback_query.from_user.id != callback_user_id:
|
||||
await callback_query.answer("Эта кнопка не для вас.", show_alert=True)
|
||||
return
|
||||
|
||||
await callback_query.message.delete()
|
||||
|
||||
|
||||
@router.callback_query(F.data.startswith("create_report"))
|
||||
async def create_report_callback(callback_query: CallbackQuery, bot: Bot):
|
||||
callback_user_id = int(callback_query.data.split(":")[1])
|
||||
if callback_query.from_user.id != callback_user_id:
|
||||
await callback_query.answer("Эта кнопка не для вас.", show_alert=True)
|
||||
return
|
||||
|
||||
user_id = callback_query.from_user.id
|
||||
|
||||
async def on_chat_unavailable():
|
||||
await callback_query.message.edit_text(
|
||||
"Я в личных сообщениях задам тебе вопросы "
|
||||
"для помощи в составлении репорта. "
|
||||
'Но перед этим ты должен нажать кнопку "Запустить"'
|
||||
)
|
||||
info = await bot.get_me()
|
||||
await callback_query.answer(
|
||||
url=f"https://t.me/{info.username}?start=create_report_apps"
|
||||
)
|
||||
|
||||
try:
|
||||
chat_member = await bot.get_chat_member(chat_id=user_id, user_id=user_id)
|
||||
if chat_member.status != "left":
|
||||
await start_report(user_id, bot)
|
||||
await callback_query.message.edit_text(
|
||||
"Я в личных сообщениях задам тебе "
|
||||
"вопросы для помощи в составлении "
|
||||
"репорта."
|
||||
)
|
||||
else:
|
||||
await on_chat_unavailable()
|
||||
except TelegramForbiddenError:
|
||||
await on_chat_unavailable()
|
||||
|
||||
|
||||
async def module_init():
|
||||
router.include_router(create_report_router)
|
||||
|
||||
register_router(router)
|
||||
register_command("create_report_apps", "Написать репорт о приложении")
|
||||
@ -1,59 +0,0 @@
|
||||
import aiogram
|
||||
|
||||
|
||||
class ReportFormatter:
|
||||
def __init__(self, html=True):
|
||||
self.html = html
|
||||
|
||||
def bold(self, string):
|
||||
if self.html:
|
||||
return f"<b>{self.text(string)}</b>"
|
||||
return self.text(string)
|
||||
|
||||
def text(self, string):
|
||||
if self.html:
|
||||
return aiogram.html.quote(string)
|
||||
return string
|
||||
|
||||
|
||||
class Report:
|
||||
def __init__(self, data: dict):
|
||||
self.data = data
|
||||
|
||||
def export(self):
|
||||
data = self.data
|
||||
|
||||
report = f"""
|
||||
Стенд с ошибкой:
|
||||
==============================
|
||||
|
||||
{data['system']}
|
||||
|
||||
Пакет:
|
||||
==============================
|
||||
|
||||
{data['app']}
|
||||
|
||||
Шаги, приводящие к ошибке:
|
||||
==============================
|
||||
|
||||
{data['problem_step_by_step']}
|
||||
|
||||
Фактический результат:
|
||||
==============================
|
||||
|
||||
{data['actual']}
|
||||
|
||||
Ожидаемый результат:
|
||||
==============================
|
||||
|
||||
{data['expected']}
|
||||
"""
|
||||
if data["additional"] != "":
|
||||
report += f"""
|
||||
Дополнительно:
|
||||
==============================
|
||||
|
||||
{data['additional']}
|
||||
"""
|
||||
return report
|
||||
@ -1,22 +0,0 @@
|
||||
# Модуль YandexGPT
|
||||
|
||||
Модуль `yandexgpt` интегрирует в бота OCAB нейросеть YandexGPT.
|
||||
|
||||
## Функциональность
|
||||
|
||||
- Позволяет боту отвечать на сообщения пользователей, используя YandexGPT.
|
||||
- Строит линию контекста для нейросети, используя историю сообщений.
|
||||
|
||||
## Конфигурация
|
||||
|
||||
- `yandexgpt::token` - API-ключ для доступа к YandexGPT.
|
||||
- `yandexgpt::catalogid` - идентификатор каталога YandexGPT.
|
||||
- `yandexgpt::prompt` - системная подсказка для YandexGPT.
|
||||
- `yandexgpt::startword` - слова, с которых должно начинаться сообщение, чтобы бот ответил.
|
||||
- `yandexgpt::inword` - слова, которые должны быть в сообщении, чтобы бот ответил.
|
||||
|
||||
## Использование
|
||||
|
||||
1. Настройте конфигурационные параметры модуля.
|
||||
2. Отправьте боту сообщение, которое соответствует условиям, указанным в параметрах `startword` и `inword`.
|
||||
3. Бот ответит на сообщение, используя YandexGPT.
|
||||
@ -1,2 +0,0 @@
|
||||
from .handlers import answer_to_message
|
||||
from .main import module_init
|
||||
@ -1,47 +0,0 @@
|
||||
# flake8: noqa
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from aiogram import Bot
|
||||
from aiogram.types import Message
|
||||
|
||||
from ocab_core.modules_system.public_api import get_module, log
|
||||
|
||||
from .yandexgpt import YandexGPT
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ocab_modules.standard.config import IConfig
|
||||
from ocab_modules.standard.database.db_api import add_message as IAddMessage
|
||||
|
||||
config: "IConfig" = get_module(
|
||||
"standard.config",
|
||||
"config",
|
||||
)
|
||||
|
||||
|
||||
def get_yandexgpt_catalog_id():
|
||||
return config.get("yandexgpt::catalogid")
|
||||
|
||||
|
||||
def get_yandexgpt_token():
|
||||
return config.get("yandexgpt::token")
|
||||
|
||||
|
||||
def get_yandexgpt_prompt():
|
||||
return config.get("yandexgpt::prompt")
|
||||
|
||||
|
||||
add_message: "IAddMessage" = get_module("standard.database", "db_api.add_message")
|
||||
|
||||
|
||||
async def answer_to_message(message: Message, bot: Bot):
|
||||
# print("answer_to_message")
|
||||
log("answer_to_message")
|
||||
yagpt = YandexGPT(get_yandexgpt_token(), get_yandexgpt_catalog_id())
|
||||
text = message.text
|
||||
prompt = get_yandexgpt_prompt()
|
||||
# response = await yagpt.async_yandexgpt(system_prompt=prompt, input_messages=text)
|
||||
response = await yagpt.yandexgpt_request(
|
||||
chat_id=message.chat.id, message_id=message.message_id, type="yandexgpt"
|
||||
)
|
||||
reply = await message.reply(response, parse_mode="Markdown")
|
||||
add_message(reply, message_ai_model="yandexgpt")
|
||||
@ -1,21 +0,0 @@
|
||||
{
|
||||
"id": "external.yandexgpt",
|
||||
"name": "Yandex GPT",
|
||||
"description": "Модуль для работы с Yandex GPT",
|
||||
"author": "OCAB Team",
|
||||
"version": "1.0.0",
|
||||
"privileged": false,
|
||||
"dependencies": {
|
||||
"required": {
|
||||
"standard.config": "^1.0.0",
|
||||
"standard.database": "^1.0.0"
|
||||
}
|
||||
},
|
||||
"pythonDependencies": {
|
||||
"required": {
|
||||
"aiohttp": "*",
|
||||
"requests": "*",
|
||||
"json": "*"
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,50 +0,0 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from ocab_core.modules_system.public_api import get_module
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ocab_modules.standard.config import IConfig
|
||||
|
||||
config: "IConfig" = get_module("standard.config", "config")
|
||||
|
||||
|
||||
def module_init():
|
||||
config.register(
|
||||
"yandexgpt::token",
|
||||
"password",
|
||||
required=True,
|
||||
)
|
||||
config.register(
|
||||
"yandexgpt::token_for_request",
|
||||
"int",
|
||||
default_value=8000,
|
||||
)
|
||||
config.register(
|
||||
"yandexgpt::token_for_answer",
|
||||
"int",
|
||||
default_value=2000,
|
||||
)
|
||||
|
||||
config.register(
|
||||
"yandexgpt::catalogid",
|
||||
"password",
|
||||
required=True,
|
||||
)
|
||||
|
||||
config.register(
|
||||
"yandexgpt::prompt",
|
||||
"string",
|
||||
default_value="Ты чат-бот ...",
|
||||
)
|
||||
|
||||
config.register(
|
||||
"yandexgpt::startword",
|
||||
"string",
|
||||
default_value="Бот| Бот, | бот | бот,",
|
||||
)
|
||||
|
||||
config.register(
|
||||
"yandexgpt::inword",
|
||||
"string",
|
||||
default_value="помогите | не работает",
|
||||
)
|
||||
@ -1,10 +0,0 @@
|
||||
# flake8: noqa
|
||||
from aiogram import F, Router
|
||||
|
||||
from .handlers import answer_to_message
|
||||
|
||||
router = Router()
|
||||
# Если сообщение содержит в начале текст "Гномик" или "гномик" или отвечает на сообщение бота, то вызывается функция answer_to_message
|
||||
router.message.register(
|
||||
answer_to_message, F.text.startswith("Гномик") | F.text.startswith("гномик")
|
||||
)
|
||||
@ -1,312 +0,0 @@
|
||||
# flake8: noqa
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import aiohttp
|
||||
import requests
|
||||
|
||||
from ocab_core.modules_system.public_api import get_module, log
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ocab_modules.standard.config import IConfig
|
||||
from ocab_modules.standard.database import db_api as IDbApi
|
||||
|
||||
db_api: "IDbApi" = get_module("standard.database", "db_api")
|
||||
|
||||
config: "IConfig" = get_module("standard.config", "config")
|
||||
|
||||
|
||||
def get_yandexgpt_token_for_answer():
|
||||
return config.get("yandexgpt::token_for_answer")
|
||||
|
||||
|
||||
def get_yandexgpt_token_for_request():
|
||||
return config.get("yandexgpt::token_for_request")
|
||||
|
||||
|
||||
def get_yandexgpt_prompt():
|
||||
return config.get("yandexgpt::prompt")
|
||||
|
||||
|
||||
class YandexGPT:
|
||||
token = None
|
||||
catalog_id = None
|
||||
languages = {
|
||||
"ru": "русский язык",
|
||||
"en": "английский язык",
|
||||
"de": "немецкий язык",
|
||||
"uk": "украинский язык",
|
||||
"es": "испанский язык",
|
||||
"be": "белорусский язык",
|
||||
}
|
||||
|
||||
def __init__(self, token, catalog_id):
|
||||
self.token = token
|
||||
self.catalog_id = catalog_id
|
||||
|
||||
async def async_request(self, url, headers, prompt) -> dict:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, headers=headers, json=prompt) as response:
|
||||
return await response.json()
|
||||
|
||||
async def async_token_check(
|
||||
self, messages, gpt, max_tokens, stream, temperature, del_msg_id=1
|
||||
):
|
||||
url = "https://llm.api.cloud.yandex.net/foundationModels/v1/tokenizeCompletion"
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Api-Key {self.token}",
|
||||
}
|
||||
answer_token = get_yandexgpt_token_for_answer()
|
||||
while True:
|
||||
try:
|
||||
request = {
|
||||
"modelUri": gpt,
|
||||
"completionOptions": {
|
||||
"stream": stream,
|
||||
"temperature": temperature,
|
||||
"maxTokens": max_tokens,
|
||||
},
|
||||
"messages": messages,
|
||||
}
|
||||
response = await self.async_request(
|
||||
url=url, headers=headers, prompt=request
|
||||
)
|
||||
except Exception as e: # TODO: Переделать обработку ошибок
|
||||
# print(e)
|
||||
log(f"Error: {e}")
|
||||
|
||||
continue
|
||||
if int(len(response["tokens"])) < (max_tokens - answer_token):
|
||||
break
|
||||
else:
|
||||
try:
|
||||
messages.pop(del_msg_id)
|
||||
except IndexError:
|
||||
Exception("IndexError: list index out of range")
|
||||
return messages
|
||||
|
||||
async def async_yandexgpt_lite(
|
||||
self,
|
||||
system_prompt,
|
||||
input_messages,
|
||||
stream=False,
|
||||
temperature=0.6,
|
||||
max_tokens=8000,
|
||||
):
|
||||
url = "https://llm.api.cloud.yandex.net/foundationModels/v1/completion"
|
||||
gpt = f"gpt://{self.catalog_id}/yandexgpt-lite/latest"
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Api-Key {self.token}",
|
||||
}
|
||||
|
||||
messages = [{"role": "system", "text": system_prompt}]
|
||||
for message in input_messages:
|
||||
messages.append(message)
|
||||
messages = await self.async_token_check(messages, gpt, max_tokens)
|
||||
|
||||
prompt = {
|
||||
"modelUri": gpt,
|
||||
"completionOptions": {
|
||||
"stream": stream,
|
||||
"temperature": temperature,
|
||||
"maxTokens": max_tokens,
|
||||
},
|
||||
"messages": messages,
|
||||
}
|
||||
|
||||
response = requests.post(url, headers=headers, json=prompt).text # nosec
|
||||
return json.loads(response)["result"]["alternatives"][0]["message"]["text"]
|
||||
|
||||
async def async_yandexgpt(
|
||||
self,
|
||||
system_prompt,
|
||||
input_messages,
|
||||
stream=False,
|
||||
temperature=0.6,
|
||||
max_tokens=None,
|
||||
):
|
||||
if max_tokens is None:
|
||||
max_tokens = get_yandexgpt_token_for_request()
|
||||
|
||||
url = "https://llm.api.cloud.yandex.net/foundationModels/v1/completion"
|
||||
gpt = f"gpt://{self.catalog_id}/yandexgpt/latest"
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Api-Key {self.token}",
|
||||
}
|
||||
|
||||
messages = []
|
||||
messages.append({"role": "system", "text": system_prompt})
|
||||
for message in input_messages:
|
||||
messages.append(message)
|
||||
|
||||
messages = await self.async_token_check(
|
||||
messages, gpt, max_tokens, stream, temperature
|
||||
)
|
||||
|
||||
request = {
|
||||
"modelUri": gpt,
|
||||
"completionOptions": {
|
||||
"stream": stream,
|
||||
"temperature": temperature,
|
||||
"maxTokens": max_tokens,
|
||||
},
|
||||
"messages": messages,
|
||||
}
|
||||
response = await self.async_request(
|
||||
url=url, headers=headers, prompt=request
|
||||
) # nosec
|
||||
return response["result"]["alternatives"][0]["message"]["text"]
|
||||
|
||||
async def async_yandexgpt_translate(self, input_language, output_language, text):
|
||||
input_language = self.languages[input_language]
|
||||
output_language = self.languages[output_language]
|
||||
|
||||
return await self.async_yandexgpt(
|
||||
f"Переведи на {output_language} сохранив оригинальный смысл текста. Верни только результат:",
|
||||
[{"role": "user", "text": text}],
|
||||
stream=False,
|
||||
temperature=0.6,
|
||||
max_tokens=8000,
|
||||
)
|
||||
|
||||
async def async_yandexgpt_spelling_check(self, input_language, text):
|
||||
input_language = self.languages[input_language]
|
||||
|
||||
return await self.async_yandexgpt(
|
||||
f"Проверьте орфографию и пунктуацию текста на {input_language}. Верни исправленный текст "
|
||||
f"без смысловых искажений:",
|
||||
[{"role": "user", "text": text}],
|
||||
stream=False,
|
||||
temperature=0.6,
|
||||
max_tokens=8000,
|
||||
)
|
||||
|
||||
async def async_yandexgpt_text_history(
|
||||
self, input_messages, stream=False, temperature=0.6, max_tokens=8000
|
||||
):
|
||||
url = "https://llm.api.cloud.yandex.net/foundationModels/v1/completion"
|
||||
gpt = f"gpt://{self.catalog_id}/summarization/latest"
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Api-Key {self.token}",
|
||||
}
|
||||
|
||||
messages = []
|
||||
for message in input_messages:
|
||||
messages.append(message)
|
||||
messages = await self.async_token_check(messages, gpt, max_tokens, del_msg_id=0)
|
||||
|
||||
prompt = {
|
||||
"modelUri": gpt,
|
||||
"completionOptions": {
|
||||
"stream": stream,
|
||||
"temperature": temperature,
|
||||
"maxTokens": max_tokens,
|
||||
},
|
||||
"messages": messages,
|
||||
}
|
||||
|
||||
response = requests.post(url, headers=headers, json=prompt).text # nosec
|
||||
return json.loads(response)["result"]["alternatives"][0]["message"]["text"]
|
||||
|
||||
async def async_yandex_cloud_text_to_speech(
|
||||
self, text, voice, emotion, speed, format, quality
|
||||
):
|
||||
tts = "tts.api.cloud.yandex.net/speech/v1/tts:synthesize"
|
||||
# TODO: Сделать функцию TTS
|
||||
return 0
|
||||
|
||||
async def async_yandex_cloud_vision(self, image, features, language):
|
||||
# TODO: Сделать функцию Vision
|
||||
return 0
|
||||
|
||||
async def collect_messages(self, message_id, chat_id):
|
||||
messages = []
|
||||
# Собираем цепочку сообщений в формате: [{"role": "user", "text": "<Имя_пользователя>: Привет!"},
|
||||
# {"role": "assistant", "text": "Привет!"}]
|
||||
while True:
|
||||
message = db_api.get_message_text(chat_id, message_id)
|
||||
if db_api.get_message_ai_model(chat_id, message_id) != None:
|
||||
messages.append({"role": "assistant", "text": message})
|
||||
else:
|
||||
sender_name = db_api.get_user_name(
|
||||
db_api.get_message_sender_id(chat_id, message_id)
|
||||
)
|
||||
messages.append({"role": "user", "text": sender_name + ": " + message})
|
||||
message_id = db_api.get_answer_to_message_id(chat_id, message_id)
|
||||
if message_id is None:
|
||||
break
|
||||
return list(reversed(messages))
|
||||
|
||||
async def collecting_messages_for_history(
|
||||
self, start_message_id, end_message_id, chat_id
|
||||
):
|
||||
messages = []
|
||||
# Собираем цепочку сообщений в формате: [{"role": "user", "text": "<Имя_пользователя>: Привет!"},
|
||||
# {"role": "assistant", "text": "Привет!"}]
|
||||
while True:
|
||||
message = db_api.get_message_text(chat_id, start_message_id)
|
||||
if db_api.get_message_ai_model(chat_id, start_message_id) != None:
|
||||
messages.append({"role": "assistant", "text": message})
|
||||
else:
|
||||
sender_name = db_api.get_user_name(
|
||||
db_api.get_message_sender_id(chat_id, start_message_id)
|
||||
)
|
||||
messages.append({"role": "user", "text": sender_name + ": " + message})
|
||||
start_message_id -= 1
|
||||
if start_message_id <= end_message_id:
|
||||
break
|
||||
return messages.reverse()
|
||||
|
||||
async def yandexgpt_request(
|
||||
self,
|
||||
message_id=None,
|
||||
type="yandexgpt-lite",
|
||||
chat_id=None,
|
||||
message_id_end=None,
|
||||
input_language=None,
|
||||
output_language=None,
|
||||
text=None,
|
||||
):
|
||||
if type == "yandexgpt-lite":
|
||||
messages = await self.collect_messages(message_id, chat_id)
|
||||
return await self.async_yandexgpt_lite(
|
||||
system_prompt=get_yandexgpt_prompt(),
|
||||
input_messages=messages,
|
||||
stream=False,
|
||||
temperature=0.6,
|
||||
max_tokens=8000,
|
||||
)
|
||||
elif type == "yandexgpt":
|
||||
# print("yandexgpt_request")
|
||||
log("yandexgpt_request")
|
||||
messages = await self.collect_messages(message_id, chat_id)
|
||||
return await self.async_yandexgpt(
|
||||
system_prompt=get_yandexgpt_prompt(),
|
||||
input_messages=messages,
|
||||
stream=False,
|
||||
temperature=0.6,
|
||||
max_tokens=get_yandexgpt_token_for_request(),
|
||||
)
|
||||
elif type == "yandexgpt-translate":
|
||||
return await self.async_yandexgpt_translate(
|
||||
input_language,
|
||||
output_language,
|
||||
text=db_api.get_message_text(chat_id, message_id),
|
||||
)
|
||||
elif type == "yandexgpt-spelling-check":
|
||||
return await self.async_yandexgpt_spelling_check(
|
||||
input_language, text=db_api.get_message_text(chat_id, message_id)
|
||||
)
|
||||
elif type == "yandexgpt-text-history":
|
||||
messages = await self.collect_messages_for_history(
|
||||
message_id, message_id_end, chat_id
|
||||
)
|
||||
return await self.async_yandexgpt_text_history(
|
||||
messages=messages, stream=False, temperature=0.6, max_tokens=8000
|
||||
)
|
||||
else:
|
||||
return "Ошибка: Неизвестный тип запроса | Error: Unknown request type"
|
||||
Loading…
Reference in New Issue
Block a user