mirror of
https://github.com/father-bot/chatgpt_telegram_bot.git
synced 2026-06-13 12:04:57 +03:00
96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
import config
|
|
|
|
import openai
|
|
openai.api_key = config.openai_api_key
|
|
|
|
|
|
CHAT_MODES = config.chat_modes
|
|
|
|
OPENAI_COMPLETION_OPTIONS = {
|
|
"temperature": 0.7,
|
|
"max_tokens": 1000,
|
|
"top_p": 1,
|
|
"frequency_penalty": 0,
|
|
"presence_penalty": 0
|
|
}
|
|
|
|
|
|
class ChatGPT:
|
|
def __init__(self, use_chatgpt_api=True):
|
|
self.use_chatgpt_api = use_chatgpt_api
|
|
|
|
async def send_message(self, message, dialog_messages=[], chat_mode="assistant"):
|
|
if chat_mode not in CHAT_MODES.keys():
|
|
raise ValueError(f"Chat mode {chat_mode} is not supported")
|
|
|
|
n_dialog_messages_before = len(dialog_messages)
|
|
answer = None
|
|
while answer is None:
|
|
try:
|
|
if self.use_chatgpt_api:
|
|
messages = self._generate_prompt_messages_for_chatgpt_api(message, dialog_messages, chat_mode)
|
|
r = await openai.ChatCompletion.acreate(
|
|
model="gpt-3.5-turbo",
|
|
messages=messages,
|
|
**OPENAI_COMPLETION_OPTIONS
|
|
)
|
|
answer = r.choices[0].message["content"]
|
|
else:
|
|
prompt = self._generate_prompt(message, dialog_messages, chat_mode)
|
|
r = await openai.Completion.acreate(
|
|
engine="text-davinci-003",
|
|
prompt=prompt,
|
|
**OPENAI_COMPLETION_OPTIONS
|
|
)
|
|
answer = r.choices[0].text
|
|
|
|
answer = self._postprocess_answer(answer)
|
|
n_used_tokens = r.usage.total_tokens
|
|
|
|
except openai.error.InvalidRequestError as e: # too many tokens
|
|
if len(dialog_messages) == 0:
|
|
raise ValueError("Dialog messages is reduced to zero, but still has too many tokens to make completion") from e
|
|
|
|
# forget first message in dialog_messages
|
|
dialog_messages = dialog_messages[1:]
|
|
|
|
n_first_dialog_messages_removed = n_dialog_messages_before - len(dialog_messages)
|
|
|
|
return answer, n_used_tokens, n_first_dialog_messages_removed
|
|
|
|
def _generate_prompt(self, message, dialog_messages, chat_mode):
|
|
prompt = CHAT_MODES[chat_mode]["prompt_start"]
|
|
prompt += "\n\n"
|
|
|
|
# add chat context
|
|
if len(dialog_messages) > 0:
|
|
prompt += "Chat:\n"
|
|
for dialog_message in dialog_messages:
|
|
prompt += f"User: {dialog_message['user']}\n"
|
|
prompt += f"ChatGPT: {dialog_message['bot']}\n"
|
|
|
|
# current message
|
|
prompt += f"User: {message}\n"
|
|
prompt += "ChatGPT: "
|
|
|
|
return prompt
|
|
|
|
def _generate_prompt_messages_for_chatgpt_api(self, message, dialog_messages, chat_mode):
|
|
prompt = CHAT_MODES[chat_mode]["prompt_start"]
|
|
|
|
messages = [{"role": "system", "content": prompt}]
|
|
for dialog_message in dialog_messages:
|
|
messages.append({"role": "user", "content": dialog_message["user"]})
|
|
messages.append({"role": "assistant", "content": dialog_message["bot"]})
|
|
messages.append({"role": "user", "content": message})
|
|
|
|
return messages
|
|
|
|
def _postprocess_answer(self, answer):
|
|
answer = answer.strip()
|
|
return answer
|
|
|
|
|
|
async def transcribe_audio(audio_file):
|
|
r = await openai.Audio.atranscribe("whisper-1", audio_file)
|
|
return r["text"] |