GrossAI.py

import requests
import os
import time
import pprint

import instructor
from pydantic import BaseModel, Field, create_model
from openai import OpenAI

import pprint

LLM = {
    "address": "http://localhost:6000/v1",
    "key": "Not needed"
}


def waitForReady():
    # This checks that my local llm is queryable.
    # If the llm isn't running, it starts it and waits for it to ready up.
    print("Starting up")
    try:
        requests.get(LLM["address"])
    except requests.ConnectionError:
        # https://github.com/oobabooga/text-generation-webui
        print(os.system(
            r"start C:\Users\theep\Downloads\text-generation-webui-main\start_windows.bat"))
        while True:
            try:
                requests.get(LLM["address"])
                return
            except:
                print(".",)
                time.sleep(1)
    finally:
        print("Ready!")


# Some super jank global garbage to make registering commands easier.
model = {}  # pydantic struct, maps name to (ArgsClass, Field definition)
handlers = {}  # maps name to callable method that takes an instance of ArgsClass


def Command(cls):
    # Given a class, 'register it' into the global model and handlers stuff.
    name = cls.__name__
    args = getattr(cls, 'Args')
    desc = getattr(cls, 'description')
    invoke = getattr(cls, 'invoke')
    assert name not in model
    model[name] = (args, Field(..., description=desc,
                               default_factory=lambda: None,
                               required=False,))
    handlers[name] = invoke
    return cls


# Some commands!
@Command
class RelativeAlarm:
    description = "Set an alarm given a message an a relative amount of time in seconds"

    class Args(BaseModel):
        alarm_message: str = Field(
            description="Just the reminder that the user wants.")
        delay_seconds: int = Field(
            description="How many seconds to wait before firing the alarm.")

    @staticmethod
    def invoke(args: Args):
        print("Called RelativeAlarm ", args)


@Command
class AbsoluteAlarm:
    description = "Set an alarm given a message and an absolute time like 3pm."

    class Args(BaseModel):
        alarm_message: str = Field(
            description="Just the reminder that the user wants.")
        alert_hour: int = Field(description="Hour of the day (24 hour time).")
        alert_minute: int = Field(description="Minute of the hour (0-59).")

    @staticmethod
    def invoke(args: Args):
        print("Called AbsoluteAlarm ", args)


@Command
class Lights:
    description = "Turns lights on or off given the name of the light and a boolean (true = on, false = off)."

    class Args(BaseModel):
        name: str = Field(description="Which lights")
        on: bool = Field(description="On (true) or off (false)")

    @staticmethod
    def invoke(args: Args):
        print("Called Lights ", args)


def handle_user_input(message, response_model):
    response = client.chat.completions.create(
        model="gpt-4-turbo-preview",
        messages=[
            {"role": "user", "content": f"A User is going to give you a command. You must convert that command to valid json format. Do not output anything else. The users command is:\n{message}"}
        ],
        response_model=response_model)

    # Invoke the handler for any properties in the response object that aren't None
    for name, handler in handlers.items():
        val = getattr(response, name)
        if val is not None:
            handler(val)

    return response


if __name__ == '__main__':
    # Generate the model and setup the llm.
    GeneratedClass = create_model("GeneratedClass", **model)
    client = instructor.patch(
        OpenAI(
            base_url=LLM["address"],
            api_key=LLM["key"]
        ),
        mode=instructor.Mode.JSON
    )

    waitForReady()

    # REPL
    while True:
        try:
            handle_user_input(input(), GeneratedClass)
        except BaseException as e:
            pprint.pprint(e)
            if isinstance(e, KeyboardInterrupt):
                exit(0)

    # Voice
    # from RealtimeSTT import AudioToTextRecorder
    # with AudioToTextRecorder(model="medium.en", spinner =False) as recorder:
    #     while True:
    #         recorder.text(onCommand)