admin/openclaw-voice
1
0
Fork 0
mirror of https://github.com/Purple-Horizons/openclaw-voice.git synced 2026-08-14 00:58:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Streaming TTS: Talk while generating

Browse Source 
- Backend: Add chat_stream() for streaming LLM responses
- TTS: Add synthesize_stream() for progressive audio
- Server: Stream response sentence-by-sentence
  - Buffer text until sentence boundary (. ! ?)
  - Synthesize and send audio immediately
  - User hears first sentence while rest generates

- Client: Audio queue system
  - Queue chunks as they arrive
  - Play sequentially without gaps
  - Progressive text display

Result: ~50% faster perceived response time
This commit is contained in:
gianni-dalerta
2026-02-01 15:38:16 -05:00
parent 0f29ecf188
commit 69a3791b80
4 changed files with 245 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/index.html
+101 -11
View File
@@ -366,6 +366,12 @@
};
}
// Audio queue for streaming playback
let audioQueue = [];
let isPlayingQueue = false;
let currentResponseElement = null;
let streamingText = '';
function handleMessage(msg) {
switch (msg.type) {
case 'listening_started':
@@ -377,22 +383,34 @@
case 'transcript':
addTranscript('You', msg.text, 'user');
setStatus('Getting response...');
// Reset streaming state
streamingText = '';
currentResponseElement = null;
audioQueue = [];
break;
case 'response_text':
// Legacy non-streaming response
addTranscript('AI', msg.text, 'assistant');
break;
case 'response_chunk':
// Streaming text chunk
streamingText += msg.text;
updateStreamingTranscript(streamingText);
setStatus('Speaking...', true);
break;
case 'audio_chunk':
// Queue audio chunk for playback
queueAudioChunk(msg.data, msg.sample_rate);
break;
case 'response_complete':
// Finalize the response
if (currentResponseElement) {
currentResponseElement.innerHTML = `<strong>AI:</strong> ${renderMarkdown(msg.text)}`;
}
break;
case 'audio_response':
playAudioWithCallback(msg.data, msg.sample_rate, () => {
// After audio finishes playing
if (continuousMode) {
setStatus('🎙️ Ready to listen...', true);
setTimeout(() => {
if (continuousMode) startRecording();
}, 300);
} else {
setStatus('Ready');
}
});
// Legacy non-streaming audio
playAudioWithCallback(msg.data, msg.sample_rate, onAudioComplete);
setStatus('Speaking...');
break;
case 'pong':
@@ -400,6 +418,78 @@
}
}
function updateStreamingTranscript(text) {
if (!currentResponseElement) {
currentResponseElement = document.createElement('p');
currentResponseElement.className = 'assistant';
transcriptEl.appendChild(currentResponseElement);
}
currentResponseElement.innerHTML = `<strong>AI:</strong> ${renderMarkdown(text)}`;
transcriptEl.scrollTop = transcriptEl.scrollHeight;
}
function queueAudioChunk(base64Data, sampleRate) {
audioQueue.push({ data: base64Data, sampleRate });
if (!isPlayingQueue) {
playNextInQueue();
}
}
async function playNextInQueue() {
if (audioQueue.length === 0) {
isPlayingQueue = false;
onAudioComplete();
return;
}
isPlayingQueue = true;
const { data, sampleRate } = audioQueue.shift();
try {
// Decode base64 to PCM
const binaryString = atob(data);
const bytes = new Uint8Array(binaryString.length);
for (let i = 0; i < binaryString.length; i++) {
bytes[i] = binaryString.charCodeAt(i);
}
// Convert to Int16 then Float32
const int16 = new Int16Array(bytes.buffer);
const float32 = new Float32Array(int16.length);
for (let i = 0; i < int16.length; i++) {
float32[i] = int16[i] / 32768.0;
}
// Create audio buffer and play
const audioCtx = new (window.AudioContext || window.webkitAudioContext)({ sampleRate });
const buffer = audioCtx.createBuffer(1, float32.length, sampleRate);
buffer.copyToChannel(float32, 0);
const source = audioCtx.createBufferSource();
source.buffer = buffer;
source.connect(audioCtx.destination);
source.onended = () => {
audioCtx.close();
playNextInQueue();
};
source.start(0);
} catch (e) {
console.error('Audio playback error:', e);
playNextInQueue();
}
}
function onAudioComplete() {
if (continuousMode) {
setStatus('🎙️ Ready to listen...', true);
setTimeout(() => {
if (continuousMode) startRecording();
}, 300);
} else {
setStatus('Ready');
}
}
function setStatus(text, active = false) {
statusEl.textContent = text;
statusEl.className = active ? 'status active' : 'status';
src/server/backend.py
+57 -2
View File
@@ -3,7 +3,7 @@ AI Backend module - connects to OpenAI, OpenClaw gateway, or custom backends.
"""
import asyncio
from typing import Optional, List, Dict
from typing import Optional, List, Dict, AsyncGenerator
from loguru import logger
@@ -65,6 +65,22 @@ class AIBackend:
# Fallback echo response
return f"I heard you say: {user_message}"
async def chat_stream(self, user_message: str) -> AsyncGenerator[str, None]:
"""
Stream a response, yielding chunks as they arrive.
Args:
user_message: The user's transcribed speech
Yields:
Text chunks as they're generated
"""
if self.backend_type == "openai" and self._client:
async for chunk in self._chat_openai_stream(user_message):
yield chunk
else:
yield f"I heard you say: {user_message}"
async def _chat_openai(self, user_message: str) -> str:
"""Chat via OpenAI API."""
# Add user message to history
@@ -81,7 +97,7 @@ class AIBackend:
response = await self._client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=150, # Keep responses short for voice
max_tokens=500, # Allow longer for voice
temperature=0.7,
)
@@ -99,6 +115,45 @@ class AIBackend:
logger.error(f"OpenAI API error: {e}")
return "Sorry, I had trouble processing that. Could you try again?"
async def _chat_openai_stream(self, user_message: str) -> AsyncGenerator[str, None]:
"""Stream chat via OpenAI API."""
# Add user message to history
self.conversation_history.append({
"role": "user",
"content": user_message,
})
# Build messages
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.conversation_history[-10:])
full_response = ""
try:
stream = await self._client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=500,
temperature=0.7,
stream=True,
)
async for chunk in stream:
if chunk.choices[0].delta.content:
text = chunk.choices[0].delta.content
full_response += text
yield text
# Add complete response to history
self.conversation_history.append({
"role": "assistant",
"content": full_response,
})
except Exception as e:
logger.error(f"OpenAI streaming error: {e}")
yield "Sorry, I had trouble processing that."
def clear_history(self):
"""Clear conversation history."""
self.conversation_history = []
src/server/main.py
+61 -24
View File
@@ -279,33 +279,70 @@ async def websocket_endpoint(websocket: WebSocket):
logger.info(f"Transcript: {transcript}")
if transcript.strip():
# Get AI response
logger.debug("Getting AI response...")
response_text = await backend.chat(transcript)
# Stream AI response with progressive TTS
logger.debug("Streaming AI response...")
# Send original text for display (with markdown)
full_response = ""
sentence_buffer = ""
audio_chunks = []
# Stream response and synthesize sentences as they complete
async for chunk in backend.chat_stream(transcript):
full_response += chunk
sentence_buffer += chunk
# Send text chunk for progressive display
await websocket.send_json({
"type": "response_chunk",
"text": chunk,
})
# Check for sentence boundaries
while any(sep in sentence_buffer for sep in ['. ', '! ', '? ', '.\n', '!\n', '?\n']):
# Find first sentence boundary
earliest_idx = len(sentence_buffer)
for sep in ['. ', '! ', '? ', '.\n', '!\n', '?\n']:
idx = sentence_buffer.find(sep)
if idx != -1 and idx < earliest_idx:
earliest_idx = idx + len(sep)
if earliest_idx < len(sentence_buffer):
sentence = sentence_buffer[:earliest_idx].strip()
sentence_buffer = sentence_buffer[earliest_idx:]
if sentence:
# Clean and synthesize this sentence
speech_text = clean_for_speech(sentence)
if speech_text:
logger.debug(f"Synthesizing: {speech_text[:50]}...")
async for audio_chunk in tts.synthesize_stream(speech_text):
audio_b64 = base64.b64encode(audio_chunk).decode()
await websocket.send_json({
"type": "audio_chunk",
"data": audio_b64,
"sample_rate": 24000,
})
else:
break
# Handle any remaining text
if sentence_buffer.strip():
speech_text = clean_for_speech(sentence_buffer.strip())
if speech_text:
async for audio_chunk in tts.synthesize_stream(speech_text):
audio_b64 = base64.b64encode(audio_chunk).decode()
await websocket.send_json({
"type": "audio_chunk",
"data": audio_b64,
"sample_rate": 24000,
})
# Signal end of response
await websocket.send_json({
"type": "response_text",
"text": response_text,
})
logger.info(f"Response: {response_text}")
# Clean text for TTS (remove markdown, hashtags, etc.)
speech_text = clean_for_speech(response_text)
logger.debug(f"Speech text: {speech_text}")
# Generate speech from cleaned text
logger.debug("Generating speech...")
audio_response = await tts.synthesize(speech_text)
# Send audio back
audio_b64 = base64.b64encode(audio_response.tobytes()).decode()
await websocket.send_json({
"type": "audio_response",
"data": audio_b64,
"sample_rate": 24000, # TTS output rate
"text": response_text, # Original for display
"type": "response_complete",
"text": full_response,
})
logger.info(f"Response complete: {full_response[:100]}...")
audio_buffer = []
await websocket.send_json({"type": "listening_stopped"})
src/server/tts.py
+26 -1
View File
@@ -4,7 +4,7 @@ Text-to-Speech module using ElevenLabs, Chatterbox, or fallbacks.
import asyncio
import os
from typing import Optional
from typing import Optional, AsyncGenerator
from pathlib import Path
import numpy as np
@@ -102,6 +102,31 @@ class ChatterboxTTS:
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, self._synthesize_sync, text)
async def synthesize_stream(self, text: str) -> AsyncGenerator[bytes, None]:
"""
Stream synthesized audio chunks.
Yields:
Raw PCM audio chunks (24kHz, 16-bit)
"""
if self._backend == "elevenlabs":
try:
# Use streaming API
audio_generator = self._elevenlabs_client.text_to_speech.convert(
voice_id=self.voice_id,
text=text,
model_id="eleven_turbo_v2_5",
output_format="pcm_24000",
)
for chunk in audio_generator:
yield chunk
except Exception as e:
logger.error(f"ElevenLabs streaming error: {e}")
else:
# Non-streaming fallback
audio = await self.synthesize(text)
yield audio.tobytes()
def _synthesize_sync(self, text: str) -> np.ndarray:
"""Synchronous synthesis."""
if self._backend == "elevenlabs":