0
0
mirror of https://gitflic.ru/project/maks1ms/ocab.git synced 2024-12-24 00:33:05 +03:00

Реализован модуль YandexGPT первой версии. Реализованы методы: генерация ответа с помощью YandexGPT и YandexGPT-lite, перевод текста с помощью YandexGPT, проверка орфографии и пунктуации с помощью YandexGPT, краткий пересказ истории чата с помощью Yandex Text Summarization Model.

This commit is contained in:
armatik 2024-01-14 00:57:33 +03:00
parent dee4f5bc79
commit 6f1efa6b7b
3 changed files with 236 additions and 0 deletions

View File

@ -1,5 +1,11 @@
TELEGRAM: TELEGRAM:
TOKEN: xxxxxxxxxxxxxxxxxxxx TOKEN: xxxxxxxxxxxxxxxxxxxx
YANDEXGPT:
TOKEN: xxxxxxxxxxxxxxxxxxxx
CATALOGID: xxxxxxxxxxxxxxxxxxxx
PROMPT: Тестовый пример промпта
ROLES: ROLES:
ADMIN: 0 ADMIN: 0
MODERATOR: 1 MODERATOR: 1

View File

@ -0,0 +1,224 @@
import requests
import json
import asyncio
import aiohttp
from ...standard.database import *
from ...standard.config.config import *
class YandexGPT:
self.token = None
self.catalog_id = None
self.language = {
"ru": "русский язык",
"en": "английский язык",
"de": "немецкий язык",
"uk": "украинский язык",
"es": "испанский язык",
"be": "белорусский язык",
}
def __init__(self, token, catalog_id):
self.token = token
self.catalog_id = catalog_id
async def async_token_check(self, messages, gpt, max_tokens, del_msg_id=1):
url = "https://llm.api.cloud.yandex.net/foundationModels/v1/tokenize"
while True:
text = ""
for message in messages:
text += message["text"]
try:
response = requests.post(url, json={"model": gpt, "text": text})
except Exception as e: # TODO: Переделать обработку ошибок
print(e)
continue
if int(response.text) < max_tokens - 2000:
break
else:
messages.pop(del_msg_id)
return messages
async def async_request(*, 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_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
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=8000):
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 = [{"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
return json.loads(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, 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
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 = get_message_text(chat_id, message_id)
if get_message_ai_model(chat_id, start_message_id) != None:
messages.append({"role": "assistant", "text": message})
else:
sender_name = get_user_name(get_message_sender_id(chat_id, start_message_id))
messages.append({"role": "user", "text": sender_name + ": " + message})
message_id = get_message_answer_to_message_id(chat_id, message_id)
if message_id is None:
break
return messages.reverse()
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 = get_message_text(chat_id, start_message_id)
if get_message_ai_model(chat_id, start_message_id) != None:
messages.append({"role": "assistant", "text": message})
else:
sender_name = get_user_name(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):
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":
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=8000
)
elif type == "yandexgpt-translate":
return await self.async_yandexgpt_translate(
input_language=get_message_language(chat_id, message_id),
output_language=get_chat_language(chat_id),
text=get_message_text(chat_id, message_id)
)
elif type == "yandexgpt-spelling-check":
return await self.async_yandexgpt_spelling_check(
input_language=get_message_language(chat_id, message_id),
text=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"

View File

@ -16,4 +16,10 @@ def get_config(is_test: bool = False) -> dict:
def get_yandexgpt_token() -> str: def get_yandexgpt_token() -> str:
return get_config()["YANDEX_GPT_TOKEN"] return get_config()["YANDEX_GPT_TOKEN"]
def get_yandexgpt_catalog_id() -> str:
return get_config()["YANDEX_GPT_CATALOG_ID"]
def get_yandexgpt_prompt() -> str:
return get_config()["YANDEX_GPT_PROMPT"]