Transcrire pendant les reponses et regrouper les interventions vocales

chatbot.py

@@ -402,7 +402,7 @@ async def search_web(user_text, force=False):
     )
 
 
-async def call_openai_api(user_text, user_name):
+async def call_openai_api(user_text, user_name, voice=False):
     # Préparer le contenu pour l'appel API
     message_to_send = {
         "role": "user",
@@ -439,6 +439,15 @@ async def call_openai_api(user_text, user_name):
             }, "required": ["query"]}
         }}], "tool_choice": "auto"}
 
+    if voice:
+        messages[0]["content"] += (
+            "\nLe message suivant regroupe des transcriptions du vocal Discord, avec leurs auteurs. "
+            "Il peut contenir des erreurs de reconnaissance, notamment sur les noms propres. "
+            "Interprète les petites erreurs à l'aide du contexte sans inventer les propos. "
+            "Si une ambiguïté change le sens, demande une précision. "
+            "Réponds naturellement à l'ensemble des interventions, en distinguant les personnes."
+        )
+
     try:
         response = await openai_client.chat.completions.create(
             model=MODEL,
@@ -559,15 +568,18 @@ async def handle_voice_command(message):
             await message.channel.send("Je suis déjà à l'écoute. Utilise !quit pour arrêter.")
             return
 
-        async def respond_to_voice(user, text):
+        async def display_transcription(user, text):
+            heard = discord.utils.escape_markdown(text).replace('\n', ' ')
+            name = discord.utils.escape_markdown(user.display_name)
+            for part in split_message(f"🎙 {name} : {heard}", max_length=1900):
+                await message.channel.send('> ' + part, allowed_mentions=discord.AllowedMentions.none())
+
+        async def respond_to_voice(batch):
+            text = "\n".join(f"{user.display_name} : {phrase}" for user, phrase in batch)
             async with message.channel.typing():
                 try:
-                    heard = discord.utils.escape_markdown(text).replace('\n', ' ')
-                    name = discord.utils.escape_markdown(user.display_name)
-                    for part in split_message(f"🎙 {name} : {heard}", max_length=1900):
-                        await message.channel.send('> ' + part, allowed_mentions=discord.AllowedMentions.none())
                     logger.info("Vocal : transcription transmise au LLM (%d caractères)", len(text))
-                    result = await call_openai_api(text, user.display_name)
+                    result = await call_openai_api(text, "Conversation vocale", voice=True)
                     reply = result.choices[0].message.content if result else None
                     reply = transform_emote(filter_message(reply or ''), True)
                     for part in split_message(reply or "Désolé, le modèle ne répond pas pour le moment."):
@@ -585,14 +597,14 @@ async def handle_voice_command(message):
             voice = await message.author.voice.channel.connect(
                 cls=voice_recv.VoiceRecvClient, self_deaf=False,
             )
-            sink = STTSink(respond_to_voice, STT_URL)
+            sink = STTSink(respond_to_voice, STT_URL, on_transcription=display_transcription)
             voice.listen(sink, after=sink.receiver_stopped)
             state = message.guild.me.voice
             logger.info("Écoute vocale démarrée : connecté=%s écoute=%s self_deaf=%s server_deaf=%s",
                         voice.is_connected(), voice.is_listening(),
                         getattr(state, 'self_deaf', None), getattr(state, 'deaf', None))
             sink.log_diagnostic('connexion')
-            await message.channel.send("🎧 J'écoute le vocal et réponds ici à une phrase à la fois. Pendant ma réponse, les autres phrases sont ignorées. !quit pour arrêter.")
+            await message.channel.send("🎧 J'écoute le vocal et réponds ici à une phrase à la fois. Les transcriptions s'affichent au fil des phrases ; celles reçues pendant ma réponse seront regroupées pour la suivante. !quit pour arrêter.")
         except Exception as exc:
             logger.warning("Connexion vocale impossible : %s", type(exc).__name__)
             if sink:

test_voice_stt.py

@@ -36,8 +36,7 @@ class VoiceTests(unittest.IsolatedAsyncioTestCase):
         self.sink.queue.task_done()
         self.assertEqual(user.id, 1)
         self.assertEqual(pcm, self.pcm)
-        self.assertFalse(self.sink.buffers)
-        self.assertTrue(self.sink.busy)
+        self.assertIn(2, self.sink.buffers)
 
     async def test_audio_is_bounded_and_short_noise_ignored(self):
         self.sink.worker_task.cancel()
@@ -76,55 +75,48 @@ class VoiceTests(unittest.IsolatedAsyncioTestCase):
             await asyncio.wait_for(self.sink.queue.join(), 2)
             self.sink.queue.put_nowait((self.user, self.pcm))
             await asyncio.wait_for(self.sink.queue.join(), 2)
-            self.reply.assert_awaited_once_with(self.user, 'Bonjour')
+            self.reply.assert_awaited_once_with([(self.user, 'Bonjour')])
             payload = post.call_args.kwargs['files']['file'][1]
             with wave.open(BytesIO(payload)) as wav:
                 self.assertEqual((wav.getnchannels(), wav.getsampwidth(), wav.getframerate()), (2, 2, 48000))
                 self.assertEqual(wav.readframes(wav.getnframes()), self.pcm)
 
-    async def test_no_backlog_during_stt_and_reply(self):
+    async def test_transcriptions_continue_and_next_reply_groups_speakers(self):
         await self.sink.close()
-        stt_started, release_stt = asyncio.Event(), asyncio.Event()
-        reply_started, release_reply = asyncio.Event(), asyncio.Event()
-
-        async def post(*args, **kwargs):
-            stt_started.set()
-            await release_stt.wait()
-            return httpx.Response(200, json={'text': 'Bonjour'},
-                                  request=httpx.Request('POST', 'http://stt'))
-
-        async def reply(*args):
-            reply_started.set()
-            await release_reply.wait()
+        started, release, second = asyncio.Event(), asyncio.Event(), asyncio.Event()
+        displayed = AsyncMock()
+        batches = []
+
+        async def reply(batch):
+            batches.append(batch)
+            if len(batches) == 1:
+                started.set()
+                await release.wait()
+            else:
+                second.set()
 
         http = AsyncMock()
-        http.__aenter__.return_value.post.side_effect = post
-        self.reply.side_effect = reply
+        http.__aenter__.return_value.post.side_effect = [
+            httpx.Response(200, json={'text': text}, request=httpx.Request('POST', 'http://stt'))
+            for text in ('Bonjour', 'Minecraft ?', 'Avec des mods')
+        ]
         other = SimpleNamespace(id=2)
         with patch('voice_stt.httpx.AsyncClient', return_value=http):
-            self.sink = STTSink(self.reply, 'http://stt/transcribe')
-            with self.sink.lock:
-                self.sink.add_pcm(self.user, self.pcm, time.monotonic() - 2)
-                self.sink.add_pcm(other, self.pcm, time.monotonic() - 2)
-            await asyncio.wait_for(stt_started.wait(), 2)
-            with self.sink.lock:
-                self.sink.add_pcm(other, self.pcm, time.monotonic())
-            release_stt.set()
-            await asyncio.wait_for(reply_started.wait(), 2)
-            with self.sink.lock:
-                self.sink.add_pcm(other, self.pcm, time.monotonic())
-            self.assertTrue(self.sink.queue.empty())
-            self.assertFalse(self.sink.buffers)
-            release_reply.set()
+            self.sink = STTSink(reply, 'http://stt/transcribe', on_transcription=displayed)
+            self.sink.queue.put_nowait((self.user, self.pcm))
+            await asyncio.wait_for(started.wait(), 2)
+            self.sink.queue.put_nowait((other, self.pcm))
+            self.sink.queue.put_nowait((self.user, self.pcm))
             await asyncio.wait_for(self.sink.queue.join(), 2)
-            self.assertFalse(self.sink.busy)
-            self.reply.assert_awaited_once_with(self.user, 'Bonjour')
+            self.assertEqual(displayed.await_count, 3)
+            self.assertEqual(len(batches), 1)
             with self.sink.lock:
-                now = time.monotonic()
-                self.sink.add_pcm(other, self.pcm, now)
-                self.assertFalse(self.sink.buffers)  # Ne pas récupérer la fin d'une phrase ignorée.
-                self.sink.add_pcm(other, self.pcm, now + 1)
+                self.sink.add_pcm(other, self.pcm, time.monotonic())
                 self.assertIn(2, self.sink.buffers)
+            release.set()
+            await asyncio.wait_for(second.wait(), 2)
+            self.assertEqual(batches[1], [(other, 'Minecraft ?'), (self.user, 'Avec des mods')])
+            self.assertFalse(self.sink.pending)
 
     async def test_close_discards_pending_audio(self):
         self.sink.queue.put_nowait((self.user, self.pcm))
@@ -133,6 +125,7 @@ class VoiceTests(unittest.IsolatedAsyncioTestCase):
         self.assertTrue(self.sink.queue.empty())
         self.assertTrue(self.sink.flush_task.done())
         self.assertTrue(self.sink.worker_task.done())
+        self.assertTrue(self.sink.reply_task.done())
         self.reply.assert_not_awaited()
 
     async def test_diagnostics_distinguish_ignored_audio_and_dave_failure(self):
@@ -198,8 +191,9 @@ class CommandTests(unittest.IsolatedAsyncioTestCase):
         await namespace['handle_voice_command'](message)
         voice.listen.assert_called_once_with(sink, after=sink.receiver_stopped)
         callback = make_sink.call_args.args[0]
-        await callback(SimpleNamespace(display_name='Alice'), 'Bonjour')
-        call_api.assert_awaited_once_with('Bonjour', 'Alice')
+        await make_sink.call_args.kwargs['on_transcription'](SimpleNamespace(display_name='Alice'), 'Bonjour')
+        await callback([(SimpleNamespace(display_name='Alice'), 'Bonjour')])
+        call_api.assert_awaited_once_with('Alice : Bonjour', 'Conversation vocale', voice=True)
         self.assertEqual(message.channel.send.call_args.args[0], 'Réponse')
         quote = message.channel.send.call_args_list[-2]
         self.assertEqual(quote.args[0], '> 🎙 Alice : Bonjour')

voice_stt.py

@@ -17,25 +17,28 @@ logger = logging.getLogger(__name__)
 
 
 class STTSink(voice_recv.AudioSink):
-    def __init__(self, on_phrase, stt_url):
+    def __init__(self, on_phrase, stt_url, on_transcription=None):
         super().__init__()
         self.on_phrase = on_phrase
+        self.on_transcription = on_transcription
+        self.pending = []
+        self.pending_event = asyncio.Event()
         self.stt_url = stt_url
         self.loop = asyncio.get_running_loop()
         self.lock = threading.Lock()
         self.buffers = {}
-        self.busy = False
-        self.ignored_speakers = {}
         self.decoders = {}
         self.closed = False
         self.last_error = 0
         self.stats_lock = threading.Lock()
         self.stats = Counter(audio=0, rtcp=0, pcm=0, audible=0, phrases=0, stt=0)
         self.last_diagnostic = time.monotonic()
-        # Uniquement le transfert vers le worker, aucune phrase en attente.
-        self.queue = asyncio.Queue(maxsize=1)
+        # File audio bornée si Whisper ne suit plus le rythme du vocal.
+        self.queue = asyncio.Queue(maxsize=32)
         self.flush_task = self.loop.create_task(self.flush_phrases())
         self.worker_task = self.loop.create_task(self.transcribe_phrases())
+        self.reply_task = self.loop.create_task(self.respond_batches())
+        self.reply_task.add_done_callback(self.task_finished)
         self.flush_task.add_done_callback(self.task_finished)
         self.worker_task.add_done_callback(self.task_finished)
 
@@ -119,13 +122,6 @@ class STTSink(voice_recv.AudioSink):
         """Appelé sous le verrou ; PCM Discord = 48 kHz, stéréo, 16 bits."""
         audible = audioop.rms(pcm, 2) >= 200
         self.count('audible' if audible else 'below_threshold')
-        last_ignored = self.ignored_speakers.get(user.id)
-        if self.busy or (last_ignored is not None and now - last_ignored < 0.8):
-            if audible:
-                self.ignored_speakers[user.id] = now
-            self.count('busy_ignored')
-            return
-        self.ignored_speakers.pop(user.id, None)
         if user.id not in self.buffers:
             if not audible:
                 return
@@ -146,19 +142,11 @@ class STTSink(voice_recv.AudioSink):
                 self.last_diagnostic = now
             ready = []
             with self.lock:
-                if self.busy:
-                    continue
                 for user_id, (user, pcm, last_voice) in list(self.buffers.items()):
                     if now - last_voice >= 0.8 or len(pcm) >= 48000 * 4 * 20:
                         del self.buffers[user_id]
                         if len(pcm) >= 48000 * 4 * 0.25:
                             ready.append((user, bytes(pcm)))
-                            self.busy = True
-                            # Oublier aussi les phrases déjà commencées par les autres.
-                            self.ignored_speakers.update(
-                                {uid: buffer[2] for uid, buffer in self.buffers.items()})
-                            self.buffers.clear()
-                            break
                         else:
                             self.count('too_short')
             for item in ready:
@@ -172,7 +160,7 @@ class STTSink(voice_recv.AudioSink):
                     logger.warning("File vocale pleine : une phrase a été ignorée")
 
     async def transcribe_phrases(self):
-        # Rester occupé jusqu'à la fin de la transcription ET de la réponse.
+        # La transcription continue pendant les réponses du LLM.
         async with httpx.AsyncClient(timeout=60) as http:
             while not self.closed:
                 user, pcm = await self.queue.get()
@@ -198,8 +186,14 @@ class STTSink(voice_recv.AudioSink):
                     if not text:
                         self.count('empty_transcription')
                     if text and not self.closed:
-                        stage = 'réponse'
-                        await self.on_phrase(user, text)
+                        try:
+                            stage = 'affichage'
+                            if self.on_transcription:
+                                await self.on_transcription(user, text)
+                        finally:
+                            if not self.closed:
+                                self.pending.append((user, text))
+                                self.pending_event.set()
                 except asyncio.CancelledError:
                     raise
                 except Exception as exc:
@@ -208,17 +202,28 @@ class STTSink(voice_recv.AudioSink):
                     logger.warning("Traitement vocal impossible : étape=%s erreur=%s HTTP=%s durée=%.2fs",
                                    stage, type(exc).__name__, status, time.monotonic() - started)
                 finally:
-                    with self.lock:
-                        self.busy = False
                     self.queue.task_done()
 
+    async def respond_batches(self):
+        while not self.closed:
+            await self.pending_event.wait()
+            self.pending_event.clear()
+            batch, self.pending = self.pending, []
+            if not batch:
+                continue
+            try:
+                await self.on_phrase(batch)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                logger.warning("Réponse au lot vocal impossible : %s", type(exc).__name__)
+
     @voice_recv.AudioSink.listener()
     def on_voice_member_disconnect(self, member, ssrc):
         if member:
             with self.lock:
                 self.buffers.pop(member.id, None)
                 self.decoders.pop(member.id, None)
-                self.ignored_speakers.pop(member.id, None)
 
     def cleanup(self):
         # La bibliothèque appelle cleanup depuis son thread de réception.
@@ -228,17 +233,19 @@ class STTSink(voice_recv.AudioSink):
             self.closed = True
             self.buffers.clear()
             self.decoders.clear()
-            self.ignored_speakers.clear()
+            self.pending.clear()
         self.log_diagnostic('arrêt')
         if not self.loop.is_closed():
             self.loop.call_soon_threadsafe(self.flush_task.cancel)
             self.loop.call_soon_threadsafe(self.worker_task.cancel)
+            self.loop.call_soon_threadsafe(self.reply_task.cancel)
 
     async def close(self):
         self.cleanup()
         self.flush_task.cancel()
         self.worker_task.cancel()
-        await asyncio.gather(self.flush_task, self.worker_task, return_exceptions=True)
+        self.reply_task.cancel()
+        await asyncio.gather(self.flush_task, self.worker_task, self.reply_task, return_exceptions=True)
         while not self.queue.empty():
             self.queue.get_nowait()
             self.queue.task_done()