Upload GPTSimple.py

GPTSimple.py

@@ -0,0 +1,101 @@
+import requests
+import json
+base_urls = {'deepinfra':"https://api.deepinfra.com/v1/openai/chat/completions", "openai":"https://api.openai.com/v1/chat/completions"}
+def print_token(token):
+    if token.token == None:
+        print()
+    else:
+        print(token.token, end="", flush=True)
+def get_direct_output(history, model, api_key, stream = False, base_url="openai"):
+    if base_url in base_urls:
+        url = base_urls[base_url]
+    else:
+        url = base_url
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {api_key}"
+    }
+
+    data = {
+        "model": model,
+        "stream":stream,
+        "messages": history
+    }
+
+    response = requests.post(url, json=data, headers=headers, stream=stream)
+    if stream:
+        return response
+    return response.json()
+class conversation:
+    class token:
+        def __init__(self, line):
+            if line['choices'][0]['finish_reason'] == "stop":
+                self.token = None
+                self.model = line["model"]
+                self.message = {'role':'assistant','content':None}
+                self.response = line
+            else:
+                self.token = line["choices"][0]['delta']['content']
+                self.model = line["model"]
+                self.message = line["choices"][0]['delta']
+                self.response = line
+    def streamingResponse(self, lines, invis):
+        message = ""
+        iters = lines.iter_lines(decode_unicode=True)
+        for line in iters:
+            if 'data: ' not in line:
+                continue
+            line_js = json.loads(line.split('data: ')[1])
+            if line_js['choices'][0]['finish_reason'] == "stop":
+                if not invis:
+                    self.history.append({'role':'assistant', 'content':message})
+                yield self.token(line_js)
+                break
+            token = self.token(line_js)
+            message += token.token
+            yield token
+    class response:
+        def __init__(self, json):
+            self.response = json
+            self.model = json['model']
+            self.id = json['id']
+            self.choices = json['choices']
+            self.text = json['choices'][0]['message']['content']
+            self.message = json['choices'][0]['message']
+            self.usage = json['usage']
+            self.prompt_tokens = json['usage']['prompt_tokens']
+            self.output_tokens = json['usage']['completion_tokens']
+            self.total_tokens = json['usage']['total_tokens']
+    def __init__(self, api_key='', model='gpt-3.5-turbo', history=None, system_prompt="You are a helpful assistant", base_url="openai"):
+        if base_url.lower() == "deepinfra" and model == "gpt-3.5-turbo":
+            model = "meta-llama/Llama-2-70b-chat-hf"
+        self.base_url = base_url.lower()
+        self.api_key = api_key
+        self.model = model
+        self.history = [{'role':'system',"content":system_prompt}]
+        if history is not None:
+            self.history = history
+    def generate(self, invisible=False, stream=False):
+        if stream:
+            res = self.streamingResponse(get_direct_output(self.history, self.model, self.api_key, stream=True, base_url=self.base_url), invisible)
+        else:
+            res = self.response(get_direct_output(self.history, self.model, self.api_key, base_url=self.base_url))
+            if not invisible:
+                self.history.append(res.message)
+        return res
+    def ask(self, message, invisible=False, stream=False):
+
+        if invisible:
+            out = self.history.copy()
+            out.append({"role":"user", "content":message})
+        else:
+            self.history.append({"role":"user", "content":message})
+            out = self.history
+        if stream:
+            res = self.streamingResponse(get_direct_output(out, self.model, self.api_key, stream=True, base_url=self.base_url), invisible)
+        else:
+            res = self.response(get_direct_output(out, self.model, self.api_key, base_url=self.base_url))
+            if not invisible:
+                self.history.append(res.message)
+        return res
+