min_speech_duration

turn_detection="vad"

turn_detection="stt"

BVCTelephony()

# Exemple Python LiveKit — téléphonie optimisée (adaptable)
import os
import asyncio
from dotenv import load_dotenv

# Importez les modules LiveKit/Agents selon votre SDK installé
from livekit import rtc
from livekit.agents import AgentServer, AgentSession, JobContext
from livekit.plugins import silero
from livekit.plugins.noise_cancellation import BVCTelephony, BVC

load_dotenv()

# Paramètres de configuration modifiables
AUDIO_GAIN_DB = 9  # +9 dB de gain initial à tester (ajustez entre 6-12 dB)
USE_BVCTELEPHONY_FOR_SIP = True
DEEPGRAM_MODEL = "deepgram/nova-2-phonecall"  # modèle optimisé pour appels téléphoniques

# Fonction utilitaire pour convertir dB en facteur linéaire
def db_to_linear(db):
    return 10 ** (db / 20.0)

# Amplification simple d'un buffer PCM 16-bit signed (little-endian)
def amplify_pcm16(pcm_bytes: bytes, gain_db: float) -> bytes:
    import struct
    factor = db_to_linear(gain_db)
    out = bytearray()
    # chaque sample 16-bit
    for i in range(0, len(pcm_bytes), 2):
        sample = struct.unpack_from("<h", pcm_bytes, i)[0]
        amplified = int(sample * factor)
        # clip sur 16-bit
        if amplified > 32767:
            amplified = 32767
        elif amplified < -32768:
            amplified = -32768
        out += struct.pack("<h", amplified)
    return bytes(out)

# Précharge VAD pour réduire latence
def prewarm(proc):
    proc.userdata["vad"] = silero.VAD.load()

server = AgentServer()
server.setup_fnc = prewarm

@server.rtc_session(agent_name="telephony-agent")
async def agent_entry(ctx: JobContext):
    # Attendre le participant pour déterminer son type
    participant = await ctx.wait_for_participant()

    # Choix du noise cancellation selon le type participant (SIP vs WebRTC)
    if participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP and USE_BVCTELEPHONY_FOR_SIP:
        noise_filter = BVCTelephony()
    else:
        noise_filter = BVC()  # Optionnel pour WebRTC, ou None

    # Récupérer le VAD préchargé
    vad_model = ctx.proc.userdata.get("vad")

    # Construire la session agent avec STT optimisé pour la téléphonie
    session = AgentSession(
        stt={
            "provider": "deepgram",
            "model": DEEPGRAM_MODEL,
            # autres options STT : language, diarization, streaming config...
        },
        vad=vad_model,
        turn_detection="vad",
    )

    # Démarrer la session en fournissant les options audio (noise_cancellation)
    await session.start(
        room=ctx.room,
        agent=None,  # remplacez par votre instance agent si nécessaire
        room_options={
            "audio_input": {
                "noise_cancellation": noise_filter
            },
        },
    )

    # Exemple d'écoute d'un flux audio entrant et amplification avant traitement VAD/STT
    # NOTE: l'API exacte pour intercepter frames audio dépend de votre SDK LiveKit spécifique.
    async for audio_frame in session.receive_audio_frames():  # pseudo-API illustrative
        # audio_frame.data doit être PCM16 bytes (adapter si format différent)
        amplified = amplify_pcm16(audio_frame.data, AUDIO_GAIN_DB)
        audio_frame.data = amplified

        # envoyez la frame amplifiée à la pipeline de VAD/STT (pseudo-API)
        await session.process_audio_frame(audio_frame)

    # Fermer proprement
    await session.stop()

if __name__ == "__main__":
    # Lance le serveur agent (CLI utilitaire fourni par le SDK)
    server.run()

min_speech_duration

turn_detection="stt"

proc.userdata["vad"]

prewarm(proc)

silero.VAD.load()

proc.userdata

ctx.proc.userdata["vad"]