kagi_fastgpt: new expensive plugin to replace GPT3

config.json.example

@@ -59,7 +59,8 @@
           "awake": true
         }
       },
-      "question_mark": "?"
+      "question_mark": "?",
+      "computing_replies": ["Hmm…"]
     },
     "compliments": {
       "sentences": ["you're breathtaking"],
@@ -70,8 +71,7 @@
     },
     "gpt3": {
       "openai_key": "",
-      "join_lines": "; ",
-      "computing_replies": ["Hmm…"]
+      "join_lines": "; "
     },
     "horoscope": {
       "commands": ["horoscope"],
@@ -90,6 +90,10 @@
       "dates": ["01-01 NYE"],
       "no_entry_reply": "Nothing special today."
     },
+    "kagifastgpt": {
+      "api_key": "",
+      "prompt": ""
+    },
     "meteofrance": {
       "commands": ["météo"],
       "aliases": {

edmond/plugins/gpt3.py

@@ -10,7 +10,7 @@ from edmond.utils import limit_text_length
 
 class Gpt3Plugin(Plugin):
 
-    REQUIRED_CONFIGS = ["openai_key", "computing_replies", "join_lines"]
+    REQUIRED_CONFIGS = ["openai_key", "join_lines"]
 
     def __init__(self, bot):
         super().__init__(bot)

edmond/plugins/kagi_fastgpt.py

@@ -0,0 +1,78 @@
+import json
+import random
+import re
+from typing import Optional, Tuple, cast
+
+import requests
+
+from edmond.plugin import Plugin
+from edmond.plugins.plus import PlusPlugin
+from edmond.utils import limit_text_length
+
+
+class KagiFastgptPlugin(Plugin):
+
+    BASE_URL = "https://kagi.com/api/v0/fastgpt"
+    REQUIRED_CONFIGS = ["api_key"]
+
+    def __init__(self, bot):
+        super().__init__(bot)
+        self.api_key = self.config["api_key"]
+        self.prompt = self.config.get("prompt", "")
+
+    def on_welcome(self, _):
+        if not self.api_key:
+            self.bot.log_w("Kagi FastGPT API key unavailable.")
+            self.is_ready = False
+
+    def reply(self, query: str, target: str):
+        computing_reply = random.choice(self.config["computing_replies"])
+        self.bot.say(target, computing_reply)
+
+        output, references = self.complete(query)
+        if output:
+            self.bot.say(target, self.sanitize(output))
+            self.register_references_for_plus(references, target)
+        else:
+            self.signal_failure(target)
+
+    def complete(self, query: str) -> Tuple[Optional[str], list]:
+        try:
+            response = requests.post(
+                self.BASE_URL,
+                headers={"Authorization": f"Bot {self.api_key}"},
+                json={
+                    "query": self.prompt + query
+                }
+            )
+        except requests.RequestException as exc:
+            self.bot.log_e(f"Request error: {exc}")
+            return None, []
+
+        data = response.json().get("data", {})
+        self.bot.log_d(f"Data received: {json.dumps(data)}")
+        output = data.get("output", "")
+        if not output:
+            self.bot.log_w("Empty FastGPT output!")
+            return None, []
+
+        references = data.get("references", [])
+        return output, references
+
+    def register_references_for_plus(
+        self,
+        references: list[str],
+        target: str
+    ) -> None:
+        if references and (plus_plugin := self.bot.get_plugin("plus")):
+            def handler(plus_event):
+                for ref in references[:3]:
+                    message = ref["title"] + " " + ref["url"]
+                    self.bot.say(plus_event.target, message)
+            cast(PlusPlugin, plus_plugin).add_handler(target, handler)
+
+    def sanitize(self, text: str) -> str:
+        text = text.strip()
+        text = re.sub(r"\n+", " — ", text)
+        text = limit_text_length(text)
+        return text

edmond/plugins/unknown_command.py

@@ -1,5 +1,5 @@
 from edmond.plugin import Plugin
-from edmond.plugins.gpt3 import Gpt3Plugin
+from edmond.plugins.kagi_fastgpt import KagiFastgptPlugin
 
 
 class UnknownCommandPlugin(Plugin):
@@ -7,12 +7,12 @@ class UnknownCommandPlugin(Plugin):
     def __init__(self, bot):
         super().__init__(bot)
         self.priority: int = -6
-        self.gpt3_plugin: Gpt3Plugin
+        self.gpt_plugin: KagiFastgptPlugin
 
     def on_welcome(self, _):
-        self.gpt3_plugin = self.bot.get_plugin("gpt3")
-        if self.gpt3_plugin is None or not self.gpt3_plugin:
-            self.bot.log_w("GPT-3 plugin is not available.")
+        self.gpt_plugin = self.bot.get_plugin("kagifastgpt")
+        if self.gpt_plugin is None or not self.gpt_plugin:
+            self.bot.log_w("GPT plugin is not available.")
             self.is_ready = False
 
     def on_pubmsg(self, event):
@@ -26,6 +26,5 @@ class UnknownCommandPlugin(Plugin):
         query = " ".join(words[:-1])
         if not query.endswith("."):
             query += "."
-        query += "\n\n"
-        self.gpt3_plugin.reply(query, event.target)
+        self.gpt_plugin.reply(query, event.target)
         return True

edmond/plugins/unknown_question.py

@@ -17,16 +17,15 @@ class UnknownQuestionPlugin(Plugin):
 
     def on_welcome(self, _):
         self.misc_plugin = self.bot.get_plugin("miscreactions")
-        self.gpt3_plugin = self.bot.get_plugin("gpt3")
+        self.gpt_plugin = self.bot.get_plugin("kagifastgpt")
 
     def on_pubmsg(self, event):
         message = self.should_read_message(event.arguments[0])
         if message is None:
             return False
 
-        if self.gpt3_plugin and message.endswith(self.config["question_mark"]):
-            message += "\n\n"
-            self.gpt3_plugin.reply(message, event.target)
+        if self.gpt_plugin and message.endswith(self.config["question_mark"]):
+            self.gpt_plugin.reply(message, event.target)
         else:
             self.classic_reply(event)
         return True

edmond/plugins/wolfram_alpha.py

@@ -23,7 +23,7 @@ class WolframAlphaPlugin(Plugin):
 
     def on_welcome(self, _):
         if not self.config["api_key"]:
-            self.bot.log_w("API key unavailable.")
+            self.bot.log_w("Wolfram API key unavailable.")
             self.is_ready = False
 
     def on_pubmsg(self, event):

edmond/plugins/youtube.py

@@ -40,7 +40,7 @@ class YoutubePlugin(Plugin):
 
     def on_welcome(self, _):
         if not self.config["api_key"]:
-            self.bot.log_w("API key unavailable.")
+            self.bot.log_w("Youtube API key unavailable.")
             self.is_ready = False
 
     def on_pubmsg(self, event):