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: : Versión 3.9 o superior. Python : para ejecutar los modelos. PyTorch : proporciona acceso al modelo Qwen. Transformadores : requerido en algunos entornos. Acelerar : para procesamiento de audio. FFmpeg y pydub : Para crear el servidor web. FastAPI : servidor ASGI para ejecutar FastAPI. Uvicorn : Para síntesis de texto a voz. Corteza : para manipular audio. Multipart y Scipy FFmpeg se puede instalar mediante en Linux o en MacOS. apt install ffmpeg brew install ffmpeg 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 para ejecutar PyTorch en Metal, pero la implementación de PyTorch Metal no es completa. mps 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 o , deberá configurar los directorios de inicio de HuggingFace y Bark en su almacenamiento de volumen ejecutando & antes de descargar los modelos. Runpod Vast export HF_HOME=/workspace/hf export XDG_CACHE_HOME=/workspace/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) 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 para comprobar que estás usando su modelo más reciente. ver todos los modelos Qwen en HuggingFace Para un entorno de producción, es posible que desee utilizar un motor de inferencia rápido como para obtener un rendimiento mucho mayor. vLLM 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, para ahorrar memoria. puedes usar una versión cuantificada del modelo Qwen 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 y . /voice /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 para hacer ping a nuestro servidor de la siguiente manera: curl # 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)
