Наиболее важным компонентом здесь является серверная часть модели большого языка (LLM), для которой мы будем использовать Ollama. широко известен как популярный инструмент для запуска и обслуживания программ LLM в автономном режиме. Если Ollama для вас новичок, я рекомендую прочитать мою предыдущую статью об автономном RAG: По сути, вам просто нужно загрузить приложение Ollama, выбрать предпочитаемую модель и запустить ее.
Диаграмма последовательности действий для голосового помощника с Whisper, Ollama и Bark.
Реализация начинается с создания TextToSpeechService
на основе Bark, включающего методы синтеза речи из текста и беспрепятственной обработки более длинных текстовых вводов следующим образом:
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__
) : класс принимает необязательный параметр device
, который указывает устройство, которое будет использоваться для модели (либо cuda
, если доступен графический процессор, либо cpu
). Он загружает модель Bark и соответствующий процессор из предварительно обученной модели suno/bark-small
. Вы также можете использовать большую версию, указав suno/bark
для загрузчика модели.
synthesize
) : этот метод принимает text
ввод и параметр voice_preset
, который определяет голос, который будет использоваться для синтеза. Вы можете проверить другое значение voice_preset
. Он использует processor
для подготовки входного текста и голосовой настройки, а затем генерирует аудиомассив с помощью метода model.generate()
. Сгенерированный аудиомассив преобразуется в массив NumPy, и вместе с аудиомассивом возвращается частота дискретизации.
long_form_synthesize
) : этот метод используется для синтеза более длинных текстовых вводов. Сначала он разбивает входной текст на предложения с помощью функции nltk.sent_tokenize
. Для каждого предложения он вызывает метод synthesize
для создания аудиомассива. Затем он объединяет сгенерированные аудиомассивы, добавляя между каждым предложением короткую паузу (0,25 секунды).
Теперь, когда у нас настроен TextToSpeechService
, нам нужно подготовить сервер Ollama для обслуживания модели большого языка (LLM). Для этого вам необходимо выполнить следующие шаги:
ollama pull llama2
.
ollama serve
.
base.en
) для расшифровки пользовательского ввода.
ConversationalChain
из библиотеки Langchain, которая предоставляет шаблон для управления диалоговым потоком. Мы настроим его для использования языковой модели Llama-2 с серверной частью 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(), )
Теперь определим необходимые функции:record_audio
: эта функция выполняется в отдельном потоке для захвата аудиоданных с микрофона пользователя с помощью sounddevice.RawInputStream
. Функция обратного вызова вызывается всякий раз, когда доступны новые аудиоданные, и помещает данные в data_queue
для дальнейшей обработки.
transcribe
: эта функция использует экземпляр Whisper для расшифровки аудиоданных из data_queue
в текст.
get_llm_response
: эта функция передает текущий контекст разговора в языковую модель Llama-2 (через Langchain ConversationalChain
) и извлекает сгенерированный текстовый ответ.
play_audio
: Эта функция принимает форму звукового сигнала, сгенерированную механизмом преобразования текста в речь Bark, и воспроизводит ее пользователю с помощью библиотеки воспроизведения звука (например, 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()
Затем мы определяем основной цикл приложения. Основной цикл приложения проводит пользователя через диалоговое взаимодействие следующим образом:
Как только пользователь нажимает Enter, функция record_audio
вызывается в отдельном потоке для захвата аудиовхода пользователя.
Когда пользователь снова нажимает Enter, чтобы остановить запись, аудиоданные расшифровываются с помощью функции transcribe
.
Расшифрованный текст затем передается в функцию get_llm_response
, которая генерирует ответ с использованием языковой модели Llama-2.
Сгенерированный ответ выводится на консоль и воспроизводится пользователю с помощью функции 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.")
Также опубликовано