Le composant le plus critique ici est le backend du Large Language Model (LLM), pour lequel nous utiliserons Ollama. est largement reconnu comme un outil populaire pour exécuter et servir des LLM hors ligne. Si Ollama est nouveau pour vous, je vous recommande de consulter mon article précédent sur RAG hors ligne : Fondamentalement, il vous suffit de télécharger l'application Ollama, d'extraire votre modèle préféré et de l'exécuter.
Diagramme de séquence pour assistant vocal avec Whisper, Ollama et Bark.
L'implémentation commence par la création d'un TextToSpeechService
basé sur Bark, incorporant des méthodes de synthèse de la parole à partir du texte et gérant de manière transparente les entrées de texte plus longues comme suit :
import nltk import torch import warnings import numpy as np from transformers import AutoProcessor, BarkModel warnings.filterwarnings( "ignore", message="torch.nn.utils.weight_norm is deprecated in favor of torch.nn.utils.parametrizations.weight_norm.", ) class TextToSpeechService: def __init__(self, device: str = "cuda" if torch.cuda.is_available() else "cpu"): """ Initializes the TextToSpeechService class. Args: device (str, optional): The device to be used for the model, either "cuda" if a GPU is available or "cpu". Defaults to "cuda" if available, otherwise "cpu". """ self.device = device self.processor = AutoProcessor.from_pretrained("suno/bark-small") self.model = BarkModel.from_pretrained("suno/bark-small") self.model.to(self.device) def synthesize(self, text: str, voice_preset: str = "v2/en_speaker_1"): """ Synthesizes audio from the given text using the specified voice preset. Args: text (str): The input text to be synthesized. voice_preset (str, optional): The voice preset to be used for the synthesis. Defaults to "v2/en_speaker_1". Returns: tuple: A tuple containing the sample rate and the generated audio array. """ inputs = self.processor(text, voice_preset=voice_preset, return_tensors="pt") inputs = {k: v.to(self.device) for k, v in inputs.items()} with torch.no_grad(): audio_array = self.model.generate(**inputs, pad_token_id=10000) audio_array = audio_array.cpu().numpy().squeeze() sample_rate = self.model.generation_config.sample_rate return sample_rate, audio_array def long_form_synthesize(self, text: str, voice_preset: str = "v2/en_speaker_1"): """ Synthesizes audio from the given long-form text using the specified voice preset. Args: text (str): The input text to be synthesized. voice_preset (str, optional): The voice preset to be used for the synthesis. Defaults to "v2/en_speaker_1". Returns: tuple: A tuple containing the sample rate and the generated audio array. """ pieces = [] sentences = nltk.sent_tokenize(text) silence = np.zeros(int(0.25 * self.model.generation_config.sample_rate)) for sent in sentences: sample_rate, audio_array = self.synthesize(sent, voice_preset) pieces += [audio_array, silence.copy()] return self.model.generation_config.sample_rate, np.concatenate(pieces)
__init__
) : La classe prend un paramètre device
facultatif, qui spécifie le périphérique à utiliser pour le modèle (soit cuda
si un GPU est disponible, soit cpu
). Il charge le modèle Bark et le processeur correspondant à partir du modèle pré-entraîné suno/bark-small
. Vous pouvez également utiliser la grande version en spécifiant suno/bark
pour le chargeur de modèles.
synthesize
) : Cette méthode prend une saisie text
et un paramètre voice_preset
, qui spécifie la voix à utiliser pour la synthèse. Vous pouvez consulter d'autres valeurs voice_preset
. Il utilise le processor
pour préparer le texte d'entrée et le préréglage vocal, puis génère le tableau audio à l'aide de la méthode model.generate()
. Le tableau audio généré est converti en tableau NumPy et la fréquence d'échantillonnage est renvoyée avec le tableau audio.
long_form_synthesize
) : Cette méthode est utilisée pour synthétiser des entrées de texte plus longues. Il convertit d'abord le texte saisi en phrases à l'aide de la fonction nltk.sent_tokenize
. Pour chaque phrase, il appelle la méthode synthesize
pour générer le tableau audio. Il concatène ensuite les tableaux audio générés, avec un court silence (0,25 seconde) ajouté entre chaque phrase.
Maintenant que TextToSpeechService
est configuré, nous devons préparer le serveur Ollama pour le service LLM (Large Language Model). Pour ce faire, vous devrez suivre ces étapes :
ollama pull llama2
.
ollama serve
.
base.en
) pour transcrire les entrées de l'utilisateur.
ConversationalChain
intégrée de la bibliothèque Langchain qui fournit un modèle pour gérer le flux conversationnel. Nous allons le configurer pour utiliser le modèle de langage Llama-2 avec le backend Ollama. import time import threading import numpy as np import whisper import sounddevice as sd from queue import Queue from rich.console import Console from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationChain from langchain.prompts import PromptTemplate from langchain_community.llms import Ollama from tts import TextToSpeechService console = Console() stt = whisper.load_model("base.en") tts = TextToSpeechService() template = """ You are a helpful and friendly AI assistant. You are polite, respectful, and aim to provide concise responses of less than 20 words. The conversation transcript is as follows: {history} And here is the user's follow-up: {input} Your response: """ PROMPT = PromptTemplate(input_variables=["history", "input"], template=template) chain = ConversationChain( prompt=PROMPT, verbose=False, memory=ConversationBufferMemory(ai_prefix="Assistant:"), llm=Ollama(), )
Maintenant, définissons les fonctions nécessaires :record_audio
: Cette fonction s'exécute dans un thread séparé pour capturer les données audio du microphone de l'utilisateur à l'aide de sounddevice.RawInputStream
. La fonction de rappel est appelée chaque fois que de nouvelles données audio sont disponibles et place les données dans une data_queue
pour un traitement ultérieur.
transcribe
: Cette fonction utilise l'instance Whisper pour transcrire les données audio de la data_queue
en texte.
get_llm_response
: Cette fonction alimente le contexte de conversation actuel au modèle de langage Llama-2 (via Langchain ConversationalChain
) et récupère la réponse textuelle générée.
play_audio
: Cette fonction prend la forme d'onde audio générée par le moteur de synthèse vocale Bark et la restitue à l'utilisateur à l'aide d'une bibliothèque de lecture sonore (par exemple, sounddevice
). def record_audio(stop_event, data_queue): """ Captures audio data from the user's microphone and adds it to a queue for further processing. Args: stop_event (threading.Event): An event that, when set, signals the function to stop recording. data_queue (queue.Queue): A queue to which the recorded audio data will be added. Returns: None """ def callback(indata, frames, time, status): if status: console.print(status) data_queue.put(bytes(indata)) with sd.RawInputStream( samplerate=16000, dtype="int16", channels=1, callback=callback ): while not stop_event.is_set(): time.sleep(0.1) def transcribe(audio_np: np.ndarray) -> str: """ Transcribes the given audio data using the Whisper speech recognition model. Args: audio_np (numpy.ndarray): The audio data to be transcribed. Returns: str: The transcribed text. """ result = stt.transcribe(audio_np, fp16=False) # Set fp16=True if using a GPU text = result["text"].strip() return text def get_llm_response(text: str) -> str: """ Generates a response to the given text using the Llama-2 language model. Args: text (str): The input text to be processed. Returns: str: The generated response. """ response = chain.predict(input=text) if response.startswith("Assistant:"): response = response[len("Assistant:") :].strip() return response def play_audio(sample_rate, audio_array): """ Plays the given audio data using the sounddevice library. Args: sample_rate (int): The sample rate of the audio data. audio_array (numpy.ndarray): The audio data to be played. Returns: None """ sd.play(audio_array, sample_rate) sd.wait()
Ensuite, nous définissons la boucle applicative principale. La boucle d'application principale guide l'utilisateur tout au long de l'interaction conversationnelle comme suit :
Une fois que l'utilisateur appuie sur Entrée, la fonction record_audio
est appelée dans un thread séparé pour capturer l'entrée audio de l'utilisateur.
Lorsque l'utilisateur appuie à nouveau sur Entrée pour arrêter l'enregistrement, les données audio sont transcrites à l'aide de la fonction transcribe
.
Le texte transcrit est ensuite transmis à la fonction get_llm_response
, qui génère une réponse en utilisant le modèle de langage Llama-2.
La réponse générée est imprimée sur la console et restituée à l'utilisateur à l'aide de la fonction play_audio
.
if __name__ == "__main__": console.print("[cyan]Assistant started! Press Ctrl+C to exit.") try: while True: console.input( "Press Enter to start recording, then press Enter again to stop." ) data_queue = Queue() # type: ignore[var-annotated] stop_event = threading.Event() recording_thread = threading.Thread( target=record_audio, args=(stop_event, data_queue), ) recording_thread.start() input() stop_event.set() recording_thread.join() audio_data = b"".join(list(data_queue.queue)) audio_np = ( np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0 ) if audio_np.size > 0: with console.status("Transcribing...", spinner="earth"): text = transcribe(audio_np) console.print(f"[yellow]You: {text}") with console.status("Generating response...", spinner="earth"): response = get_llm_response(text) sample_rate, audio_array = tts.long_form_synthesize(response) console.print(f"[cyan]Assistant: {response}") play_audio(sample_rate, audio_array) else: console.print( "[red]No audio recorded. Please ensure your microphone is working." ) except KeyboardInterrupt: console.print("\n[red]Exiting...") console.print("[blue]Session ended.")
Également publié