curved-balloon-55303
12/26/2025, 2:04 PMtall-airline-16756
12/26/2025, 2:04 PMtall-airline-16756
12/26/2025, 2:04 PMtts_node in your Agent and applying audio DSP before publishing the frames.
What you can do
• Override tts_node to receive the synthesized rtc.AudioFrame stream.
• Modify each frame (example techniques):
◦ Muffling: apply a low-pass filter or attenuate high frequencies.
◦ Distortion: apply soft/hard clipping or drive + saturation.
◦ Phone/voicemail simulation: downsample, reduce bandwidth, add gentle band-pass around 300–3400 Hz, and add subtle noise or compression.
• Convert frame bytes to a numeric buffer (e.g., int16/float32), run your filter/process, then repackage into rtc.AudioFrame with original sample rate/channels.
Minimal Python outline (conceptual)
async def tts_node(self, text: AsyncIterable[str], model_settings: ModelSettings) -> AsyncIterable[rtc.AudioFrame]:
async for frame in Agent.default.tts_node(self, text, model_settings):
# convert frame.data to numpy array, apply filter/distortion, convert back
yield rtc.AudioFrame(data=modified_bytes, sample_rate=frame.sample_rate, num_channels=frame.num_channels, samples_per_channel=frame.samples_per_channel)
Libraries and techniques to use
• Simple: numpy + scipy.signal (low-pass, band-pass, filtering, resampling).
• Higher-level: pydub for quick effects or an audio DSP library for more control.
• Keep sample rate and channel count consistent with the original frame.
Performance tips
• Do CPU-light processing in the hot path; prefer efficient implementations (vectorized numpy, compiled DSP).
• If effects are costly, consider an async worker or separate processing node to avoid blocking real-time audio.
• Test with the actual sample rates/frames LiveKit uses to avoid artifacts.
If you want, tell me:
• Which SDK/language you’re using (Python, Node, Go, etc.),
• The exact effect you prefer (e.g., “telephone muffled”, “slight saturation”, “heavy distortion”),
and I’ll give a focused code example and recommended parameters.
Sources: Pipeline nodes and hooks | LiveKit docscurved-balloon-55303
12/26/2025, 2:09 PMtall-airline-16756
12/26/2025, 2:09 PMtts_node. I’ll explain each step and provide runnable Python-style pseudocode you can adapt.
Quick summary of the approach
• Intercept the TTS audio frames in tts_node.
• Convert frame bytes to numeric samples.
• Apply three simple DSP steps that create the “telephone” sound:
a. Band-pass filter (keeps ~300–3400 Hz).
b. Downsample / reduce bandwidth (optional mild resampling).
c. Add light compression and low-level noise for realism.
• Convert samples back to bytes and yield modified rtc.AudioFrame.
Before you start
• Confirm your TTS frames are PCM 16-bit (int16). If different, adjust types.
• Keep sample rate and channel count consistent when yielding frames.
Step-by-step explanation + example code 1) Import libs
• We use numpy for numeric work and scipy.signal for filters. These are efficient and easy for beginners.
import numpy as np
from scipy import signal
2) Helper: convert frame bytes -> int16 numpy array
• Audio frames arrive as interleaved PCM int16 bytes. We convert to a 1-D numpy array of int16, then to float32 in range [-1, 1] for DSP.
def bytes_to_float_samples(frame):
samples = np.frombuffer(frame.data, dtype=np.int16) # int16 array
samples = samples.astype(np.float32) / 32768.0 # normalize to [-1,1]
if frame.num_channels > 1:
samples = samples.reshape(-1, frame.num_channels) # shape: (frames, channels)
return samples
3) Helper: convert float samples -> bytes
• After processing, convert floats back to int16 bytes carefully (clipping to avoid overflow).
def float_samples_to_bytes(samples):
clipped = np.clip(samples, -1.0, 1.0)
int16 = (clipped * 32767.0).astype(np.int16)
return int16.tobytes()
4) Band-pass filter (the core “telephone” effect)
• Telephone sound focuses on ~300–3400 Hz. We create a band-pass filter using a 4th-order Butterworth filter. Filtering removes low rumble and very high frequencies that make speech crisp.
def make_bandpass(fs, low=300.0, high=3400.0, order=4):
nyq = 0.5 * fs
low_n = low / nyq
high_n = high / nyq
b, a = signal.butter(order, [low_n, high_n], btype='band')
return b, a
def apply_bandpass(samples, b, a):
# If stereo, apply filter to each channel separately
if samples.ndim == 2:
out = np.zeros_like(samples)
for ch in range(samples.shape[1]):
out[:, ch] = signal.lfilter(b, a, samples[:, ch])
return out
else:
return signal.lfilter(b, a, samples)
Why this helps: keeping 300–3400 Hz simulates the frequency range of phone lines, making audio sound muffled and narrower.
5) Optional: mild downsampling / bandwidth reduction
• Real phone codecs reduce sample rate or bitrate. You can simulate by decimating slightly (e.g., from 48k -> 16k) and then resampling back. This creates subtle aliasing and loss of high detail.
def downsample_then_upsample(samples, orig_fs, target_fs=16000):
if orig_fs == target_fs:
return samples
# simple approach using decimate (anti-aliasing)
factor = orig_fs // target_fs
if factor <= 1:
return samples
# If stereo, process channels separately
if samples.ndim == 2:
ch_out = []
for ch in range(samples.shape[1]):
dec = signal.decimate(samples[:, ch], factor, ftype='iir', zero_phase=True)
up = signal.resample(dec, len(samples[:, ch])) # resample back to original length
ch_out.append(up)
return np.stack(ch_out, axis=1)
else:
dec = signal.decimate(samples, factor, ftype='iir', zero_phase=True)
return signal.resample(dec, len(samples))
Note: decimation factor must be integer (e.g., 48000 -> 16000 factor 3). If your sample rate is 48000, using 16000 is common for phone-like sound.
6) Add gentle compression and low-level noise
• Compression reduces dynamic range so quieter parts are audible like a phone. Adding a small amount of noise (e.g., -40 dB) makes it sound less “studio-clean”.
def simple_compressor(samples, threshold=0.6, ratio=4.0):
# very simple soft-knee style: if |x|>threshold, reduce gain
gain = np.ones_like(samples)
over = np.abs(samples) > threshold
gain[over] = 1.0 / (1.0 + (ratio - 1.0) * (np.abs(samples[over]) - threshold))
return samples * gain
def add_noise(samples, noise_db=-40.0):
rms = np.sqrt(np.mean(samples**2))
noise_amp = rms * 10**(noise_db / 20.0)
noise = np.random.normal(0.0, noise_amp, samples.shape)
return samples + noise
7) Putting it together inside tts_node
• Get frames from the default TTS node, process and yield modified frames. Keep timing and frame sizes intact.
async def tts_node(self, text: AsyncIterable[str], model_settings: ModelSettings) -> AsyncIterable[rtc.AudioFrame]:
# Prepare once per call if sample rate known; otherwise inside loop per frame
async for frame in Agent.default.tts_node(self, text, model_settings):
fs = frame.sample_rate
samples = bytes_to_float_samples(frame) # shape: (N,) or (N, channels)
# 1) Band-pass
b, a = make_bandpass(fs, low=300.0, high=3400.0, order=4)
samples = apply_bandpass(samples, b, a)
# 2) Downsample/resample for that telephone codec flavor (optional)
samples = downsample_then_upsample(samples, orig_fs=fs, target_fs=16000)
# 3) Compression + noise
samples = simple_compressor(samples, threshold=0.6, ratio=3.0)
samples = add_noise(samples, noise_db=-38.0)
# Convert back to bytes and yield a new frame keeping original metadata
modified_bytes = float_samples_to_bytes(samples)
yield rtc.AudioFrame(
data=modified_bytes,
sample_rate=frame.sample_rate,
num_channels=frame.num_channels,
samples_per_channel=frame.samples_per_channel,
)
Practical tips & caveats
• CPU: real-time per-frame filtering can be CPU-heavy. Test performance; if needed, precompute filters and reuse, vectorize operations, or run processing in a separate worker.
• Latency: avoid heavy resampling per frame; try to operate on whole utterances if TTS gives longer buffers.
• Quality tuning: adjust band edges (300–3400), compressor threshold, and noise level to taste.
• Stereo vs mono: phone voice is typically mono. If you receive stereo, you can collapse to mono before processing: samples = samples.mean(axis=1).
If you’d like, I can:
• Provide a complete runnable example tailored to your SDK (Python environment or Node).
• Produce a simpler minimal example that only applies band-pass (lighter CPU).
• Suggest exact filter parameters and test audio files to audition changes.
Which would you prefer?