양방향 음성 채팅이 가능한 AI를 호스팅하는 건 생각보다 쉽습니다!

LLM과 음성 기능을 통합함으로써 개인화된 고객 상호작용에서 새로운 기회가 창출되었습니다.

이 가이드에서는 Python, Transformers, Qwen2-Audio-7B-Instruct 및 Bark를 사용하여 양방향 음성 상호작용을 지원하는 로컬 LLM 서버를 설정하는 방법을 안내합니다.

필수 조건

시작하기 전에 다음이 설치되어 있어야 합니다.

• Python : 버전 3.9 이상.
• PyTorch : 모델을 실행합니다.
• 변압기 : Qwen 모델에 대한 액세스를 제공합니다.
• 가속화 : 일부 환경에서는 필수입니다.
• FFmpeg 및 pydub : 오디오 처리용.
• FastAPI : 웹 서버를 생성합니다.
• Uvicorn : FastAPI를 실행하는 ASGI 서버입니다.
• Bark : 텍스트 음성 합성에 사용됨.
• Multipart & Scipy : 오디오를 조작합니다.

FFmpeg는 Linux에서는 apt install ffmpeg 통해, MacOS에서는 brew install ffmpeg 통해 설치할 수 있습니다.

pip를 사용하여 Python 종속성을 설치할 수 있습니다: pip install torch transformers accelerate pydub fastapi uvicorn bark python-multipart scipy

1단계: 환경 설정

먼저 Python 환경을 설정하고 PyTorch 장치를 선택해 보겠습니다.

 import torch device = 'cuda' if torch.cuda.is_available() else 'cpu'

이 코드는 CUDA 호환(Nvidia) GPU를 사용할 수 있는지 확인하고 이에 따라 장치를 설정합니다.

해당 GPU를 사용할 수 없으면 PyTorch는 훨씬 느린 CPU에서 실행됩니다.

최신 Apple Silicon 기기의 경우 PyTorch를 Metal에서 실행하기 위해 기기를 mps 로 설정할 수도 있지만, PyTorch Metal 구현은 포괄적이지 않습니다.

2단계: 모델 로딩

대부분의 오픈소스 LLM은 텍스트 입력과 텍스트 출력만 지원합니다. 그러나 음성-입력-음성-출력 시스템을 만들고 싶기 때문에 두 가지 모델을 더 사용하여 (1) LLM에 입력하기 전에 음성을 텍스트로 변환하고 (2) LLM 출력을 다시 음성으로 변환해야 합니다.

Qwen Audio와 같은 다중 모달 LLM을 사용하면 음성 입력을 텍스트 응답으로 처리하는 하나의 모델만 사용하고, 그런 다음 두 번째 모델을 사용하여 LLM 출력을 다시 음성으로 변환하기만 하면 됩니다.

이러한 다중 모드 접근 방식은 처리 시간 및 (V)RAM 소비 측면에서 효율적일 뿐만 아니라, 입력 오디오가 아무런 마찰 없이 LLM으로 직접 전송되기 때문에 일반적으로 더 나은 결과를 낳습니다.

Runpod 또는 Vast 와 같은 클라우드 GPU 호스트에서 실행하는 경우 모델을 다운로드하기 전에 export HF_HOME=/workspace/hf & export XDG_CACHE_HOME=/workspace/bark 실행하여 HuggingFace 홈 및 Bark 디렉터리를 볼륨 저장소로 설정해야 합니다.

 from transformers import AutoProcessor, Qwen2AudioForConditionalGeneration model_name = "Qwen/Qwen2-Audio-7B-Instruct" processor = AutoProcessor.from_pretrained(model_name) model = Qwen2AudioForConditionalGeneration.from_pretrained(model_name, device_map="auto").to(device)

우리는 계산 요구 사항을 줄이기 위해 여기서 Qwen Audio 모델 시리즈의 작은 7B 변형을 사용하기로 했습니다. 그러나 이 기사를 읽을 때쯤이면 Qwen이 더 강력하고 더 큰 오디오 모델을 출시했을 수도 있습니다. HuggingFace에서 모든 Qwen 모델을 보고 최신 모델을 사용하고 있는지 다시 확인할 수 있습니다.

운영 환경에서는 훨씬 더 높은 처리량을 위해 vLLM 과 같은 빠른 추론 엔진을 사용하는 것이 좋습니다.

3단계: Bark 모델 로딩

Bark는 여러 언어와 음향 효과를 지원하는 최첨단 오픈소스 텍스트-음성 변환 AI 모델입니다.

 from bark import SAMPLE_RATE, generate_audio, preload_models preload_models()

Bark 외에도 다른 오픈소스 또는 독점 텍스트-음성 모델을 사용할 수도 있습니다. 독점적인 모델이 더 성능이 좋을 수 있지만 비용이 훨씬 더 많이 든다는 점을 명심하세요. TTS arena는 최신 비교를 유지합니다 .

Qwen Audio 7B와 Bark가 메모리에 로드되면 대략적인 (V)RAM 사용량은 24GB이므로 하드웨어가 이를 지원하는지 확인하세요. 그렇지 않으면 Qwen 모델의 양자화된 버전을 사용하여 메모리를 절약할 수 있습니다 .

4단계: FastAPI 서버 설정

들어오는 오디오나 텍스트 입력을 처리하고 오디오 응답을 반환하는 두 개의 경로를 갖는 FastAPI 서버를 생성하겠습니다.

 from fastapi import FastAPI, UploadFile, Form from fastapi.responses import StreamingResponse import uvicorn app = FastAPI() @app.post("/voice") async def voice_interaction(file: UploadFile): # TODO return @app.post("/text") async def text_interaction(text: str = Form(...)): # TODO return if __name__ == "__main__":  uvicorn.run(app, host="0.0.0.0", port=8000)

이 서버는 /voice & /text 엔드포인트에서 POST 요청을 통해 오디오 파일을 수락합니다.

5단계: 오디오 입력 처리

ffmpeg를 사용하여 수신 오디오를 처리하고 Qwen 모델에 맞게 준비합니다.

 from pydub import AudioSegment from io import BytesIO import numpy as np def audiosegment_to_float32_array(audio_segment: AudioSegment, target_rate: int = 16000) -> np.ndarray: audio_segment = audio_segment.set_frame_rate(target_rate).set_channels(1) samples = np.array(audio_segment.get_array_of_samples(), dtype=np.int16) samples = samples.astype(np.float32) / 32768.0 return samples def load_audio_as_array(audio_bytes: bytes) -> np.ndarray: audio_segment = AudioSegment.from_file(BytesIO(audio_bytes)) float_array = audiosegment_to_float32_array(audio_segment, target_rate=16000) return float_array

6단계: Qwen을 사용하여 텍스트 응답 생성

처리된 오디오로 Qwen 모델을 사용하여 텍스트 응답을 생성할 수 있습니다. 여기에는 텍스트 및 오디오 입력을 모두 처리해야 합니다.

전처리기는 입력 내용을 모델의 채팅 템플릿(Qwen의 경우 ChatML)으로 변환합니다.

 def generate_response(conversation): text = processor.apply_chat_template(conversation, add_generation_prompt=True, tokenize=False) audios = [] for message in conversation: if isinstance(message["content"], list): for ele in message["content"]: if ele["type"] == "audio": audio_array = load_audio_as_array(ele["audio_url"]) audios.append(audio_array) if audios: inputs = processor( text=text, audios=audios, return_tensors="pt", padding=True ).to(device) else: inputs = processor( text=text, return_tensors="pt", padding=True ).to(device) generate_ids = model.generate(**inputs, max_length=256) generate_ids = generate_ids[:, inputs.input_ids.size(1):] response = processor.batch_decode( generate_ids, skip_special_tokens=True, clean_up_tokenization_spaces=False )[0] return response

model.generate 함수에서 온도와 같은 생성 매개변수를 자유롭게 사용해 보세요.

7단계: Bark로 텍스트를 음성으로 변환

마지막으로, 생성된 텍스트 응답을 다시 음성으로 변환합니다.

 from scipy.io.wavfile import write as write_wav def text_to_speech(text): audio_array = generate_audio(text) output_buffer = BytesIO() write_wav(output_buffer, SAMPLE_RATE, audio_array) output_buffer.seek(0) return output_buffer

8단계: API에 모든 것을 통합하기

오디오 또는 텍스트 입력을 처리하고, 응답을 생성하고, 합성된 음성을 WAV 파일로 반환하도록 엔드포인트를 업데이트합니다.

 @app.post("/voice") async def voice_interaction(file: UploadFile): audio_bytes = await file.read() conversation = [ { "role": "user", "content": [ { "type": "audio", "audio_url": audio_bytes } ] } ] response_text = generate_response(conversation) audio_output = text_to_speech(response_text) return StreamingResponse(audio_output, media_type="audio/wav") @app.post("/text") async def text_interaction(text: str = Form(...)): conversation = [ {"role": "user", "content": [{"type": "text", "text": text}]} ] response_text = generate_response(conversation) audio_output = text_to_speech(response_text) return StreamingResponse(audio_output, media_type="audio/wav")

대화에 시스템 메시지를 추가하면 도우미의 응답을 더 효과적으로 제어할 수 있습니다.

9단계: 테스트하기

다음과 같이 curl 사용하여 서버에 ping을 보낼 수 있습니다.

 # Audio input curl -X POST http://localhost:8000/voice --output output.wav -F "[email protected]" # Text input curl -X POST http://localhost:8000/text --output output.wav -H "Content-Type: application/x-www-form-urlencoded" -d "text=Hey"

결론

이러한 단계를 따르면 최첨단 모델을 사용하여 양방향 음성 상호 작용이 가능한 간단한 로컬 서버를 설정하게 됩니다. 이 설정은 보다 복잡한 음성 지원 애플리케이션을 구축하기 위한 기반이 될 수 있습니다.

응용 프로그램

AI 기반 언어 모델을 수익화하는 방법을 모색 중이라면 다음과 같은 잠재적인 적용 분야를 고려해 보세요.

• 챗봇(예: 캐릭터 AI , NSFW AI 채팅 )
• 전화 상담원(예: Synthflow , Bland )
• 고객 지원 자동화(예: Zendesk , Forethought )
• 법률 보조원( Harvey AI , Leya AI )

전체 코드

 import torch from fastapi import FastAPI, UploadFile, Form from fastapi.responses import StreamingResponse import uvicorn from transformers import AutoProcessor, Qwen2AudioForConditionalGeneration from bark import SAMPLE_RATE, generate_audio, preload_models from scipy.io.wavfile import write as write_wav from pydub import AudioSegment from io import BytesIO import numpy as np device = 'cuda' if torch.cuda.is_available() else 'cpu' model_name = "Qwen/Qwen2-Audio-7B-Instruct" processor = AutoProcessor.from_pretrained(model_name) model = Qwen2AudioForConditionalGeneration.from_pretrained(model_name, device_map="auto").to(device) preload_models() app = FastAPI() def audiosegment_to_float32_array(audio_segment: AudioSegment, target_rate: int = 16000) -> np.ndarray: audio_segment = audio_segment.set_frame_rate(target_rate).set_channels(1) samples = np.array(audio_segment.get_array_of_samples(), dtype=np.int16) samples = samples.astype(np.float32) / 32768.0 return samples def load_audio_as_array(audio_bytes: bytes) -> np.ndarray: audio_segment = AudioSegment.from_file(BytesIO(audio_bytes)) float_array = audiosegment_to_float32_array(audio_segment, target_rate=16000) return float_array def generate_response(conversation): text = processor.apply_chat_template(conversation, add_generation_prompt=True, tokenize=False) audios = [] for message in conversation: if isinstance(message["content"], list): for ele in message["content"]: if ele["type"] == "audio": audio_array = load_audio_as_array(ele["audio_url"]) audios.append(audio_array) if audios: inputs = processor( text=text, audios=audios, return_tensors="pt", padding=True ).to(device) else: inputs = processor( text=text, return_tensors="pt", padding=True ).to(device) generate_ids = model.generate(**inputs, max_length=256) generate_ids = generate_ids[:, inputs.input_ids.size(1):] response = processor.batch_decode( generate_ids, skip_special_tokens=True, clean_up_tokenization_spaces=False )[0] return response def text_to_speech(text): audio_array = generate_audio(text) output_buffer = BytesIO() write_wav(output_buffer, SAMPLE_RATE, audio_array) output_buffer.seek(0) return output_buffer @app.post("/voice") async def voice_interaction(file: UploadFile): audio_bytes = await file.read() conversation = [ { "role": "user", "content": [ { "type": "audio", "audio_url": audio_bytes } ] } ] response_text = generate_response(conversation) audio_output = text_to_speech(response_text) return StreamingResponse(audio_output, media_type="audio/wav") @app.post("/text") async def text_interaction(text: str = Form(...)): conversation = [ {"role": "user", "content": [{"type": "text", "text": text}]} ] response_text = generate_response(conversation) audio_output = text_to_speech(response_text) return StreamingResponse(audio_output, media_type="audio/wav") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)