¡Alojar tu propia IA con chat de voz bidireccional es más fácil de lo que crees!

La integración de LLM con capacidades de voz ha creado nuevas oportunidades en interacciones personalizadas con los clientes.

Esta guía lo guiará a través de la configuración de un servidor LLM local que admita interacciones de voz bidireccionales mediante Python, Transformers, Qwen2-Audio-7B-Instruct y Bark.

Prerrequisitos

Antes de comenzar, deberás tener instalado lo siguiente:

• Python : Versión 3.9 o superior.
• PyTorch : para ejecutar los modelos.
• Transformadores : proporciona acceso al modelo Qwen.
• Acelerar : requerido en algunos entornos.
• FFmpeg y pydub : para procesamiento de audio.
• FastAPI : Para crear el servidor web.
• Uvicorn : servidor ASGI para ejecutar FastAPI.
• Corteza : Para síntesis de texto a voz.
• Multipart y Scipy : para manipular audio.

FFmpeg se puede instalar mediante apt install ffmpeg en Linux o brew install ffmpeg en MacOS.

Puede instalar las dependencias de Python usando pip: pip install torch transformers accelerate pydub fastapi uvicorn bark python-multipart scipy

Paso 1: Configuración del entorno

Primero, configuremos nuestro entorno Python y elijamos nuestro dispositivo PyTorch:

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

Este código verifica si hay una GPU compatible con CUDA (Nvidia) disponible y configura el dispositivo en consecuencia.

Si no hay ninguna GPU disponible, PyTorch se ejecutará en la CPU, que es mucho más lenta.

En el caso de los dispositivos Apple Silicon más nuevos, el dispositivo también se puede configurar en mps para ejecutar PyTorch en Metal, pero la implementación de PyTorch Metal no es completa.

Paso 2: Carga del modelo

La mayoría de los LLM de código abierto solo admiten entrada y salida de texto. Sin embargo, dado que queremos crear un sistema de entrada y salida de voz, esto requeriría que usemos dos modelos más para (1) convertir el habla en texto antes de ingresarlo en nuestro LLM y (2) convertir la salida del LLM nuevamente en habla.

Al utilizar un LLM multimodal como Qwen Audio, podemos usar un modelo para procesar la entrada de voz en una respuesta de texto y luego solo tener que usar un segundo modelo para convertir la salida LLM nuevamente en voz.

Este enfoque multimodal no solo es más eficiente en términos de tiempo de procesamiento y consumo de (V)RAM, sino que también suele producir mejores resultados ya que el audio de entrada se envía directamente al LLM sin ninguna fricción.

Si está ejecutando un host de GPU en la nube como Runpod o Vast , deberá configurar los directorios de inicio de HuggingFace y Bark en su almacenamiento de volumen ejecutando export HF_HOME=/workspace/hf & export XDG_CACHE_HOME=/workspace/bark antes de descargar los modelos.

 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)

Elegimos utilizar la variante pequeña 7B de la serie de modelos Qwen Audio para reducir nuestros requisitos computacionales. Sin embargo, es posible que Qwen haya lanzado modelos de audio más potentes y grandes cuando estés leyendo este artículo. Puedes ver todos los modelos Qwen en HuggingFace para comprobar que estás usando su modelo más reciente.

Para un entorno de producción, es posible que desee utilizar un motor de inferencia rápido como vLLM para obtener un rendimiento mucho mayor.

Paso 3: Carga del modelo Bark

Bark es un modelo de inteligencia artificial de texto a voz de código abierto de última generación que admite varios idiomas y efectos de sonido.

 from bark import SAMPLE_RATE, generate_audio, preload_models preload_models()

Además de Bark, también puedes utilizar otros modelos de conversión de texto a voz de código abierto o propietarios. Ten en cuenta que, si bien los propietarios pueden tener un mejor rendimiento, tienen un costo mucho mayor. El área de TTS mantiene una comparación actualizada .

Con Qwen Audio 7B y Bark cargados en la memoria, el uso aproximado de (V)RAM es de 24 GB, así que asegúrate de que tu hardware lo admita. De lo contrario, puedes usar una versión cuantificada del modelo Qwen para ahorrar memoria.

Paso 4: Configuración del servidor FastAPI

Crearemos un servidor FastAPI con dos rutas para manejar entradas de audio o texto entrantes y devolver respuestas de audio.

 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)

Este servidor acepta archivos de audio a través de solicitudes POST en los puntos finales /voice y /text .

Paso 5: Procesamiento de la entrada de audio

Usaremos ffmpeg para procesar el audio entrante y prepararlo para el modelo 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

Paso 6: Generar una respuesta textual con Qwen

Con el audio procesado, podemos generar una respuesta textual utilizando el modelo Qwen. Para ello, será necesario gestionar entradas de texto y audio.

El preprocesador convertirá nuestra entrada a la plantilla de chat del modelo (ChatML en el caso de Qwen).

 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

Siéntete libre de jugar con los parámetros de generación como la temperatura en la función model.generate .

Paso 7: Convertir texto a voz con Bark

Finalmente, convertiremos la respuesta de texto generada nuevamente en voz.

 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

Paso 8: Integrar todo en las API

Actualice los puntos finales para procesar la entrada de audio o texto, generar una respuesta y devolver la voz sintetizada como un archivo 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")

También puede optar por agregar un mensaje del sistema a las conversaciones para obtener más control sobre las respuestas del asistente.

Paso 9: Probar las cosas

Podemos usar curl para hacer ping a nuestro servidor de la siguiente manera:

 # 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"

Conclusión

Si sigue estos pasos, habrá configurado un servidor local simple capaz de realizar interacciones de voz bidireccionales mediante modelos de última generación. Esta configuración puede servir como base para crear aplicaciones de voz más complejas.

Aplicaciones

Si está explorando formas de monetizar modelos de lenguaje impulsados por IA, considere estas posibles aplicaciones:

• Chatbots (por ejemplo, IA de personajes , NSFW AI Chat );
• Agentes telefónicos (por ejemplo, Synthflow , Bland )
• Automatización de la atención al cliente (por ejemplo, Zendesk , Forethought )
• Asistentes legales ( Harvey AI , Leya AI )

Código completo

 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)