Телеграмм бот с ChatGPT
@ArgonFox_bot
Кто знает что с этим делать тот знает.
1000 токенов в запросе
Пока оплачен
Позволяю пользоваться
код на питоне :
import logging
import asyncio
from telegram import Update
from telegram.ext import Application, CommandHandler, CallbackContext, MessageHandler, filters
import openai
import time
from collections import deque
# Вставьте ваш API токен Telegram и OpenAI
TELEGRAM_TOKEN = 'YOUR_TELEGRAM_TOKEN'
OPENAI_API_KEY = 'YOUR_OPENAI_API_KEY'
# Настройка OpenAI
openai.api_key = OPENAI_API_KEY
# Настройка логирования
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
# Начальные параметры
parameters = {
'temperature': 1.0,
'max_tokens': 1000,# Обновлено
'top_p': 1.0,
'n': 1,
'model': 'gpt-3.5-turbo'
}
# Очередь для хранения временных меток запросов
request_times = deque()
# Период времени для контроля (например, 60 секунд)
TIME_WINDOW = 60
MAX_REQUESTS = 10
def request_allowed():
current_time = time.time()
while request_times and request_times[0] < current_time - TIME_WINDOW:
request_times.popleft()
if len(request_times) < MAX_REQUESTS:
request_times.append(current_time)
return True
else:
return False
async def start(update: Update, context: CallbackContext) -> None:
await update.message.reply_text('Добро пожаловать в дом, где Лис любит Совушку. Начинаем генерацию?')
async def set_temperature(update: Update, context: CallbackContext) -> None:
try:
temp = float(context.args[0])
if 0 <= temp <= 2:
parameters['temperature'] = temp
await update.message.reply_text(f"Температура установлена на {temp}.")
else:
await update.message.reply_text("Температура должна быть в диапазоне от 0 до 2.")
except (IndexError, ValueError):
await update.message.reply_text("Использование: /set_temperature <значение>")
async def set_max_tokens(update: Update, context: CallbackContext) -> None:
try:
tokens = int(context.args[0])
if tokens > 0:
parameters['max_tokens'] = tokens
await update.message.reply_text(f"Максимальное количество токенов установлено на {tokens}.")
else:
await update.message.reply_text("Максимальное количество токенов должно быть положительным числом.")
except (IndexError, ValueError):
await update.message.reply_text("Использование: /set_max_tokens <значение>")
async def set_top_p(update: Update, context: CallbackContext) -> None:
try:
top_p = float(context.args[0])
if 0 <= top_p <= 1:
parameters['top_p'] = top_p
await update.message.reply_text(f"Top-p установлено на {top_p}.")
else:
await update.message.reply_text("Top-p должно быть в диапазоне от 0 до 1.")
except (IndexError, ValueError):
await update.message.reply_text("Использование: /set_top_p <значение>")
async def set_n(update: Update, context: CallbackContext) -> None:
try:
n = int(context.args[0])
if n > 0:
parameters['n'] = n
await update.message.reply_text(f"Количество ответов установлено на {n}.")
else:
await update.message.reply_text("Количество ответов должно быть положительным числом.")
except (IndexError, ValueError):
await update.message.reply_text("Использование: /set_n <значение>")
async def set_model(update: Update, context: CallbackContext) -> None:
try:
model = context.args[0]
available_models = ['gpt-3.5-turbo', 'gpt-4']
if model in available_models:
parameters['model'] = model
await update.message.reply_text(f"Модель установлена на {model}.")
else:
await update.message.reply_text(f"Модель должна быть одной из следующих: {', '.join(available_models)}.")
except IndexError:
await update.message.reply_text("Использование: /set_model <модель>")
except Exception as e:
logging.error("Ошибка при установке модели: %s", e)
await update.message.reply_text("Произошла ошибка при установке модели.")
async def show_status(update: Update, context: CallbackContext) -> None:
status = (f"Текущие параметры:\n"
f"Модель: {parameters['model']}\n"
f"Температура: {parameters['temperature']}\n"
f"Максимальное количество токенов: {parameters['max_tokens']}\n"
f"Top-p: {parameters['top_p']}\n"
f"Количество ответов: {parameters['n']}")
await update.message.reply_text(status)
async def respond(update: Update, context: CallbackContext) -> None:
if not request_allowed():
await update.message.reply_text("Достигнут лимит запросов. Попробуйте позже.")
return
user_message = update.message.text
try:
logging.info(f"Запрос к OpenAI: Модель={parameters['model']}, Токены={parameters['max_tokens']}, Температура={parameters['temperature']}, Top-p={parameters['top_p']}, Количество ответов={parameters['n']}")
response = openai.ChatCompletion.create(
model=parameters['model'],
messages=[{"role": "user", "content": user_message}],
max_tokens=parameters['max_tokens'],
temperature=parameters['temperature'],
top_p=parameters['top_p'],
n=parameters['n']
)
if response and response.choices and response.choices[0].message:
await send_long_message(update, response.choices[0].message['content'].strip())
else:
logging.error("Пустой ответ от OpenAI")
await update.message.reply_text("Получен пустой ответ от OpenAI.")
if 'usage' in response:
usage = response['usage']
logging.info(f"Использовано токенов: {usage['total_tokens']}")
except openai.error.RateLimitError:
logging.error("Превышен лимит запросов. Попробуйте позже.")
await update.message.reply_text("Превышен лимит запросов. Попробуйте позже.")
except openai.error.OpenAIError as e:
logging.error("Ошибка OpenAI API: %s", e)
await update.message.reply_text("Произошла ошибка при взаимодействии с API.")
except Exception as e:
logging.error("Произошла неожиданная ошибка: %s", e)
await update.message.reply_text("Произошла неожиданная ошибка. Пожалуйста, попробуйте снова позже.")
async def send_long_message(update: Update, text: str) -> None:
chunks = [text[i:i + 4096] for i in range(0, len(text), 4096)]
for chunk in chunks:
try:
await update.message.reply_text(chunk)
await asyncio.sleep(1)# Задержка в 1 секунду между сообщениями
except Exception as e:
logging.error("Ошибка при отправке сообщения: %s", e)
break
def main() -> None:
application = Application.builder().token(TELEGRAM_TOKEN).build()
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler("set_temperature", set_temperature))
application.add_handler(CommandHandler("set_max_tokens", set_max_tokens))
application.add_handler(CommandHandler("set_top_p", set_top_p))
application.add_handler(CommandHandler("set_n", set_n))
application.add_handler(CommandHandler("set_model", set_model))
application.add_handler(CommandHandler("show_status", show_status))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, respond))
application.run_polling()
if __name__ == '__main__':
main()
