Merge pull request #3 from roderick-bishop11/strong_types

Stronger typing, exa search, README

assistant/README.md

@@ -0,0 +1,40 @@
+# E.L.O. proof of concept - aka "assistant"
+
+## Objective
+
+ELO = Enhanced Logical Operative
+To create an environment control plane entity powered by AI.
+
+## Scope
+
+This project is meant to be POC work. "Garage time", if you will.
+
+No R.A.G. for this project, however MoE may be explored.
+
+## Inspiration
+
+Tony Stark, J.A.R.V.I.S. and Robert Elo Bishop(my grandfather). ELO is meant to be more than an assistant. If you think of Operating Systems working on computers, think of ELO as an *operative*. Another way to define the concept is a Semi-Autonomous AI powered operative working on your behalf. Outfit with integrations, standard library of defined functions, local access, and intuitive interfacing.
+
+## Desired state
+
+"Assistant (this folder) is meant to be the POC for ELO. In this I'd like to accomplish or deliver on this basic feature set:
+
+- A support window for ELO to show/render result sets from queries
+- A control window where ELO will display chat, diagnostics, etc.
+- Chat GPT integration or Local LLM generation; The option to choose/switch models.
+- Internet Search (whitelisted sources)
+- Wake and shut down word/phrases/intents
+
+## tech stack/integrations
+
+Python- language
+Poetry- build system
+pyttsx4/Elevel labs- Text to speech
+OpenAI API- interact with models
+Exa- Search for AI
+Slint-Desktop UI for python
+
+## Changelog
+
+## contributors
+

assistant/assistant/__init__.py

@@ -1,3 +1,6 @@
+import threading
+import time
+
 import speech_recognition as sr
 import pyttsx4
 import datetime
@@ -6,88 +9,152 @@
 from elevenlabs import play
 import os
 from dotenv import load_dotenv
+from exa_py import Exa
 
 load_dotenv()
 
-gptKey = os.getenv("OPENAI_API_KEY")
-elevenKey = os.getenv("ELEVENLABS_API_KEY")
-client = OpenAI(api_key=gptKey)
+gptKey: str = os.getenv("OPENAI_API_KEY")
+elevenKey: str = os.getenv("ELEVENLABS_API_KEY")
+exaKey: str = os.getenv("EXA_KEY")
+client: OpenAI = OpenAI(api_key=gptKey)
+
+# initialize voice components
+recognizer: sr.Recognizer = sr.Recognizer()
+ttsEngine: pyttsx4.Engine = pyttsx4.init()
 elevenlabs.set_api_key(api_key=elevenKey)
 
-# initialize the recognizer and engine
-recognizer = sr.Recognizer()
-ttsEngine = pyttsx4.init()
+# initialize integration components
+exa: Exa = Exa(exaKey)
 
 
-def speak(text):
+def speakAndPrint(text) -> None:
+    # audio = elevenlabs.generate(
+    #     text=text, voice="Adam", model="eleven_monolingual_v1", api_key=elevenKey
+    # )
+    # play(audio)
+    print(text)
+    ttsEngine.say(text=text)
+    ttsEngine.runAndWait()
 
-    audio = elevenlabs.generate(
-            text=text, voice="Adam", model="eleven_monolingual_v1", api_key=elevenKey
-        )
-    play(audio)
-    # ttsEngine.say(audio)
-    # ttsEngine.runAndWait()
-
-def wishMe():
-    hour = int(datetime.datetime.now().hour)
-    if hour>= 0 and hour<12:
+
+def wishMe() -> None:
+    hour: int = int(datetime.datetime.now().hour)
+    if 0 <= hour < 12:
         greeting = "Good Morning, Rod! How may I help you"
-  
-    elif hour>= 12 and hour<18:
+
+    elif 12 <= hour < 18:
         greeting = "Good afternoon, Sir"
-  
+
     else:
         greeting = "Good evening, Sir"
-    speak(greeting)
+    speakAndPrint(greeting)
+
 
 # the assistant will listen to you, go speech to text.
-def listen():
+def listen() -> str:
+    print(sr.Microphone.list_working_microphones())
     with sr.Microphone(device_index=1) as source:
-         
+
         print("Listening...")
         recognizer.pause_threshold = 1
         audio = recognizer.listen(source)
-  
+
     try:
-        print("Recognizing...")    
-        query = recognizer.recognize_google(audio, language ='en-in')
+        print("Recognizing...")
+        query = recognizer.recognize_google(audio, language='en-in')
         print(f"User said: {query}\n")
-  
+
     except Exception as e:
-        print(e)    
-        print("Unable to Recognize your voice.")  
+        print(e)
+        print("Unable to Recognize your voice.")
         return "None"
-     
+
     return query
 
-#submit text to LLM
-def textToModel(query):
+
+# submit text to LLM
+def textToModel(query: str) -> None:
     completion = client.chat.completions.create(
-    model="gpt-3.5-turbo",
-     messages=[
-        {"role": "system", "content": "You are a personal assistant, named ELO. ELO stands for Enhanced Logical Operative."},
-        {"role": "user", "content": query}
+        model="gpt-3.5-turbo",
+        messages=[
+            {"role": "system",
+             "content": "You are a personal assistant, named ELO. ELO stands for Enhanced Logical Operative."},
+            {"role": "user", "content": query}
         ]
     )
-    reply = completion.choices[0].message.content
-    print(reply)
-    speak(reply)
+    reply = summarize(completion.choices[0].message.content)
+    speakAndPrint(reply)
+
+
+def summarize(response: str) -> str:
+    if response.__len__() > 300:
+        print(response)
+        return "My LLM backend response is rather lengthy, I'll put it in your console for reading, sir."
+    else:
+        return response
+
 
-def listenForStop(query):
+def listenForKeywords(query: str) -> bool:
+    skipLLM = False
     if query == "stop":
-        speak("Okay, hope I was able to help!")
+        speakAndPrint("Okay, hope I was able to help!")
         exit()
-    elif query == "Thanks ELO that is all I needed":
-        speak("Okay, hope I was able to help!")
+    elif "thanks elo that is all I needed" in query:
+        speakAndPrint("Okay, hope I was able to help!")
         exit()
+    elif 'search' in query:
+        speakAndPrint("I can run a search query for that if you'd like")
+        if confirm():
+            speakAndPrint("What would you like me to search?")
+            desiredSearch = listen()
+            contents = exa.search_and_contents(desiredSearch)
+            speakAndPrint("I've come up with a couple of results and displayed them in your console")
+            print(contents.results)
+            skipLLM = True
+
+    return skipLLM
+
+
+# confirmation as a function seems intuitive.
+def confirm() -> bool:
+    confirmationQuery = listen()
+    if confirmationQuery == "yes" or "ok" or "sure":
+        speakAndPrint("Confirmed")
+        return True
+    else:
+        speakAndPrint("alright.")
+        return False
+
 
-def main():
+def calibrate() -> int:
+    speakAndPrint("Hi. Elo is booting up. Please wait.")
+    print("Initializing...")
+    microphoneDict = sr.Microphone.list_working_microphones()
+    print("A list of mics! %s" % microphoneDict)
+    print("A list of mic values! %s" % microphoneDict.values())
+    for mic in microphoneDict.values():
+        if mic == 'Shure MV7':
+            return 1
+    else:
+        speakAndPrint("Please connect your iphone's mic sir.")
+        iphone = microphoneDict.get('Skywalker (2) Microphone')
+        time.sleep(5)
+        return iphone
+
+
+def main() -> None:
+    startup = calibrate()
+    if startup == 0:
+        speakAndPrint("Not able to connect to audio source")
+        exit()
     wishMe()
     while True:
-        query = listen()
-        listenForStop(query)
-        textToModel(query)
+        query = listen().lower()
+        if not listenForKeywords(query):
+            textToModel(query)
+        else:
+            break
+
 
 if __name__ == "__main__":
     main()
-

assistant/pyproject.toml

@@ -14,8 +14,10 @@ setuptools = "^69.1.1"
 openai = "^1.12.0"
 elevenlabs = "^0.2.27"
 pygame = "^2.5.2"
+exa-py = "^1.0.8"
+icecream = "^2.1.3"
 
 
 [build-system]
 requires = ["poetry-core"]
-build-backend = "poetry.core.masonry.api"
+build-backend = "poetry.core.masonry.api"