Build a Low-Latency ChatGPT Voice Assistant in Python

pip install pvporcupine pvcheetah pvorca pvrecorder pvspeaker openai

import pvporcupine
from pvrecorder import PvRecorder

ACCESS_KEY = "${ACCESS_KEY}"

# Path to your Porcupine wake-word model file (.ppn) that triggers activation
# e.g., "./models/hey-chatgpt.ppn" 
KEYWORD_PATH = "${KEYWORD_PATH}"

porcupine = pvporcupine.create(access_key=ACCESS_KEY, keyword_paths=[KEYWORD_PATH])
recorder = PvRecorder(frame_length=porcupine.frame_length)
recorder.start()

print("Listening for wake word...")
while True:
    pcm = recorder.read()
    keyword_index = porcupine.process(pcm)
    if keyword_index >= 0:
        print("Wake word detected.")
        break

recorder.stop()

import pvcheetah

ACCESS_KEY = "${ACCESS_KEY}"

cheetah = pvcheetah.create(
            access_key=ACCESS_KEY,
            endpoint_duration_sec= 1.0)

recorder = PvRecorder(frame_length=cheetah.frame_length)
recorder.start()

print("Speak your message…")
transcript = ""
while True:
    pcm = recorder.read()
    partial_transcript, is_endpoint = cheetah.process(pcm)
    transcript += partial_transcript
    print(partial_transcript, end="", flush=True)
    if is_endpoint:
        final_transcript = cheetah.flush()
        transcript += final_transcript
        print(final_transcript)
        break

recorder.stop()
cheetah.delete()

from openai import OpenAI

OPENAI_API_KEY = "${OPENAI_API_KEY}"

def ask_chatgpt(api_key: str, prompt: str, model: str = "gpt-4o") -> str:
    try:
        client = OpenAI(api_key=api_key)
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            timeout=45
        )
        return response.choices[0].message.content
    except Exception as e:
        return f"I'm having trouble connecting to ChatGPT. Error: {e}"
        
 # transcript = "example user request"       
reply = ask_chatgpt(OPENAI_API_KEY, transcript)

import pvorca
from pvspeaker import PvSpeaker
from collections import deque

ACCESS_KEY = "${ACCESS_KEY}"
orca = pvorca.create(access_key=ACCESS_KEY)
speaker = PvSpeaker(sample_rate=orca.sample_rate, bits_per_sample=16)

# Synthesize speech
# reply = response from chatgpt
pcm_out, _ = orca.synthesize(reply)

# Play audio
speaker.start()

pcm_buffer = deque()
pcm_buffer.append(pcm_out)

while len(pcm_buffer) > 0:
    pcm = pcm_buffer.popleft()
    written = speaker.write(pcm)
    if written < len(pcm):
        pcm_buffer.appendleft(pcm[written:])

speaker.flush()
speaker.stop()

# Cleanup
speaker.delete()
orca.delete()

import argparse
from collections import deque
import sys
from openai import OpenAI
import pvporcupine
import pvcheetah
import pvorca
from pvrecorder import PvRecorder
from pvspeaker import PvSpeaker


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Porcupine + Cheetah + Orca voice interface for ChatGPT"
    )
    parser.add_argument("--access_key", required=True, help="Picovoice AccessKey")
    parser.add_argument("--keyword_paths", nargs='+', required=True, 
                       help="Path(s) to .ppn wake-word model(s)")
    parser.add_argument("--openai_key", required=True, help="OpenAI API key")
    args = parser.parse_args()

    porcupine = None
    cheetah = None
    orca = None
    recorder = None
    speaker = None

    try:
        # Initialize engines
        porcupine = pvporcupine.create(
            access_key=args.access_key, 
            keyword_paths=args.keyword_paths)
        cheetah = pvcheetah.create(access_key=args.access_key, endpoint_duration_sec=1.0)
        orca = pvorca.create(access_key=args.access_key)

        print(f'Porcupine version: {porcupine.version}')
        print(f'Cheetah version: {cheetah.version}')
        print(f'Orca version: {orca.version}\n')

        # Initialize speaker
        speaker = PvSpeaker(
            sample_rate=orca.sample_rate, 
            bits_per_sample=16)

        # Initialize recorder
        recorder = PvRecorder(frame_length=porcupine.frame_length)
        recorder.start()

        print("Ready. Say the wake word… (Ctrl+C to stop)")

        # Wait for wake word
        while True:
            pcm = recorder.read()
            keyword_index = porcupine.process(pcm)
            if keyword_index >= 0:
                print("[EVENT] Wake word detected")
                break

        recorder.stop()
        recorder.delete()
        recorder = PvRecorder(frame_length=cheetah.frame_length)
        recorder.start()

        # Stream STT with Cheetah
        print("Speak your message…")
        transcript = ""
        while True:
            pcm = recorder.read()
            partial_transcript, is_endpoint = cheetah.process(pcm)
            transcript += partial_transcript
            print(partial_transcript, end="", flush=True)
            if is_endpoint:
                final_transcript = cheetah.flush()
                transcript += final_transcript
                print(final_transcript)
                break

        print("[TRANSCRIPT]", transcript)

        # Call ChatGPT
        client = OpenAI(api_key=args.openai_key)
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": transcript}],
            timeout=45
        )

        reply = response.choices[0].message.content
        print("[REPLY]", reply)

        # Synthesize speech with Orca
        pcm_out, _ = orca.synthesize(reply)

        # Play audio
        speaker.start()

        pcm_buffer = deque()
        pcm_buffer.append(pcm_out)

        while len(pcm_buffer) > 0:
            pcm = pcm_buffer.popleft()
            written = speaker.write(pcm)
            if written < len(pcm):
                pcm_buffer.appendleft(pcm[written:])

        speaker.flush()
        speaker.stop()

    except KeyboardInterrupt:
        print("\n[EXIT] Stopping…")
    except pvporcupine.PorcupineActivationLimitError:
        print("AccessKey has reached its processing limit")
    except pvcheetah.CheetahActivationLimitError:
        print("AccessKey has reached its processing limit")
    except pvorca.OrcaActivationLimitError:
        print("AccessKey has reached its processing limit")
    finally:
        # Cleanup
        if speaker is not None:
            speaker.delete()
        
        if recorder is not None:
            recorder.delete()
        
        if orca is not None:
            orca.delete()
        
        if cheetah is not None:
            cheetah.delete()
        
        if porcupine is not None:
            porcupine.delete()

    return 0


if __name__ == "__main__":
    sys.exit(main())

python3 voice_chatgpt.py \
  --access_key "$ACCESS_KEY" \
  --keyword_paths ./models/hey-chatgpt.ppn \
  --openai_key "$OPENAI_API_KEY" \