Transcript
@@ -209,7 +236,8 @@
const wsUrl = `${wsProtocol}//${window.location.host}/ws/voice/${sessionId}`;
let ws = null;
- let audioContext = null;
+ let inputAudioContext = null;
+ let outputAudioContext = null;
let microphone = null;
let scriptProcessor = null;
let isConnected = false;
@@ -218,6 +246,7 @@
const statusEl = document.getElementById('status');
const statusTextEl = document.getElementById('status-text');
+ const thinkingEl = document.getElementById('thinking');
const connectBtn = document.getElementById('connect-btn');
const disconnectBtn = document.getElementById('disconnect-btn');
const transcriptEl = document.getElementById('transcript');
@@ -229,6 +258,10 @@
statusTextEl.textContent = text;
}
+ function showThinking(show) {
+ thinkingEl.style.display = show ? 'inline-flex' : 'none';
+ }
+
function showError(message) {
errorEl.textContent = message;
errorEl.style.display = 'block';
@@ -238,6 +271,21 @@
errorEl.style.display = 'none';
}
+ function addTranscript(text, type = 'transcript') {
+ const item = document.createElement('div');
+ item.className = 'transcript-item';
+
+ const content = document.createElement('div');
+ content.className = type === 'transcript' ? 'transcript-transcript' : 'transcript-response';
+ content.textContent = text;
+
+ item.appendChild(content);
+ transcriptContentEl.appendChild(item);
+
+ // Auto-scroll to bottom
+ transcriptEl.scrollTop = transcriptEl.scrollHeight;
+ }
+
async function connect() {
if (isConnected) return;
@@ -263,10 +311,13 @@
};
ws.onmessage = (event) => {
- const data = JSON.parse(event.data);
-
- if (data.type === 'welcome') {
- console.log('Server greeting:', data.message);
+ if (event.data instanceof Blob) {
+ // Binary audio data
+ handleAudioData(event.data);
+ } else {
+ // JSON text data
+ const data = JSON.parse(event.data);
+ handleWebsocketMessage(data);
}
};
@@ -288,6 +339,61 @@
}
}
+ function handleWebsocketMessage(data) {
+ switch (data.type) {
+ case 'welcome':
+ console.log('Server greeting:', data.message);
+ break;
+
+ case 'transcript':
+ addTranscript(data.text, 'transcript');
+ break;
+
+ case 'response':
+ addTranscript(data.text, 'response');
+ showThinking(false);
+ break;
+
+ case 'tts_audio':
+ console.log('TTS audio header received:', data.samples, 'samples @', data.sample_rate, 'Hz');
+ break;
+
+ case 'ping':
+ // Keepalive - ignore
+ break;
+
+ default:
+ console.warn('Unknown message type:', data.type);
+ }
+ }
+
+ async function handleAudioData(blob) {
+ try {
+ const arrayBuffer = await blob.arrayBuffer();
+ const audioFloat32Array = new Float32Array(arrayBuffer);
+
+ // Decode audio using output AudioContext
+ const audioBuffer = await outputAudioContext.decodeAudioData(audioFloat32Array.buffer);
+
+ // Play the audio
+ playAudioBuffer(audioBuffer);
+
+ } catch (error) {
+ console.error('Audio playback error:', error);
+ }
+ }
+
+ async function playAudioBuffer(audioBuffer) {
+ const source = outputAudioContext.createBufferSource();
+ source.buffer = audioBuffer;
+
+ // Connect to destination
+ source.connect(outputAudioContext.destination);
+
+ // Start playback
+ source.start();
+ }
+
async function disconnect() {
if (!ws) return;
@@ -325,7 +431,13 @@
async function initAudio() {
try {
- audioContext = new (window.AudioContext || window.webkitAudioContext)({
+ // Create input audio context for microphone (16kHz)
+ inputAudioContext = new (window.AudioContext || window.webkitAudioContext)({
+ sampleRate: 16000
+ });
+
+ // Create output audio context for playback (will be set to server sample rate)
+ outputAudioContext = new (window.AudioContext || window.webkitAudioContext)({
sampleRate: 16000
});
@@ -341,8 +453,8 @@
});
console.log('Microphone acquired, stream tracks:', stream.getTracks().length);
- microphone = audioContext.createMediaStreamSource(stream);
- console.log('MediaStreamSource created, sample rate:', audioContext.sampleRate);
+ microphone = inputAudioContext.createMediaStreamSource(stream);
+ console.log('MediaStreamSource created, sample rate:', inputAudioContext.sampleRate);
// Use ScriptProcessor for reliable audio capture
initScriptProcessor();
@@ -353,28 +465,11 @@
}
}
- async function initAudioWorklet() {
- // Load worklet module
- const workletUrl = `${window.location.origin}/static/voice-worklet.js`;
-
- await audioContext.audioWorklet.addModule(workletUrl);
-
- const processor = new AudioWorkletNode(audioContext, 'voice-processor');
-
- microphone.connect(processor);
-
- processor.port.onmessage = (event) => {
- if (event.data.type === 'audio') {
- sendAudio(event.data.audio);
- }
- };
- }
-
function initScriptProcessor() {
- scriptProcessor = audioContext.createScriptProcessor(4096, 1, 1);
+ scriptProcessor = inputAudioContext.createScriptProcessor(4096, 1, 1);
microphone.connect(scriptProcessor);
- scriptProcessor.connect(audioContext.destination);
+ scriptProcessor.connect(inputAudioContext.destination);
scriptProcessor.onaudioprocess = (event) => {
const inputData = event.inputBuffer.getChannelData(0);
@@ -393,8 +488,12 @@
scriptProcessor = null;
}
- if (audioContext && audioContext.state !== 'closed') {
- audioContext.close();
+ if (inputAudioContext && inputAudioContext.state !== 'closed') {
+ inputAudioContext.close();
+ }
+
+ if (outputAudioContext && outputAudioContext.state !== 'closed') {
+ outputAudioContext.close();
}
}
diff --git a/server/voice_ws.py b/server/voice_ws.py
index e0a866a..123653d 100644
--- a/server/voice_ws.py
+++ b/server/voice_ws.py
@@ -21,6 +21,14 @@ from server.stt import DeepgramSTT
from server.tts import VeniceKokoroTTS
from openclaw_client.client import OpenClawClient, OpenClawConfig
+# Simple energy-based VAD to avoid sending silence to Deepgram
+def _is_speech(audio: np.ndarray, threshold: float = 0.01) -> bool:
+ """Check if audio buffer contains speech (above energy threshold)."""
+ if len(audio) == 0:
+ return False
+ energy = float(np.sqrt(np.mean(audio ** 2)))
+ return energy > threshold
+
logger = logging.getLogger(__name__)
@@ -42,6 +50,12 @@ class VoiceSession:
self.channel_count = 1
self.bits_per_sample = 32
+ # Concurrency
+ self.audio_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=100)
+
+ # WebSocket connection
+ self.websocket: Optional[WebSocket] = None
+
# Engines (self-contained, don't share with run.py)
self.stt = None
self.tts = None
@@ -51,6 +65,9 @@ class VoiceSession:
self.connected = False
self.transcript = []
+ # Consumer task
+ self.consumer_task: Optional[asyncio.Task] = None
+
logger.info(f"Created voice session {session_id}")
async def initialize(self):
@@ -97,6 +114,13 @@ class VoiceSession:
"""Clean up resources."""
self.connected = False
+ if self.consumer_task and not self.consumer_task.done():
+ self.consumer_task.cancel()
+ try:
+ await self.consumer_task
+ except asyncio.CancelledError:
+ pass
+
if self.openclaw:
await self.openclaw.disconnect()
@@ -106,70 +130,115 @@ class VoiceSession:
"""Generate random session ID."""
return "".join(random.choices(string.ascii_letters + string.digits, k=8))
- async def process_audio_chunk(self, data: bytes):
- """Process incoming audio chunk."""
- async with self._buffer_lock:
- self.audio_buffer.extend(data)
+ async def _consumer_task(self):
+ """Consumer task that processes audio from queue."""
+ start_time = asyncio.get_event_loop().time()
- # Calculate duration
- chunk_size = len(data)
- chunk_duration = chunk_size / (self.sample_rate * self.channel_count * 4)
+ while self.connected:
+ try:
+ # Wait for audio chunk (with timeout)
+ try:
+ data = await asyncio.wait_for(self.audio_queue.get(), timeout=0.1)
+ except asyncio.TimeoutError:
+ # Check if enough time has passed for buffer to accumulate
+ elapsed = asyncio.get_event_loop().time() - start_time
+ if elapsed > 1.0 and len(self.audio_buffer) == 0:
+ # No audio received for 1 second, reset
+ start_time = asyncio.get_event_loop().time()
+ continue
- self.buffer_duration += chunk_duration
+ # Accumulate audio (no lock needed — only consumer touches buffer)
+ self.audio_buffer.extend(data)
- # Buffer until ~1 second
- if self.buffer_duration >= 0.8: # Slightly less than 1 second
- await self._transcribe_buffered_audio()
+ # Calculate duration
+ chunk_size = len(data)
+ chunk_duration = chunk_size / (self.sample_rate * self.channel_count * 4)
+
+ self.buffer_duration += chunk_duration
+
+ # Buffer until ~0.8 seconds
+ if self.buffer_duration >= 0.8:
+ await self._transcribe_buffered_audio()
+ start_time = asyncio.get_event_loop().time()
+
+ except asyncio.CancelledError:
+ logger.info(f"Consumer task cancelled for session {self.session_id}")
+ break
+
+ except Exception as e:
+ logger.error(f"Consumer task error: {e}", exc_info=True)
+
+ logger.info(f"Consumer task exited for session {self.session_id}")
async def _transcribe_buffered_audio(self):
"""Transcribe accumulated audio and send to OpenClaw."""
- async with self._buffer_lock:
- if not self.audio_buffer:
- return
+ if not self.audio_buffer:
+ return
- # Convert bytearray to numpy array
- audio_data = np.frombuffer(bytes(self.audio_buffer), dtype=np.float32)
+ # Copy and clear buffer immediately (only consumer touches it)
+ audio_bytes = bytes(self.audio_buffer)
+ self.audio_buffer.clear()
+ self.buffer_duration = 0.0
+ # Convert bytearray to numpy array
+ audio_data = np.frombuffer(audio_bytes, dtype=np.float32)
+
+ # Skip silence — don't waste Deepgram credits on empty audio
+ if not _is_speech(audio_data):
+ logger.debug(f"Session {self.session_id}: silence detected, skipping STT")
+ return
+
+ try:
# Transcribe
- try:
- result = await self.stt.transcribe_async(audio_data)
+ result = await self.stt.transcribe_async(audio_data)
- if result.text.strip():
- # Send to OpenClaw
- response = await self.openclaw.send_message(
- agent="main",
- message=result.text,
- speaker="voice_user",
- )
+ if result.text.strip():
+ # Send intermediate transcript status
+ if self.connected:
+ await self._send_status("transcript", result.text)
- # Log transcript
- timestamp = asyncio.get_event_loop().time()
- entry = {
- "timestamp": timestamp,
- "session_id": self.session_id,
- "transcript": result.text,
- "response": response,
- }
+ # Send to OpenClaw
+ response = await self.openclaw.send_message(
+ agent="main",
+ message=result.text,
+ speaker="voice_user",
+ )
- self.transcript.append(entry)
+ # Send intermediate response status
+ if self.connected:
+ await self._send_status("response", response)
- # Write to file
- with open(self.transcript_file, "a") as f:
- f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+ # Log transcript
+ timestamp = asyncio.get_event_loop().time()
+ entry = {
+ "timestamp": timestamp,
+ "session_id": self.session_id,
+ "transcript": result.text,
+ "response": response,
+ }
- logger.info(
- f"Session {self.session_id}: "
- f'"{result.text[:50]}..." -> "{response[:50]}..."'
- )
+ self.transcript.append(entry)
- # Clear buffer
- self.audio_buffer.clear()
- self.buffer_duration = 0.0
+ # Write to file
+ with open(self.transcript_file, "a") as f:
+ f.write(json.dumps(entry, ensure_ascii=False) + "\n")
- except Exception as e:
- logger.error(f"Transcription error: {e}")
+ logger.info(
+ f"Session {self.session_id}: "
+ f'"{result.text[:50]}..." -> "{response[:50]}..."'
+ )
- async def synthesize_response(self, text: str):
+ # Generate TTS audio
+ audio = await self._synthesize_response(response)
+
+ # Send TTS audio back to browser
+ if audio and self.connected:
+ await self._send_tts_audio(audio)
+
+ except Exception as e:
+ logger.error(f"Transcription error: {e}", exc_info=True)
+
+ async def _synthesize_response(self, text: str):
"""Synthesize TTS audio from response text."""
try:
audio = await self.tts.generate_async(
@@ -181,17 +250,59 @@ class VoiceSession:
return audio
except Exception as e:
- logger.error(f"TTS synthesis error: {e}")
+ logger.error(f"TTS synthesis error: {e}", exc_info=True)
return None
- def get_transcript(self) -> list:
- """Get transcript history."""
- return self.transcript
+ async def _send_status(self, status_type: str, text: str):
+ """Send status message to WebSocket."""
+ try:
+ message = {
+ "type": status_type,
+ "text": text,
+ }
+ await self._send_json(message)
+ except Exception as e:
+ logger.error(f"Failed to send {status_type} status: {e}")
+
+ async def _send_tts_audio(self, audio: np.ndarray):
+ """Send TTS audio back to browser as binary PCM with JSON header."""
+ try:
+ # Convert to 16-bit PCM
+ pcm_data = (audio * 32767).astype(np.int16).tobytes()
+
+ # Create JSON header
+ header = {
+ "type": "tts_audio",
+ "samples": len(pcm_data) // 2, # 2 bytes per sample
+ "sample_rate": self.sample_rate,
+ }
+
+ # Send header as JSON text
+ await self._send_json(header)
+
+ # Send PCM audio as binary
+ await self._send_bytes(pcm_data)
+
+ logger.info(f"Sent TTS audio: {len(pcm_data)} bytes, {header['samples']} samples")
+
+ except Exception as e:
+ logger.error(f"Failed to send TTS audio: {e}", exc_info=True)
+
+ async def _send_json(self, data: dict):
+ """Send JSON message to WebSocket."""
+ await self._send_bytes(json.dumps(data).encode("utf-8"))
+
+ async def _send_bytes(self, data: bytes):
+ """Send bytes to WebSocket."""
+ await self._send_json({"type": "websocket.send", "bytes": len(data)})
+ if self.websocket:
+ await self.websocket.send_bytes(data)
async def handle_voice_websocket(websocket: WebSocket, session_id: str):
"""Handle WebSocket connection for voice session."""
session = VoiceSession(session_id)
+ session.websocket = websocket
await websocket.accept()
session.connected = True
@@ -220,7 +331,10 @@ async def handle_voice_websocket(websocket: WebSocket, session_id: str):
keepalive_task = asyncio.create_task(keepalive())
- # Receive and process audio
+ # Start consumer task
+ session.consumer_task = asyncio.create_task(session._consumer_task())
+
+ # Receive and process audio (non-blocking)
chunk_count = 0
while session.connected:
try:
@@ -237,7 +351,13 @@ async def handle_voice_websocket(websocket: WebSocket, session_id: str):
chunk_count += 1
if chunk_count <= 5 or chunk_count % 100 == 0:
logger.info(f"Audio chunk #{chunk_count}: {len(msg['bytes'])} bytes")
- await session.process_audio_chunk(msg["bytes"])
+
+ # Put audio chunk into queue (non-blocking)
+ try:
+ session.audio_queue.put_nowait(msg["bytes"])
+ except asyncio.QueueFull:
+ logger.warning(f"Audio queue full for session {session_id}, dropping chunk")
+
elif "text" in msg:
pass
else: