Utilitaires de conversion audio en Python

Bibliothèque d’utilitaires audio

Pour construire une application Voice Agent robuste, vous avez besoin de fonctions utilitaires pour convertir l’audio entre les différents formats. Cette leçon rassemble les outils Python essentiels pour le traitement audio en temps réel.

Conversion PCM16 vers base64 et inversement

Les deux fonctions les plus utilisées : encoder l’audio capturé en base64 pour l’envoyer, et décoder l’audio reçu en base64 pour le jouer.

import base64
import numpy as np

def audio_to_base64(audio_data: np.ndarray) -> str:
    """Convertit un tableau numpy float32 [-1, 1] en base64 PCM16."""
    audio_int16 = (audio_data * 32767).astype(np.int16)
    return base64.b64encode(audio_int16.tobytes()).decode("utf-8")

def base64_to_audio(base64_audio: str) -> np.ndarray:
    """Convertit une chaîne base64 PCM16 en tableau numpy float32."""
    audio_bytes = base64.b64decode(base64_audio)
    audio_int16 = np.frombuffer(audio_bytes, dtype=np.int16)
    return audio_int16.astype(np.float32) / 32768.0

Utilisation dans le flux Voice Agent

import pyaudio
import json

# Capture micro → envoi
def capture_and_send(ws, stream, chunk_size=4096):
    raw_data = stream.read(chunk_size, exception_on_overflow=False)
    audio_array = np.frombuffer(raw_data, dtype=np.int16)
    audio_float = audio_array.astype(np.float32) / 32768.0
    b64 = audio_to_base64(audio_float)

    ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": b64
    }))

# Réception → lecture
def play_received(player_stream, base64_audio):
    audio_float = base64_to_audio(base64_audio)
    audio_int16 = (audio_float * 32767).astype(np.int16)
    player_stream.write(audio_int16.tobytes())

Rééchantillonnage (resampling)

Si votre microphone capture à une fréquence différente de celle attendue par l’API, vous devez rééchantillonner :

from scipy.signal import resample

def resample_audio(
    audio: np.ndarray,
    original_rate: int,
    target_rate: int
) -> np.ndarray:
    """Rééchantillonne l'audio d'une fréquence à une autre."""
    if original_rate == target_rate:
        return audio
    num_samples = int(len(audio) * target_rate / original_rate)
    return resample(audio, num_samples)

# Exemple : micro à 44100 Hz → API à 24000 Hz
audio_44100 = capture_microphone()
audio_24000 = resample_audio(audio_44100, 44100, 24000)
b64 = audio_to_base64(audio_24000)

Conversion entre formats G.711 et PCM

Pour les intégrations téléphoniques, vous devrez convertir entre G.711 et PCM linéaire. Attention à la version de Python : audioop a été retiré de la bibliothèque standard en 3.13 — sur un Python récent, installez le paquet audioop-lts, qui fournit la même interface :

import audioop

def pcm_to_mulaw(pcm_data: bytes) -> bytes:
    """Convertit PCM16 en G.711 mu-law."""
    return audioop.lin2ulaw(pcm_data, 2)

def mulaw_to_pcm(mulaw_data: bytes) -> bytes:
    """Convertit G.711 mu-law en PCM16."""
    return audioop.ulaw2lin(mulaw_data, 2)

def pcm_to_alaw(pcm_data: bytes) -> bytes:
    """Convertit PCM16 en G.711 A-law."""
    return audioop.lin2alaw(pcm_data, 2)

def alaw_to_pcm(alaw_data: bytes) -> bytes:
    """Convertit G.711 A-law en PCM16."""
    return audioop.alaw2lin(alaw_data, 2)

Relais téléphonique complet

Voici un exemple de relais entre un flux SIP G.711 et l’API Voice Agent :

async def relay_sip_to_voice_agent(sip_stream, ws):
    """Relaie l'audio d'un appel SIP vers l'API Voice Agent."""
    async for mulaw_chunk in sip_stream:
        # Convertir mu-law → PCM → base64
        pcm_data = mulaw_to_pcm(mulaw_chunk)
        b64 = base64.b64encode(pcm_data).decode("utf-8")

        await ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": b64
        }))

async def relay_voice_agent_to_sip(ws, sip_stream):
    """Relaie la réponse audio de l'API vers le flux SIP."""
    async for message in ws:
        event = json.loads(message)
        if event["type"] == "response.output_audio.delta":
            pcm_data = base64.b64decode(event["delta"])
            mulaw_data = pcm_to_mulaw(pcm_data)
            await sip_stream.send(mulaw_data)

Détection de silence côté client

Si vous gérez le push-to-talk ou si vous voulez une détection de silence locale avant d’envoyer, cette fonction mesure l’énergie du signal :

def is_silence(
    audio: np.ndarray,
    threshold: float = 0.01
) -> bool:
    """Détecte si un chunk audio est du silence."""
    rms = np.sqrt(np.mean(audio ** 2))
    return rms < threshold

# Usage
audio_chunk = capture_microphone()
if not is_silence(audio_chunk, threshold=0.02):
    b64 = audio_to_base64(audio_chunk)
    ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": b64
    }))

Sauvegarde et chargement de fichiers WAV

Pour le débogage et les tests, vous pouvez sauvegarder et charger des fichiers WAV :

import wave

def save_wav(
    filename: str,
    audio: np.ndarray,
    sample_rate: int = 24000
):
    """Sauvegarde un tableau numpy en fichier WAV."""
    audio_int16 = (audio * 32767).astype(np.int16)
    with wave.open(filename, "wb") as wf:
        wf.setnchannels(1)
        wf.setsampwidth(2)
        wf.setframerate(sample_rate)
        wf.writeframes(audio_int16.tobytes())

def load_wav(filename: str) -> tuple[np.ndarray, int]:
    """Charge un fichier WAV en tableau numpy float32."""
    with wave.open(filename, "rb") as wf:
        sample_rate = wf.getframerate()
        frames = wf.readframes(wf.getnframes())
        audio_int16 = np.frombuffer(frames, dtype=np.int16)
        return audio_int16.astype(np.float32) / 32768.0, sample_rate

Points clés à retenir

• audio_to_base64 et base64_to_audio sont les deux fonctions essentielles pour le flux Voice Agent
• Le rééchantillonnage est nécessaire quand la fréquence du microphone ne correspond pas à celle de l’API
• Le module audioop gère les conversions G.711 mu-law et A-law — retiré de la bibliothèque standard en Python 3.13, installez audioop-lts depuis PyPI pour les versions récentes
• La détection de silence côté client permet d’optimiser la bande passante en n’envoyant que l’audio utile
• Les utilitaires WAV facilitent le débogage en permettant de sauvegarder et rejouer les flux audio

Testez vos connaissances

WebSocket, VAD, audio : l’agent vocal de bout en bout.

1. Comment s'établit une session Voice Agent ?

Réponse : Connexion WebSocket temps réel, authentifiée par tokens éphémères (jamais la clé principale côté client), puis configuration via session.update — confirmée par session.created/updated.

2. Que gère la VAD ?

Réponse : La détection automatique de la parole : elle segmente quand l’utilisateur parle et s’arrête — l’agent sait quand écouter et quand répondre, sans bouton.

3. Comment l'audio circule-t-il ?

Réponse : L’entrée s’envoie par input_audio_buffer.append (avec validation et vidage du buffer), la réponse arrive en événements audio à jouer — formats PCM ou G.711 selon le canal.

4. Que permettent outils et MCP en conversation vocale ?

Réponse : Le function calling et les outils intégrés (web, X, collections) pendant l’échange vocal — l’agent consulte et agit en parlant, événements MCP compris.

5. Et pour la téléphonie ?

Réponse : SIP et LiveKit avec G.711 : l’agent vocal se branche sur le réseau téléphonique — les utilitaires Python de conversion audio font le pont entre formats.

Session sécurisée, VAD, buffers maîtrisés, outils en vocal : l’assistant téléphonique complet est à portée — le cours vous a donné chaque pièce.