آواز کی صلاحیتوں کے ساتھ LLMs کے انضمام نے ذاتی نوعیت کے صارفین کی بات چیت میں نئے مواقع پیدا کیے ہیں۔
یہ گائیڈ آپ کو ایک مقامی LLM سرور قائم کرنے کے لیے لے جائے گا جو Python، Transformers، Qwen2-Audio-7B-Instruct، اور Bark کا استعمال کرتے ہوئے دو طرفہ صوتی تعاملات کو سپورٹ کرتا ہے۔
اس سے پہلے کہ ہم شروع کریں، آپ کے پاس درج ذیل انسٹال ہوں گے:
FFmpeg کو لینکس پر apt install ffmpeg
یا MacOS پر brew install ffmpeg
کے ذریعے انسٹال کیا جا سکتا ہے۔
آپ pip کا استعمال کرتے ہوئے Python انحصار کو انسٹال کر سکتے ہیں: pip install torch transformers accelerate pydub fastapi uvicorn bark python-multipart scipy
پہلے، آئیے اپنا Python ماحول ترتیب دیں اور اپنے PyTorch ڈیوائس کا انتخاب کریں:
import torch device = 'cuda' if torch.cuda.is_available() else 'cpu'
یہ کوڈ چیک کرتا ہے کہ آیا CUDA-مطابقت پذیر (Nvidia) GPU دستیاب ہے اور اس کے مطابق آلہ سیٹ کرتا ہے۔
اگر ایسا کوئی GPU دستیاب نہیں ہے، تو PyTorch اس کے بجائے CPU پر چلے گا جو بہت سست ہے۔
ایپل سلیکون کے نئے آلات کے لیے، میٹل پر پائی ٹارچ چلانے کے لیے ڈیوائس کو
mps
پر بھی سیٹ کیا جا سکتا ہے، لیکن پائی ٹارچ میٹل کا نفاذ جامع نہیں ہے۔
زیادہ تر اوپن سورس ایل ایل ایم صرف ٹیکسٹ ان پٹ اور ٹیکسٹ آؤٹ پٹ کو سپورٹ کرتے ہیں۔ تاہم، چونکہ ہم وائس ان وائس آؤٹ سسٹم بنانا چاہتے ہیں، اس کے لیے ہمیں مزید دو ماڈلز استعمال کرنے کی ضرورت ہوگی تاکہ (1) اسپیچ کو ہمارے LLM میں فیڈ ہونے سے پہلے ٹیکسٹ میں تبدیل کریں اور (2) LLM آؤٹ پٹ کو واپس تبدیل کریں۔ تقریر میں
کیوین آڈیو جیسے ملٹی موڈل ایل ایل ایم کا استعمال کرکے، ہم اسپیچ ان پٹ کو ٹیکسٹ ریسپانس میں پروسیس کرنے کے لیے ایک ماڈل کے ساتھ بھاگ سکتے ہیں، اور پھر صرف ایک دوسرے ماڈل کا استعمال کرنا ہوگا ایل ایل ایم آؤٹ پٹ کو دوبارہ اسپیچ میں تبدیل کرنا ہے۔
یہ ملٹی موڈل اپروچ نہ صرف پروسیسنگ ٹائم اور (V)RAM کی کھپت کے لحاظ سے زیادہ کارآمد ہے، بلکہ عام طور پر بہتر نتائج بھی دیتا ہے کیونکہ ان پٹ آڈیو بغیر کسی رگڑ کے سیدھے LLM کو بھیجا جاتا ہے۔
اگر آپ کلاؤڈ GPU ہوسٹ جیسے Runpod یا Vast پر چل رہے ہیں، تو آپ ڈاؤن لوڈ کرنے سے پہلے
export HF_HOME=/workspace/hf
اورexport XDG_CACHE_HOME=/workspace/bark
چلا کر HuggingFace home & 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 آڈیو ماڈل سیریز کے چھوٹے 7B ویرینٹ کو استعمال کرنے کا انتخاب کیا۔ تاہم، جب تک آپ یہ مضمون پڑھ رہے ہوں گے Qwen نے مضبوط اور بڑے آڈیو ماڈلز جاری کر دیے ہوں گے۔ آپ Qwen کے تمام ماڈلز کو HuggingFace پر دیکھ سکتے ہیں تاکہ یہ چیک کر سکیں کہ آپ ان کا تازہ ترین ماڈل استعمال کر رہے ہیں۔
پیداواری ماحول کے لیے، آپ بہت زیادہ تھرو پٹ کے لیے وی ایل ایل ایم جیسا تیز انفرنس انجن استعمال کرنا چاہتے ہیں۔
Bark ایک جدید ترین اوپن سورس ٹیکسٹ ٹو اسپیچ AI ماڈل ہے جو متعدد زبانوں کے ساتھ ساتھ صوتی اثرات کو بھی سپورٹ کرتا ہے۔
from bark import SAMPLE_RATE, generate_audio, preload_models preload_models()
Bark کے علاوہ، آپ دوسرے اوپن سورس یا ملکیتی ٹیکسٹ ٹو اسپیچ ماڈل بھی استعمال کرسکتے ہیں۔ اس بات کو ذہن میں رکھیں کہ اگرچہ ملکیت والے زیادہ پرفارمنس ہوسکتے ہیں، وہ بہت زیادہ قیمت پر آتے ہیں۔ TTS میدان ایک تازہ ترین موازنہ رکھتا ہے ۔
Qwen Audio 7B اور Bark دونوں میموری میں لوڈ ہونے کے ساتھ، تقریباً (V)RAM کا استعمال 24GB ہے، اس لیے یقینی بنائیں کہ آپ کا ہارڈویئر اس کی حمایت کرتا ہے۔ بصورت دیگر، آپ میموری کو بچانے کے لیے Qwen ماڈل کا کوانٹائزڈ ورژن استعمال کر سکتے ہیں ۔
ہم آنے والے آڈیو یا ٹیکسٹ ان پٹ کو سنبھالنے اور آڈیو جوابات واپس کرنے کے لیے دو راستوں کے ساتھ ایک 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)
یہ سرور آڈیو فائلوں کو POST کے ذریعے /voice
اور /text
اینڈ پوائنٹ پر قبول کرتا ہے۔
ہم آنے والی آڈیو پر کارروائی کرنے اور اسے Qwen ماڈل کے لیے تیار کرنے کے لیے ffmpeg کا استعمال کریں گے۔
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
پروسیس شدہ آڈیو کے ساتھ، ہم 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
model.generate
فنکشن پر درجہ حرارت جیسے جنریشن پیرامیٹرز کے ساتھ کھیلنے کے لئے آزاد محسوس کریں۔
آخر میں، ہم تیار کردہ متن کے جواب کو تقریر میں تبدیل کر دیں گے۔
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
آڈیو یا ٹیکسٹ ان پٹ پر کارروائی کرنے کے لیے اختتامی پوائنٹس کو اپ ڈیٹ کریں، جواب پیدا کریں، اور ترکیب شدہ تقریر کو 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")
آپ اسسٹنٹ کے جوابات پر مزید کنٹرول حاصل کرنے کے لیے گفتگو میں سسٹم میسج شامل کرنے کا بھی انتخاب کر سکتے ہیں۔
ہم اپنے سرور کو پنگ کرنے کے لیے 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"
ان اقدامات پر عمل کرتے ہوئے، آپ نے جدید ترین ماڈلز کا استعمال کرتے ہوئے دو طرفہ صوتی تعاملات کے قابل ایک سادہ مقامی سرور ترتیب دیا ہے۔ یہ سیٹ اپ زیادہ پیچیدہ آواز سے چلنے والی ایپلی کیشنز کی تعمیر کے لیے ایک بنیاد کے طور پر کام کر سکتا ہے۔
اگر آپ 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)