feat(fishaudio): add realtime STT

.changeset/fish-realtime-stt.md

@@ -0,0 +1,6 @@
+---
+'@livekit/agents': minor
+'@livekit/agents-plugin-fishaudio': minor
+---
+
+Add Fish Audio realtime streaming speech-to-text with turn detection, word timings, keyterms, conversation context, and reconnect handling.

agents/etc/agents.api.md

@@ -885,6 +885,7 @@ export class APIConnectionError extends APIError {
 
 // @public
 export interface APIConnectOptions {
+    livenessTimeoutMs?: number;
     maxRetry: number;
     retryIntervalMs: number;
     timeoutMs: number;
@@ -7611,6 +7612,8 @@ abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent> {
     get terminalError(): Error | undefined;
     // (undocumented)
     updateInputStream(audioStream: ReadableStream_2<AudioFrame>): void;
+    // @internal
+    _waitClosed(): Promise<void>;
 }
 
 // @public (undocumented)
@@ -9908,7 +9911,7 @@ export const zipFunctionCallsAndOutputs: (event: FunctionToolsExecutedEvent) =>
 // src/llm/tool_context.ts:756:3 - (ae-unresolved-link) The @link reference could not be resolved: The reference is ambiguous because "ToolFlag" has more than one declaration; you need to add a TSDoc member reference selector
 // src/metrics/base.ts:226:3 - (ae-forgotten-export) The symbol "RealtimeModelMetricsInputTokenDetails" needs to be exported by the entry point index.d.ts
 // src/metrics/base.ts:230:3 - (ae-forgotten-export) The symbol "RealtimeModelMetricsOutputTokenDetails" needs to be exported by the entry point index.d.ts
-// src/stt/stt.ts:395:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "STT"
+// src/stt/stt.ts:401:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "STT"
 // src/utils.ts:468:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "cancelled"
 // src/voice/agent_session.ts:411:3 - (ae-unresolved-link) The @link reference could not be resolved: This type of declaration is not supported yet by the resolver
 // src/voice/agent_session.ts:1091:5 - (ae-forgotten-export) The symbol "RecordingOptions" needs to be exported by the entry point index.d.ts

agents/src/stt/stt.ts

@@ -345,6 +345,8 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
   protected closed = false;
   #stt: STT;
   #terminalError?: Error;
+  #closedPromise: Promise<void>;
+  #resolveClosed!: () => void;
   private deferredInputStream: DeferredReadableStream<AudioFrame>;
   private logger = log();
   private _connOptions: APIConnectOptions;
@@ -362,6 +364,9 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
     this._connOptions = connectionOptions;
     this.deferredInputStream = new DeferredReadableStream<AudioFrame>();
     this.neededSampleRate = sampleRate;
+    this.#closedPromise = new Promise((resolve) => {
+      this.#resolveClosed = resolve;
+    });
     this.monitorMetrics();
     this.pumpInput();
 
@@ -377,6 +382,7 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
         this.#terminalError = toError(error);
       } finally {
         this.queue.close();
+        this.#resolveClosed();
       }
     };
     startSoon(() => {
@@ -440,7 +446,12 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
           }
 
           if (retryInterval > 0) {
-            await delay(retryInterval);
+            try {
+              await delay(retryInterval, { signal: this.abortController.signal });
+            } catch {
+              if (this.abortController.signal.aborted) return;
+              throw error;
+            }
           }
 
           this._numRetries += 1;
@@ -622,6 +633,11 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
     return this.#terminalError;
   }
 
+  /** Wait until the stream's provider task and retry loop have fully settled. @internal */
+  async _waitClosed(): Promise<void> {
+    await this.#closedPromise;
+  }
+
   /** Close both the input and output of the STT stream */
   close() {
     if (!this.input.closed) this.input.close();

agents/src/types.ts

@@ -58,12 +58,19 @@ export interface APIConnectOptions {
   retryIntervalMs: number;
   /** Timeout for connecting to the API in milliseconds. Default: 10000 */
   timeoutMs: number;
+  /**
+   * Maximum time without inbound transport activity before a streaming connection fails.
+   * Providers that support liveness monitoring probe an idle connection after one-third of this
+   * interval. Default: 6000.
+   */
+  livenessTimeoutMs?: number;
 }
 
 export const DEFAULT_API_CONNECT_OPTIONS: APIConnectOptions = {
   maxRetry: 3,
   retryIntervalMs: 2000,
   timeoutMs: 10000,
+  livenessTimeoutMs: 6000,
 };
 
 /** Matches agents (Python), whose _interval_for_retry returns 0.1 seconds here. */

plugins/fishaudio/README.md

@@ -9,13 +9,42 @@ The Agents Framework is designed for building realtime, programmable
 participants that run on servers. Use it to create conversational, multi-modal
 voice agents that can see, hear, and understand.
 
-This package contains the Fish Audio plugin, which allows for voice synthesis.
-Refer to the [Fish Audio documentation](https://docs.fish.audio) for
-information on how to use it.
+This package contains the Fish Audio plugin for voice synthesis and realtime
+speech-to-text. Refer to the [Fish Audio documentation](https://docs.fish.audio)
+for information on how to use it.
 
 The plugin uses the Fish Audio Live TTS WebSocket API for streaming synthesis,
 and the HTTP `/v1/tts` endpoint for one-shot synthesis. Output is delivered as
 raw 16-bit little-endian PCM (24 kHz by default).
 
+## Installation
+
+```bash
+pnpm add @livekit/agents-plugin-fishaudio
+```
+
+Set `FISH_API_KEY` in the environment, then configure either service:
+
+```ts
+import { voice } from '@livekit/agents';
+import { STT, TTS } from '@livekit/agents-plugin-fishaudio';
+
+const session = new voice.AgentSession({
+  stt: new STT(),
+  tts: new TTS(),
+  // llm, vad, etc.
+});
+```
+
+Fish Audio can detect the end of each user turn itself:
+
+```ts
+const session = new voice.AgentSession({
+  stt: new STT(),
+  turnHandling: { turnDetection: 'stt' },
+  // llm, tts, etc.
+});
+```
+
 See the [repository](https://github.com/livekit/agents-js) for more information
 about the framework as a whole.

plugins/fishaudio/etc/agents-plugin-fishaudio.api.md

@@ -4,11 +4,16 @@
 
 ```ts
 
-import type { AudioFrame } from '@livekit/rtc-node';
+import { AudioFrame } from '@livekit/rtc-node';
 import { AudioResampler } from '@livekit/rtc-node';
+import { Logger } from 'pino';
 import type { ReadableStream as ReadableStream_2 } from 'node:stream/web';
+import type { ReadableStreamDefaultReader as ReadableStreamDefaultReader_2 } from 'node:stream/web';
 import type { Span } from '@opentelemetry/api';
+import { TransformStream as TransformStream_2 } from 'node:stream/web';
 import type { TypedEventEmitter } from '@livekit/typed-emitter';
+import type { VideoFrame as VideoFrame_2 } from '@livekit/rtc-node';
+import type { WritableStreamDefaultWriter as WritableStreamDefaultWriter_2 } from 'node:stream/web';
 
 // Warning: (ae-forgotten-export) The symbol "tts" needs to be exported by the entry point index.d.ts
 //
@@ -33,6 +38,65 @@ export type MP3Bitrate = 64 | 128 | 192;
 // @public
 export type OpusBitrate = -1000 | 24000 | 32000 | 48000 | 64000;
 
+// Warning: (ae-forgotten-export) The symbol "stt" needs to be exported by the entry point index.d.ts
+//
+// @public
+export class SpeechStream extends stt.SpeechStream {
+    constructor(sttInstance: STT, connOptions?: APIConnectOptions);
+    close(): void;
+    label: string;
+    // @internal (undocumented)
+    protected run(): Promise<void>;
+}
+
+// @public
+export class STT extends stt.STT {
+    constructor(opts?: STTOptions);
+    close(): Promise<void>;
+    label: string;
+    get model(): string;
+    get provider(): string;
+    // Warning: (ae-forgotten-export) The symbol "ConversationItemAddedEvent" needs to be exported by the entry point index.d.ts
+    //
+    // @internal (undocumented)
+    _pushConversationItem(ev: ConversationItemAddedEvent): void;
+    // Warning: (ae-forgotten-export) The symbol "AudioBuffer_2" needs to be exported by the entry point index.d.ts
+    //
+    // @internal (undocumented)
+    protected _recognize(_buffer: AudioBuffer_2): Promise<stt.SpeechEvent>;
+    stream(options?: {
+        connOptions?: APIConnectOptions;
+    }): SpeechStream;
+    updateOptions(opts: STTUpdateOptions): void;
+    // @internal (undocumented)
+    _updateSessionKeyterms(keyterms: string[]): void;
+}
+
+// @public
+export type STTModels = 'transcribe-stream-1';
+
+// @public
+export interface STTOptions {
+    apiKey?: string;
+    baseURL?: string;
+    eagerEot?: boolean;
+    interimResults?: boolean;
+    keyterms?: string[];
+    model?: STTModels | string;
+    sampleRate?: number;
+    silenceTimeoutMs?: number;
+    tagAudioEvents?: boolean;
+    turnDetectionMode?: TurnDetectionMode;
+}
+
+// @public
+export interface STTUpdateOptions {
+    eagerEot?: boolean;
+    keyterms?: string[];
+    silenceTimeoutMs?: number;
+    tagAudioEvents?: boolean;
+}
+
 // @public (undocumented)
 export class SynthesizeStream extends tts.SynthesizeStream {
     constructor(tts: TTS, opts: ResolvedTTSOptions, connOptions?: APIConnectOptions);
@@ -118,6 +182,9 @@ export interface TTSOptions {
     volume?: number;
 }
 
+// @public
+export type TurnDetectionMode = 'auto' | 'vad' | 'manual';
+
 // (No @packageDocumentation comment for this package)
 
 ```

plugins/fishaudio/package.json

@@ -1,7 +1,18 @@
 {
   "name": "@livekit/agents-plugin-fishaudio",
   "version": "1.9.5",
-  "description": "Fish Audio plugin for LiveKit Node Agents",
+  "description": "Agent Framework plugin for voice synthesis and speech recognition with Fish Audio's API.",
+  "keywords": [
+    "voice",
+    "ai",
+    "realtime",
+    "audio",
+    "video",
+    "livekit",
+    "fishaudio",
+    "stt",
+    "tts"
+  ],
   "main": "dist/index.js",
   "require": "dist/index.cjs",
   "types": "dist/index.d.ts",

plugins/fishaudio/src/index.ts

@@ -4,6 +4,7 @@
 import { Plugin } from '@livekit/agents';
 
 export * from './models.js';
+export { STT, SpeechStream, type STTOptions, type STTUpdateOptions } from './stt.js';
 export * from './tts.js';
 
 class FishAudioPlugin extends Plugin {

plugins/fishaudio/src/models.ts

@@ -5,6 +5,12 @@
 /** @public */
 export type TTSModels = 's1' | 's2-pro' | 's2.1-pro' | 'drama-3';
 
+/** Fish Audio realtime speech-to-text models. @public */
+export type STTModels = 'transcribe-stream-1';
+
+/** How Fish Audio detects the end of a user turn. @public */
+export type TurnDetectionMode = 'auto' | 'vad' | 'manual';
+
 /** @public */
 export type LatencyMode = 'normal' | 'balanced' | 'low';