feat: diarisation des locuteurs en temps réel (ECAPA-TDNN + clustering incrémental)

- Nouveau module diarizer.py : EmbeddingModel (SpeechBrain ECAPA-TDNN) + SpeakerDiarizer par session
- Backend : chargement modèle au lifespan, migration DB (colonne speaker), worker enrichi, exports avec locuteur
- Frontend : badges locuteurs colorés (8 couleurs), toggle activation, copie avec [Locuteur X]
- Config : DIARIZATION=on/off, DIARIZATION_THRESHOLD=0.70, PyTorch CPU-only dans Dockerfile
- Rétrocompatible : désactivé par défaut (DIARIZATION=off)

.env.example

@@ -11,3 +11,11 @@ ASR_API_KEY=sk-local
 
 # Jeton Hugging Face, seulement si le téléchargement du modèle l'exige.
 HF_TOKEN=
+
+# Diarisation des locuteurs : on / off (défaut : off).
+# Nécessite le modèle ECAPA-TDNN (~25 Mo, téléchargé au premier lancement).
+DIARIZATION=off
+
+# Seuil de similarité cosinus pour regrouper deux segments au même locuteur.
+# Plus bas = plus permissif (risque de fusion), plus haut = plus strict (risque de split).
+DIARIZATION_THRESHOLD=0.70

app/Dockerfile

@@ -7,7 +7,9 @@ RUN apt-get update \
 WORKDIR /srv
 
 COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
+# PyTorch CPU-only (plus léger ~200 MB vs ~2 GB avec CUDA)
+RUN pip install --no-cache-dir torch torchaudio --index-url https://download.pytorch.org/whl/cpu \
+    && pip install --no-cache-dir -r requirements.txt
 
 COPY . .
 RUN chmod +x entrypoint.sh

app/diarizer.py

@@ -0,0 +1,164 @@
+"""Identification des locuteurs par embeddings ECAPA-TDNN + clustering incrémental.
+
+Chaque segment de parole (PCM 16 kHz mono) est projeté dans un espace de
+représentation de dimension 192 via ECAPA-TDNN (SpeechBrain).  On maintient
+un profil par locuteur (moyenne mobile exponentielle des embeddings) et on
+attribue chaque nouveau segment au locuteur le plus proche par similarité
+cosinus, ou on crée un nouveau locuteur si le score est sous le seuil.
+"""
+
+from __future__ import annotations
+
+import struct
+from dataclasses import dataclass, field
+
+import numpy as np
+
+# Imports lourds (torch, speechbrain) chargés paresseusement dans load_model()
+# pour ne pas pénaliser le démarrage quand la diarisation est désactivée.
+
+
+@dataclass
+class _SpeakerProfile:
+    """Profil incrémental d'un locuteur au sein d'une session."""
+
+    label: str
+    embedding: np.ndarray            # centroïde courant (moyenne mobile)
+    count: int = 0                   # nombre d'observations
+    _ema_alpha: float = field(default=0.3, repr=False)
+
+    def update(self, new_emb: np.ndarray) -> None:
+        """Met à jour le centroïde via une moyenne mobile exponentielle."""
+        self.count += 1
+        if self.count == 1:
+            self.embedding = new_emb.copy()
+        else:
+            self.embedding = (
+                self._ema_alpha * new_emb
+                + (1 - self._ema_alpha) * self.embedding
+            )
+        # Renormaliser pour que la similarité cosinus reste cohérente
+        norm = np.linalg.norm(self.embedding)
+        if norm > 0:
+            self.embedding /= norm
+
+
+def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
+    """Similarité cosinus entre deux vecteurs unitaires."""
+    return float(np.dot(a, b))
+
+
+class EmbeddingModel:
+    """Encapsule le modèle SpeechBrain ECAPA-TDNN (singleton partagé).
+
+    Chargé une seule fois au démarrage de l'application, puis réutilisé
+    par chaque session WebSocket via des instances de SpeakerDiarizer.
+    """
+
+    def __init__(self) -> None:
+        import torch  # noqa: F811 — import local volontaire
+        from speechbrain.inference.speaker import EncoderClassifier
+
+        self._device = "cpu"
+        self._model = EncoderClassifier.from_hparams(
+            source="speechbrain/spkrec-ecapa-voxceleb",
+            savedir="/data/models/ecapa-tdnn",
+            run_opts={"device": self._device},
+        )
+        self._torch = torch
+
+    def extract(self, pcm: bytes, sample_rate: int = 16000) -> np.ndarray:
+        """Extrait un embedding 192-d à partir d'un segment PCM 16-bit mono.
+
+        Returns:
+            np.ndarray de forme (192,), normalisé L2.
+
+        Raises:
+            ValueError: si le segment audio est trop court (< 200 ms).
+        """
+        n_samples = len(pcm) // 2
+        min_samples = sample_rate // 5  # 200 ms minimum
+        if n_samples < min_samples:
+            raise ValueError(
+                f"Segment trop court ({n_samples} samples, min {min_samples})"
+            )
+
+        # PCM 16-bit little-endian → float32 [-1, 1]
+        samples = struct.unpack(f"<{n_samples}h", pcm)
+        waveform = self._torch.tensor(samples, dtype=self._torch.float32) / 32768.0
+        waveform = waveform.unsqueeze(0)  # (1, T)
+
+        with self._torch.no_grad():
+            embedding = self._model.encode_batch(waveform)
+
+        emb = embedding.squeeze().cpu().numpy()  # (192,)
+        # Normaliser L2
+        norm = np.linalg.norm(emb)
+        if norm > 0:
+            emb /= norm
+        return emb
+
+
+class SpeakerDiarizer:
+    """Identifie le locuteur de chaque segment audio au sein d'une session.
+
+    Chaque instance correspond à une réunion/session et maintient ses propres
+    profils de locuteurs. Le modèle d'embedding est partagé (EmbeddingModel).
+    """
+
+    def __init__(
+        self,
+        model: EmbeddingModel,
+        threshold: float = 0.70,
+        max_speakers: int = 8,
+    ) -> None:
+        self._model = model
+        self._threshold = threshold
+        self._max_speakers = max_speakers
+        self._profiles: list[_SpeakerProfile] = []
+
+    def identify(self, pcm: bytes, sample_rate: int = 16000) -> str:
+        """Identifie le locuteur d'un segment PCM.
+
+        Returns:
+            Label du locuteur ("Locuteur 1", "Locuteur 2", etc.)
+            ou chaîne vide si le segment est trop court pour être analysé.
+        """
+        try:
+            embedding = self._model.extract(pcm, sample_rate)
+        except ValueError:
+            # Segment trop court — on ne peut pas identifier le locuteur
+            return ""
+
+        # Comparer avec les profils existants
+        best_score = -1.0
+        best_profile: _SpeakerProfile | None = None
+
+        for profile in self._profiles:
+            score = _cosine_similarity(embedding, profile.embedding)
+            if score > best_score:
+                best_score = score
+                best_profile = profile
+
+        if best_profile is not None and best_score >= self._threshold:
+            best_profile.update(embedding)
+            return best_profile.label
+
+        # Nouveau locuteur (sauf si on a atteint le max)
+        if len(self._profiles) >= self._max_speakers:
+            # Forcer l'attribution au profil le plus proche
+            if best_profile is not None:
+                best_profile.update(embedding)
+                return best_profile.label
+            # Cas dégénéré : aucun profil et max atteint (ne devrait pas arriver)
+            return "Locuteur 1"
+
+        new_label = f"Locuteur {len(self._profiles) + 1}"
+        new_profile = _SpeakerProfile(label=new_label, embedding=embedding)
+        new_profile.update(embedding)
+        self._profiles.append(new_profile)
+        return new_label
+
+    def reset(self) -> None:
+        """Réinitialise les profils pour une nouvelle session."""
+        self._profiles.clear()

app/main.py

@@ -19,14 +19,17 @@ ASR_BASE_URL = os.environ.get("ASR_BASE_URL", "http://asr:8000/v1").rstrip("/")
 ASR_MODEL = os.environ.get("ASR_MODEL", "Qwen/Qwen3-ASR-1.7B")
 ASR_API_KEY = os.environ.get("ASR_API_KEY", "sk-local")
 ASR_LANGUAGE = os.environ.get("ASR_LANGUAGE", "").strip()
+DIARIZATION = os.environ.get("DIARIZATION", "off").strip().lower() == "on"
+DIARIZATION_THRESHOLD = float(os.environ.get("DIARIZATION_THRESHOLD", "0.70"))
 
 db: aiosqlite.Connection | None = None
 http: httpx.AsyncClient | None = None
+embedding_model = None  # EmbeddingModel chargé au lifespan si DIARIZATION=True
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    global db, http
+    global db, http, embedding_model
     os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
     db = await aiosqlite.connect(DB_PATH)
     db.row_factory = aiosqlite.Row
@@ -42,12 +45,27 @@ async def lifespan(app: FastAPI):
             meeting_id INTEGER NOT NULL REFERENCES meetings(id) ON DELETE CASCADE,
             t0 REAL NOT NULL,
             t1 REAL NOT NULL,
-            text TEXT NOT NULL
+            text TEXT NOT NULL,
+            speaker TEXT NOT NULL DEFAULT ''
         );
         """
     )
+    # Migration : ajouter la colonne speaker si elle n'existe pas
+    try:
+        await db.execute("ALTER TABLE segments ADD COLUMN speaker TEXT NOT NULL DEFAULT ''")
+        await db.commit()
+    except Exception:
+        pass  # la colonne existe déjà
     await db.execute("PRAGMA foreign_keys = ON")
     await db.commit()
+
+    # Charger le modèle de diarisation (lourd, ~30s au premier lancement)
+    if DIARIZATION:
+        print("Chargement du modèle de diarisation ECAPA-TDNN…", flush=True)
+        from diarizer import EmbeddingModel
+        embedding_model = await asyncio.to_thread(EmbeddingModel)
+        print("Modèle de diarisation prêt.", flush=True)
+
     http = httpx.AsyncClient(timeout=120)
     yield
     await http.aclose()
@@ -100,7 +118,7 @@ async def transcribe(pcm: bytes) -> str:
 async def ws_transcribe(ws: WebSocket):
     await ws.accept()
 
-    # Premier message : {"type": "start", "title": "..."}
+    # Premier message : {"type": "start", "title": "...", "diarization": bool}
     try:
         start = json.loads(await ws.receive_text())
         assert start.get("type") == "start"
@@ -109,6 +127,8 @@ async def ws_transcribe(ws: WebSocket):
         return
 
     title = (start.get("title") or "").strip() or datetime.now().strftime("Réunion du %d/%m/%Y %H:%M")
+    # La diarisation est activée si le serveur la supporte ET le client la demande
+    session_diarization = DIARIZATION and start.get("diarization", True)
     cur = await db.execute(
         "INSERT INTO meetings (title, created_at) VALUES (?, ?)",
         (title, datetime.now(timezone.utc).isoformat()),
@@ -121,11 +141,30 @@ async def ws_transcribe(ws: WebSocket):
     queue: asyncio.Queue[Segment | None] = asyncio.Queue()
 
     async def worker():
-        """Transcrit les segments dans l'ordre et pousse le texte au client."""
+        """Identifie le locuteur puis transcrit, dans l'ordre."""
+        # Chaque session a son propre diarizer (profils locuteurs isolés)
+        diarizer = None
+        if session_diarization and embedding_model is not None:
+            from diarizer import SpeakerDiarizer
+            diarizer = SpeakerDiarizer(
+                model=embedding_model,
+                threshold=DIARIZATION_THRESHOLD,
+            )
+
         while True:
             seg = await queue.get()
             if seg is None:
                 return
+
+            # Diarisation (~30-80 ms CPU, dans un thread pour ne pas bloquer)
+            speaker = ""
+            if diarizer is not None:
+                try:
+                    speaker = await asyncio.to_thread(diarizer.identify, seg.pcm)
+                except Exception as exc:
+                    print(f"Diarisation échouée : {exc}", flush=True)
+
+            # Transcription ASR (~1-5 s réseau)
             try:
                 text = await transcribe(seg.pcm)
             except Exception as exc:
@@ -133,12 +172,17 @@ async def ws_transcribe(ws: WebSocket):
                 continue
             if not text:
                 continue
+
             await db.execute(
-                "INSERT INTO segments (meeting_id, t0, t1, text) VALUES (?, ?, ?, ?)",
-                (meeting_id, seg.t0, seg.t1, text),
+                "INSERT INTO segments (meeting_id, t0, t1, text, speaker) VALUES (?, ?, ?, ?, ?)",
+                (meeting_id, seg.t0, seg.t1, text, speaker),
             )
             await db.commit()
-            await ws.send_json({"type": "segment", "t0": seg.t0, "t1": seg.t1, "text": text})
+            await ws.send_json({
+                "type": "segment",
+                "t0": seg.t0, "t1": seg.t1,
+                "text": text, "speaker": speaker,
+            })
 
     worker_task = asyncio.create_task(worker())
     try:
@@ -200,7 +244,8 @@ async def get_meeting_or_404(meeting_id: int) -> dict:
 async def get_meeting(meeting_id: int):
     meeting = await get_meeting_or_404(meeting_id)
     rows = await db.execute_fetchall(
-        "SELECT t0, t1, text FROM segments WHERE meeting_id = ? ORDER BY id", (meeting_id,)
+        "SELECT t0, t1, text, speaker FROM segments WHERE meeting_id = ? ORDER BY id",
+        (meeting_id,),
     )
     meeting["segments"] = [dict(r) for r in rows]
     return meeting
@@ -237,16 +282,23 @@ async def export_meeting(meeting_id: int, format: str = "txt"):
         body, mime, ext = json.dumps(meeting, ensure_ascii=False, indent=2), "application/json", "json"
     elif format == "md":
         lines = [f"# {title}", ""]
-        lines += [f"**[{fmt_ts(s['t0'])}]** {s['text']}" for s in segs]
+        for s in segs:
+            prefix = f"**{s['speaker']} —** " if s.get("speaker") else ""
+            lines.append(f"**[{fmt_ts(s['t0'])}]** {prefix}{s['text']}")
         body, mime, ext = "\n\n".join(lines) + "\n", "text/markdown", "md"
     elif format == "srt":
-        blocks = [
-            f"{i}\n{fmt_ts(s['t0'], srt=True)} --> {fmt_ts(s['t1'], srt=True)}\n{s['text']}"
-            for i, s in enumerate(segs, 1)
-        ]
+        blocks = []
+        for i, s in enumerate(segs, 1):
+            speaker_line = f"<i>{s['speaker']}</i>\n" if s.get("speaker") else ""
+            blocks.append(
+                f"{i}\n{fmt_ts(s['t0'], srt=True)} --> {fmt_ts(s['t1'], srt=True)}\n"
+                f"{speaker_line}{s['text']}"
+            )
         body, mime, ext = "\n\n".join(blocks) + "\n", "application/x-subrip", "srt"
     elif format == "txt":
-        body, mime, ext = "\n".join(s["text"] for s in segs) + "\n", "text/plain", "txt"
+        def _txt_line(s: dict) -> str:
+            return f"[{s['speaker']}] {s['text']}" if s.get("speaker") else s["text"]
+        body, mime, ext = "\n".join(_txt_line(s) for s in segs) + "\n", "text/plain", "txt"
     else:
         raise HTTPException(400, "Format inconnu (txt, md, srt, json)")
 

app/requirements.txt

@@ -3,3 +3,6 @@ uvicorn[standard]~=0.34
 httpx~=0.28
 aiosqlite~=0.21
 webrtcvad-wheels~=2.0
+speechbrain>=1.0
+torch>=2.0,<3.0
+torchaudio>=2.0,<3.0

app/segmenter.py

@@ -28,6 +28,7 @@ class Segment:
     pcm: bytes
     t0: float  # secondes depuis le début de la réunion
     t1: float
+    speaker: str = ""  # identifié par le diarizer (vide si désactivé)
 
 
 class SpeechSegmenter:

app/static/app.js

@@ -15,6 +15,19 @@ const state = {
 
 const BATCH_SAMPLES = 4096; // ~256 ms de PCM 16 kHz par message WebSocket
 
+// --- Diarisation : palette de couleurs par locuteur ---
+const SPEAKER_COLORS = 8; // nombre de classes CSS .speaker-0 à .speaker-7
+const speakerMap = {};     // "Speaker 1" → 0, "Speaker 2" → 1, ...
+let speakerCounter = 0;
+
+function getSpeakerColorIndex(speaker) {
+  if (!(speaker in speakerMap)) {
+    speakerMap[speaker] = speakerCounter % SPEAKER_COLORS;
+    speakerCounter++;
+  }
+  return speakerMap[speaker];
+}
+
 // ----------------------------------------------------------- enregistrement
 
 async function startRecording() {
@@ -35,7 +48,11 @@ async function startRecording() {
 
   const proto = location.protocol === 'https:' ? 'wss' : 'ws';
   state.ws = new WebSocket(`${proto}://${location.host}/ws`);
-  state.ws.onopen = () => state.ws.send(JSON.stringify({ type: 'start', title: $('title').value }));
+  state.ws.onopen = () => state.ws.send(JSON.stringify({
+    type: 'start',
+    title: $('title').value,
+    diarization: $('diarization-cb').checked,
+  }));
   state.ws.onmessage = onServerMessage;
   state.ws.onclose = () => { if (state.recording) stopRecording(true); };
 
@@ -136,12 +153,20 @@ function fmtTs(seconds) {
 
 function clearTranscript() {
   $('transcript').innerHTML = '';
+  // Reset le mapping locuteurs pour chaque nouvelle session
+  for (const key in speakerMap) delete speakerMap[key];
+  speakerCounter = 0;
 }
 
 function appendSegment(seg) {
   const div = document.createElement('div');
   div.className = 'segment';
-  div.innerHTML = `<span class="ts">${fmtTs(seg.t0)}</span><span class="text"></span>`;
+  let speakerHtml = '';
+  if (seg.speaker) {
+    const ci = getSpeakerColorIndex(seg.speaker);
+    speakerHtml = `<span class="speaker speaker-${ci}">${seg.speaker}</span>`;
+  }
+  div.innerHTML = `<span class="ts">${fmtTs(seg.t0)}</span>${speakerHtml}<span class="text"></span>`;
   div.querySelector('.text').textContent = seg.text;
   $('transcript').appendChild(div);
   $('transcript').scrollTop = $('transcript').scrollHeight;
@@ -199,9 +224,12 @@ async function deleteCurrentMeeting() {
 }
 
 async function copyTranscript() {
-  const text = [...document.querySelectorAll('#transcript .segment .text')]
-    .map((el) => el.textContent).join('\n');
-  await navigator.clipboard.writeText(text);
+  const lines = [...document.querySelectorAll('#transcript .segment')].map((el) => {
+    const speaker = el.querySelector('.speaker');
+    const text = el.querySelector('.text').textContent;
+    return speaker ? `[${speaker.textContent}] ${text}` : text;
+  });
+  await navigator.clipboard.writeText(lines.join('\n'));
   $('copy-btn').textContent = '✓ Copié';
   setTimeout(() => ($('copy-btn').textContent = '📋 Copier'), 1500);
 }

app/static/index.html

@@ -24,6 +24,10 @@
         <input id="title" type="text" placeholder="Titre de la réunion (optionnel)">
         <button id="record-btn" class="record">● Démarrer</button>
         <span id="timer">00:00</span>
+        <label id="diarization-toggle" class="toggle" title="Identifier les locuteurs">
+          <input type="checkbox" id="diarization-cb">
+          <span>👥 Locuteurs</span>
+        </label>
       </section>
 
       <section id="transcript-panel">

app/static/style.css

@@ -133,6 +133,46 @@ button.record.recording { background: var(--rec); }
   flex-shrink: 0;
 }
 
+/* --- Diarisation : badges locuteurs --- */
+.speaker {
+  font-size: 0.78rem;
+  font-weight: 600;
+  padding: 2px 8px;
+  border-radius: 6px;
+  flex-shrink: 0;
+  white-space: nowrap;
+}
+.speaker-0 { background: #1a2440; color: #4f7cff; }
+.speaker-1 { background: #3a181a; color: #e5484d; }
+.speaker-2 { background: #132a1e; color: #30a46c; }
+.speaker-3 { background: #2a2410; color: #f5a623; }
+.speaker-4 { background: #2a1a3a; color: #8e4ec6; }
+.speaker-5 { background: #0f2a28; color: #12a594; }
+.speaker-6 { background: #3a1a2a; color: #e54666; }
+.speaker-7 { background: #1e2024; color: #889096; }
+
+/* Toggle diarisation */
+.toggle {
+  display: flex;
+  align-items: center;
+  gap: 6px;
+  cursor: pointer;
+  font-size: 0.85rem;
+  color: var(--muted);
+  user-select: none;
+  padding: 6px 12px;
+  border-radius: 10px;
+  background: var(--panel-2);
+  border: 1px solid #2c3245;
+  transition: border-color 0.2s, color 0.2s;
+}
+.toggle:hover { border-color: var(--accent); }
+.toggle input { display: none; }
+.toggle:has(input:checked) {
+  border-color: var(--accent);
+  color: var(--text);
+}
+
 @media (max-width: 720px) {
   .layout { flex-direction: column; }
   aside { width: 100%; max-height: 30vh; border-right: none; border-bottom: 1px solid #262b3a; }

docker-compose.yml

@@ -13,6 +13,9 @@ services:
       # Code langue ISO ("fr", "en"...) ou vide pour la détection automatique
       - ASR_LANGUAGE=${ASR_LANGUAGE:-}
       - DB_PATH=/data/liveflow.db
+      # Diarisation des locuteurs (off par défaut pour rétrocompatibilité)
+      - DIARIZATION=${DIARIZATION:-off}
+      - DIARIZATION_THRESHOLD=${DIARIZATION_THRESHOLD:-0.70}
     volumes:
       - liveflow-data:/data
     depends_on: