minyma/__init__.py

@@ -3,6 +3,7 @@ import click
 import signal
 import sys
 from importlib.metadata import version
+from minyma.plugin import PluginLoader
 from minyma.oai import OpenAIConnector
 from minyma.vdb import ChromaDB
 from flask import Flask
@@ -15,7 +16,7 @@ def signal_handler(sig, frame):
 
 
 def create_app():
-    global oai, vdb
+    global oai, vdb, plugins
 
     from minyma.config import Config
     import minyma.api.common as api_common
@@ -24,6 +25,7 @@ def create_app():
     app = Flask(__name__)
     vdb = ChromaDB(path.join(Config.DATA_PATH, "chroma"))
     oai = OpenAIConnector(Config.OPENAI_API_KEY, vdb)
+    plugins = PluginLoader()
 
     app.register_blueprint(api_common.bp)
     app.register_blueprint(api_v1.bp)

minyma/oai.py

@@ -1,7 +1,11 @@
-from typing import Any
+import json
+from textwrap import indent
+from dataclasses import dataclass
+from typing import Any, List
 import openai
 
 from minyma.vdb import VectorDB
+import minyma
 
 # Stolen LangChain Prompt
 PROMPT_TEMPLATE = """
@@ -15,6 +19,33 @@ Question: {question}
 Helpful Answer:
 """
 
+INITIAL_PROMPT_TEMPLATE = """
+You are a helpful assistant. You are connected to various external functions that can provide you with more personalized and up-to-date information and have already been granted the permissions to execute these functions at will. DO NOT say you don't have access to real time information, instead attempt to call one or more of the listed functions:
+
+{functions}
+
+The user will not see your response. You must only respond with a comma separated list of function calls: "FUNCTION_CALLS: function(), function(), etc". It must be prepended by "FUNCTION_CALLS:".
+
+User Message: {question}
+"""
+
+FOLLOW_UP_PROMPT_TEMPLATE = """
+You are a helpful assistant. This is a follow up message to provide you with more context on a previous user request. Only respond to the user using the following information:
+
+{response}
+
+User Message: {question}
+"""
+
+@dataclass
+class ChatCompletion:
+    id: str
+    object: str
+    created: int
+    model: str
+    choices: List[dict]
+    usage: dict
+
 class OpenAIConnector:
     def __init__(self, api_key: str, vdb: VectorDB):
         self.vdb = vdb
@@ -23,6 +54,65 @@ class OpenAIConnector:
         openai.api_key = api_key
 
     def query(self, question: str) -> Any:
+        # Get Available Functions
+        functions = "\n".join(list(map(lambda x: "- %s" % x["def"], minyma.plugins.plugin_defs().values())))
+
+        # Create Initial Prompt
+        prompt = INITIAL_PROMPT_TEMPLATE.format(question = question, functions = functions)
+        messages = [{"role": "user", "content": prompt}]
+
+        print("[OpenAIConnector] Running Initial OAI Query")
+
+        # Run Initial
+        response: ChatCompletion = openai.ChatCompletion.create( # type: ignore
+          model=self.model,
+          messages=messages
+        )
+
+        if len(response.choices) == 0:
+            print("[OpenAIConnector] No Results -> TODO", response)
+
+        content = response.choices[0]["message"]["content"]
+
+        # Get Called Functions (TODO - Better Validation -> Failback Prompt?)
+        all_funcs = list(
+            map(
+                lambda x: x.strip() if x.endswith(")") else x.strip() + ")",
+                content.split("FUNCTION_CALLS:")[1].strip().split("),")
+            )
+        )
+
+        print("[OpenAIConnector] Completed Initial OAI Query:\n", indent(json.dumps({ "usage": response.usage, "function_calls": all_funcs }, indent=2), ' ' * 2))
+
+        # Execute Functions
+        func_responses = {}
+        for func in all_funcs:
+            func_responses[func] = minyma.plugins.execute(func)
+
+        # Build Response Text
+        response_content_arr = []
+        for key, val in func_responses.items():
+            response_content_arr.append("- %s\n%s" % (key, val))
+        response_content = "\n".join(response_content_arr)
+
+        # Create Follow Up Prompt
+        prompt = FOLLOW_UP_PROMPT_TEMPLATE.format(question = question, response = response_content)
+        messages = [{"role": "user", "content": prompt}]
+
+        print("[OpenAIConnector] Running Follup Up OAI Query")
+
+        # Run Follow Up
+        response: ChatCompletion = openai.ChatCompletion.create( # type: ignore
+          model=self.model,
+          messages=messages
+        )
+
+        print("[OpenAIConnector] Completed Follup Up OAI Query:\n", indent(json.dumps({ "usage": response.usage }, indent=2), ' ' * 2))
+
+        # Return Response
+        return { "llm": response }
+
+    def old_query(self, question: str) -> Any:
         # Get related documents from vector db
         related = self.vdb.get_related(question)
 

minyma/plugin.py

@@ -0,0 +1,124 @@
+import re
+import inspect
+import os
+import importlib.util
+
+class MinymaPlugin:
+    pass
+
+class PluginLoader:
+    def __init__(self):
+        self.plugins = self.get_plugins()
+        self.definitions = self.plugin_defs()
+
+
+    def execute(self, func_cmd):
+        print("[PluginLoader] Execute Function:", func_cmd)
+
+        pattern = r'([a-z_]+)\('
+
+        func_name_search = re.search(pattern, func_cmd)
+        if not func_name_search:
+            return
+
+        func_name = func_name_search.group(1)
+
+        # Not Safe
+        if func_name in self.definitions:
+            args = re.sub(pattern, '(', func_cmd)
+            func = self.definitions[func_name]["func"]
+            return eval("func%s" % args)
+
+
+    def plugin_defs(self):
+        defs = {}
+        for plugin in self.plugins:
+            plugin_name = plugin.name
+
+            for func_obj in plugin.functions:
+                func_name = func_obj.__name__
+                function_name = "%s_%s" % (plugin_name, func_name)
+
+                signature = inspect.signature(func_obj)
+                params = list(
+                    map(
+                        lambda x: "%s: %s" % (x.name, x.annotation.__name__),
+                        signature.parameters.values()
+                    )
+                )
+
+                func_def = "%s(%s)" % (function_name, ", ".join(params))
+                defs[function_name] = { "func": func_obj, "def": func_def }
+
+        return defs
+
+    def derive_function_definitions(self):
+        """Dynamically generate function definitions"""
+        function_definitions = []
+
+        for plugin in self.plugins:
+            plugin_name = plugin.name
+
+            for method_obj in plugin.functions:
+                method_name = method_obj.__name__
+                signature = inspect.signature(method_obj)
+                parameters = list(signature.parameters.values())
+
+                # Extract Parameter Information
+                params_info = {}
+                for param in parameters:
+                    param_name = param.name
+                    param_type = param.annotation
+                    params_info[param_name] = {"type": TYPE_DEFS[param_type]}
+
+                # Store Function Information
+                method_info = {
+                    "name": "%s_%s" %(plugin_name, method_name),
+                    "description": inspect.getdoc(method_obj),
+                    "parameters": {
+                        "type": "object",
+                        "properties": params_info
+                    }
+                }
+                function_definitions.append(method_info)
+
+        return function_definitions
+
+
+    def get_plugins(self):
+        """Dynamically load plugins"""
+        # Derive Plugin Folder
+        loader_dir = os.path.dirname(os.path.abspath(__file__))
+        plugin_folder = os.path.join(loader_dir, "plugins")
+
+        # Find Minyma Plugins
+        plugin_classes = []
+        for filename in os.listdir(plugin_folder):
+
+            # Exclude Files
+            if not filename.endswith(".py") or filename == "__init__.py":
+                continue
+
+            # Derive Module Path
+            module_name = os.path.splitext(filename)[0]
+            module_path = os.path.join(plugin_folder, filename)
+
+            # Load Module Dynamically
+            spec = importlib.util.spec_from_file_location(module_name, module_path)
+            if spec is None or spec.loader is None:
+                raise ImportError("Unable to dynamically load plugin - %s" % filename)
+
+            # Load & Exec Module
+            module = importlib.util.module_from_spec(spec)
+            spec.loader.exec_module(module)
+
+            # Only Process MinymaPlugin SubClasses
+            for _, member in inspect.getmembers(module):
+                if inspect.isclass(member) and issubclass(member, MinymaPlugin) and member != MinymaPlugin:
+                    plugin_classes.append(member)
+
+        # Instantiate Plugins
+        plugins = []
+        for cls in plugin_classes:
+            plugins.append(cls())
+        return plugins

minyma/plugins/README.md

@@ -0,0 +1,3 @@
+# Plugins
+
+These are plugins that provide OpenAI with functions. Each plugin can define multiple plugins. The plugin loader will automatically derive the function definition. Each function will have the plugin name prepended.

minyma/plugins/duckduckgo.py

@@ -0,0 +1,33 @@
+import json
+import requests
+from bs4 import BeautifulSoup
+from minyma.plugin import MinymaPlugin
+
+HEADERS = {
+    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:105.0)"
+    " Gecko/20100101 Firefox/105.0",
+}
+
+class DuckDuckGoPlugin(MinymaPlugin):
+    """Search DuckDuckGo"""
+
+    def __init__(self):
+        self.name = "duck_duck_go"
+        self.functions = [self.search]
+
+    def search(self, query: str):
+        """Search DuckDuckGo"""
+        resp = requests.get("https://html.duckduckgo.com/html/?q=%s" % query, headers=HEADERS)
+        soup = BeautifulSoup(resp.text, features="html.parser")
+
+        results = []
+        for item in soup.select(".result > div"):
+            title_el = item.select_one(".result__title > a")
+            title = title_el.text.strip() if title_el and title_el.text is not None else ""
+
+            description_el = item.select_one(".result__snippet")
+            description = description_el.text.strip() if description_el and description_el.text is not None else ""
+
+            results.append({"title": title, "description": description})
+
+        return json.dumps(results[:5])

minyma/plugins/vehicle_lookup.py

@@ -0,0 +1,90 @@
+import json
+import requests
+from bs4 import BeautifulSoup
+from minyma.plugin import MinymaPlugin
+
+HEADERS = {
+    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:105.0)"
+    " Gecko/20100101 Firefox/105.0",
+}
+
+class VehicleLookupPlugin(MinymaPlugin):
+    """Search Vehicle Information"""
+
+    def __init__(self):
+        self.name = "vehicle_state_plate"
+        self.functions = [self.lookup]
+
+    def __query_api(self, url, json=None, headers=None):
+        # Perform Request
+        if json is not None:
+            resp = requests.post(url, json=json, headers=headers)
+        else:
+            resp = requests.get(url, headers=headers)
+
+        # Parse Text
+        text = resp.text.strip()
+
+        # Parse JSON
+        try:
+            json = resp.json()
+            return json, text, None
+        except requests.JSONDecodeError:
+            error = None
+            if resp.status_code != 200:
+                error = "Invalid HTTP Response: %s" % resp.status_code
+            else:
+                error = "Invalid JSON"
+            return None, text, error
+
+
+    def lookup(self, state_abbreviation: str, licence_plate: str):
+        CARVANA_URL = (
+            "https://apim.carvana.io/trades/api/v5/vehicleconfiguration/plateLookup/%s/%s"
+            % (state_abbreviation, licence_plate)
+        )
+
+        # Query API
+        json_resp, text_resp, error = self.__query_api(CARVANA_URL, headers=HEADERS)
+
+        # Invalid JSON
+        if json_resp is None:
+            return {
+                "error": error,
+                "response": text_resp,
+            }
+
+        try:
+
+            # Check Result
+            status_resp = json_resp.get("status", "Unknown")
+            if status_resp != "Succeeded":
+                if status_resp == "MissingResource":
+                    error = "No Results"
+                else:
+                    error = "API Error: %s" % status_resp
+                return {"error": error, "response": text_resp}
+
+            # Parse Result
+            vehicle_info = json_resp.get("content")
+            vin = vehicle_info.get("vin")
+            year = vehicle_info.get("vehicles")[0].get("year")
+            make = vehicle_info.get("vehicles")[0].get("make")
+            model = vehicle_info.get("vehicles")[0].get("model")
+            trim = vehicle_info.get("vehicles")[0].get("trim")
+
+        except Exception as e:
+            return {
+                "error": "Unknown Error: %s" % e,
+                "response": text_resp,
+            }
+
+        return json.dumps({
+            "result": {
+                "vin": vin,
+                "year": year,
+                "make": make,
+                "model": model,
+                "trim": trim,
+            },
+        })

minyma/templates/index.html

@@ -163,7 +163,8 @@
             let responseEl = document.createElement("p");
             responseEl.setAttribute(
               "class",
-              "whitespace-break-spaces border-b pb-3 mb-3"
+              "whitespace-break-spaces"
+              // "whitespace-break-spaces border-b pb-3 mb-3"
             );
             responseEl.innerText = data.response;
 
@@ -214,7 +215,7 @@
             content.setAttribute("class", "w-full");
             content.innerHTML = "";
             content.append(responseEl);
-            content.append(contextEl);
+            // content.append(contextEl);
           })
           .catch((e) => {
             console.log("ERROR:", e);

pyproject.toml

@@ -15,7 +15,8 @@ dependencies = [
   "tqdm",
   "chromadb",
   "sqlite-utils",
-  "click"
+  "click",
+  "beautifulsoup4"
 ]
 
 [project.scripts]