feat(mistral): implement Mistral TTS plugin and STT configs

Author: CarltonBags

plugins/mistral/src/models.ts

@@ -16,8 +16,7 @@ export type MistralChatModels =
 
 export type MistralSTTModels =
   | 'voxtral-mini-transcribe-realtime-2602' //realtime streaming
-  | 'voxtral-small-latest' //chat completions
-  | 'voxtral-mini-latest' //chat completions
-  | 'voxtral-mini-transcribe'; //chat completions
+  | 'voxtral-mini-2602' //batch transcription
+  | 'voxtral-mini-transcribe-2507'; //batch transcription (deprecated)
 
 export type MistralTTSModels = 'voxtral-mini-tts-2603';

plugins/mistral/src/stt.ts

@@ -199,6 +199,7 @@ export class SpeechStream extends stt.SpeechStream {
 
     let connection: any;
     let sendAudioTask: Promise<void> | undefined;
+    let sendError: unknown;
 
     try {
       connection = await this.#client.connect(this.#stt.options.liveModel, {
@@ -233,8 +234,11 @@ export class SpeechStream extends stt.SpeechStream {
             );
             await connection.sendAudio(new Uint8Array(pcmBuffer));
           }
-        } catch (err) {
-          // Stream writing closed or errored
+        } catch (err: unknown) {
+          if (!stopRequested) {
+            sendError = err;
+            connection.close().catch(() => {});
+          }
         } finally {
           if (!connection.isClosed) {
             await connection.flushAudio().catch(() => {});
@@ -342,7 +346,13 @@ export class SpeechStream extends stt.SpeechStream {
           });
         }
       }
+
+      if (sendError) {
+        throw sendError;
+      }
     } catch (error: unknown) {
+      error = sendError ?? error;
+
       // An aborted signal means the stream was intentionally closed — do not
       // wrap into APIConnectionError, which would trigger the retry loop.
       if (this.abortController.signal.aborted) throw error;
@@ -386,7 +396,7 @@ export class SpeechStream extends stt.SpeechStream {
       stopRequested = true;
       resolveAbortTask();
       if (connection) {
-        await connection.close();
+        await connection.close().catch(() => {});
       }
       if (sendAudioTask) {
         await sendAudioTask;

plugins/mistral/src/tts.ts

@@ -10,6 +10,7 @@ import {
   tts,
 } from '@livekit/agents';
 import { Mistral } from '@mistralai/mistralai';
+import * as crypto from 'node:crypto';
 import type { MistralTTSModels } from './models.js';
 
 // Confirmed from WAV header: Mistral TTS PCM output is 24000 Hz, mono, 16-bit signed
@@ -92,8 +93,12 @@ export class TTS extends tts.TTS {
     }));
   }
 
-  synthesize(text: string, connOptions?: APIConnectOptions): ChunkedStream {
-    return new ChunkedStream(this, text, this.#client, this.#opts, connOptions);
+  synthesize(
+    text: string,
+    connOptions?: APIConnectOptions,
+    abortSignal?: AbortSignal,
+  ): ChunkedStream {
+    return new ChunkedStream(this, text, this.#client, this.#opts, connOptions, abortSignal);
   }
 
   stream(): tts.SynthesizeStream {
@@ -117,8 +122,9 @@ export class ChunkedStream extends tts.ChunkedStream {
     client: Mistral,
     opts: TTSOptions,
     connOptions?: APIConnectOptions,
+    abortSignal?: AbortSignal,
   ) {
-    super(text, ttsInstance, connOptions);
+    super(text, ttsInstance, connOptions, abortSignal);
     this.#client = client;
     this.#opts = opts;
     this.#text = text;
@@ -127,15 +133,21 @@ export class ChunkedStream extends tts.ChunkedStream {
   protected async run(): Promise<void> {
     const logger = log();
     try {
-      const eventStream = await this.#client.audio.speech.complete({
-        input: this.#text,
-        model: this.#opts.model ?? 'voxtral-mini-tts-2603',
-        voiceId: this.#opts.voiceId,
-        responseFormat: 'pcm',
-        stream: true,
-      });
+      const eventStream = await this.#client.audio.speech.complete(
+        {
+          input: this.#text,
+          model: this.#opts.model ?? 'voxtral-mini-tts-2603',
+          voiceId: this.#opts.voiceId,
+          responseFormat: 'pcm',
+          stream: true,
+        },
+        {
+          fetchOptions: { signal: this.abortController?.signal },
+        },
+      );
 
-      const requestId = this.#text.slice(0, 8);
+      const requestId = crypto.randomUUID();
+      const segmentId = crypto.randomUUID();
       const audioByteStream = new AudioByteStream(MISTRAL_TTS_SAMPLE_RATE, MISTRAL_TTS_CHANNELS);
 
       let lastFrame: import('@livekit/rtc-node').AudioFrame | undefined;
@@ -152,7 +164,7 @@ export class ChunkedStream extends tts.ChunkedStream {
           const pcmBytes = Buffer.from(event.data.audioData, 'base64');
           const frames = audioByteStream.write(pcmBytes);
           for (const frame of frames) {
-            sendLastFrame(requestId, false);
+            sendLastFrame(segmentId, false);
             lastFrame = frame;
           }
         } else if (event.data.type === 'speech.audio.done') {
@@ -163,11 +175,11 @@ export class ChunkedStream extends tts.ChunkedStream {
       // Flush any remaining buffered audio
       const flushFrames = audioByteStream.flush();
       for (const frame of flushFrames) {
-        sendLastFrame(requestId, false);
+        sendLastFrame(segmentId, false);
         lastFrame = frame;
       }
 
-      sendLastFrame(requestId, true);
+      sendLastFrame(segmentId, true);
       this.queue.close();
     } catch (error: unknown) {
       if (this.abortController?.signal.aborted) return;
@@ -204,8 +216,6 @@ export class ChunkedStream extends tts.ChunkedStream {
         message: `Mistral TTS: ${err.message ?? 'unknown error'}`,
         options: { retryable: true },
       });
-    } finally {
-      this.queue.close();
     }
   }
 }