Comment créer un agent vocal IA en streaming ?

Résumer ce contenu avec :

Perplexity ChatGPT Claude Grok Mistral

Je crée un agent vocal IA fluide en arrêtant de penser STT puis LLM puis TTS. Le vrai sujet, c’est l’orchestration en streaming. On va parler latence, tours de parole, interruptions, appels d’outils et code Python testable sans API payante.

Pourquoi le simple STT LLM TTS bloque ?

Le simple chaînage STT, LLM, TTS bloque parce qu’il force chaque étape à attendre la précédente, et ça crée une latence beaucoup trop visible pour une vraie conversation.

Sur le papier, l’idée paraît propre. L’utilisateur parle. Le STT, c’est le speech-to-text, transforme sa voix en texte. Le LLM, le modèle de langage, génère une réponse. Puis le TTS, text-to-speech, transforme cette réponse en voix. C’est net sur un schéma. C’est rassurant. Mais en production, ça casse vite.

AI

Comment créer un agent IA local en streaming ?

août 30, 2026

Une conversation humaine n’est pas séquentielle. On hésite. On se coupe. On dit “attends non”. On reformule au milieu d’une phrase. On attend aussi une réaction presque immédiate, même petite. Un “oui”, un silence actif, un début de réponse. Si l’agent attend toute la phrase, puis toute la transcription, puis toute la réponse du modèle, puis toute la synthèse vocale, on n’est plus dans une conversation. On est dans un talkie-walkie lent.

Le vrai sujet, c’est l’orchestration. L’agent doit gérer plusieurs choses en même temps :

• Le flux audio entrant, en continu.
• Les transcriptions partielles, même imparfaites.
• La détection de fin de tour, c’est-à-dire savoir quand l’utilisateur a vraiment fini de parler.
• La génération du LLM en streaming, token par token.
• La synthèse vocale progressive, sans attendre toute la réponse.
• Les interruptions utilisateur, ce qu’on appelle aussi le barge-in.
• Les appels d’outils, par exemple chercher une info dans un CRM ou déclencher une action.

Chez des clients, j’ai rarement vu le modèle IA être le problème principal. Le vrai irritant, c’était plutôt le délai entre la fin de la phrase utilisateur et le début de réponse. Ou pire, l’agent qui continue à parler alors que l’utilisateur vient de l’interrompre. Ça, ça tue l’expérience instantanément.

AI

Comment utiliser LanceDB pour une base vectorielle IA ?

août 30, 2026

Pour avoir un repère simple, une latence naturelle entre deux humains tourne autour de 200 à 300 ms. Au-delà de 500 ms, l’échange commence déjà à sembler lent. Au-delà de 3 secondes, on perd clairement l’utilisateur. Et aujourd’hui, un système peut avoir un time-to-first-token, donc le délai avant le premier morceau de réponse du LLM, entre 0,8 et 3 secondes. L’architecture devient donc déterminante.

Le fil logique est simple : on va quitter la chaîne séquentielle STT puis LLM puis TTS, passer à une architecture streaming, et construire les composants un par un avec du Python simple.

Pourquoi le streaming change tout ?

Le streaming change tout parce qu’il permet à chaque composant de commencer à travailler avec des morceaux incomplets, au lieu d’attendre un résultat final. C’est ça le vrai déclic. On ne passe plus un gros bloc d’une étape à l’autre, on fait circuler un flux vivant.

AI

Comment réussir l'observabilité IA des agents en production ?

août 30, 2026

Dans une architecture séquentielle, le chemin est très propre. On enregistre ou on reçoit l’audio complet. On attend la transcription finale avec le STT, le speech-to-text, donc la conversion voix vers texte. On envoie ensuite tout le texte au LLM, le modèle de langage. On attend sa réponse complète. Puis seulement là, on lance le TTS, le text-to-speech, donc la génération audio.

C’est simple à coder. C’est facile à déboguer. Je l’utilise encore pour des prototypes ou des traitements non interactifs. Mais dès qu’on veut une sensation conversationnelle, c’est trop lent. L’utilisateur parle, puis il attend. L’agent réfléchit, puis il attend encore pour parler. Ça casse le rythme.

Avec le streaming, le STT envoie des transcriptions partielles. Le LLM peut commencer à générer dès que l’intention est assez claire. Le TTS peut produire de l’audio dès les premiers tokens utiles, un token étant un petit morceau de texte manipulé par le modèle. Là, on se rapproche d’une vraie conversation.

AI

Comment créer un workflow Hermes Agent IA LLM à faible coût?

août 30, 2026

Le prix à payer, c’est plus de discipline. Il faut gérer des buffers, donc des zones temporaires où on stocke les morceaux. Il faut savoir annuler une réponse en cours. Il faut prioriser les événements. Il faut aussi gérer le barge-in, c’est-à-dire le moment où l’utilisateur reprend la parole pendant que l’agent parle. Et croyez-moi, en prod, ça arrive tout le temps.

import asyncio

async def sequential_agent(audio):
    # Chaque étape attend la précédente
    transcript = await stt_final(audio)
    answer = await llm_complete(transcript)
    speech = await tts_complete(answer)
    return speech

async def streaming_agent(audio_chunks):
    # Chaque étape travaille dès qu'un morceau utile arrive
    async for partial_text in stt_stream(audio_chunks):
        async for token in llm_stream(partial_text):
            async for audio_chunk in tts_stream(token):
                yield audio_chunk

Ce code est volontairement incomplet. Les fonctions stt_stream, llm_stream et tts_stream sont simplifiées pour montrer l’idée, pas pour couvrir tous les cas réels. Mais il montre bien le changement mental. On ne manipule plus des blocs. On orchestre des flux.

Comment construire le STT streaming ?

Je construis le STT streaming comme un composant événementiel. Il reçoit de petits paquets audio, souvent autour de 50 ms, et il publie des transcriptions partielles puis finales. Le STT, pour “speech-to-text”, c’est simplement la brique qui transforme la voix en texte. Dans un agent vocal IA, c’est le premier maillon sérieux de la chaîne.

En production, ce composant doit écouter en continu, garder une connexion persistante, souvent via WebSocket, c’est-à-dire un canal ouvert dans les deux sens entre votre app et le service STT. Il reçoit l’audio au fil de l’eau, puis renvoie des événements.

• PARTIAL veut dire “hypothèse temporaire”. C’est utile pour afficher du texte vite, détecter une intention probable, ou préparer la suite. Mais ça peut changer.
• FINAL veut dire “transcription stabilisée”. Là, je peux commencer à déclencher des actions plus engageantes.

Le piège classique, c’est les entités. Un nom propre, un numéro de commande ou une référence client mal transcrite peut casser l’appel d’outil juste après. Dans mes projets, j’isole toujours les entités sensibles. Et si ça touche à une commande, un compte, un paiement ou une action irréversible, je préfère demander confirmation avant d’appeler un outil. Ça évite des erreurs bêtes, mais coûteuses.

from __future__ import annotations

import asyncio
import time
from dataclasses import dataclass
from enum import Enum
from typing import AsyncIterator, Iterable


class TranscriptEventType(str, Enum):
    PARTIAL = "partial"
    FINAL = "final"


@dataclass
class TranscriptEvent:
    text: str
    event_type: TranscriptEventType
    confidence: float
    timestamp: float


class MockStreamingSTT:
    """Simule un STT streaming.

    En production, cette classe serait remplacée par une connexion WebSocket
    vers un service de speech-to-text. Ici, je veux juste montrer les événements.
    """

    def __init__(self, scripted_transcript: str) -> None:
        self.scripted_transcript = scripted_transcript

    async def stream(self, audio_chunks: Iterable[bytes]) -> AsyncIterator[TranscriptEvent]:
        words = self.scripted_transcript.split()
        current_words: list[str] = []

        for index, _chunk in enumerate(audio_chunks):
            await asyncio.sleep(0.05)  # Simule un paquet audio d'environ 50 ms

            if index < len(words):
                current_words.append(words[index])

                # Un événement PARTIAL peut changer ensuite
                yield TranscriptEvent(
                    text=" ".join(current_words),
                    event_type=TranscriptEventType.PARTIAL,
                    confidence=0.65 + min(index * 0.03, 0.25),
                    timestamp=time.time(),
                )

        # Le FINAL représente la transcription stabilisée
        yield TranscriptEvent(
            text=self.scripted_transcript,
            event_type=TranscriptEventType.FINAL,
            confidence=0.94,
            timestamp=time.time(),
        )


def fake_audio_chunks(count: int) -> list[bytes]:
    # Chaque chunk représente un petit morceau d'audio brut simulé
    return [b"audio-chunk" for _ in range(count)]


async def main() -> None:
    stt = MockStreamingSTT("je veux suivre la commande 4582")

    async for event in stt.stream(fake_audio_chunks(8)):
        print(f"{event.event_type.value.upper()} | {event.confidence:.2f} | {event.text}")


if __name__ == "__main__":
    asyncio.run(main())

La sortie montre plusieurs lignes PARTIAL qui grandissent au fur et à mesure, puis une ligne FINAL avec la phrase complète. C’est exactement le comportement que je veux tester ici, sans micro, sans API payante, sans vraie reconnaissance vocale.

Le point important, c’est de ne pas appeler trop tôt un outil sur un PARTIAL. Si le modèle entend d’abord “commande 458” puis corrige en “commande 4582”, votre outil peut partir chercher la mauvaise commande. J’utilise les PARTIAL pour préparer l’interface et anticiper. J’utilise les FINAL, ou une confirmation explicite, pour les actions qui comptent vraiment.

Comment gérer les tours de parole ?

Je gère les tours de parole en combinant plusieurs signaux : la fin de phrase probable, les silences, la stabilité du STT et le contexte. Pas avec un simple timer. Le STT, c’est le Speech-To-Text, donc le service qui transforme la voix en texte.

C’est un des points les plus sensibles d’un agent vocal IA en streaming. Si l’agent répond trop tôt, il coupe l’utilisateur. C’est insupportable. S’il attend trop longtemps, il donne l’impression d’être lent, même si le modèle répond vite derrière. C’est exactement là que l’orchestration devient plus importante que le modèle.

En pratique, je regarde plusieurs signaux en même temps :

• Le silence audio, parce qu’une vraie pause peut indiquer une fin de tour.
• La ponctuation probable, par exemple une phrase qui ressemble à une question complète.
• La stabilité entre plusieurs PARTIAL, c’est-à-dire plusieurs transcriptions temporaires qui ne changent plus.
• L’événement FINAL du STT, quand le moteur estime que l’énoncé est terminé.
• Les mots de liaison, comme “et”, “mais”, “parce que”, qui indiquent souvent que la personne n’a pas fini.
• Le contexte métier, parce que “je veux modifier ma réservation…” ne doit pas partir trop vite si l’utilisateur ajoute “…pour demain”.

Voici une version volontairement simple, sans micro et sans dépendance externe. Je simule juste des événements de transcription et des délais.

import asyncio
import time
from dataclasses import dataclass
from enum import Enum


class TranscriptEventType(str, Enum):
    PARTIAL = "partial"
    FINAL = "final"


@dataclass
class TranscriptEvent:
    text: str
    event_type: TranscriptEventType
    confidence: float
    timestamp: float


class TurnDetector:
    def __init__(self, stability_delay: float = 0.6) -> None:
        self.stability_delay = stability_delay
        self.last_text = ""
        self.last_change_at = 0.0

    def update(self, event: TranscriptEvent) -> bool:
        # Un FINAL est le signal le plus simple pour dire que le tour est fini
        if event.event_type == TranscriptEventType.FINAL:
            return True

        # Si le texte change, on réinitialise le délai de stabilité
        if event.text != self.last_text:
            self.last_text = event.text
            self.last_change_at = event.timestamp
            return False

        # Si le texte ne bouge plus, on peut considérer que l'utilisateur a fini
        stable_for = event.timestamp - self.last_change_at
        return stable_for >= self.stability_delay and event.confidence >= 0.85


async def demo() -> None:
    detector = TurnDetector(stability_delay=0.5)
    events = [
        TranscriptEvent("je veux", TranscriptEventType.PARTIAL, 0.72, time.time()),
        TranscriptEvent("je veux suivre", TranscriptEventType.PARTIAL, 0.80, time.time() + 0.2),
        TranscriptEvent("je veux suivre ma commande", TranscriptEventType.PARTIAL, 0.88, time.time() + 0.8),
        TranscriptEvent("je veux suivre ma commande", TranscriptEventType.FINAL, 0.95, time.time() + 1.0),
    ]

    for event in events:
        await asyncio.sleep(0.1)
        is_done = detector.update(event)
        print(event.text, "=> tour terminé ?", is_done)


if __name__ == "__main__":
    asyncio.run(demo())

Ce détecteur est simplifié, mais il montre bien l’idée. En production, je combine souvent plusieurs signaux, parce qu’un utilisateur peut faire une pause au milieu d’une phrase. Et c’est là que les faux positifs coûtent cher : l’agent parle, l’utilisateur continue, et toute l’expérience devient bancale.

Comment orchestrer LLM TTS et outils ?

J’orchestre le LLM, le TTS et les outils avec une boucle centrale qui sait streamer, annuler, bufferiser et reprendre proprement. C’est vraiment le cœur de l’agent vocal. Pas le modèle. Pas la voix. La boucle.

Dès que le tour utilisateur est considéré comme terminé, le texte part vers le LLM en streaming. Le LLM renvoie des tokens, c’est-à-dire des petits morceaux de texte, parfois un mot, parfois un bout de phrase. Le TTS, la synthèse vocale, n’a pas besoin d’attendre la réponse complète. Il peut commencer dès qu’il reçoit un segment assez stable. C’est souvent ça qui donne l’impression que l’agent est rapide.

Le point sensible, c’est l’interruption. Si l’utilisateur reparle pendant que l’agent parle, il faut stopper le TTS, annuler ou suspendre le LLM, vider les buffers audio, puis donner la priorité au nouvel input. On appelle ça le barge-in. En clair, l’utilisateur coupe la parole à l’agent, et l’agent doit l’accepter sans se battre avec lui.

Les outils ajoutent une autre couche. L’agent peut chercher une commande, vérifier une disponibilité, déclencher une action. Mais je n’appelle jamais un outil critique sur une transcription partielle non confirmée. Pour une adresse, un montant, une annulation, je préfère demander confirmation. Les appels doivent être traçables, idempotents si possible, c’est-à-dire répétables sans double effet, et toujours avec timeout. Un outil lent casse l’expérience vocale.

import asyncio
import re
from typing import AsyncIterator


async def mock_llm_stream(prompt: str) -> AsyncIterator[str]:
    # Simule une réponse LLM token par token
    for token in prompt.split():
        await asyncio.sleep(0.08)
        yield token + " "


async def mock_tts_stream(text_stream: AsyncIterator[str], interrupted: asyncio.Event) -> AsyncIterator[bytes]:
    # Simule un TTS qui transforme chaque morceau de texte en audio
    async for token in text_stream:
        if interrupted.is_set():
            print("TTS stoppé car l'utilisateur a repris la parole")
            return
        await asyncio.sleep(0.05)
        yield f"AUDIO({token})".encode()


async def get_order_status(order_id: str) -> str:
    # Simule un appel d'outil métier
    await asyncio.sleep(0.2)
    return f"La commande {order_id} est en cours de livraison"


def extract_order_id(text: str) -> str | None:
    match = re.search(r"\b\d{3,}\b", text)
    return match.group(0) if match else None


async def agent_reply(user_text: str, interrupted: asyncio.Event) -> None:
    order_id = extract_order_id(user_text)

    if "commande" in user_text.lower() and order_id:
        tool_result = await get_order_status(order_id)
        response_text = f"{tool_result}. Vous devriez la recevoir bientôt."
    else:
        response_text = "J'ai bien compris votre demande, je vais vous aider."

    async def response_stream() -> AsyncIterator[str]:
        async for token in mock_llm_stream(response_text):
            if interrupted.is_set():
                return
            yield token

    async for audio in mock_tts_stream(response_stream(), interrupted):
        print(audio)


async def simulate_user_interrupt(interrupted: asyncio.Event) -> None:
    await asyncio.sleep(0.45)
    interrupted.set()


async def main() -> None:
    interrupted = asyncio.Event()

    # Lance une interruption simulée pendant que l'agent parle
    asyncio.create_task(simulate_user_interrupt(interrupted))

    await agent_reply("je veux suivre la commande 4582", interrupted)


if __name__ == "__main__":
    asyncio.run(main())

Ici, l’agent détecte une demande de suivi, extrait le numéro 4582, appelle l’outil, puis commence à streamer la réponse vers le TTS. Après 0,45 seconde, l’interruption est déclenchée. Le TTS s’arrête. Dans une vraie architecture, ce signal vient d’une détection d’activité vocale, ou des événements entrants du STT, la transcription speech-to-text.

Et maintenant vous le construisez comment ?

Un agent vocal IA crédible ne se résume pas à coller un STT, un LLM et un TTS. Cette chaîne marche pour une démo, rarement pour une vraie conversation. Le point dur, c’est l’orchestration : streamer tôt, décider quand l’utilisateur a fini, lancer le LLM sans attendre inutilement, parler progressivement, interrompre proprement, appeler les bons outils au bon moment. Je retiens surtout ça : la fluidité vient moins du modèle que de l’architecture autour. Si vous construisez avec cette logique, vous gagnez une expérience plus naturelle, plus rapide et beaucoup moins frustrante pour vos utilisateurs.

FAQ

• Qu’est-ce qu’un agent vocal IA en streaming ?
  C’est un agent qui traite la voix en continu au lieu d’attendre la fin de chaque étape. Le STT produit des transcriptions partielles, le LLM génère progressivement, puis le TTS commence à parler dès que possible. C’est cette logique qui rend l’échange plus naturel.
• Pourquoi le pattern séquentiel est trop lent ?
  Parce qu’il empile les attentes. On attend la fin de l’audio, puis la transcription finale, puis la réponse complète du LLM, puis l’audio complet du TTS. Dès qu’on dépasse quelques centaines de millisecondes, l’utilisateur sent le délai. Au-delà de plusieurs secondes, il décroche.
• À quoi servent les transcriptions PARTIAL et FINAL ?
  Une transcription PARTIAL sert à réagir vite, mais elle peut encore changer. Une transcription FINAL est plus stable et plus fiable. Pour un appel d’outil sensible, comme retrouver une commande avec un numéro, je préfère attendre un FINAL ou demander confirmation.
• Comment éviter qu’un agent vocal coupe l’utilisateur ?
  Il faut une vraie détection de fin de tour. Pas juste un timer. Je combine les silences, la stabilité du texte, les événements finaux du STT et le contexte de la phrase. Sinon l’agent répond trop tôt ou attend trop longtemps, et dans les deux cas l’expérience devient pénible.
• Comment gérer une interruption pendant que l’agent parle ?
  Il faut détecter que l’utilisateur reprend la parole, couper le TTS, annuler ou suspendre le streaming LLM, vider les buffers audio et redonner la priorité à la nouvelle demande. C’est ce qu’on appelle souvent le barge-in. Sans ça, l’agent donne l’impression de ne pas écouter.

A propos de l’auteur

Je suis Franck Scandolera, expert et formateur en Tracking avancé server-side, Analytics Engineering, automatisation No/Low Code avec n8n, intégration de l’IA en entreprise et SEO/GEO. J’accompagne des équipes qui veulent passer des prototypes IA aux vrais systèmes utilisables en production, avec des sujets très concrets comme les flux de données, les appels d’outils, l’orchestration et la mesure de performance. J’ai travaillé avec des références comme Logis Hôtel, Yelloh Village, BazarChic, la Fédération Française de Football ou Texdecor. Je dirige l’agence webAnalyste et l’organisme Formations Analytics. Si vous voulez structurer vos projets IA ou automatisation, contactez-moi.

Franck Scandolera

⭐ Data Analyst, Analytics Engineer et expert dans l’automatisation IA  ⭐

Ref clients : Logis Hôtel, Yelloh Village, BazarChic, Fédération Football Français, Texdecor…

Mon terrain de jeu :

Data Analyst & Analytics engineering : tracking propre RGPD, entrepôt de données (GTM server, BigQuery…), modèles (dbt/Dataform), dashboards décisionnels (Looker, SQL, Python).
Automatisation IA des taches Data, Marketing, RH, compta etc : conception de workflows intelligents robustes (n8n, Make, App Script, scraping) connectés aux API de vos outils et LLM (OpenAI, Mistral, Claude…).
Engineering IA pour créer des applications et agent IA sur mesure : intégration de LLM (OpenAI, Mistral…), RAG, assistants métier, génération de documents complexes, APIs, backends Node.js/Python.