Introduction : Pourquoi un Assistant IA Local ?
Dans un monde où nos données personnelles sont constamment collectées par les géants de la tech, avoir un assistant IA 100% local et privé devient un véritable atout. Imaginez un assistant inspiré de JARVIS ou FRIDAY d'Iron Man, capable de :
- 🎮 Contrôler tous vos appareils à distance (PC, téléphone, laptop)
- 🗣️ Répondre vocalement en plusieurs langues
- 📺 Gérer vos médias (vidéos, musique, streaming)
- 📱 Répondre à vos messages automatiquement
- 📸 Prendre des photos via caméra
- 🌍 Utiliser la localisation GPS
- ⚙️ Exécuter des commandes système
- 🔄 Automatiser vos routines quotidiennes
- 🔒 Zéro données envoyées à des tiers
Dans ce guide, nous allons construire cet assistant de A à Z en utilisant des technologies open-source modernes. Tout tournera sur votre propre serveur, garantissant une confidentialité absolue.
Vue d'Ensemble de l'Architecture
Notre assistant repose sur une architecture modulaire et scalable :
┌──────────────────────────────────────────────────────────────┐
│ VPS/Serveur Linux │
│ │
│ ┌─────────────────────────────────────┐ │
│ │ Nginx (Reverse Proxy) │ │
│ │ Port: 443 (HTTPS) │ │
│ │ - SSL Let's Encrypt │ │
│ │ - Rate limiting │ │
│ │ - WebSocket support │ │
│ └─────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────┐ │
│ │ FastAPI Server + WebSocket │ │
│ │ - API REST │ │
│ │ - Temps réel WebSocket │ │
│ │ - Authentification JWT │ │
│ │ - System prompt IA │ │
│ └─────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────┐ │
│ │ Ollama (LLM Local) │ │
│ │ - Llama 3.1 8B / Mistral 7B │ │
│ │ - Inférences CPU/GPU │ │
│ └─────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────┐ │
│ │ SQLite Database │ │
│ │ - Historique conversations │ │
│ │ - Contexte utilisateur │ │
│ │ - Logs et préférences │ │
│ └─────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘
↕
HTTPS chiffré (TLS/SSL)
↕
┌───────────────────────────────────┐
│ Clients (Windows/Android/Linux) │
│ │
│ Agent Local: │
│ - WebSocket client │
│ - Contrôle système │
│ - TTS/STT vocal │
│ - Caméra / GPS │
│ - Auto-reconnexion │
└───────────────────────────────────┘
Choix du Modèle LLM
Le cœur de notre assistant est le modèle de langage. Voici une comparaison des meilleures options pour un serveur avec 16GB RAM et CPU multi-core :
| Modèle | RAM | Vitesse (32 cores) | Qualité | Recommandation |
|---|---|---|---|---|
| Llama 3.3 70B Q2 | ~12 GB | 15-25 sec | Excellent | ❌ Trop lent |
| Llama 3.1 8B Q4 | ~6 GB | 3-5 sec | Très bon | ⭐ OPTIMAL |
| Llama 3.2 3B | ~3 GB | 1-2 sec | Correct | ✅ Backup |
| Mistral 7B v0.3 | ~5 GB | 3-4 sec | Très bon | ✅ Alternative |
| Phi-3 Mini 3.8B | ~3 GB | 1-2 sec | Bon | ✅ Ultra-rapide |
Recommandation : Llama 3.1 8B Q4 offre le meilleur compromis vitesse/qualité/RAM pour un assistant réactif.
⚠️ Note importante : Avec le contexte long (historique conversations), la RAM peut monter jusqu'à 10 GB. Surveillez et limitez la taille du contexte si nécessaire.
Configuration du Serveur
Prérequis Serveur
Spécifications minimales recommandées :
| Composant | Minimum | Recommandé |
|---|---|---|
| CPU | 4 cores | 8+ cores (meilleure vitesse) |
| RAM | 12 GB | 16 GB+ |
| Stockage | 20 GB | 50 GB (pour logs et historique) |
| OS | Linux (Ubuntu/Debian) | Ubuntu 22.04+ LTS |
| GPU | Aucun (CPU only) | NVIDIA (accélération) |
Installation des dépendances
# Mise à jour système
sudo apt update && sudo apt upgrade -y
# Installation Docker
curl -fsSL https://get.docker.com -o get-docker.sh
sudo sh get-docker.sh
# Installation Docker Compose
sudo apt install docker-compose-plugin -y
# Vérification
docker --version
docker compose version
# Installation Nginx
sudo apt install nginx certbot python3-certbot-nginx -y
Structure du Projet Docker
Arborescence des fichiers
/opt/assistant-ia/ ← RACINE PROJET
├── docker-compose.yml # Orchestration
├── .env # Variables d'environnement
├── .env.example # Template
│
├── services/
│ └── api/
│ ├── Dockerfile
│ ├── requirements.txt
│ └── app/
│ ├── main.py # Point d'entrée FastAPI
│ ├── config.py # Configuration
│ ├── ai_personality.py # System prompt IA
│ ├── routes/
│ │ ├── chat.py # Conversation
│ │ ├── devices.py # Gestion appareils
│ │ ├── actions.py # Commandes système
│ │ └── auth.py # Authentification
│ ├── models/
│ │ ├── conversation.py
│ │ ├── device.py
│ │ └── user.py
│ └── services/
│ ├── ollama_client.py
│ ├── websocket_manager.py
│ └── command_executor.py
│
├── config/
│ ├── ai_personality.json # Personnalité IA
│ └── routines.json # Automatisations
│
├── data/ # Base de données SQLite
├── logs/ # Logs applicatifs
└── shared/ # Fichiers partagés
├── outputs/ # Photos, screenshots
└── uploads/ # Fichiers uploadés
Fichier docker-compose.yml
version: '3.8'
services:
# Service Ollama (LLM)
ollama:
image: ollama/ollama:latest
container_name: ai-assistant-ollama
restart: unless-stopped
volumes:
- ollama-models:/root/.ollama
environment:
- OLLAMA_HOST=0.0.0.0:11434
- OLLAMA_ORIGINS=*
networks:
- assistant-network
deploy:
resources:
limits:
memory: 10G
reservations:
memory: 6G
# Service FastAPI
api:
build: ./services/api
container_name: ai-assistant-api
restart: unless-stopped
ports:
- "127.0.0.1:8888:8888"
volumes:
- ./services/api/app:/app
- ./config:/config:ro
- ./data:/data
- ./logs:/logs
- ./shared:/shared
environment:
- OLLAMA_URL=http://ollama:11434
- DATABASE_PATH=/data/assistant.db
- LOG_LEVEL=INFO
- JWT_SECRET=${JWT_SECRET}
- ALLOWED_ORIGINS=${ALLOWED_ORIGINS}
depends_on:
- ollama
networks:
- assistant-network
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8888/health"]
interval: 30s
timeout: 10s
retries: 3
networks:
assistant-network:
driver: bridge
volumes:
ollama-models:
Fichier .env.example
# JWT Secret (générer avec: openssl rand -hex 32)
JWT_SECRET=your_secret_key_here_change_this
# CORS Origins (vos domaines/IPs autorisés)
ALLOWED_ORIGINS=https://yourdomain.com,https://192.168.1.100
# Configuration Ollama
OLLAMA_MODEL=llama3.1:8b
OLLAMA_CONTEXT_LENGTH=4096
# Base de données
DATABASE_PATH=/data/assistant.db
# Logging
LOG_LEVEL=INFO
Création du Système de Personnalité
Le System Prompt
Le system prompt définit le comportement de votre assistant. Voici un exemple de personnalité efficace :
Fichier: config/ai_personality.json
{
"name": "Assistant",
"version": "1.0",
"personality": {
"tone": "professionnel avec touche d'humour",
"style": "concis et direct",
"proactivity": true
},
"system_prompt": "Tu es un assistant IA personnel avancé, inspiré des IA de science-fiction comme JARVIS.\n\nRÈGLES DE COMPORTEMENT:\n- Tu es proactif : si tu détectes un problème, tu le signales\n- Tu es concis : pas de blabla, des réponses directes et efficaces\n- Tu gères les appareils connectés quand demandé\n- Tu confirmes toujours les actions exécutées\n- Tu peux refuser les commandes dangereuses en expliquant pourquoi\n\nLANGUE:\n- Tu détectes automatiquement la langue de l'utilisateur\n- Tu réponds dans la même langue\n- Tu peux mixer si demandé\n\nCONTEXTE:\n- Tu tournes sur un serveur local sécurisé\n- Tu contrôles plusieurs appareils simultanément\n- Tu es 100% local, aucune donnée ne quitte le serveur\n- Date actuelle : {current_date}\n- Appareils connectés : {connected_devices}\n\nFORMAT RÉPONSE:\n- Pour les actions : [ACTION: type] description\n- Pour les confirmations : ✅ Action effectuée\n- Pour les erreurs : ❌ Erreur : description\n- Pour les suggestions : 💡 Suggestion : description",
"capabilities": [
"device_control",
"media_management",
"messaging",
"camera_control",
"gps_location",
"system_commands",
"routines_automation"
]
}
Implémentation Python
Fichier: services/api/app/ai_personality.py
import json
from datetime import datetime
from typing import Dict, List, Optional
class AIPersonality:
def __init__(self, config_path: str = "/config/ai_personality.json"):
with open(config_path, 'r', encoding='utf-8') as f:
self.config = json.load(f)
def get_system_prompt(
self,
connected_devices: Optional[List[str]] = None,
user_context: Optional[Dict] = None
) -> str:
"""Génère le system prompt avec contexte dynamique"""
# Récupère le prompt de base
prompt = self.config['system_prompt']
# Injection des variables dynamiques
current_date = datetime.now().strftime("%d %B %Y, %H:%M")
devices_list = ", ".join(connected_devices) if connected_devices else "Aucun"
prompt = prompt.replace("{current_date}", current_date)
prompt = prompt.replace("{connected_devices}", devices_list)
# Ajout du contexte utilisateur si disponible
if user_context:
context_str = "\n\nCONTEXTE UTILISATEUR:"
for key, value in user_context.items():
context_str += f"\n- {key}: {value}"
prompt += context_str
return prompt
def get_capabilities(self) -> List[str]:
"""Retourne les capacités de l'assistant"""
return self.config.get('capabilities', [])
def format_response(self, response_type: str, message: str) -> str:
"""Formate une réponse selon le type"""
formats = {
'action': f"[ACTION: {message}]",
'success': f"✅ {message}",
'error': f"❌ Erreur : {message}",
'suggestion': f"💡 Suggestion : {message}",
'info': f"ℹ️ {message}"
}
return formats.get(response_type, message)
# Utilisation
personality = AIPersonality()
system_prompt = personality.get_system_prompt(
connected_devices=["PC-Bureau", "Téléphone-Android"],
user_context={"location": "Maison", "time_of_day": "matin"}
)
API FastAPI - Backend Principal
Point d'entrée main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import uvicorn
import os
from typing import List
from routes import chat, devices, actions, auth
from services.websocket_manager import ConnectionManager
from services.ollama_client import OllamaClient
from ai_personality import AIPersonality
# Initialisation
app = FastAPI(
title="Assistant IA Personnel API",
version="1.0.0",
description="API Backend pour assistant IA local"
)
# CORS
app.add_middleware(
CORSMiddleware,
allow_origins=os.getenv("ALLOWED_ORIGINS", "*").split(","),
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Managers globaux
manager = ConnectionManager()
ollama = OllamaClient(base_url=os.getenv("OLLAMA_URL", "http://ollama:11434"))
personality = AIPersonality()
# Routes
app.include_router(chat.router, prefix="/api/chat", tags=["Chat"])
app.include_router(devices.router, prefix="/api/devices", tags=["Devices"])
app.include_router(actions.router, prefix="/api/actions", tags=["Actions"])
app.include_router(auth.router, prefix="/api/auth", tags=["Auth"])
@app.get("/health")
async def health_check():
"""Health check endpoint"""
ollama_status = await ollama.check_health()
return {
"status": "healthy",
"ollama": "connected" if ollama_status else "disconnected",
"connected_clients": len(manager.active_connections)
}
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
"""WebSocket pour communication temps réel"""
await manager.connect(websocket, client_id)
try:
while True:
# Réception message client
data = await websocket.receive_json()
# Traitement selon le type
if data.get("type") == "chat":
# Génération réponse IA
system_prompt = personality.get_system_prompt(
connected_devices=manager.get_connected_devices()
)
response = await ollama.generate(
prompt=data.get("message"),
system_prompt=system_prompt
)
await manager.send_personal_message({
"type": "response",
"message": response
}, client_id)
elif data.get("type") == "command":
# Exécution commande
result = await execute_command(data.get("command"))
await manager.send_personal_message({
"type": "command_result",
"result": result
}, client_id)
except WebSocketDisconnect:
manager.disconnect(client_id)
await manager.broadcast({
"type": "device_disconnected",
"device": client_id
})
if __name__ == "__main__":
uvicorn.run(
"main:app",
host="0.0.0.0",
port=8888,
reload=True,
log_level="info"
)
Client Ollama
Fichier: services/api/app/services/ollama_client.py
import aiohttp
import asyncio
from typing import Optional, Dict, List
class OllamaClient:
def __init__(self, base_url: str = "http://localhost:11434"):
self.base_url = base_url
self.model = "llama3.1:8b"
async def check_health(self) -> bool:
"""Vérifie si Ollama est disponible"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.base_url}/api/tags") as response:
return response.status == 200
except:
return False
async def generate(
self,
prompt: str,
system_prompt: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2048
) -> str:
"""Génère une réponse avec Ollama"""
messages = []
if system_prompt:
messages.append({
"role": "system",
"content": system_prompt
})
messages.append({
"role": "user",
"content": prompt
})
payload = {
"model": self.model,
"messages": messages,
"stream": False,
"options": {
"temperature": temperature,
"num_predict": max_tokens
}
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/api/chat",
json=payload,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status == 200:
result = await response.json()
return result['message']['content']
else:
return f"Erreur Ollama: {response.status}"
except asyncio.TimeoutError:
return "❌ Timeout: Le modèle met trop de temps à répondre"
except Exception as e:
return f"❌ Erreur: {str(e)}"
async def generate_stream(
self,
prompt: str,
system_prompt: Optional[str] = None
):
"""Génère une réponse en streaming"""
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
payload = {
"model": self.model,
"messages": messages,
"stream": True
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/api/chat",
json=payload
) as response:
async for line in response.content:
if line:
yield line.decode('utf-8')
Configuration Nginx (Reverse Proxy)
Nginx sert de reverse proxy pour sécuriser l'accès à votre API et gérer SSL.
Fichier: /etc/nginx/sites-available/assistant-ia
# Redirection HTTP → HTTPS
server {
listen 80;
server_name yourdomain.com;
location / {
return 301 https://$server_name$request_uri;
}
}
# Configuration HTTPS
server {
listen 443 ssl http2;
server_name yourdomain.com;
# SSL Configuration (Let's Encrypt)
ssl_certificate /etc/letsencrypt/live/yourdomain.com/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/yourdomain.com/privkey.pem;
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers HIGH:!aNULL:!MD5;
# Security Headers
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
add_header X-Content-Type-Options "nosniff" always;
add_header X-Frame-Options "DENY" always;
# Rate Limiting
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;
limit_req zone=api_limit burst=20 nodelay;
# Proxy vers FastAPI
location /api/ {
proxy_pass http://127.0.0.1:8888;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection 'upgrade';
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_cache_bypass $http_upgrade;
}
# WebSocket
location /ws/ {
proxy_pass http://127.0.0.1:8888;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_read_timeout 86400;
}
# Health check
location /health {
proxy_pass http://127.0.0.1:8888/health;
access_log off;
}
}
Activation de la configuration :
# Créer le lien symbolique
sudo ln -s /etc/nginx/sites-available/assistant-ia /etc/nginx/sites-enabled/
# Tester la configuration
sudo nginx -t
# Obtenir certificat SSL (remplacer yourdomain.com)
sudo certbot --nginx -d yourdomain.com
# Redémarrer Nginx
sudo systemctl restart nginx
Client Agent (Python)
L'agent client s'installe sur vos appareils (Windows/Linux/Android) pour communiquer avec le serveur.
Agent de base (compatible multi-plateformes)
import asyncio
import websockets
import json
import platform
import psutil
from typing import Dict, Callable
class AssistantAgent:
def __init__(self, server_url: str, device_id: str, token: str):
self.server_url = server_url
self.device_id = device_id
self.token = token
self.ws = None
self.handlers: Dict[str, Callable] = {}
def register_handler(self, action_type: str, handler: Callable):
"""Enregistre un handler pour un type d'action"""
self.handlers[action_type] = handler
async def connect(self):
"""Connexion WebSocket avec auto-reconnexion"""
while True:
try:
uri = f"{self.server_url}/ws/{self.device_id}"
headers = {"Authorization": f"Bearer {self.token}"}
async with websockets.connect(uri, extra_headers=headers) as websocket:
self.ws = websocket
print(f"✅ Connecté au serveur: {self.server_url}")
# Envoi info appareil
await self.send_device_info()
# Écoute des messages
async for message in websocket:
await self.handle_message(message)
except websockets.ConnectionClosed:
print("❌ Connexion fermée. Reconnexion dans 5s...")
await asyncio.sleep(5)
except Exception as e:
print(f"❌ Erreur: {e}. Reconnexion dans 5s...")
await asyncio.sleep(5)
async def send_device_info(self):
"""Envoie les infos de l'appareil"""
info = {
"type": "device_info",
"data": {
"platform": platform.system(),
"hostname": platform.node(),
"cpu_count": psutil.cpu_count(),
"memory_total": psutil.virtual_memory().total,
"disk_usage": psutil.disk_usage('/').percent
}
}
await self.ws.send(json.dumps(info))
async def handle_message(self, message: str):
"""Traite un message reçu du serveur"""
try:
data = json.loads(message)
action_type = data.get("type")
if action_type in self.handlers:
result = await self.handlers[action_type](data)
# Envoi résultat
await self.ws.send(json.dumps({
"type": "action_result",
"action": action_type,
"result": result
}))
else:
print(f"⚠️ Action non supportée: {action_type}")
except json.JSONDecodeError:
print(f"❌ Message invalide: {message}")
async def send_message(self, message_type: str, data: dict):
"""Envoie un message au serveur"""
if self.ws:
payload = {
"type": message_type,
"data": data
}
await self.ws.send(json.dumps(payload))
# Exemple d'utilisation
async def handle_system_command(data):
"""Handler pour commandes système"""
import subprocess
command = data.get("command")
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30
)
return {
"success": True,
"output": result.stdout,
"error": result.stderr
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
async def handle_media_control(data):
"""Handler pour contrôle média"""
action = data.get("action")
if action == "play":
# Logique pour lire média
return {"success": True, "message": "Lecture démarrée"}
elif action == "pause":
return {"success": True, "message": "Média en pause"}
return {"success": False, "error": "Action non reconnue"}
# Lancement
async def main():
agent = AssistantAgent(
server_url="wss://yourdomain.com",
device_id="pc-bureau-001",
token="your_jwt_token_here"
)
# Enregistrement handlers
agent.register_handler("system_command", handle_system_command)
agent.register_handler("media_control", handle_media_control)
# Connexion
await agent.connect()
if __name__ == "__main__":
asyncio.run(main())
Cas d'Usage Pratiques
Scénario 1 : Contrôle médias cross-device
User (depuis téléphone): "Lance le prochain épisode de ma série sur mon PC"
Assistant IA:
1. Identifie l'appareil cible (PC-Bureau)
2. Recherche le dernier épisode regardé
3. Trouve le fichier suivant
4. Envoie commande de lecture au PC
5. Confirme: "✅ Episode 12 lancé sur ton PC"
Scénario 2 : Routine matinale automatisée
[7h00 - Déclenchement automatique]
Assistant IA:
1. Vérifie la météo locale
2. Lit les notifications importantes
3. Lance la playlist "Morning"
4. Ouvre VS Code sur le PC
5. Affiche l'agenda du jour
Message: "Bonjour ! Il fait 22°C aujourd'hui. Tu as 3 notifications Telegram et 2 réunions planifiées. Bon courage !"
Scénario 3 : Réponse automatique aux messages
User: "Si quelqu'un m'envoie un message dans les 2 prochaines heures, réponds que je suis en réunion"
Assistant IA:
1. Active le mode "auto-réponse"
2. Surveille les messages entrants
3. Détecte nouveau message de "Maman"
4. Répond automatiquement: "Je suis actuellement en réunion, je te réponds dès que possible !"
5. Notifie l'utilisateur: "💬 Réponse automatique envoyée à Maman"
Scénario 4 : Prise de photo à distance
User (depuis PC): "Prends une photo avec la caméra de mon téléphone"
Assistant IA:
1. Envoie commande au téléphone Android
2. Active la caméra arrière
3. Capture l'image
4. Upload vers /shared/outputs/photo_20260216_143052.jpg
5. Confirme: "✅ Photo sauvegardée. Veux-tu que je te l'affiche ?"
Fonctionnalités Avancées
1. Text-to-Speech (TTS) - Voix de l'assistant
Pour que votre assistant puisse parler, utilisez Piper TTS (local et rapide) :
# Installation Piper
pip install piper-tts
# Télécharger une voix française
wget https://huggingface.co/rhasspy/piper-voices/resolve/main/fr/fr_FR/siwis/medium/fr_FR-siwis-medium.onnx
wget https://huggingface.co/rhasspy/piper-voices/resolve/main/fr/fr_FR/siwis/medium/fr_FR-siwis-medium.onnx.json
Intégration Python :
from piper import PiperVoice
import wave
class TTSManager:
def __init__(self, model_path: str):
self.voice = PiperVoice.load(model_path)
def speak(self, text: str, output_file: str = "output.wav"):
"""Convertit texte en audio"""
with wave.open(output_file, 'wb') as wav_file:
self.voice.synthesize(text, wav_file)
# Lecture audio
import sounddevice as sd
import soundfile as sf
data, samplerate = sf.read(output_file)
sd.play(data, samplerate)
sd.wait()
return output_file
# Utilisation
tts = TTSManager("fr_FR-siwis-medium.onnx")
tts.speak("Bonjour ! Je suis votre assistant personnel.")
2. Speech-to-Text (STT) - Commandes vocales
Pour reconnaître la voix de l'utilisateur, utilisez Whisper :
# Installation
pip install openai-whisper
# ou version optimisée:
pip install faster-whisper
Implémentation :
from faster_whisper import WhisperModel
class STTManager:
def __init__(self, model_size: str = "base"):
# Modèles: tiny, base, small, medium, large
self.model = WhisperModel(model_size, device="cpu", compute_type="int8")
def transcribe(self, audio_file: str, language: str = "fr") -> str:
"""Transcrit audio en texte"""
segments, info = self.model.transcribe(
audio_file,
language=language,
beam_size=5
)
text = " ".join([segment.text for segment in segments])
return text.strip()
# Utilisation
stt = STTManager("base")
text = stt.transcribe("voice_command.wav", language="fr")
print(f"Commande détectée: {text}")
3. Détection de Wake Word
Pour activer l'assistant avec "Hey Assistant" :
# Installation
pip install pvporcupine
# Code de détection
import pvporcupine
import pyaudio
import struct
class WakeWordDetector:
def __init__(self, keyword_path: str, access_key: str):
self.porcupine = pvporcupine.create(
access_key=access_key,
keyword_paths=[keyword_path]
)
self.audio_stream = pyaudio.PyAudio().open(
rate=self.porcupine.sample_rate,
channels=1,
format=pyaudio.paInt16,
input=True,
frames_per_buffer=self.porcupine.frame_length
)
def listen(self):
"""Écoute en continu pour le wake word"""
print("🎤 En écoute du wake word...")
while True:
pcm = self.audio_stream.read(self.porcupine.frame_length)
pcm = struct.unpack_from("h" * self.porcupine.frame_length, pcm)
keyword_index = self.porcupine.process(pcm)
if keyword_index >= 0:
print("✅ Wake word détecté !")
return True
# Utilisation
detector = WakeWordDetector(
keyword_path="path/to/hey-assistant.ppn",
access_key="your_picovoice_key"
)
if detector.listen():
# Wake word détecté, traiter la commande vocale
process_voice_command()
Base de Données SQLite - Persistance
Schéma de la base de données
import sqlite3
from datetime import datetime
def init_database(db_path: str = "/data/assistant.db"):
"""Initialise la base de données"""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Table conversations
cursor.execute('''
CREATE TABLE IF NOT EXISTS conversations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
device_id TEXT NOT NULL,
user_message TEXT NOT NULL,
assistant_response TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
tokens_used INTEGER
)
''')
# Table devices
cursor.execute('''
CREATE TABLE IF NOT EXISTS devices (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
platform TEXT,
last_seen DATETIME,
capabilities TEXT,
status TEXT DEFAULT 'offline'
)
''')
# Table routines
cursor.execute('''
CREATE TABLE IF NOT EXISTS routines (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
trigger_type TEXT NOT NULL,
trigger_value TEXT,
actions TEXT NOT NULL,
enabled BOOLEAN DEFAULT 1,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
# Table user_context
cursor.execute('''
CREATE TABLE IF NOT EXISTS user_context (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
conn.commit()
conn.close()
print("✅ Base de données initialisée")
# Classe helper
class Database:
def __init__(self, db_path: str = "/data/assistant.db"):
self.db_path = db_path
def save_conversation(self, device_id: str, user_msg: str, assistant_msg: str):
"""Sauvegarde une conversation"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO conversations (device_id, user_message, assistant_response)
VALUES (?, ?, ?)
''', (device_id, user_msg, assistant_msg))
conn.commit()
conn.close()
def get_conversation_history(self, device_id: str, limit: int = 10):
"""Récupère l'historique"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT user_message, assistant_response, timestamp
FROM conversations
WHERE device_id = ?
ORDER BY timestamp DESC
LIMIT ?
''', (device_id, limit))
results = cursor.fetchall()
conn.close()
return [
{"user": r[0], "assistant": r[1], "time": r[2]}
for r in results
]
def update_device_status(self, device_id: str, status: str):
"""Met à jour le statut d'un appareil"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO devices (id, last_seen, status)
VALUES (?, ?, ?)
''', (device_id, datetime.now(), status))
conn.commit()
conn.close()
Routines Automatisées
Configuration des routines
Fichier: config/routines.json
{
"routines": [
{
"name": "morning_routine",
"enabled": true,
"trigger": {
"type": "time",
"value": "07:00"
},
"conditions": [
{
"type": "day_of_week",
"value": ["monday", "tuesday", "wednesday", "thursday", "friday"]
}
],
"actions": [
{
"type": "tts_speak",
"message": "Bonjour ! Bonne journée !",
"device": "all"
},
{
"type": "media_play",
"playlist": "Morning Motivation",
"device": "pc-bureau"
},
{
"type": "open_app",
"app": "vscode",
"device": "pc-bureau"
}
]
},
{
"name": "leaving_home",
"enabled": true,
"trigger": {
"type": "location",
"condition": "leaving",
"radius_meters": 100
},
"actions": [
{
"type": "system_command",
"command": "shutdown /s /t 60",
"device": "pc-bureau"
},
{
"type": "notification",
"message": "PC s'éteindra dans 1 minute",
"device": "smartphone"
}
]
},
{
"name": "night_mode",
"enabled": true,
"trigger": {
"type": "time",
"value": "23:00"
},
"actions": [
{
"type": "device_mode",
"mode": "do_not_disturb",
"device": "all"
},
{
"type": "tts_speak",
"message": "Bonne nuit !",
"device": "smartphone"
}
]
}
]
}
Moteur d'exécution des routines
import json
import asyncio
from datetime import datetime, time
from typing import List, Dict
class RoutineEngine:
def __init__(self, config_path: str = "/config/routines.json"):
with open(config_path, 'r') as f:
self.config = json.load(f)
self.running_routines = {}
async def start(self):
"""Démarre le moteur de routines"""
for routine in self.config['routines']:
if routine['enabled']:
asyncio.create_task(self.run_routine(routine))
async def run_routine(self, routine: Dict):
"""Exécute une routine selon son trigger"""
trigger_type = routine['trigger']['type']
if trigger_type == "time":
await self.time_based_routine(routine)
elif trigger_type == "location":
await self.location_based_routine(routine)
elif trigger_type == "event":
await self.event_based_routine(routine)
async def time_based_routine(self, routine: Dict):
"""Routine basée sur l'heure"""
trigger_time = datetime.strptime(routine['trigger']['value'], "%H:%M").time()
while True:
now = datetime.now().time()
# Vérifier si c'est l'heure
if now.hour == trigger_time.hour and now.minute == trigger_time.minute:
# Vérifier les conditions
if self.check_conditions(routine.get('conditions', [])):
await self.execute_actions(routine['actions'])
# Attendre 60 secondes pour éviter les exécutions multiples
await asyncio.sleep(60)
# Vérifier chaque minute
await asyncio.sleep(60)
def check_conditions(self, conditions: List[Dict]) -> bool:
"""Vérifie si toutes les conditions sont remplies"""
for condition in conditions:
if condition['type'] == "day_of_week":
current_day = datetime.now().strftime("%A").lower()
if current_day not in condition['value']:
return False
return True
async def execute_actions(self, actions: List[Dict]):
"""Exécute une liste d'actions"""
print(f"🔄 Exécution de {len(actions)} actions...")
for action in actions:
action_type = action['type']
if action_type == "tts_speak":
await self.action_speak(action)
elif action_type == "media_play":
await self.action_media_play(action)
elif action_type == "system_command":
await self.action_system_command(action)
elif action_type == "notification":
await self.action_notification(action)
async def action_speak(self, action: Dict):
"""Action: faire parler l'assistant"""
# Intégration avec TTS
print(f"🔊 Speaking: {action['message']}")
async def action_media_play(self, action: Dict):
"""Action: lire des médias"""
print(f"▶️ Playing: {action['playlist']} on {action['device']}")
async def action_system_command(self, action: Dict):
"""Action: exécuter une commande système"""
print(f"⚙️ Executing: {action['command']} on {action['device']}")
# Lancement
async def main():
engine = RoutineEngine()
await engine.start()
# Garder le programme en exécution
while True:
await asyncio.sleep(3600)
if __name__ == "__main__":
asyncio.run(main())
Sécurité et Bonnes Pratiques
1. Authentification JWT
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
import os
SECRET_KEY = os.getenv("JWT_SECRET")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 10080 # 7 jours
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def create_access_token(data: dict):
"""Crée un JWT token"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def verify_token(token: str):
"""Vérifie un JWT token"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
2. Rate Limiting
from collections import defaultdict
from datetime import datetime, timedelta
class RateLimiter:
def __init__(self, max_requests: int = 100, window_seconds: int = 60):
self.max_requests = max_requests
self.window = timedelta(seconds=window_seconds)
self.requests = defaultdict(list)
def is_allowed(self, client_id: str) -> bool:
"""Vérifie si le client peut faire une requête"""
now = datetime.now()
# Nettoyer les anciennes requêtes
self.requests[client_id] = [
req_time for req_time in self.requests[client_id]
if now - req_time < self.window
]
# Vérifier la limite
if len(self.requests[client_id]) >= self.max_requests:
return False
# Ajouter la nouvelle requête
self.requests[client_id].append(now)
return True
# Utilisation dans FastAPI
limiter = RateLimiter(max_requests=100, window_seconds=60)
@app.post("/api/chat")
async def chat(request: Request, message: str):
client_id = request.client.host
if not limiter.is_allowed(client_id):
raise HTTPException(status_code=429, detail="Too many requests")
# Traiter la requête
return {"response": "..."}
3. Chiffrement des données sensibles
from cryptography.fernet import Fernet
import base64
import os
class Encryptor:
def __init__(self, key: str = None):
if key is None:
key = os.getenv("ENCRYPTION_KEY")
self.fernet = Fernet(key.encode())
def encrypt(self, data: str) -> str:
"""Chiffre une chaîne"""
encrypted = self.fernet.encrypt(data.encode())
return base64.b64encode(encrypted).decode()
def decrypt(self, encrypted_data: str) -> str:
"""Déchiffre une chaîne"""
decoded = base64.b64decode(encrypted_data.encode())
decrypted = self.fernet.decrypt(decoded)
return decrypted.decode()
# Génération d'une clé (à faire une seule fois)
# key = Fernet.generate_key()
# print(key.decode()) # Sauvegarder dans .env
# Utilisation
encryptor = Encryptor()
token = encryptor.encrypt("mon_token_secret")
print(f"Token chiffré: {token}")
Monitoring et Maintenance
Script de monitoring
#!/bin/bash
# monitor.sh - Surveillance système
echo "=== Monitoring Assistant IA ==="
echo ""
# Docker containers
echo "📦 Containers Docker:"
docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Ports}}"
echo ""
# RAM usage
echo "💾 Utilisation RAM:"
docker stats --no-stream --format "table {{.Container}}\t{{.MemUsage}}\t{{.CPUPerc}}"
echo ""
# Disk usage
echo "💿 Espace disque:"
du -sh /opt/assistant-ia/*
echo ""
# Logs errors
echo "❌ Dernières erreurs (10 lignes):"
tail -n 10 /opt/assistant-ia/logs/api.log | grep -i error
echo ""
# Connected devices
echo "📱 Appareils connectés:"
curl -s https://yourdomain.com/health | jq '.connected_clients'
echo ""
echo "✅ Monitoring terminé"
Script de backup automatique
#!/bin/bash
# backup.sh - Sauvegarde automatique
BACKUP_DIR="/backup/assistant-ia"
DATE=$(date +%Y%m%d_%H%M%S)
echo "🔄 Démarrage backup..."
# Créer le dossier de backup
mkdir -p $BACKUP_DIR
# Backup base de données
cp /opt/assistant-ia/data/assistant.db $BACKUP_DIR/assistant_${DATE}.db
echo "✅ Base de données sauvegardée"
# Backup configuration
tar -czf $BACKUP_DIR/config_${DATE}.tar.gz /opt/assistant-ia/config/
echo "✅ Configuration sauvegardée"
# Backup logs (derniers 7 jours)
find /opt/assistant-ia/logs/ -mtime -7 -type f -exec cp {} $BACKUP_DIR/ \;
echo "✅ Logs sauvegardés"
# Nettoyer anciens backups (> 30 jours)
find $BACKUP_DIR -mtime +30 -type f -delete
echo "✅ Anciens backups nettoyés"
echo "✅ Backup terminé: $BACKUP_DIR"
Automatisation avec cron :
# Éditer crontab
crontab -e
# Ajouter:
# Backup quotidien à 3h du matin
0 3 * * * /opt/assistant-ia/backup.sh >> /opt/assistant-ia/logs/backup.log 2>&1
# Monitoring toutes les heures
0 * * * * /opt/assistant-ia/monitor.sh >> /opt/assistant-ia/logs/monitor.log 2>&1
Troubleshooting
Problème: Ollama est lent
Solutions :
- ✅ Vérifier RAM disponible:
docker stats - ✅ Réduire
OLLAMA_CONTEXT_LENGTHà 2048 - ✅ Utiliser un modèle plus petit (Llama 3.2 3B)
- ✅ Ajouter swap:
sudo fallocate -l 4G /swapfile && sudo mkswap /swapfile && sudo swapon /swapfile - ✅ Limiter l'historique conversations à 5 messages
Problème: WebSocket se déconnecte
Solutions :
- ✅ Vérifier logs:
tail -f /opt/assistant-ia/logs/api.log - ✅ L'agent a auto-reconnexion (5 secondes)
- ✅ Vérifier firewall port 443
- ✅ Augmenter
proxy_read_timeoutdans Nginx à 86400 (24h)
Problème: SSL ne fonctionne pas
Solutions :
- ✅ Vérifier DNS:
ping yourdomain.com - ✅ Ports 80/443 ouverts:
sudo ufw allow 80/tcp && sudo ufw allow 443/tcp - ✅ Logs Nginx:
tail -f /var/log/nginx/error.log - ✅ Renouveler certificat:
sudo certbot renew
Déploiement Complet - Checklist
✅ Phase 1: Infrastructure (Jour 1)
- [ ] Serveur Linux avec 16GB+ RAM configuré
- [ ] Docker et Docker Compose installés
- [ ] Nginx installé
- [ ] Domaine pointant vers le serveur
- [ ] Ports 80/443 ouverts
✅ Phase 2: Backend (Jours 2-3)
- [ ] Structure /opt/assistant-ia créée
- [ ] docker-compose.yml configuré
- [ ] .env avec secrets générés
- [ ] FastAPI avec routes de base
- [ ] Ollama avec modèle téléchargé
- [ ] Base SQLite initialisée
- [ ] SSL configuré avec Certbot
- [ ] Test:
curl https://yourdomain.com/health
✅ Phase 3: Client Agents (Jours 4-5)
- [ ] Agent Python de base fonctionnel
- [ ] WebSocket connexion stable
- [ ] Handlers commandes système
- [ ] TTS/STT intégré
- [ ] Agent déployé sur appareils
✅ Phase 4: Fonctionnalités (Jours 6-14)
- [ ] Contrôle médias (VLC, Spotify)
- [ ] Gestion messages (Telegram)
- [ ] Caméra et photos
- [ ] Localisation GPS
- [ ] Routines automatisées
- [ ] Wake word detection
✅ Phase 5: Production (Semaine 3+)
- [ ] Monitoring automatique
- [ ] Backups quotidiens
- [ ] Rate limiting
- [ ] Logs rotatifs
- [ ] Documentation utilisateur
Évolutions Futures
Intégrations possibles
- 🏠 Home Assistant: Contrôle domotique (lumières, thermostats)
- 📊 Grafana: Dashboards de monitoring
- 🔔 Notifications Push: via Firebase ou NTFY
- 🎮 Game integration: Contrôle jeux PC
- 📧 Email: Lecture et réponse automatique
- 📅 Calendrier: Google Calendar, Outlook
- 🚗 Véhicule connecté: Via API constructeur
- 🌐 Web scraping: Surveillance prix, news
Améliorations IA
- 🧠 RAG (Retrieval-Augmented Generation): Mémoire longue durée avec ChromaDB
- 👁️ Vision: LLaVA pour analyse d'images
- 🎵 Reconnaissance musicale: Shazam-like local
- 🗺️ Navigation: Assistant de trajet
- 📚 Knowledge Base: RAG sur vos documents personnels
Conclusion
Vous avez maintenant tous les éléments pour construire votre propre assistant IA personnel 100% local et privé. Ce système offre :
- ✅ Confidentialité totale - Vos données restent sur votre serveur
- ✅ Contrôle complet - Personnalisez chaque aspect
- ✅ Multi-plateformes - Windows, Linux, Android
- ✅ Évolutif - Ajoutez vos propres fonctionnalités
- ✅ Économique - Pas d'abonnement mensuel
- ✅ Open Source - Technologies libres et gratuites
Points clés à retenir :
- 🔹 Llama 3.1 8B offre le meilleur compromis pour un serveur 16GB RAM
- 🔹 Docker simplifie le déploiement et la maintenance
- 🔹 SQLite est suffisant pour commencer, PostgreSQL pour scale
- 🔹 WebSocket permet la communication temps réel
- 🔹 Nginx + SSL est essentiel pour la sécurité
- 🔹 Les routines automatisées transforment l'assistant en JARVIS
💡 Conseil: Commencez simple avec les fonctionnalités de base, puis ajoutez progressivement les features avancées. Rome ne s'est pas construite en un jour !
Prochaines étapes :
- Provisionner votre serveur Linux
- Suivre la checklist de déploiement
- Tester avec un appareil
- Déployer sur tous vos appareils
- Personnaliser selon vos besoins
N'hésitez pas à adapter ce guide à votre infrastructure et vos besoins spécifiques. L'assistant IA du futur, c'est celui que vous construisez !
Article rédigé par un développeur passionné d'IA et de privacy.
Dernière mise à jour : Février 2026
Technologies utilisées : Python, FastAPI, Docker, Ollama, Llama, WebSocket, Nginx, SQLite