feat: 初始化

base/func_chatgpt.py

@@ -0,0 +1,106 @@
+#! /usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import logging
+from datetime import datetime
+
+import httpx
+from openai import APIConnectionError, APIError, AuthenticationError, OpenAI
+
+
+class ChatGPT():
+    def __init__(self, conf: dict) -> None:
+        key = conf.get("key")
+        api = conf.get("api")
+        proxy = conf.get("proxy")
+        prompt = conf.get("prompt")
+        self.model = conf.get("model", "gpt-3.5-turbo")
+        self.LOG = logging.getLogger("ChatGPT")
+        if proxy:
+            self.client = OpenAI(api_key=key, base_url=api, http_client=httpx.Client(proxy=proxy))
+        else:
+            self.client = OpenAI(api_key=key, base_url=api)
+        self.conversation_list = {}
+        self.system_content_msg = {"role": "system", "content": prompt}
+
+    def __repr__(self):
+        return 'ChatGPT'
+
+    @staticmethod
+    def value_check(conf: dict) -> bool:
+        if conf:
+            if conf.get("key") and conf.get("api") and conf.get("prompt"):
+                return True
+        return False
+
+    def get_answer(self, question: str, wxid: str) -> str:
+        # wxid或者roomid,个人时为微信id，群消息时为群id
+        self.updateMessage(wxid, question, "user")
+        rsp = ""
+        try:
+            ret = self.client.chat.completions.create(model=self.model,
+                                                      messages=self.conversation_list[wxid],
+                                                      temperature=0.2)
+            rsp = ret.choices[0].message.content
+            rsp = rsp[2:] if rsp.startswith("\n\n") else rsp
+            rsp = rsp.replace("\n\n", "\n")
+            self.updateMessage(wxid, rsp, "assistant")
+        except AuthenticationError:
+            self.LOG.error("OpenAI API 认证失败，请检查 API 密钥是否正确")
+        except APIConnectionError:
+            self.LOG.error("无法连接到 OpenAI API，请检查网络连接")
+        except APIError as e1:
+            self.LOG.error(f"OpenAI API 返回了错误：{str(e1)}")
+        except Exception as e0:
+            self.LOG.error(f"发生未知错误：{str(e0)}")
+
+        return rsp
+
+    def updateMessage(self, wxid: str, question: str, role: str) -> None:
+        now_time = str(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+
+        time_mk = "当需要回答时间时请直接参考回复:"
+        # 初始化聊天记录,组装系统信息
+        if wxid not in self.conversation_list.keys():
+            question_ = [
+                self.system_content_msg,
+                {"role": "system", "content": "" + time_mk + now_time}
+            ]
+            self.conversation_list[wxid] = question_
+
+        # 当前问题
+        content_question_ = {"role": role, "content": question}
+        self.conversation_list[wxid].append(content_question_)
+
+        for cont in self.conversation_list[wxid]:
+            if cont["role"] != "system":
+                continue
+            if cont["content"].startswith(time_mk):
+                cont["content"] = time_mk + now_time
+
+        # 只存储10条记录，超过滚动清除
+        i = len(self.conversation_list[wxid])
+        if i > 10:
+            print("滚动清除微信记录：" + wxid)
+            # 删除多余的记录，倒着删，且跳过第一个的系统消息
+            del self.conversation_list[wxid][1]
+
+
+if __name__ == "__main__":
+    from configuration import Config
+    config = Config().CHATGPT
+    if not config:
+        exit(0)
+
+    chat = ChatGPT(config)
+
+    while True:
+        q = input(">>> ")
+        try:
+            time_start = datetime.now()  # 记录开始时间
+            print(chat.get_answer(q, "wxid"))
+            time_end = datetime.now()  # 记录结束时间
+
+            print(f"{round((time_end - time_start).total_seconds(), 2)}s")  # 计算的时间差为程序的执行时间，单位为秒/s
+        except Exception as e:
+            print(e)