gptManager.py

import re
from persistStorage import saveLog

class ChatgptManager:
  def __init__(self, openAIClient, model="gpt-3.5-turbo-1106", tokenLimit=8000, throwError=False, contextHistoryLen=3):
    self.client = openAIClient
    self.tokenLimit = tokenLimit
    self.model = model
    self.throwError = throwError
    self.messages = []
    self.contextHistoryLen = contextHistoryLen #sending previous n user inputs and their assistant messages. 
  
  def setSystemPrompt(self, systemPrompt):
    systemMessage = {"role":"system", "content":systemPrompt}
    if len(self.messages)==0:
      self.messages = [systemMessage]
    else:
      del self.messages[0]
      self.messages.insert(0, systemMessage)
  
  def getResponseForChatHistory(self, chatHistory, userInput):
    """
    Assume incoming chat History as [("user message", "bot message"), ("user message", "bot message")]
    and systemp prompt is fixed.
    """
    messages = [self.messages[0]]
    for history in chatHistory:
      userMessage, botMessage = history
      messages.append({"role":"user", "content":userMessage})
      messages.append({"role":"assistant", "content":botMessage})
    messages = self.getRecentContextOnly(messages)
    messages.append({"role":"user", "content":userInput})
    try:
      completion = self.client.chat.completions.create(
        model=self.model,
        messages=messages,
        temperature=0,
      )
      gptResponse = completion.choices[0].message.content
    except Exception as e:
      if not self.throwError:
        errorText = "Error while connecting with gpt " + str(e)[:100] + "..."
        return errorText
    return gptResponse

  def getResponseForUserInput(self, userInput):
    #send only recent history to gpt.
    self.messages = self.getRecentContextOnly(self.messages)
    userMessage = {"role":"user", "content":userInput}
    self.messages.append(userMessage)
    print(self.messages, "messages being sent to gpt for completion.")
    print(help(self.client.chat.completions.create),'\n\n\n')
    try:
      completion = self.client.chat.completions.create(
        model=self.model,
        messages=self.messages,
        temperature=0,
      )
      gptResponse = completion.choices[0].message.content
    except Exception as e:
      if not self.throwError:
        errorText = "Error while connecting with gpt " + str(e)[:100] + "..."
        return errorText    

    self.messages.append({"role": "assistant", "content": gptResponse})
    return gptResponse

  def getRecentContextOnly(self, messages):
    #take only systemp prompt and recent self.contextHistoryLen user input and self.contextHistoryLen assistant messages
    if len(messages)<2*self.contextHistoryLen+1:
        return messages[:]
    return [messages[0]] + messages[-2*self.contextHistoryLen:]