Overblog Tous les blogs Top blogs Technologie & Science Tous les blogs Technologie & Science
Editer l'article Suivre ce blog Administration + Créer mon blog
MENU
Aquila Networks : Sécurité & Réseaux Étudiant en Master SSI à l'ESMT

Serveur Vocal Interactif Intelligent (SVI IA)

El Hadji Diakhate
 Créer un Assistant Vocal Intelligent avec Asterisk et l'IA

Introduction

Dans ce tutoriel, je vais vous montrer comment j'ai créé un **Serveur Vocal Interactif Intelligent (SVI IA)** en combinant Asterisk, l'API Groq (gratuite) et Python.

L'utilisateur appelle le numéro **3000**, pose une question vocalement, et le système répond avec une voix synthétique générée par l'IA.

**Exemples de questions :**

- "Donne-moi mes notes"
- "Quels sont mes cours ?"
- "Explique-moi la sécurité réseau"



Architecture du système
 

 

 

 Prérequis

installation Asterisk
Asterisk avec PJSIP installé et MySQL installé fonctionnel
 googletts.agi installé dans `/var/lib/asterisk/agi-bin/`
Un softphone (Zoiper, MicroSIP, Linphone)

 

 ÉTAPE 1 — Créer sa clé API Groq (100% gratuit)

 

1. Aller sur https://console.groq.com
2. Créer un compte gratuit
3. Cliquer sur "Clés API" → "Créer une clé API"
4. Nommer la clé (ex: `SVI-IA`)
5. Choisir "Aucune date d'expiration"
6. Cliquer "Soumettre"
7. Copier immédiatement la clé — elle ne sera plus affichée

> ⚠️ La clé ressemble à : `gsk_xxxxxxxxxxxxxxxxxxxx`
> Groq offre 14 400 requêtes gratuites par jour — largement suffisant.

 

ÉTAPE 2 — Installer les packages Python


# pip install groq gtts pydub requests --break-system-packages
```

 

 

Vérifier les installations

```bash
# python3 -c "from groq import Groq; print('groq OK')"
# python3 -c "from gtts import gTTS; print('gtts OK')"
# python3 -c "from pydub import AudioSegment; print('pydub OK')"
```

 

 Installer ffmpeg (nécessaire pour la conversion audio)

```bash
# apt install -y ffmpeg
# ffmpeg -version
```

 

 

ÉTAPE 3 — Vérifier le modèle Groq

```bash

vim /tmp/test_groq.py

```


from groq import Groq

client = Groq(api_key="gsk_VOTRE_CLE_ICI")

models = client.models.list()
print("=== Modèles disponibles ===")
for m in models.data:
    print(f"  - {m.id}")

print("\n=== Test du modèle ===")
r = client.chat.completions.create(
    model="llama-3.3-70b-versatile",
    max_tokens=100,
    messages=[{"role": "user", "content": "Dis bonjour en français"}]
)
print("Réponse :", r.choices[0].message.content)
 

```bash

python3 /tmp/test_groq.py
```

 

> ⚠️ En 2025/2026, les modèles `llama3-8b-8192` et `mixtral-8x7b-32768` sont **décommissionnés**.
> Utilisez uniquement **`llama-3.3-70b-versatile`**.

 

 ÉTAPE 4 — Créer la base de données MySQL

```bash
sudo mysql -u root -p
```

```sql
CREATE DATABASE asterisk_svi;
USE asterisk_svi;

CREATE TABLE etudiants (
    id INT AUTO_INCREMENT PRIMARY KEY,
    extension VARCHAR(10) UNIQUE NOT NULL,
    nom VARCHAR(100),
    prenom VARCHAR(100),
    filiere VARCHAR(100)
);

CREATE TABLE notes (
    id INT AUTO_INCREMENT PRIMARY KEY,
    etudiant_id INT,
    matiere VARCHAR(100),
    note DECIMAL(4,2),
    FOREIGN KEY (etudiant_id) REFERENCES etudiants(id)
);

CREATE TABLE emploi_du_temps (
    id INT AUTO_INCREMENT PRIMARY KEY,
    etudiant_id INT,
    jour VARCHAR(20),
    heure_debut TIME,
    heure_fin TIME,
    matiere VARCHAR(100),
    salle VARCHAR(50),
    FOREIGN KEY (etudiant_id) REFERENCES etudiants(id)
);

-- Données de test (sans apostrophes dans les chaînes SQL)
INSERT INTO etudiants VALUES
    (1,'1000','Diakhate','Elhadji','Securite des systemes information'),
    (2,'1001','Diakhate','Fatou','Ingenierie des systemes information');

INSERT INTO notes VALUES
    (NULL,1,'Securite Reseau',15.5),
    (NULL,1,'Protocoles IP',17.0),
    (NULL,1,'Terminaux mobiles communicants',18.0),
    (NULL,2,'Securite Reseau',14.0),
    (NULL,2,'Protocoles IP',16.5);

INSERT INTO emploi_du_temps VALUES
    (NULL,1,'Lundi','08:00','10:00','Securite Reseau','HA1'),
    (NULL,1,'Mercredi','10:00','12:00','Terminaux mobiles communicants','Salle Cyber'),
    (NULL,1,'Vendredi','14:00','16:00','Protocoles IP','HA8'),
    (NULL,2,'Mardi','08:00','10:00','Securite Reseau','Salle Cyber');

SELECT * FROM etudiants;
SELECT * FROM notes;
SELECT * FROM emploi_du_temps;
quit
```

> ⚠️ Ne pas mettre d'apostrophes dans les données SQL (ex: `d'information`).
> MySQL interprète l'apostrophe comme la fin de la chaîne → erreur de syntaxe.

 

ÉTAPE 5 — Créer le dossier de travail Asterisk

```bash
mkdir -p /var/spool/asterisk/tmp
chown -R asterisk:asterisk /var/spool/asterisk/tmp
chmod 777 /var/spool/asterisk/tmp
ls -la /var/spool/asterisk/tmp
```

> ⚠️ **Règle importante** : Asterisk ne peut PAS écrire dans `/tmp` par défaut.
> Toujours utiliser `/var/spool/asterisk/tmp/` avec `chown asterisk:asterisk`.

 

ÉTAPE 6 — Créer l'API Flask

```bash
mkdir -p /opt/svi_ia

 

# vim /opt/svi_ia/app.py

 
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from flask import Flask, request, jsonify
import mysql.connector
from groq import Groq

app = Flask(__name__)

GROQ_API_KEY = "gsk_VOTRE_CLE_ICI"

DB_CONFIG = {
    "host":     "localhost",
    "user":     "root",
    "password": "VOTRE_MOT_DE_PASSE_MYSQL",
    "database": "asterisk_svi"
}

def get_student_info(extension):
    try:
        conn   = mysql.connector.connect(**DB_CONFIG)
        cursor = conn.cursor(dictionary=True)
        cursor.execute(
            "SELECT * FROM etudiants WHERE extension = %s", (extension,)
        )
        etudiant = cursor.fetchone()
        if not etudiant:
            return None, None, None
        cursor.execute(
            "SELECT matiere, note FROM notes WHERE etudiant_id = %s",
            (etudiant["id"],)
        )
        notes = cursor.fetchall()
        cursor.execute("""
            SELECT jour, heure_debut, heure_fin, matiere, salle
            FROM emploi_du_temps WHERE etudiant_id = %s
            ORDER BY FIELD(jour,'Lundi','Mardi','Mercredi','Jeudi','Vendredi')
        """, (etudiant["id"],))
        emploi = cursor.fetchall()
        cursor.close()
        conn.close()
        return etudiant, notes, emploi
    except Exception as e:
        print(f"Erreur MySQL : {e}")
        return None, None, None

def build_context(etudiant, notes, emploi):
    ctx  = "Tu es un assistant vocal de l'ESMT.\n"
    ctx += "Reponds en francais, 2-3 phrases max, sans tirets ni listes.\n\n"
    if etudiant:
        ctx += f"Etudiant : {etudiant['prenom']} {etudiant['nom']}, "
        ctx += f"Filiere : {etudiant.get('filiere','')}\n\n"
    if notes:
        ctx += "Notes :\n"
        for n in notes:
            ctx += f"  - {n['matiere']} : {n['note']}/20\n"
        ctx += "\n"
    if emploi:
        ctx += "Emploi du temps :\n"
        for e in emploi:
            ctx += f"  - {e['jour']} {str(e['heure_debut'])[:5]}"
            ctx += f"-{str(e['heure_fin'])[:5]} : {e['matiere']}"
            ctx += f" salle {e['salle']}\n"
    return ctx

@app.route("/ask", methods=["POST"])
def ask():
    data      = request.get_json()
    question  = data.get("question", "")
    extension = data.get("extension", "")
    if not question:
        return jsonify({"reponse": "Question vide."})
    etudiant, notes, emploi = get_student_info(extension)
    system_context = build_context(etudiant, notes, emploi)
    try:
        client   = Groq(api_key=GROQ_API_KEY)
        response = client.chat.completions.create(
            model="llama-3.3-70b-versatile",
            max_tokens=300,
            messages=[
                {"role": "system", "content": system_context},
                {"role": "user",   "content": question}
            ]
        )
        reponse = response.choices[0].message.content
    except Exception as e:
        print(f"Erreur Groq : {e}")
        reponse = "Je n'ai pas pu contacter le service d'intelligence artificielle."
    return jsonify({"reponse": reponse})

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5000, debug=False)

```

Tester l'API Flask

```bash
# Démarrer Flask
python3 /opt/svi_ia/app.py &
 

 

 Tester

curl -X POST http://127.0.0.1:5000/ask \
  -H "Content-Type: application/json" \
  -d '{"question":"Donne-moi mes notes","extension":"1000"}'
```

Résultat attendu :

```json
{"reponse": "Elhadji Diakhate, tes notes sont : 15,50/20 en Securite Reseau, 17/20 en Protocoles IP et 18/20 en Terminaux mobiles communicants."}
```

---

ÉTAPE 7 — Créer le script AGI

```bash
vim /var/lib/asterisk/agi-bin/svi_ia.agi

 

 


#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import sys
import os
import requests
from groq import Groq
from gtts import gTTS
from pydub import AudioSegment

GROQ_API_KEY = "gsk_VOTRE_CLE_ICI"
WORKDIR      = "/var/spool/asterisk/tmp"

class AGI:
    def __init__(self):
        self.env = {}
        self._read_env()

    def _read_env(self):
        while True:
            line = sys.stdin.readline().strip()
            if not line:
                break
            if ':' in line:
                key, val = line.split(':', 1)
                self.env[key.strip()] = val.strip()

    def send(self, cmd):
        sys.stdout.write(cmd + '\n')
        sys.stdout.flush()
        return sys.stdin.readline().strip()

    def verbose(self, msg, level=1):
        self.send(f'VERBOSE "{msg}" {level}')

    def hangup(self):
        self.send('HANGUP')

def speech_to_text(wav_path):
    try:
        client = Groq(api_key=GROQ_API_KEY)
        with open(wav_path, "rb") as f:
            t = client.audio.transcriptions.create(
                model="whisper-large-v3",
                file=f,
                language="fr",
                response_format="text"
            )
        return t.strip()
    except Exception as e:
        sys.stderr.write(f"Erreur STT : {e}\n")
        return ""

def ask_api(question, extension):
    try:
        resp = requests.post(
            "http://127.0.0.1:5000/ask",
            json={"question": question, "extension": extension},
            timeout=20
        )
        return resp.json().get("reponse", "Pas de reponse.")
    except Exception as e:
        sys.stderr.write(f"Erreur API : {e}\n")
        return "Le service est temporairement indisponible."

def text_to_speech(text):
    try:
        tts = gTTS(text=text, lang="fr")
        tts.save(f"{WORKDIR}/reponse.mp3")
        sound = AudioSegment.from_mp3(f"{WORKDIR}/reponse.mp3")
        sound = sound.set_frame_rate(8000).set_channels(1)
        sound.export(f"{WORKDIR}/reponse.wav", format="wav")
    except Exception as e:
        sys.stderr.write(f"Erreur TTS : {e}\n")

def main():
    agi        = AGI()
    caller_ext = agi.env.get('agi_callerid', '1000')
    agi.verbose(f"Extension : {caller_ext}")

    wav_path = f"{WORKDIR}/question.wav"

    if not os.path.exists(wav_path) or os.path.getsize(wav_path) < 1000:
        agi.verbose("Fichier audio absent ou trop petit")
        text_to_speech("Je n'ai pas entendu votre question. Au revoir.")
        return

    agi.verbose("Transcription STT Groq...")
    question = speech_to_text(wav_path)
    agi.verbose(f"Question : {question}")

    if len(question) < 3:
        text_to_speech("Je n'ai pas compris. Veuillez rappeler.")
        return

    agi.verbose("Appel IA...")
    reponse = ask_api(question, caller_ext)
    agi.verbose(f"Reponse : {reponse}")

    agi.verbose("Generation audio...")
    text_to_speech(reponse)
    agi.verbose("Termine.")

if __name__ == "__main__":
    main()
 

 

chmod +x /var/lib/asterisk/agi-bin/svi_ia.agi
chown asterisk:asterisk /var/lib/asterisk/agi-bin/svi_ia.agi
ls -la /var/lib/asterisk/agi-bin/svi_ia.agi
```

 

ÉTAPE 8 — Modifier le dialplan Asterisk

```bash
vim /etc/asterisk/extensions.conf
```

 

 

**Dans `[from-internal]` ajouter :**
```
exten => 3000,1,Goto(svi-ia,3000,1)
```

**Ajouter ce contexte à la fin du fichier :**
```
; ============================================================
; SVI Intelligent IA — extension 3000
; ============================================================
[svi-ia]
exten => 3000,1,Answer()
 same => n,AGI(googletts.agi,"Bienvenue dans votre assistant intelligent de l'ESMT",fr,any)
 same => n,AGI(googletts.agi,"Posez votre question apres le signal",fr,any)
 same => n,Record(/var/spool/asterisk/tmp/question.wav,3,30)
 same => n,AGI(svi_ia.agi)
 same => n,Playback(/var/spool/asterisk/tmp/reponse)
 same => n,Hangup()
```

```bash
# asterisk -rx "dialplan reload"
# asterisk -rx "dialplan show svi-ia"
```

 

ÉTAPE 9 — Démarrage automatique Flask avec systemd

```bash
vim  /etc/systemd/system/svi-flask.service 


[Unit]
Description=SVI IA Flask API
After=network.target mysql.service

[Service]
ExecStart=/usr/bin/python3 /opt/svi_ia/app.py
Restart=always
User=root

[Install]
WantedBy=multi-user.target

systemctl daemon-reload
systemctl enable svi-flask
systemctl start svi-flask
systemctl status svi-flask
```

ÉTAPE 10 — Tests complets

 

Test final : appel réel

```
1. Ouvrir Zoiper ou MicroSIP
2. Se connecter avec l'extension 1000
3. Composer le 3000
4. Écouter : "Bienvenue dans votre assistant intelligent de l'ESMT"
5. Écouter : "Posez votre question après le signal"
6. Parler après le bip :
   → "Donne-moi mes notes"
   → "Quels sont mes cours ?"
7. Écouter la réponse intelligente de l'IA 🎉
```

Conclusion

Ce projet montre comment combiner la téléphonie VoIP avec l'intelligence artificielle pour créer un service vocal intelligent. Grâce à Groq (gratuit), aucun modèle lourd n'est nécessaire en local — tout passe par le cloud.

 

Serveur Vocal Interactif Intelligent (SVI IA)

Pour être informé des derniers articles, inscrivez vous :
Commentaires