Script EAGI Python Complet pour Agent Vocal IA — Code Open-Source

Introduction : Pourquoi EAGI est supérieur à AGI pour les agents vocaux IA

Dans l'univers de la téléphonie sur IP et des serveurs vocaux interactifs (SVI), Asterisk reste un pilier incontournable. Son interface de passerelle, AGI (Asterisk Gateway Interface), a longtemps permis aux développeurs d'étendre ses fonctionnalités avec des scripts externes. Cependant, l'avènement des agents vocaux IA conversationnels, qui exigent une interaction fluide et en temps réel, a mis en lumière les limites fondamentales de l'AGI standard.

L'AGI fonctionne sur un modèle synchrone de requête/réponse. Un script AGI envoie une commande (par exemple, STREAM FILE welcome) et attend qu'elle soit terminée avant de pouvoir continuer. Il ne peut pas écouter l'utilisateur pendant qu'il parle, ce qui rend impossible une fonctionnalité essentielle : le "barge-in", ou la capacité pour l'utilisateur d'interrompre l'agent. Cette limitation crée des conversations robotiques et frustrantes, à l'opposé de l'expérience naturelle que nous cherchons à construire.

C'est ici qu'intervient l'EAGI (Enhanced Asterisk Gateway Interface). La différence fondamentale, et révolutionnaire, est l'ajout d'un canal de communication audio bidirectionnel. En plus des flux `stdin` et `stdout` pour les commandes, EAGI ouvre un troisième descripteur de fichier (file descriptor 3) qui est directement connecté au flux audio de l'appel. Cela signifie qu'un script EAGI Python pour agent vocal IA peut simultanément :

• Lire l'audio de l'appelant en temps réel.
• Envoyer de l'audio généré (la voix de l'IA) pour être joué à l'appelant.

Cette capacité de duplex intégral est la clé pour construire des agents vocaux réactifs, capables de gérer les interruptions, de réduire la latence perçue et d'offrir une expérience conversationnelle véritablement humaine. Ce guide vous fournira un code open-source complet et détaillé pour construire un tel agent, en s'appuyant sur la puissance de l'EAGI et d'un écosystème d'IA de pointe (Whisper, LLM backend, mixael-TTS).

Architecture du script eagi_conversation.py

Avant de plonger dans le code, il est crucial de comprendre l'architecture de notre agent vocal. Le script eagi_conversation.py agit comme un orchestrateur central qui connecte Asterisk à divers services d'intelligence artificielle. L'ensemble du processus est conçu pour être aussi "streamé" que possible afin de minimiser la latence.

Voici le flux de données et de contrôle, étape par étape :

1. Initiation par Asterisk : Un appel arrive sur votre serveur Asterisk. Le plan de numérotation (extensions.conf) exécute notre eagi_conversation.py via la commande EAGI().
2. Connexion EAGI : Le script Python s'initialise, établit la communication avec Asterisk via `stdin`/`stdout` et, surtout, ouvre le descripteur de fichier 3 pour l'audio.
3. Phase d'Écoute (STT) : Le script lit en continu le flux audio (PCM 8kHz, 16-bit signed-linear) provenant de l'appelant via le descripteur 3. Il utilise un simple algorithme de détection d'activité vocale (VAD) pour savoir quand l'utilisateur commence à parler et quand il a fini.
4. Transcription : Une fois qu'un segment de parole est capturé, il est envoyé à une API de Speech-to-Text (STT), comme l'API de OpenAI Whisper ou une instance locale via des projets comme `whisper.cpp`. L'API renvoie la transcription textuelle.
5. Génération de Réponse (LLM) : Le texte de l'utilisateur est envoyé à un grand modèle de langage (LLM). Nous utiliserons LLM backend pour faire tourner localement des modèles comme Llama 3 ou Mistral. L'API d'LLM backend est configurée pour streamer la réponse, c'est-à-dire renvoyer les mots (tokens) dès qu'ils sont générés, sans attendre la fin de la phrase.
6. Synthèse Vocale (TTS) : Les tokens du LLM sont regroupés en phrases complètes. Chaque phrase est ensuite envoyée à une API de Text-to-Speech (TTS) qui supporte également le streaming, comme Coqui mixael-TTS. Cette API commence à renvoyer les premiers morceaux de l'audio synthétisé (chunks) avant même d'avoir reçu la fin de la phrase.
7. Phase de Parole (Playback) : Les chunks audio PCM reçus de l'API TTS sont immédiatement écrits sur le descripteur de fichier 3, ce qui les envoie directement à l'appelant.
8. Gestion du Barge-in : Pendant toute la phase de parole (étape 7), un thread séparé continue d'écouter l'audio de l'appelant (étape 3). Si l'utilisateur parle par-dessus l'IA, le script le détecte, arrête immédiatement la lecture de l'audio synthétisé et retourne à la phase d'écoute, créant une interruption naturelle.

Cette architecture en pipeline et en streaming est la seule façon d'atteindre des latences de "premier son" inférieures à 500ms, seuil psychologique pour une conversation fluide. Un agent vocal IA code source bien conçu doit impérativement intégrer ces principes.

Code Python Complet et Commenté : Le Cœur de l'Agent Vocal

Nous allons maintenant construire le fichier eagi_conversation.py. Le code ci-dessous est présenté sous forme de pseudo-code structuré et commenté. Il est conçu pour être lisible et pédagogique, tout en contenant toute la logique nécessaire pour un déploiement fonctionnel. Vous pouvez l'adapter et le copier-coller pour démarrer votre projet.

Prérequis et Import des Bibliothèques

Assurez-vous d'avoir installé les bibliothèques nécessaires. Un simple `pip install requests` suffira pour la communication API. Les autres sont des modules standards de Python.

# eagi_conversation.py - Script EAGI Python pour agent vocal IA
# -*- coding: utf-8 -*-

import sys
import os
import requests
import struct
import array
import threading
import time
import logging
from collections import deque

# Ce script est un exemple complet pour un agent vocal IA utilisant EAGI avec Asterisk.
# Il gère la conversation en temps réel, incluant le barge-in.

Constantes de Configuration Essentielles

Centraliser la configuration est une bonne pratique. Cela facilite l'ajustement des paramètres sans avoir à fouiller dans le code. Ces valeurs sont des points de départ raisonnables.

# --- Configuration ---
LOG_FILE = "/tmp/eagi_debug.log"

# Paramètres audio
SAMPLE_RATE = 8000  # Asterisk utilise 8000Hz pour les appels standards
CHUNK_SIZE = 160    # 20ms de données audio (8000 * 0.020)
AUDIO_FORMAT = 'h'  # PCM 16-bit signed-linear (short integer)

# Paramètres VAD (Voice Activity Detection)
SILENCE_THRESHOLD = 250  # Seuil RMS pour détecter la parole. À ajuster selon le micro.
SILENCE_CHUNKS_NEEDED = 50 # 50 chunks de 20ms = 1 seconde de silence pour terminer l'enregistrement

# Paramètres Barge-in
BARGE_IN_THRESHOLD = 350 # Seuil RMS pour détecter une interruption. Légèrement plus haut que le silence.

# Endpoints des API IA (remplacez par vos URLs)
WHISPER_API_URL = "http://127.0.0.1:8080/inference" # URL d'une API compatible Whisper
OLLAMA_API_URL = "http://127.0.0.1:11434/api/generate" # URL de l'API LLM backend
OLLAMA_MODEL = "llama3:8b-instruct" # Modèle à utiliser
mixael-TTS_API_URL = "http://127.0.0.1:8020/tts_stream" # URL d'une API de streaming TTS (ex: mixael-TTS-api-server)

# Buffers
PLAY_BUFFER_SIZE = 28800 # Buffer de lecture pour lisser le streaming TTS (environ 1.8s d'audio à 16-bit)
# Note: La valeur de 28800 est grande. Elle peut servir de buffer max pour l'enregistrement
# avant envoi à Whisper, ou de buffer de pré-chargement pour le TTS. Pour une latence minimale
# au démarrage, un buffer de lecture plus petit (ex: 3200) est souvent préférable.

La Classe EAGIConversation : Orchestration de la Conversation

Nous encapsulons toute la logique dans une classe pour une meilleure organisation et gestion de l'état (par exemple, les variables de l'appel).

class EAGIConversation:
    """
    Gère une session de conversation complète via EAGI.
    Ce EAGI Python script est le cœur de notre agent vocal IA.
    """

Méthode __init__() : Initialisation et Connexion

Le constructeur prépare l'environnement : lecture des variables AGI, configuration du logging et, surtout, ouverture des descripteurs de fichiers pour la communication avec Asterisk.

    def __init__(self, agi_vars):
        self.agi_vars = agi_vars
        self.caller_id = agi_vars.get('agi_callerid', 'unknown')
        
        # Configuration du logging
        logging.basicConfig(filename=LOG_FILE, level=logging.INFO,
                            format='%(asctime)s - %(levelname)s - %(message)s')
        self.log(f"Nouvelle conversation EAGI initiée pour {self.caller_id}")

        # Descripteurs de fichiers pour EAGI
        self.stdin = sys.stdin
        self.stdout = sys.stdout
        # Le descripteur de fichier 3 est la clé de l'EAGI pour l'audio
        self.audio_fd = 3 
        
        # Événement pour gérer le barge-in
        self.barge_in_event = threading.Event()

Méthode send_command(cmd) : Communication avec Asterisk

Une fonction utilitaire pour envoyer des commandes AGI (comme `ANSWER` ou `HANGUP`) et lire la réponse d'Asterisk.

    def send_command(self, cmd):
        self.log(f"CMD > {cmd}")
        self.stdout.write(cmd + '\n')
        self.stdout.flush()
        response = self.stdin.readline().strip()
        self.log(f"RES < {response}")
        return response

Méthode record_speech() : Capture Audio en Temps Réel

Cette méthode lit l'audio depuis le canal EAGI, détecte la parole, et retourne les données audio une fois que l'utilisateur a fini de parler.

    def record_speech(self):
        self.log("Phase d'écoute : en attente de la parole de l'utilisateur.")
        speech_started = False
        silent_chunks = 0
        audio_data = bytearray()

        while True:
            try:
                # Lire un chunk d'audio depuis Asterisk (fd 3)
                chunk = os.read(self.audio_fd, CHUNK_SIZE * 2) # *2 car 16-bit
                if not chunk:
                    break

                # Détecter si l'utilisateur parle
                is_silent = self.detect_silence(chunk)

                if not speech_started and not is_silent:
                    self.log("Parole détectée, début de l'enregistrement.")
                    speech_started = True
                    audio_data.extend(chunk)
                elif speech_started:
                    audio_data.extend(chunk)
                    if is_silent:
                        silent_chunks += 1
                    else:
                        silent_chunks = 0 # Reset si la parole reprend
                    
                    if silent_chunks >= SILENCE_CHUNKS_NEEDED:
                        self.log(f"Silence détecté pendant {SILENCE_CHUNKS_NEEDED * 20}ms. Fin de l'enregistrement.")
                        return audio_data

            except Exception as e:
                self.log(f"Erreur pendant l'enregistrement : {e}")
                return None
        return audio_data

Méthode detect_silence(chunk) : Voice Activity Detection (VAD)

Un VAD simple mais efficace basé sur le calcul de l'énergie (RMS) du signal audio. C'est une brique fondamentale de notre Asterisk EAGI code.

    def detect_silence(self, chunk):
        # Convertit le chunk binaire en un tableau d'entiers 16-bit
        samples = array.array(AUDIO_FORMAT, chunk)
        # Calcule la Root Mean Square (RMS) - une mesure de l'énergie
        sum_squares = sum(s * s for s in samples)
        rms = (sum_squares / len(samples)) ** 0.5
        return rms < SILENCE_THRESHOLD

Méthode detect_bargein() : Gestion de l'Interruption (Barge-in)

Cette méthode tourne dans un thread séparé pendant que l'IA parle. Elle écoute l'utilisateur et déclenche un événement si une interruption est détectée.

    def detect_bargein(self):
        self.log("Thread de barge-in démarré.")
        while not self.barge_in_event.is_set():
            try:
                chunk = os.read(self.audio_fd, CHUNK_SIZE * 2)
                if not chunk:
                    break
                
                samples = array.array(AUDIO_FORMAT, chunk)
                rms = (sum(s * s for s in samples) / len(samples)) ** 0.5

                if rms > BARGE_IN_THRESHOLD:
                    self.log(f"Barge-in détecté ! RMS={rms:.2f}")
                    self.barge_in_event.set()
                    break
            except (IOError, OSError):
                # Le thread peut être interrompu, c'est normal
                break
        self.log("Thread de barge-in terminé.")

Méthode transcribe(audio_data) : Appel à l'API Whisper

Envoie les données audio brutes à une API de transcription. La gestion des erreurs est importante ici.

    def transcribe(self, audio_data):
        self.log(f"Envoi de {len(audio_data)} bytes à l'API de transcription.")
        try:
            # Note: l'API doit accepter du PCM 16-bit 8000Hz.
            # Il faudra peut-être ajouter des headers Content-Type.
            response = requests.post(WHISPER_API_URL, data=audio_data, headers={'Content-Type': 'audio/s16le; rate=8000'}, timeout=10)
            response.raise_for_status()
            transcription = response.json().get('text', '').strip()
            self.log(f"Transcription reçue : '{transcription}'")
            return transcription
        except requests.exceptions.RequestException as e:
            self.log(f"Erreur API Whisper : {e}")
            return ""

Méthode generate_response(text) : Génération de Réponse avec LLM backend

Cette fonction est un générateur. Elle appelle l'API de streaming d'LLM backend et `yield` les tokens de réponse au fur et à mesure qu'ils arrivent, permettant un traitement en temps réel.

    def generate_response(self, text):
        self.log(f"Envoi du texte au LLM : '{text}'")
        payload = {
            "model": OLLAMA_MODEL,
            "prompt": f"Vous êtes un agent vocal serviable. Répondez de manière concise à la question suivante: {text}",
            "stream": True # Activer le streaming est crucial !
        }
        try:
            with requests.post(OLLAMA_API_URL, json=payload, stream=True, timeout=30) as response:
                response.raise_for_status()
                for line in response.iter_lines():
                    if line:
                        json_line = json.loads(line)
                        token = json_line.get("response", "")
                        yield token
                        if json_line.get("done"):
                            break
        except requests.exceptions.RequestException as e:
            self.log(f"Erreur API LLM backend : {e}")
            yield "Je rencontre un problème technique, veuillez réessayer."

Méthode synthesize(text) : Synthèse Vocale avec mixael-TTS

Similaire à la génération de réponse, cette méthode appelle l'API de streaming TTS et `yield` les chunks audio PCM dès qu