Поддержка преобразования речи в текст была в OpenAI API уже давно, а вот из текста в речь, а также распознавание изображений было добавлено совсем недавно. В связи с чем продолжаю свою серию туториалов по разработке собственного ChatGPT бота в Telegram.
Свой ChatGPT бот в Telegram в 2023
Первая статья в серии, в ней начинаем с нуля и собираем всю необходимую инфраструктуру, на выходе имеем чат-бот, который может отвечать на текстовые сообщения.
Добавляем DALL-E 3 в свой ChatGPT бот в Telegram
Вторая статья серии, на основе готового текстового бота добавляем возможность генерации изображений по текстовому описанию.
Для того чтобы наш бот мог распознавать изображения, нам надо отредактировать несколько методов в облачной функции.
import base64
Для начала добавляем библиотеку base64
в самое начало кода функции, она нам понадобится позже.
@bot.message_handler(func=lambda message: True, content_types=["text", "photo"])
def echo_message(message):
typing_process = multiprocessing.Process(target=typing, args=(message.chat.id,))
typing_process.start()
try:
text = message.text
image_content = None
photo = message.photo
if photo is not None:
photo = photo[0]
file_info = bot.get_file(photo.file_id)
image_content = bot.download_file(file_info.file_path)
text = message.caption
if text is None or len(text) == 0:
text = "Что на картинке?"
ai_response = process_text_message(text, message.chat.id, image_content)
except Exception as e:
bot.reply_to(message, f"Произошла ошибка, попробуйте позже! {e}")
return
typing_process.terminate()
bot.reply_to(message, ai_response)
Во-первых, добавляем поддержку "photo" в качестве входящего сообщения:@bot.message_handler(func=lambda message: True, content_types=["text", "photo"])
Во-вторых, скачиваем изображение, если оно было получено, и передаем функции process_text_message
.
def process_text_message(text, chat_id, image_content=None) -> str:
model = "gpt-3.5-turbo"
max_tokens = None
# read current chat history
s3client = get_s3_client()
history = []
try:
history_object_response = s3client.get_object(
Bucket=YANDEX_BUCKET, Key=f"{chat_id}.json"
)
history = json.loads(history_object_response["Body"].read())
except:
pass
history_text_only = history.copy()
history_text_only.append({"role": "user", "content": text})
if image_content is not None:
model = "gpt-4-vision-preview"
max_tokens = 4000
base64_image_content = base64.b64encode(image_content).decode("utf-8")
base64_image_content = f"data:image/jpeg;base64,{base64_image_content}"
history.append(
{
"role": "user",
"content": [
{"type": "text", "text": text},
{"type": "image_url", "image_url": {"url": base64_image_content}},
],
}
)
else:
history.append({"role": "user", "content": text})
try:
chat_completion = client.chat.completions.create(
model=model, messages=history, max_tokens=max_tokens
)
except Exception as e:
if type(e).__name__ == "InvalidRequestError":
clear_history_for_chat(chat_id)
return process_text_message(text, chat_id)
else:
raise e
ai_response = chat_completion.choices[0].message.content
history_text_only.append({"role": "assistant", "content": ai_response})
# save current chat history
s3client.put_object(
Bucket=YANDEX_BUCKET,
Key=f"{chat_id}.json",
Body=json.dumps(history_text_only),
)
return ai_response
В метод process_text_message
добавляем аргумент image_content
и, если он ненулевой, переключаемся между моделями gpt-3.5-turbo (эту будем использовать для текстовых вопросов) и gpt-4-vision-preview (для обработки присланного изображения, а также обозначаем количество токенов, которые можно использовать для ответа max_tokens
. Если этого не делать, то по умолчанию окно контекста слишком маленькое, и все токены уходят на обработку изображения.
base64_image_content = base64.b64encode(image_content).decode("utf-8")
base64_image_content = f"data:image/jpeg;base64,{base64_image_content}"
Здесь мы преобразуем содержимое файла изображения в base64-encoded формат, чтобы в таком виде послать его в API.
Сохраняем изменения и ждем, пока функция соберется.
Проверим, как работает зрение нашего бота! Для этого пошлем фото и в подписи зададим интересующий нас вопрос:
Можно посылать фото и без подписи, тогда вопрос "что на фото?" добавится автоматически при запросе к API.
Снова редактируем облачную функцию.
from telebot.types import InputFile
На этот раз импортируем специальный класс библиотеки telebot
. Он нужен нам для того, чтобы пересылать аудио-файл в OpenAI API для распознавания текста.
@bot.message_handler(
func=lambda msg: msg.voice.mime_type == "audio/ogg", content_types=["voice"]
)
def voice(message):
file_info = bot.get_file(message.voice.file_id)
downloaded_file = bot.download_file(file_info.file_path)
try:
response = client.audio.transcriptions.create(
file=("file.ogg", downloaded_file, "audio/ogg"),
model="whisper-1",
)
ai_response = process_text_message(response.text, message.chat.id)
ai_voice_response = client.audio.speech.create(
input=ai_response,
voice="nova",
model="tts-1-hd",
response_format="opus",
)
with open("/tmp/ai_voice_response.ogg", "wb") as f:
f.write(ai_voice_response.content)
except Exception as e:
bot.reply_to(message, f"Произошла ошибка, попробуйте позже! {e}")
return
with open("/tmp/ai_voice_response.ogg", "rb") as f:
bot.send_voice(
message.chat.id,
voice=InputFile(f),
reply_to_message_id=message.message_id,
)
Теперь добавляем метод, который будет обрабатывать входящие голосовые.
Сначала мы обращаемся к API для того, чтобы преобразовать речь в текст из нашего сообщения, затем посылаем этот текст в качестве обычного запроса к языковой модели, полученный ответ преобразовываем обратно в речь, и, наконец, посылаем в чат-бот в виде аудио.
Таким образом, на текстовые сообщения бот будет отвечать текстом, а на голосовые - голосовыми.
Для удобства привожу весь исходный код функции в одном месте, чтобы не переключаться между туториалами:
import logging
import telebot
import os
import openai
import json
import boto3
import time
import multiprocessing
import base64
from telebot.types import InputFile
TG_BOT_TOKEN = os.environ.get("TG_BOT_TOKEN")
TG_BOT_CHATS = os.environ.get("TG_BOT_CHATS").split(",")
PROXY_API_KEY = os.environ.get("PROXY_API_KEY")
YANDEX_KEY_ID = os.environ.get("YANDEX_KEY_ID")
YANDEX_KEY_SECRET = os.environ.get("YANDEX_KEY_SECRET")
YANDEX_BUCKET = os.environ.get("YANDEX_BUCKET")
logger = telebot.logger
telebot.logger.setLevel(logging.INFO)
bot = telebot.TeleBot(TG_BOT_TOKEN, threaded=False)
client = openai.Client(
api_key=os.getenv("PROXY_API_KEY"),
base_url="https://api.proxyapi.ru/openai/v1",
)
def get_s3_client():
session = boto3.session.Session(
aws_access_key_id=YANDEX_KEY_ID, aws_secret_access_key=YANDEX_KEY_SECRET
)
return session.client(
service_name="s3", endpoint_url="https://storage.yandexcloud.net"
)
def typing(chat_id):
while True:
bot.send_chat_action(chat_id, "typing")
time.sleep(5)
@bot.message_handler(commands=["help", "start"])
def send_welcome(message):
bot.reply_to(
message,
("Привет! Я ChatGPT бот. Спроси меня что-нибудь!"),
)
@bot.message_handler(commands=["new"])
def clear_history(message):
clear_history_for_chat(message.chat.id)
bot.reply_to(message, "История чата очищена!")
@bot.message_handler(commands=["image"])
def image(message):
prompt = message.text.split("/image")[1].strip()
if len(prompt) == 0:
bot.reply_to(message, "Введите запрос после команды /image")
return
try:
response = client.images.generate(
prompt=prompt, n=1, size="1024x1024", model="dall-e-3"
)
except:
bot.reply_to(message, "Произошла ошибка, попробуйте позже!")
return
bot.send_photo(
message.chat.id,
response.data[0].url,
reply_to_message_id=message.message_id,
)
@bot.message_handler(func=lambda message: True, content_types=["text", "photo"])
def echo_message(message):
typing_process = multiprocessing.Process(target=typing, args=(message.chat.id,))
typing_process.start()
try:
text = message.text
image_content = None
photo = message.photo
if photo is not None:
photo = photo[0]
file_info = bot.get_file(photo.file_id)
image_content = bot.download_file(file_info.file_path)
text = message.caption
if text is None or len(text) == 0:
text = "Что на картинке?"
ai_response = process_text_message(text, message.chat.id, image_content)
except Exception as e:
bot.reply_to(message, f"Произошла ошибка, попробуйте позже! {e}")
return
typing_process.terminate()
bot.reply_to(message, ai_response)
@bot.message_handler(
func=lambda msg: msg.voice.mime_type == "audio/ogg", content_types=["voice"]
)
def voice(message):
file_info = bot.get_file(message.voice.file_id)
downloaded_file = bot.download_file(file_info.file_path)
try:
response = client.audio.transcriptions.create(
file=("file.ogg", downloaded_file, "audio/ogg"),
model="whisper-1",
)
ai_response = process_text_message(response.text, message.chat.id)
ai_voice_response = client.audio.speech.create(
input=ai_response,
voice="nova",
model="tts-1-hd",
response_format="opus",
)
with open("/tmp/ai_voice_response.ogg", "wb") as f:
f.write(ai_voice_response.content)
except Exception as e:
bot.reply_to(message, f"Произошла ошибка, попробуйте позже! {e}")
return
with open("/tmp/ai_voice_response.ogg", "rb") as f:
bot.send_voice(
message.chat.id,
voice=InputFile(f),
reply_to_message_id=message.message_id,
)
def process_text_message(text, chat_id, image_content=None) -> str:
model = "gpt-3.5-turbo"
max_tokens = None
# read current chat history
s3client = get_s3_client()
history = []
try:
history_object_response = s3client.get_object(
Bucket=YANDEX_BUCKET, Key=f"{chat_id}.json"
)
history = json.loads(history_object_response["Body"].read())
except:
pass
history_text_only = history.copy()
history_text_only.append({"role": "user", "content": text})
if image_content is not None:
model = "gpt-4-vision-preview"
max_tokens = 4000
base64_image_content = base64.b64encode(image_content).decode("utf-8")
base64_image_content = f"data:image/jpeg;base64,{base64_image_content}"
history.append(
{
"role": "user",
"content": [
{"type": "text", "text": text},
{"type": "image_url", "image_url": {"url": base64_image_content}},
],
}
)
else:
history.append({"role": "user", "content": text})
try:
chat_completion = client.chat.completions.create(
model=model, messages=history, max_tokens=max_tokens
)
except Exception as e:
if type(e).__name__ == "InvalidRequestError":
clear_history_for_chat(chat_id)
return process_text_message(text, chat_id)
else:
raise e
ai_response = chat_completion.choices[0].message.content
history_text_only.append({"role": "assistant", "content": ai_response})
# save current chat history
s3client.put_object(
Bucket=YANDEX_BUCKET,
Key=f"{chat_id}.json",
Body=json.dumps(history_text_only),
)
return ai_response
def clear_history_for_chat(chat_id):
try:
s3client = get_s3_client()
s3client.put_object(
Bucket=YANDEX_BUCKET,
Key=f"{chat_id}.json",
Body=json.dumps([]),
)
except:
pass
def handler(event, context):
message = json.loads(event["body"])
update = telebot.types.Update.de_json(message)
if update.message.from_user.username.lower() in TG_BOT_CHATS:
bot.process_new_updates([update])
return {
"statusCode": 200,
"body": "ok",
}
Итак, наш ChatGPT бот в Telegram умеет отвечать на вопросы, генерировать и распознавать изображения и даже обмениваться голосовыми! Ждем новых релизов от OpenAI, чтобы сделать наш бот еще умнее!