O componente mais crítico aqui é o backend do Large Language Model (LLM), para o qual usaremos o Ollama. é amplamente reconhecido como uma ferramenta popular para executar e servir LLMs offline. Se Ollama é novo para você, recomendo verificar meu artigo anterior sobre RAG offline: Basicamente, você só precisa baixar o aplicativo Ollama, extrair seu modelo preferido e executá-lo.
Diagrama de sequência para assistente de voz com Whisper, Ollama e Bark.
A implementação começa com a criação de um TextToSpeechService
baseado em Bark, incorporando métodos para sintetizar fala a partir de texto e lidar perfeitamente com entradas de texto mais longas, como segue:
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__
) : A classe usa um parâmetro opcional device
, que especifica o dispositivo a ser usado para o modelo ( cuda
se uma GPU estiver disponível, ou cpu
). Ele carrega o modelo Bark e o processador correspondente do modelo pré-treinado suno/bark-small
. Você também pode usar a versão grande especificando suno/bark
para o carregador de modelo.
synthesize
) : Este método recebe uma entrada text
e um parâmetro voice_preset
, que especifica a voz a ser usada para a síntese. Você pode verificar outro valor voice_preset
. Ele usa o processor
para preparar o texto de entrada e a predefinição de voz e, em seguida, gera a matriz de áudio usando o método model.generate()
. A matriz de áudio gerada é convertida em uma matriz NumPy e a taxa de amostragem é retornada junto com a matriz de áudio.
long_form_synthesize
) : Este método é usado para sintetizar entradas de texto mais longas. Primeiro, ele transforma o texto de entrada em frases usando a função nltk.sent_tokenize
. Para cada frase, ele chama o método synthesize
para gerar o array de áudio. Em seguida, ele concatena as matrizes de áudio geradas, com um breve silêncio (0,25 segundos) adicionado entre cada frase.
Agora que temos o TextToSpeechService
configurado, precisamos preparar o servidor Ollama para o serviço do modelo de linguagem grande (LLM). Para fazer isso, você precisará seguir estas etapas:
ollama pull llama2
.
ollama serve
.
base.en
) para transcrever a entrada do usuário.
ConversationalChain
integrado da biblioteca Langchain, que fornece um modelo para gerenciar o fluxo conversacional. Iremos configurá-lo para usar o modelo de linguagem Llama-2 com o 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(), )
Agora, vamos definir as funções necessárias:record_audio
: Esta função é executada em um thread separado para capturar dados de áudio do microfone do usuário usando sounddevice.RawInputStream
. A função de retorno de chamada é chamada sempre que novos dados de áudio estão disponíveis e coloca os dados em data_queue
para processamento posterior.
transcribe
: esta função utiliza a instância Whisper para transcrever os dados de áudio de data_queue
em texto.
get_llm_response
: Esta função alimenta o contexto da conversa atual para o modelo de linguagem Llama-2 (por meio do Langchain ConversationalChain
) e recupera a resposta de texto gerada.
play_audio
: Esta função pega a forma de onda de áudio gerada pelo mecanismo de conversão de texto em fala Bark e a reproduz para o usuário usando uma biblioteca de reprodução de som (por exemplo, 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()
Em seguida, definimos o loop principal da aplicação. O loop principal do aplicativo orienta o usuário através da interação conversacional da seguinte forma:
Depois que o usuário pressiona Enter, a função record_audio
é chamada em um thread separado para capturar a entrada de áudio do usuário.
Quando o usuário pressiona Enter novamente para interromper a gravação, os dados de áudio são transcritos usando a função transcribe
.
O texto transcrito é então passado para a função get_llm_response
, que gera uma resposta usando o modelo de linguagem Llama-2.
A resposta gerada é impressa no console e reproduzida para o usuário usando a função 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.")
Também publicado