[add] migrate chromadb to plugin

README.md

@@ -13,7 +13,7 @@
 
 ---
 
-AI Chat Bot with Plugins (RAG VectorDB - ChromaDB, DuckDuckGo Search, Home Assistant, Vehicle Lookup, YouTube)
+AI Chat Bot with Plugins (RAG VectorDB - ChromaDB, DuckDuckGo Search, Home Assistant, Vehicle Lookup)
 
 [![Build Status](https://drone.va.reichard.io/api/badges/evan/minyma/status.svg)](https://drone.va.reichard.io/evan/minyma)
 
@@ -37,20 +37,6 @@ Assistant: Some common symptoms of COVID-19 mentioned in the context are
 "Normalizing & Loading Data" section. We include a PubMed data normalizer as an
 example.
 
-### YouTube
-
-This utilizes `yt-dlp` to download a videos subtitles. Ask questions about YouTube videos!
-
-```
-User:      Tell me about this youtube video: https://www.youtube.com/watch?v=ZWgr7qP6yhY
-Assistant: The YouTube video you provided is a review of the new MacBook Pro by
-           Apple. The host discusses the laptop's features, including its new
-           color and chip. They mention that the laptop still retains its ports,
-           HDMI, and high-quality display, but also notes some shortcomings like
-           the notch and lack of face unlock. The host shares their impressions
-           of the new black color [...]
-```
-
 ### DuckDuckGo
 
 This utilizes DuckDuckGo Search by scraping the top 5 results.

minyma/api/v1.py

@@ -19,17 +19,36 @@ def get_response():
 
     resp = minyma.oai.query(message)
 
+    # Derive LLM Data
+    # llm_resp = resp.get("llm", {})
+    # llm_choices = llm_resp.get("choices", [])
+
+    # Derive VDB Data
+    # vdb_resp = resp.get("vdb", {})
+    # combined_context  = [{
+    #         "id": vdb_resp.get("ids")[i],
+    #         "distance": vdb_resp.get("distances")[i],
+    #         "doc": vdb_resp.get("docs")[i],
+    #         "metadata": vdb_resp.get("metadatas")[i],
+    # } for i, _ in enumerate(vdb_resp.get("docs", []))]
+
     # Return Data
     return resp
 
-"""
-TODO - Embeds and loads data into the local ChromaDB.
 
-{
-  "input": "string",
-  "normalizer": "string",
-}
+
 """
-bp.route("/embed", methods=["POST"])
-def post_embeddings():
-    pass
+Return the raw vector db related response
+"""
+@bp.route("/related", methods=["POST"])
+def get_related():
+    data = request.get_json()
+    if not data:
+        return {"error": "Missing Message"}
+
+    message = str(data.get("message"))
+    if message == "":
+        return {"error": "Empty Message"}
+
+    related_documents = minyma.vdb.get_related(message)
+    return related_documents

minyma/config.py

@@ -1,12 +1,11 @@
 import os
 
 
-def get_env(key, default=None, required=False) -> str | None:
+def get_env(key, default=None, required=False) -> str:
     """Wrapper for gathering env vars."""
     if required:
         assert key in os.environ, "Missing Environment Variable: %s" % key
-    env = os.environ.get(key, default)
-    return str(env) if env is not None else None
+    return str(os.environ.get(key, default))
 
 
 class Config:
@@ -20,7 +19,7 @@ class Config:
         OpenAI API Key - Required
     """
 
-    CHROMA_DATA_PATH: str | None = get_env("CHROMA_DATA_PATH", required=False)
-    HOME_ASSISTANT_API_KEY: str | None = get_env("HOME_ASSISTANT_API_KEY", required=False)
-    HOME_ASSISTANT_URL: str | None = get_env("HOME_ASSISTANT_URL", required=False)
-    OPENAI_API_KEY: str | None = get_env("OPENAI_API_KEY", required=True)
+    CHROMA_DATA_PATH: str = get_env("CHROMA_DATA_PATH", required=False)
+    HOME_ASSISTANT_API_KEY: str = get_env("HOME_ASSISTANT_API_KEY", required=False)
+    HOME_ASSISTANT_URL: str = get_env("HOME_ASSISTANT_URL", required=False)
+    OPENAI_API_KEY: str = get_env("OPENAI_API_KEY", required=True)

minyma/oai.py

@@ -1,20 +1,18 @@
-from dataclasses import dataclass
-from textwrap import indent
-from typing import Any, List
 import json
-import minyma
+from textwrap import indent
+from dataclasses import dataclass
+from typing import Any, List
 import openai
+import minyma
 
 INITIAL_PROMPT_TEMPLATE = """
-You are connected to various functions that can be used to answer the users questions. Your options are only "functions". Functions should be an array of strings containing the desired function calls (e.g. "function_name()").
-
-Available Functions:
+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}
 
-You must respond in JSON only with no other fluff or bad things will happen. The JSON keys must ONLY be "functions". Be sure to call the functions with the right arguments.
+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: {message}
+User Message: {question}
 """
 
 FOLLOW_UP_PROMPT_TEMPLATE = """
@@ -22,7 +20,7 @@ You are a helpful assistant. This is a follow up message to provide you with mor
 
 {response}
 
-User Message: {message}
+User Message: {question}
 """
 
 @dataclass
@@ -34,15 +32,13 @@ class ChatCompletion:
     choices: List[dict]
     usage: dict
 
-
 class OpenAIConnector:
     def __init__(self, api_key: str):
         self.model = "gpt-3.5-turbo"
         self.word_cap = 1000
         openai.api_key = api_key
 
-
-    def query(self, message: str) -> Any:
+    def query(self, question: str) -> Any:
         # Track Usage
         prompt_tokens = 0
         completion_tokens = 0
@@ -52,7 +48,7 @@ class OpenAIConnector:
         functions = "\n".join(list(map(lambda x: "- %s" % x["def"], minyma.plugins.plugin_defs().values())))
 
         # Create Initial Prompt
-        prompt = INITIAL_PROMPT_TEMPLATE.format(message = message, functions = indent(functions, ' ' * 2))
+        prompt = INITIAL_PROMPT_TEMPLATE.format(question = question, functions = functions)
         messages = [{"role": "user", "content": prompt}]
 
         print("[OpenAIConnector] Running Initial OAI Query")
@@ -67,7 +63,14 @@ class OpenAIConnector:
             print("[OpenAIConnector] No Results -> TODO", response)
 
         content = response.choices[0]["message"]["content"]
-        all_funcs = json.loads(content).get("functions")
+
+        # 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("),")
+            )
+        )
 
         # Update Usage
         prompt_tokens += response.usage.get("prompt_tokens", 0)
@@ -76,33 +79,20 @@ class OpenAIConnector:
 
         print("[OpenAIConnector] Completed Initial OAI Query:\n", indent(json.dumps({ "usage": response.usage, "function_calls": all_funcs }, indent=2), ' ' * 2))
 
-        # Build Response Text & Metadata
-        func_metadata = {}
-        func_response = []
-
+        # Execute Requested Functions
+        func_responses = {}
         for func in all_funcs:
-            # Execute Requested Function
-            resp = minyma.plugins.execute(func)
+            func_responses[func] = minyma.plugins.execute(func)
 
-            # Unknown Response
-            if resp is None:
-                print("[OpenAIConnector] Invalid Function Response: %s" % func)
-                continue
-
-            # Get Response
-            content = resp.get("content")
-            metadata = resp.get("metadata")
-            error = resp.get("error")
-
-            # Append Responses & Metadata
-            indented_val = indent(content or error or "Unknown Error", ' ' * 2)
-            func_response.append("- %s\n%s" % (func, indented_val))
-            func_metadata[func] = { "metadata": metadata, "error": error }
-
-        func_response = "\n".join(func_response)
+        # Build Response Text
+        response_content_arr = []
+        for key, val in func_responses.items():
+            indented_val = indent(val, ' ' * 2)
+            response_content_arr.append("- %s\n%s" % (key, indented_val))
+        response_content = "\n".join(response_content_arr)
 
         # Create Follow Up Prompt
-        prompt = FOLLOW_UP_PROMPT_TEMPLATE.format(message = message, response = func_response)
+        prompt = FOLLOW_UP_PROMPT_TEMPLATE.format(question = question, response = response_content)
         messages = [{"role": "user", "content": prompt}]
 
         print("[OpenAIConnector] Running Follup Up OAI Query")
@@ -126,7 +116,7 @@ class OpenAIConnector:
         # Return Response
         return {
             "response": content,
-            "functions": func_metadata,
+            "functions": func_responses,
             "usage": {
                 "prompt_tokens": prompt_tokens,
                 "completion_tokens": completion_tokens,

minyma/plugins/chroma_db.py

@@ -13,9 +13,8 @@ class ChromaDBPlugin(MinymaPlugin):
     def __init__(self, config):
         self.name = "chroma_db"
         self.config = config
-        self.word_cap = 1000
 
-        if config.CHROMA_DATA_PATH is None:
+        if not config.CHROMA_DATA_PATH:
             self.functions = []
         else:
             self.vdb = ChromaDB(config.CHROMA_DATA_PATH)
@@ -26,28 +25,17 @@ class ChromaDBPlugin(MinymaPlugin):
         # Get Related
         related = self.vdb.get_related(collection_name, query)
 
-        # Get Metadata
-        metadata = [{
-                "id": related.get("ids")[i],
-                "distance": related.get("distances")[i],
-                "metadata": related.get("metadatas")[i],
-        } for i, _ in enumerate(related.get("docs", []))]
-
         # Normalize Data
         return list(
             map(
-                lambda x: " ".join(x.split()[:self.word_cap]),
+                lambda x: " ".join(x.split()[:self.vdb.word_cap]),
                 related.get("docs", [])
             )
-        ), metadata
+        )
 
 
     def lookup_pubmed_data(self, query: str):
         COLLECTION_NAME = "pubmed"
-        documents, metadata = self.__lookup_data(COLLECTION_NAME, query)
+        documents = self.__lookup_data(COLLECTION_NAME, query)
         context = '\n'.join(documents)
-        return {
-            "content": context,
-            "metadata": metadata,
-            "error": None
-        }
+        return context

minyma/plugins/duckduckgo.py

@@ -14,14 +14,13 @@ class DuckDuckGoPlugin(MinymaPlugin):
     def __init__(self, config):
         self.config = config
         self.name = "duck_duck_go"
-        self.functions = [self.search_duck_duck_go]
+        self.functions = [self.duck_duck_go_search]
 
-    def search_duck_duck_go(self, query: str):
+    def duck_duck_go_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")
 
-        # Get Results
         results = []
         for item in soup.select(".result > div"):
             title_el = item.select_one(".result__title > a")
@@ -32,18 +31,4 @@ class DuckDuckGoPlugin(MinymaPlugin):
 
             results.append({"title": title, "description": description})
 
-        # Derive Metadata (Title)
-        metadata = {
-            "titles": list(
-                map(
-                    lambda x: x.get("title"),
-                    results[:5]
-                )
-            )
-        }
-
-        return {
-            "content": json.dumps(results[:5]),
-            "metadata": metadata,
-            "error": None
-        }
+        return json.dumps(results[:5])

minyma/plugins/home_assistant.py

@@ -10,15 +10,18 @@ class HomeAssistantPlugin(MinymaPlugin):
     def __init__(self, config):
         self.config = config
         self.name = "home_assistant"
-        self.functions = []
 
-        if config.HOME_ASSISTANT_API_KEY and config.HOME_ASSISTANT_URL:
-            self.functions = [self.home_automation_command]
+
+        if not config.HOME_ASSISTANT_API_KEY or not config.HOME_ASSISTANT_URL:
             if not config.HOME_ASSISTANT_API_KEY:
                 print("[HomeAssistantPlugin] Missing HOME_ASSISTANT_API_KEY")
             if not config.HOME_ASSISTANT_URL:
                 print("[HomeAssistantPlugin] Missing HOME_ASSISTANT_URL")
 
+            self.functions = []
+        else:
+            self.functions = [self.home_automation_command]
+
     def home_automation_command(self, natural_language_command: str):
         url = urllib.parse.urljoin(self.config.HOME_ASSISTANT_URL, "/api/conversation/process")
         headers = {
@@ -31,17 +34,6 @@ class HomeAssistantPlugin(MinymaPlugin):
 
         # Parse JSON
         try:
-            r = resp.json()
-            text = r["response"]["speech"]["plain"]["speech"]
-
-            return {
-                "content": text,
-                "metadata": r,
-                "error": None
-            }
+            return json.dumps(resp.json())
         except requests.JSONDecodeError:
-            return {
-                "content": None,
-                "metadata": None,
-                "error": "Command Failed"
-            }
+            return json.dumps({ "error": "Command Failed" })

minyma/plugins/vehicle_lookup.py

@@ -50,11 +50,10 @@ class VehicleLookupPlugin(MinymaPlugin):
 
         # Invalid JSON
         if json_resp is None:
-            return{
-                "content": None,
-                "metadata": text_resp,
+            return json.dumps({
                 "error": error,
-            }
+                "response": text_resp,
+            })
 
         try:
             # Check Result
@@ -64,11 +63,7 @@ class VehicleLookupPlugin(MinymaPlugin):
                     error = "No Results"
                 else:
                     error = "API Error: %s" % status_resp
-                return {
-                    "content": None,
-                    "metadata": json_resp,
-                    "error": error,
-                }
+                return {"error": error, "response": text_resp}
 
             # Parse Result
             vehicle_info = json_resp.get("content")
@@ -79,20 +74,17 @@ class VehicleLookupPlugin(MinymaPlugin):
             trim = vehicle_info.get("vehicles")[0].get("trim")
 
         except Exception as e:
-            return {
-                "content": None,
-                "metadata": text_resp,
+            return json.dumps({
                 "error": "Unknown Error: %s" % e,
-            }
+                "response": text_resp,
+            })
 
-        return {
-            "content": json.dumps({
+        return json.dumps({
+            "result": {
                 "vin": vin,
                 "year": year,
                 "make": make,
                 "model": model,
                 "trim": trim,
-            }),
-            "metadata": json_resp,
-            "error": None
-        }
+            },
+        })

minyma/plugins/youtube.py

@@ -1,53 +0,0 @@
-import os
-from yt_dlp import YoutubeDL
-import xml.etree.ElementTree as ET
-from minyma.plugin import MinymaPlugin
-
-class YouTubePlugin(MinymaPlugin):
-    """Transcribe YouTube Video"""
-
-    def __init__(self, config):
-        self.config = config
-        self.name = "youtube"
-        self.functions = [self.transcribe_youtube]
-
-
-    def transcribe_youtube(self, youtube_video_id: str):
-        URLS = [youtube_video_id]
-
-        vid = YoutubeDL({
-            "skip_download": True,
-            "writesubtitles": True,
-            "writeautomaticsub": True,
-            "subtitleslangs": ["en"],
-            "subtitlesformat": "ttml",
-            "outtmpl": "transcript"
-        })
-
-        vid.download(URLS)
-        content = self.convert_ttml_to_plain_text("transcript.en.ttml")
-        os.remove("transcript.en.ttml")
-
-        return {
-            "content": content,
-            "metadata": URLS,
-            "error": "TTML Conversion Error" if content is None else None
-        }
-
-
-    def convert_ttml_to_plain_text(self, ttml_file_path):
-        try:
-            # Parse the TTML file
-            tree = ET.parse(ttml_file_path)
-            root = tree.getroot()
-
-            # Process Text
-            plain_text = ""
-            for elem in root.iter():
-                if elem.text:
-                    plain_text += elem.text + " "
-
-            return plain_text.strip()
-        except ET.ParseError as e:
-            print("[YouTubePlugin] TTML Conversion Error:", e)
-            return None

pyproject.toml

@@ -16,8 +16,7 @@ dependencies = [
   "chromadb",
   "sqlite-utils",
   "click",
-  "beautifulsoup4",
-  "yt-dlp"
+  "beautifulsoup4"
 ]
 
 [project.scripts]