ในบทความนี้ผมจะพาทุกคนสร้าง Voice Assistant Pipeline ที่สมบูรณ์แบบตั้งแต่ Speech-to-Text (STT) ด้วย Whisper v4 ไปจนถึง Text-to-Speech (TTS) โดยใช้ HolySheep AI API ซึ่งให้ความเร็วต่ำกว่า 50ms พร้อมราคาที่ประหยัดมาก — อัตราแลกเปลี่ยน ¥1 เท่ากับ $1 ครับ ถ้าพร้อมแล้วไปลุยกันเลย
ภาพรวมสถาปัตยกรรม
สถาปัตยกรรมที่เราจะสร้างประกอบด้วย 3 ส่วนหลัก:
- Whisper v4 (STT) — แปลงเสียงพูดเป็นข้อความ รองรับภาษาไทยและหลายภาษา
- LLM Processing — ประมวลผลข้อความและสร้างคำตอบ
- TTS Engine — สังเคราะห์เสียงพูดจากข้อความคำตอบ
การติดตั้งและ Setup
# สร้าง virtual environment
python -m venv voice_assistant_env
source voice_assistant_env/bin/activate # Linux/Mac
voice_assistant_env\Scripts\activate # Windows
ติดตั้ง dependencies
pip install openai-whisper torch torchaudio pyaudio numpy scipy
pip install python-dotenv aiofiles asyncio
ติดตั้ง FFmpeg (จำเป็นสำหรับ Whisper)
Ubuntu/Debian:
sudo apt-get install ffmpeg
macOS:
brew install ffmpeg
Windows:
winget install ffmpegการใช้งาน Whisper v4 สำหรับ STT
Whisper v4 เป็นโมเดล Speech-to-Text ที่ทรงพลังมากจาก OpenAI รองรับกว่า 100 ภาษารวมถึงภาษาไทย ในตัวอย่างนี้ผมจะใช้โมเดล whisper-large-v3 ซึ่งให้ความแม่นยำสูงสุด
import whisper
import numpy as np
import soundfile as sf
class WhisperSTT:
def __init__(self, model_name="whisper-large-v3", device="cuda"):
"""
Initialize Whisper model for Speech-to-Text
model_name: base, small, medium, large-v3, large-v3-turbo
device: cuda สำหรับ GPU, cpu สำหรับ CPU
"""
self.model = whisper.load_model(model_name, device=device)
print(f"✓ Whisper {model_name} loaded on {device}")
def transcribe_audio(self, audio_path, language="th", task="transcribe"):
"""
แปลงไฟล์เสียงเป็นข้อความ
Args:
audio_path: ที่อยู่ไฟล์เสียง (wav, mp3, m4a, flac)
language: รหัสภาษา (th=ไทย, en=อังกฤษ, auto=auto detect)
task: transcribe หรือ translate
Returns:
dict: {'text': 'ข้อความ', 'language': 'ภาษา', 'segments': [...]}
"""
# โหลดและตรวจสอบไฟล์เสียง
audio = whisper.load_audio(audio_path)
audio = whisper.pad_or_trim(audio)
# สร้าง log-Mel spectrogram
mel = whisper.log_mel_spectrogram(audio).to(self.model.device)
# ตรวจจับภาษา (ถ้าไม่ระบุ)
if language == "auto":
_, probs = self.model.detect_language(mel)
detected_lang = max(probs, key=probs.get)
print(f"Detected language: {detected_lang} ({probs[detected_lang]:.2%})")
language = detected_lang
# ถอดเสียง
options = {
"task": task,
"language": language,
"temperature": 0.0,
"best_of": 5,
"beam_size": 5,
}
result = self.model.transcribe(audio_path, **options)
return {
"text": result["text"].strip(),
"language": result.get("language", language),
"segments": result.get("segments", []),
"duration": result.get("duration", 0)
}
ใช้งาน
stt = WhisperSTT(model_name="whisper-large-v3", device="cuda")
result = stt.transcribe_audio("recording.wav", language="th")
print(f"Transcribed: {result['text']}")Streaming Audio Capture ด้วย PyAudio
import pyaudio
import numpy as np
import wave
import threading
import time
class AudioStreamer:
"""จับเสียงแบบ real-time streaming สำหรับ voice assistant"""
def __init__(self, chunk_size=1024, sample_rate=16000, channels=1):
self.chunk_size = chunk_size
self.sample_rate = sample_rate
self.channels = channels
self.format = pyaudio.paInt16
self.audio = pyaudio.PyAudio()
self.stream = None
self.frames = []
self.is_recording = False
self.silence_threshold = 500 # Amplitude threshold for silence detection
self.silence_duration = 1.5 # Seconds of silence to stop recording
self.max_duration = 30 # Maximum recording duration in seconds
def start_streaming(self, callback=None):
"""เริ่มจับเสียง streaming"""
self.stream = self.audio.open(
format=self.format,
channels=self.channels,
rate=self.sample_rate,
input=True,
frames_per_buffer=self.chunk_size,
stream_callback=callback
)
self.stream.start_streaming()
print(f"🎤 Recording started at {self.sample_rate}Hz")
def record_until_silence(self):
"""บันทึกเสียงจนกว่าจะเงียบ"""
self.frames = []
self.is_recording = True
silence_counter = 0
self.stream = self.audio.open(
format=self.format,
channels=self.channels,
rate=self.sample_rate,
input=True,
frames_per_buffer=self.chunk_size
)
start_time = time.time()
while self.is_recording:
data = self.stream.read(self.chunk_size, exception_on_overflow=False)
audio_data = np.frombuffer(data, dtype=np.int16)
# คำนวณ amplitude
amplitude = np.abs(audio_data).mean()
# ตรวจจับความเงียบ
if amplitude < self.silence_threshold:
silence_counter += self.chunk_size / self.sample_rate
else:
silence_counter = 0
self.frames.append(data)
# หยุดเมื่อเงียบนานพอ หรือเกินเวลาสูงสุด
if silence_counter >= self.silence_duration or \
(time.time() - start_time) >= self.max_duration:
self.stop_recording()
return self.get_audio_bytes()
def stop_recording(self):
"""หยุดบันทึก"""
self.is_recording = False
if self.stream:
self.stream.stop_stream()
self.stream.close()
self.stream = None
def get_audio_bytes(self):
"""แปลง frames เป็น bytes"""
return b''.join(self.frames)
def save_to_wav(self, filename):
"""บันทึกเป็นไฟล์ WAV"""
with wave.open(filename, 'wb') as wf:
wf.setnchannels(self.channels)
wf.setsampwidth(self.audio.get_sample_size(self.format))
wf.setframerate(self.sample_rate)
wf.writeframes(self.get_audio_bytes())
print(f"💾 Saved to {filename}")
def __del__(self):
self.audio.terminate()
ใช้งาน
streamer = AudioStreamer(sample_rate=16000)
print("🎤 Say something...")
audio_bytes = streamer.record_until_silence()
streamer.save_to_wav("input.wav")TTS ด้วย HolySheep AI API
สำหรับ Text-to-Speech เราจะใช้ HolySheep AI ซึ่งให้ความเร็วในการตอบสนองต่ำกว่า 50ms พร้อมราคาที่ประหยัดมาก — Claude Sonnet 4.5 อยู่ที่ $15/MTok เท่านั้น ถ้าใครยังไม่มี API Key สามารถสมัครที่นี่เพื่อรับเครดิตฟรีเมื่อลงทะเบียน
import requests
import base64
import io
import soundfile as sf
import numpy as np
class HolySheepTTS:
"""Text-to-Speech using HolySheep AI API"""
def __init__(self, api_key: str):
"""
Initialize HolySheep TTS Client
Args:
api_key: YOUR_HOLYSHEEP_API_KEY from https://www.holysheep.ai
"""
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def synthesize_speech(self, text: str, voice: str = "alloy",
response_format: str = "mp3",
speed: float = 1.0) -> bytes:
"""
แปลงข้อความเป็นเสียง
Args:
text: ข้อความที่ต้องการแปลงเป็นเสียง
voice: เสียงที่ใช้ (alloy, echo, fable, onyx, nova, shimmer)
response_format: รูปแบบไฟล์ (mp3, opus, aac, flac)
speed: ความเร็วในการพูด (0.25 - 4.0)
Returns:
bytes: Audio data
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "tts-1",
"input": text,
"voice": voice,
"response_format": response_format,
"speed": speed
}
response = requests.post(
f"{self.base_url}/audio/speech",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
return response.content
else:
raise Exception(f"TTS Error {response.status_code}: {response.text}")
def synthesize_to_file(self, text: str, output_path: str, **kwargs):
"""บันทึกเสียงลงไฟล์"""
audio_data = self.synthesize_speech(text, **kwargs)
with open(output_path, 'wb') as f:
f.write(audio_data)
print(f"✓ Saved TTS to {output_path}")
return output_path
def synthesize_streaming(self, text: str, chunk_size: int = 1024):
"""
Streaming TTS - ส่งเสียงทีละส่วน (เหมาะสำหรับ real-time)
Yields:
bytes: Audio chunks
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "tts-1",
"input": text,
"voice": "alloy",
"response_format": "mp3"
}
with requests.post(
f"{self.base_url}/audio/speech",
headers=headers,
json=payload,
stream=True,
timeout=60
) as response:
for chunk in response.iter_content(chunk_size=chunk_size):
if chunk:
yield chunk
ใช้งาน
tts = HolySheepTTS(api_key="YOUR_HOLYSHEEP_API_KEY")
ทดสอบ TTS ภาษาไทย
thai_text = "สวัสดีครับ ผมคือ Voice Assistant ที่สร้างจาก Whisper และ HolySheep AI"
tts.synthesize_to_file(thai_text, "output_thai.mp3", voice="alloy")
ทดสอบ TTS ภาษาจีน
chinese_text = "你好，我是语音助手"
tts.synthesize_to_file(chinese_text, "output_chinese.mp3", voice="nova")สร้าง Voice Assistant Pipeline สมบูรณ์
import asyncio
import queue
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Optional, Callable
@dataclass
class VoiceMessage:
"""โครงสร้างข้อมูลข้อความเสียง"""
text: str
audio_data: bytes
timestamp: float
language: str = "auto"
class VoiceAssistantPipeline:
"""
Voice Assistant Pipeline สมบูรณ์แบบ
- Whisper v4 สำหรับ Speech-to-Text
- LLM Processing ด้วย HolySheep AI
- TTS สำหรับ Speech Synthesis
"""
def __init__(self, holysheep_api_key: str):
self.stt = WhisperSTT(model_name="whisper-large-v3", device="cuda")
self.tts = HolySheepTTS(api_key=holysheep_api_key)
self.audio_streamer = AudioStreamer(sample_rate=16000)
# Thread pool สำหรับ async processing
self.executor = ThreadPoolExecutor(max_workers=4)
# Queue สำหรับ message passing
self.input_queue = queue.Queue()
self.output_queue = queue.Queue()
# สถานะ
self.is_running = False
# System prompt
self.system_prompt = """คุณคือผู้ช่วยเสียงอัจฉริยะ ชื่อ "พี่หนำซำ"
ตอบกลับด้วยความเป็นมิตร กระชับ และเป็นธรรมชาติ
ความยาวข้อความไม่ควรเกิน 2-3 ประโยค
หากผู้ใช้ถามเป็นภาษาอังกฤษ ให้ตอบเป็นภาษาอังกฤษ"""
# Conversation history
self.conversation_history = []
def _call_llm(self, user_message: str) -> str:
"""เรียก LLM ผ่าน HolySheep AI API"""
import openai
client = openai.OpenAI(
api_key=self.holysheep_api_key,
base_url="https://api.holysheep.ai/v1"
)
# เพิ่ม system prompt และ history
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.conversation_history[-10:]) # Keep last 10 messages
messages.append({"role": "user", "content": user_message})
response = client.chat.completions.create(
model="gpt-4.1",
messages=messages,
temperature=0.7,
max_tokens=500
)
return response.choices[0].message.content
def process_voice_input(self, audio_bytes: bytes) -> VoiceMessage:
"""ประมวลผลเสียงเป็นข้อความ"""
# บันทึก sementara
temp_path = "temp_input.wav"
with wave.open(temp_path, 'wb') as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(16000)
wf.writeframes(audio_bytes)
# Transcribe
result = self.stt.transcribe_audio(temp_path, language="auto")
return VoiceMessage(
text=result["text"],
audio_data=audio_bytes,
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง