Jan 01, 1970
여기서 가장 중요한 구성 요소는 Ollama를 사용할 LLM(Large Language Model) 백엔드입니다. 오프라인에서 LLM을 실행하고 제공하는 데 널리 사용되는 도구로 널리 알려져 있습니다. Ollama가 처음이라면 오프라인 RAG에 대한 이전 기사인 기본적으로 Ollama 애플리케이션을 다운로드하고 선호하는 모델을 가져와서 실행하기만 하면 됩니다.
Whisper, Ollama 및 Bark를 사용한 음성 어시스턴트의 시퀀스 다이어그램.
구현은 다음과 같이 텍스트에서 음성을 합성하고 더 긴 텍스트 입력을 원활하게 처리하는 방법을 통합하여 Bark를 기반으로 TextToSpeechService
만드는 것으로 시작됩니다.
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
매개변수를 사용합니다(GPU를 사용할 수 있는 경우 cuda
또는 cpu
). suno/bark-small
사전 훈련된 모델에서 Bark 모델과 해당 프로세서를 로드합니다. 모델 로더에 suno/bark
지정하여 대형 버전을 사용할 수도 있습니다.
synthesize
) : 이 메서드는 text
입력과 합성에 사용할 음성을 지정하는 voice_preset
매개변수를 사용합니다. 다른 voice_preset
값을 확인할 수 있습니다. processor
사용하여 입력 텍스트와 음성 사전 설정을 준비한 다음 model.generate()
메서드를 사용하여 오디오 배열을 생성합니다. 생성된 오디오 배열은 NumPy 배열로 변환되고 샘플 속도는 오디오 배열과 함께 반환됩니다.
long_form_synthesize
) : 이 방법은 긴 텍스트 입력을 합성하는 데 사용됩니다. 먼저 nltk.sent_tokenize
함수를 사용하여 입력 텍스트를 문장으로 토큰화합니다. 각 문장에 대해 synthesize
메서드를 호출하여 오디오 배열을 생성합니다. 그런 다음 각 문장 사이에 짧은 묵음(0.25초)을 추가하여 생성된 오디오 배열을 연결합니다.
이제 TextToSpeechService
가 설정되었으므로 LLM(대형 언어 모델) 제공을 위해 Ollama 서버를 준비해야 합니다. 이렇게 하려면 다음 단계를 따라야 합니다.
ollama pull llama2
.
ollama serve
명령을 실행하여 시작합니다.
base.en
)을 사용합니다.
ConversationalChain
사용합니다. Ollama 백엔드와 함께 Llama-2 언어 모델을 사용하도록 구성하겠습니다. 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
기능을 사용하여 오디오 데이터가 녹음됩니다.
그런 다음 복사된 텍스트는 Llama-2 언어 모델을 사용하여 응답을 생성하는 get_llm_response
함수로 전달됩니다.
생성된 응답은 콘솔에 인쇄되고 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.")
게시됨