ShawnHymel · September 17, 2024 18:28
diff --git a/ollama-light-assistant.py b/ollama-light-assistant.py
 """
 Ollama Light Assistant

 Test the ability for LLMs to call tools (i.e. functions) from within the Ollama
 environment on a Raspberry Pi with speech-to-text. Tested with a Raspberry Pi 5
 (8 GB) and this USB microphone: https://www.adafruit.com/product/3367.

 Connect an LED to GPIO 17 (physical pin 11 on the board) in an active low
 configuration:

    3.3V -> 220 Ohm resistor -> LED -> GPIO 17

 Install dependencies:

    $ curl -fsSL https://ollama.com/install.sh | sh
    $ ollama serve
    $ sudo apt install -y libportaudio2

 Create a Python virtual environment and install necessary packages. We need
 system site packages enabled so that gpiozero actually works.

    $ python -m venv venv-ollama --system-site-packages
    $ source venv-ollama/bin/activate
    $ python -m pip install ollama vosk sounddevice

 Download the LLM and Vosk STT model:

    $ ollama pull allenporter/xlam:1b
    $ python -c "from vosk import Model; Model(lang='en-us')"

 Get the USB device number of your attached microphone:

    $ python -c "import sounddevice; print(sounddevice.query_devices())"

 For example, my device number is '0' for the "USB PnP Sound Device." Change the
 AUDIO_INPUT_INDEX value to this number in the code below.

 Run the script with:

    $ python ollama-light-assistant.py

 Give a voice command like "turn on the light" to have the LLM call the 
 led_write() function.

 For the xlam:1b model, you must be SUPER specific with your commands. For
 example, it will accept "turn on the light" but not "turn the light on." You 
 can try the llama3.1:8b model for more flexibility, but it's VERY slow on the 
 Pi 5, taking up to 3 minutes to respond.

 Note: LLMs are SUPER slow on the Raspberry Pi right now, and tool calling is
 janky. Once we get better/smaller LLMs, more RAM, and/or AI accelerators, this
 should (in theory) be much faster and more robust.

 References:
    Ollama tools: https://ollama.com/blog/tool-support
    Tool-supported models: https://ollama.com/search?c=tools

 Author: Shawn Hymel
 Date: September 17, 2024
 License: BSD-0 (https://opensource.org/license/0bsd)
 """

 from collections import deque
 import json
 import queue

 from gpiozero import LED
 import ollama
 import sounddevice as sd
 from vosk import Model, KaldiRecognizer

 # Settings
 LED_PIN = 17
 AUDIO_INPUT_INDEX = 0
 MODEL = "allenporter/xlam:1b" # You could also try "llama3.1:8b"
 OLLAMA_HOST = "http://localhost:11434"
 MAX_MESSAGES = 5
 PREAMBLE = "You are a helpful assistant that can turn the light on and off."

 # Define tools for the model to use (i.e. functions)
 TOOLS = [
    {
        'type': 'function',
        'function': {
            'name': "led_write",
            'description': "Turn the light off or on",
            'parameters': {
                'type': 'object',
                'properties': {
                    'value': {
                        'type': 'number',
                        'description': "The value to write to the light pin " \
                            "to turn it off and on. 0 for off, 1 for on.",
                    },
                },
                'required': ['value'],
            },
        }
    }
 ]

 # -----------------------------------------------------------------------------
 # Classes

 class FixedSizeQueue:
    """
    Fixed size array with FIFO and optional preamble.
    """
    def __init__(self, max_size, preamble=None):
        self.queue = deque(maxlen=max_size)
        self.preamble = {
            'role': 'system',
            'content': preamble
        }

    def push(self, item):
        self.queue.append(item)

    def get(self):
        if self.preamble['content'] is None:
            return list(self.queue)
        else:
            return [self.preamble] + list(self.queue)

 # -----------------------------------------------------------------------------
 # Functions

 def led_write(led, value):
    """
    Turn the LED on or off. Remember that the LED is active low!
    """
    if int(value) > 0:
        led.off()
        print("The LED is now on")
    else:
        led.on()
        print("The LED is now off")

 def send(chat, msg_history, client, model, tools, led):
    """
    Send a message to the LLM server and print the response.
    """

    # Add user message to the conversation history
    msg_history.push({
        'role': 'user',
        'content': chat
    })

    # Send message to LLM server
    response = client.chat(
        model=model,
        messages=msg_history.get(),
        tools=tools,
        stream=False
    )

    # Print the full response
    print(f"Response: {response['message']}")

    # Add the model's response to the conversation history
    msg_history.push({
        'role': 'assistant',
        'content': response['message']['content']
    })

    # Check if the model used any of the provided tools
    if response['message'].get('tool_calls') is None:
        print("Tools not used.")
        return

    # Call the function(s) the model used
    else:
        print("Tools used. Calling:")
        for tool in response['message']['tool_calls']:
            print(tool)
            if tool['function']['name'] == "led_write":
                led_write(led, tool['function']['arguments']['value'])

 def record_callback(indata, frames, time, status, q):
    """
    Callback for recording audio from the microphone.
    """
    if status:
        print(status, file=sys.stderr)
    q.put(bytes(indata))

 # -----------------------------------------------------------------------------
 # Main

 if __name__ == "__main__":

    # Get the sample rate of the input device
    device_info = sd.query_devices(sd.default.device[0], 'input')
    sample_rate = int(device_info['default_samplerate'])

    # Build the STT model
    stt_model = Model(lang='en-us')
    stt_recognizer = KaldiRecognizer(stt_model, sample_rate)
    stt_recognizer.SetWords(False)

    # Configure chat history and connect to the LLM server
    msg_history = FixedSizeQueue(MAX_MESSAGES, PREAMBLE)
    chat_client = ollama.Client(host=OLLAMA_HOST)

    # Initialize the audio recording queue
    q = queue.Queue()

    # Initialize the LED. Remember that the LED is active low! So, led.on()
    # is actually turning the LED off.
    led = LED(LED_PIN)
    led.on()

    while True:

        # Listen for user input
        print("Listening...")
        result_text = ""
        with sd.RawInputStream(
            dtype='int16',
            channels=1,
            callback=lambda in_data, frames, time, status: record_callback(
                in_data,
                frames,
                time,
                status,
                q
            )
        ):

            # Collect audio data until we have a full phrase
            while True:
                data = q.get()
                if stt_recognizer.AcceptWaveform(data):

                    # Perform speech-to-text (STT) on the audio data
                    result = json.loads(stt_recognizer.Result())
                    result_text = result.get("text", "")
                    break

        # Send the user's message to the LLM server
        if not result_text:
            print("No speech detected")
        else:
            print(f"Speech detected: {result_text}")
            send(result_text, msg_history, chat_client, MODEL, TOOLS, led)
	"""
	Ollama Light Assistant

	Test the ability for LLMs to call tools (i.e. functions) from within the Ollama
	environment on a Raspberry Pi with speech-to-text. Tested with a Raspberry Pi 5
	(8 GB) and this USB microphone: https://www.adafruit.com/product/3367.

	Connect an LED to GPIO 17 (physical pin 11 on the board) in an active low
	configuration:

	3.3V -> 220 Ohm resistor -> LED -> GPIO 17

	Install dependencies:

	$ curl -fsSL https://ollama.com/install.sh \| sh
	$ ollama serve
	$ sudo apt install -y libportaudio2

	Create a Python virtual environment and install necessary packages. We need
	system site packages enabled so that gpiozero actually works.

	$ python -m venv venv-ollama --system-site-packages
	$ source venv-ollama/bin/activate
	$ python -m pip install ollama vosk sounddevice

	Download the LLM and Vosk STT model:

	$ ollama pull allenporter/xlam:1b
	$ python -c "from vosk import Model; Model(lang='en-us')"

	Get the USB device number of your attached microphone:

	$ python -c "import sounddevice; print(sounddevice.query_devices())"

	For example, my device number is '0' for the "USB PnP Sound Device." Change the
	AUDIO_INPUT_INDEX value to this number in the code below.

	Run the script with:

	$ python ollama-light-assistant.py

	Give a voice command like "turn on the light" to have the LLM call the
	led_write() function.

	For the xlam:1b model, you must be SUPER specific with your commands. For
	example, it will accept "turn on the light" but not "turn the light on." You
	can try the llama3.1:8b model for more flexibility, but it's VERY slow on the
	Pi 5, taking up to 3 minutes to respond.

	Note: LLMs are SUPER slow on the Raspberry Pi right now, and tool calling is
	janky. Once we get better/smaller LLMs, more RAM, and/or AI accelerators, this
	should (in theory) be much faster and more robust.

	References:
	Ollama tools: https://ollama.com/blog/tool-support
	Tool-supported models: https://ollama.com/search?c=tools

	Author: Shawn Hymel
	Date: September 17, 2024
	License: BSD-0 (https://opensource.org/license/0bsd)
	"""

	from collections import deque
	import json
	import queue

	from gpiozero import LED
	import ollama
	import sounddevice as sd
	from vosk import Model, KaldiRecognizer

	# Settings
	LED_PIN = 17
	AUDIO_INPUT_INDEX = 0
	MODEL = "allenporter/xlam:1b" # You could also try "llama3.1:8b"
	OLLAMA_HOST = "http://localhost:11434"
	MAX_MESSAGES = 5
	PREAMBLE = "You are a helpful assistant that can turn the light on and off."

	# Define tools for the model to use (i.e. functions)
	TOOLS = [
	{
	'type': 'function',
	'function': {
	'name': "led_write",
	'description': "Turn the light off or on",
	'parameters': {
	'type': 'object',
	'properties': {
	'value': {
	'type': 'number',
	'description': "The value to write to the light pin " \
	"to turn it off and on. 0 for off, 1 for on.",
	},
	},
	'required': ['value'],
	},
	}
	}
	]

	# -----------------------------------------------------------------------------
	# Classes

	class FixedSizeQueue:
	"""
	Fixed size array with FIFO and optional preamble.
	"""
	def __init__(self, max_size, preamble=None):
	self.queue = deque(maxlen=max_size)
	self.preamble = {
	'role': 'system',
	'content': preamble
	}

	def push(self, item):
	self.queue.append(item)

	def get(self):
	if self.preamble['content'] is None:
	return list(self.queue)
	else:
	return [self.preamble] + list(self.queue)

	# -----------------------------------------------------------------------------
	# Functions

	def led_write(led, value):
	"""
	Turn the LED on or off. Remember that the LED is active low!
	"""
	if int(value) > 0:
	led.off()
	print("The LED is now on")
	else:
	led.on()
	print("The LED is now off")

	def send(chat, msg_history, client, model, tools, led):
	"""
	Send a message to the LLM server and print the response.
	"""

	# Add user message to the conversation history
	msg_history.push({
	'role': 'user',
	'content': chat
	})

	# Send message to LLM server
	response = client.chat(
	model=model,
	messages=msg_history.get(),
	tools=tools,
	stream=False
	)

	# Print the full response
	print(f"Response: {response['message']}")

	# Add the model's response to the conversation history
	msg_history.push({
	'role': 'assistant',
	'content': response['message']['content']
	})

	# Check if the model used any of the provided tools
	if response['message'].get('tool_calls') is None:
	print("Tools not used.")
	return

	# Call the function(s) the model used
	else:
	print("Tools used. Calling:")
	for tool in response['message']['tool_calls']:
	print(tool)
	if tool['function']['name'] == "led_write":
	led_write(led, tool['function']['arguments']['value'])

	def record_callback(indata, frames, time, status, q):
	"""
	Callback for recording audio from the microphone.
	"""
	if status:
	print(status, file=sys.stderr)
	q.put(bytes(indata))

	# -----------------------------------------------------------------------------
	# Main

	if __name__ == "__main__":

	# Get the sample rate of the input device
	device_info = sd.query_devices(sd.default.device[0], 'input')
	sample_rate = int(device_info['default_samplerate'])

	# Build the STT model
	stt_model = Model(lang='en-us')
	stt_recognizer = KaldiRecognizer(stt_model, sample_rate)
	stt_recognizer.SetWords(False)

	# Configure chat history and connect to the LLM server
	msg_history = FixedSizeQueue(MAX_MESSAGES, PREAMBLE)
	chat_client = ollama.Client(host=OLLAMA_HOST)

	# Initialize the audio recording queue
	q = queue.Queue()

	# Initialize the LED. Remember that the LED is active low! So, led.on()
	# is actually turning the LED off.
	led = LED(LED_PIN)
	led.on()

	while True:

	# Listen for user input
	print("Listening...")
	result_text = ""
	with sd.RawInputStream(
	dtype='int16',
	channels=1,
	callback=lambda in_data, frames, time, status: record_callback(
	in_data,
	frames,
	time,
	status,
	q
	)
	):

	# Collect audio data until we have a full phrase
	while True:
	data = q.get()
	if stt_recognizer.AcceptWaveform(data):

	# Perform speech-to-text (STT) on the audio data
	result = json.loads(stt_recognizer.Result())
	result_text = result.get("text", "")
	break

	# Send the user's message to the LLM server
	if not result_text:
	print("No speech detected")
	else:
	print(f"Speech detected: {result_text}")
	send(result_text, msg_history, chat_client, MODEL, TOOLS, led)