feat: support azure openai service

bot/azure_chatgpt.py

@@ -0,0 +1,43 @@
+import openai
+import requests
+from bot.chatgpt import ChatGPTBot
+from common.singleton import singleton
+from config import conf
+from utils.log import logger
+from common.reply import Reply, ReplyType
+
+
+@singleton
+class AzureChatGPTBot(ChatGPTBot):
+    def __init__(self):
+        super().__init__()
+        openai.api_type = "azure"
+        openai.api_version = "2023-03-15-preview"
+        self.args["deployment_id"] = conf().get("azure_deployment_id")
+
+    def reply_img(self, query):
+        create_image_size = conf().get("create_image_size", "256x256")
+        try:
+            response = openai.Image.create(prompt=query, n=1, size=create_image_size)
+            image_url = response["data"][0]["url"]
+            logger.info(f"[ChatGPT] Image={image_url}")
+            return Reply(ReplyType.IMAGE, image_url)
+        except Exception as e:
+            logger.error(f"[ChatGPT] Create image failed: {e}")
+            return Reply(ReplyType.TEXT, "Image created failed")
+
+    def create_img(self, query):
+        url = f"{openai.api_base}dalle/text-to-image?api-version=2022-08-03-preview"
+        headers = {"api-key": openai.api_key, "Content-Type": "application/json"}
+        create_image_size = conf().get("create_image_size", "256x256")
+        try:
+            body = {"caption": query, "resolution": create_image_size}
+            submission = requests.post(url, headers=headers, json=body)
+            operation_location = submission.headers["Operation-Location"]
+            response = requests.get(operation_location, headers=headers)
+            image_url = response.json()["result"]["contentUrl"]
+            logger.info(f"[ChatGPT] Image={image_url}")
+            return Reply(ReplyType.IMAGE, image_url)
+        except Exception as e:
+            logger.error(f"[ChatGPT] Create image failed: {e}")
+            return Reply(ReplyType.TEXT, "Image created failed")

bot/chatgpt.py

@@ -17,6 +17,10 @@ def __init__(self):
             openai.api_base = api_base
         if proxy:
             openai.proxy = proxy
+        self.args = {
+            "model": conf().get("model"),
+            "temperature": conf().get("temperature"),
+        }
 
     def reply(self, context: Context):
         query = context.query
@@ -46,16 +50,13 @@ def reply_img(self, query):
             return Reply(ReplyType.TEXT, "Image created failed")
 
     def reply_text(self, session):
-        model = conf().get("model")
-        temperature = conf().get("temperature")
         try:
             response = openai.ChatCompletion.create(
-                model=model,
                 messages=session,
-                temperature=temperature,
                 top_p=1.0,
                 frequency_penalty=0.0,
                 presence_penalty=0.0,
+                **self.args,
             )
             return {
                 "total_tokens": response["usage"]["total_tokens"],

common/singleton.py

@@ -1,9 +1,11 @@
 def singleton(cls):
     instances = {}
 
-    def get_instance(*args, **kwargs):
-        if cls not in instances:
-            instances[cls] = cls(*args, **kwargs)
-        return instances[cls]
+    class Singleton(cls):
+        def __new__(cls, *args, **kwargs):
+            if cls not in instances:
+                instances[cls] = super(Singleton, cls).__new__(cls, *args, **kwargs)
+            return instances[cls]
 
-    return get_instance
+    Singleton.__name__ = cls.__name__
+    return Singleton

config.template.json

@@ -1,6 +1,7 @@
 {
   "openai_api_key": "YOUR API SECRET KEY",
   "model": "gpt-3.5-turbo",
+  "azure_deployment_id": "",
   "role_desc": "You are a helpful assistant.",
   "session_expired_duration": 3600,
   "max_tokens": 1000,