refactor: add handleSpeechTurn method to process audio input and integrate with agent response

Author: NoOne7135

agentTurnService.ts

@@ -1,4 +1,4 @@
-import type { AdminUser, IAdminForth } from "adminforth";
+import type { AdminUser, AudioAdapter, IAdminForth } from "adminforth";
 import { logger } from "adminforth";
 import { randomUUID } from "crypto";
 import { HumanMessage, SystemMessage } from "langchain";
@@ -11,6 +11,7 @@ import type { AgentEventEmitter } from "./agentEvents.js";
 import { buildAgentTurnSystemPrompt } from "./agent/systemPrompt.js";
 import type { CurrentPageContext } from "./agent/tools/getUserLocation.js";
 import { isAbortError, getErrorMessage } from "./errors.js";
+import { sanitizeSpeechText } from "./sanitizeSpeechText.js";
 import type { AgentSessionStore } from "./sessionStore.js";
 import type { PluginOptions } from "./types.js";
 
@@ -54,6 +55,15 @@ export type HandleTurnInput = Omit<RunAndPersistAgentResponseInput, "failureLogM
   abortLogMessage?: string;
 };
 
+export type HandleSpeechTurnInput = Omit<HandleTurnInput, "prompt"> & {
+  audioAdapter: AudioAdapter;
+  audio: {
+    buffer: Buffer;
+    filename: string;
+    mimeType: string;
+  };
+};
+
 type AgentTurnServiceOptions = {
   getAdminforth: () => IAdminForth;
   getPluginInstanceId: () => string;
@@ -342,6 +352,167 @@ export class AgentTurnService {
 
     return agentResponse;
   }
+
+  async handleSpeechTurn(input: HandleSpeechTurnInput) {
+    let transcription;
+
+    try {
+      transcription = await input.audioAdapter.transcribe({
+        buffer: input.audio.buffer,
+        filename: input.audio.filename,
+        mimeType: input.audio.mimeType,
+        language: "auto",
+        abortSignal: input.abortSignal,
+      });
+    } catch (error) {
+      if (input.abortSignal?.aborted || isAbortError(error)) {
+        logger.info("Agent speech transcription aborted by the client");
+        await input.emit({ type: "finish" });
+        return null;
+      }
+
+      logger.error(`Agent speech transcription failed:\n${getErrorMessage(error)}`);
+      await input.emit({
+        type: "error",
+        error: "Speech transcription failed. Check server logs for details.",
+      });
+      await input.emit({ type: "finish" });
+      return null;
+    }
+
+    if (input.abortSignal?.aborted) {
+      await input.emit({ type: "finish" });
+      return null;
+    }
+
+    const prompt = transcription.text;
+    if (!prompt) {
+      await input.emit({
+        type: "error",
+        error: "Speech transcription is empty",
+      });
+      await input.emit({ type: "finish" });
+      return null;
+    }
+
+    await input.emit({
+      type: "transcript",
+      text: transcription.text,
+      language: transcription.language,
+    });
+
+    const agentResponse = await this.runAndPersistAgentResponse({
+      prompt,
+      sessionId: input.sessionId,
+      modeName: input.modeName,
+      userTimeZone: input.userTimeZone,
+      currentPage: input.currentPage,
+      abortSignal: input.abortSignal,
+      adminUser: input.adminUser,
+      emit: async (event) => {
+        if (event.type === "tool-call") {
+          await input.emit(event);
+        }
+      },
+      failureLogMessage: input.failureLogMessage ?? "Agent speech response failed",
+      abortLogMessage: input.abortLogMessage ?? "Agent speech response aborted by the client",
+    });
+
+    if (agentResponse.aborted) {
+      await input.emit({ type: "finish" });
+      return agentResponse;
+    }
+
+    if (agentResponse.failed) {
+      await input.emit({
+        type: "error",
+        error: agentResponse.text,
+      });
+      await input.emit({ type: "finish" });
+      return agentResponse;
+    }
+
+    try {
+      await input.emit({
+        type: "speech-response",
+        transcript: {
+          text: transcription.text,
+          language: transcription.language,
+        },
+        response: {
+          text: agentResponse.text,
+        },
+        sessionId: input.sessionId,
+        turnId: agentResponse.turnId,
+      });
+      const speech = await input.audioAdapter.synthesize({
+        text: sanitizeSpeechText(agentResponse.text),
+        stream: true,
+        streamFormat: "audio",
+        format: "pcm",
+        abortSignal: input.abortSignal,
+      });
+
+      await input.emit({
+        type: "audio-start",
+        mimeType: speech.mimeType,
+        format: speech.format,
+        sampleRate: 24000,
+        channelCount: 1,
+        bitsPerSample: 16,
+      });
+
+      const reader = speech.audioStream.getReader();
+      const cancelAudioStream = () => {
+        void reader.cancel().catch(() => undefined);
+      };
+
+      try {
+        input.abortSignal?.addEventListener("abort", cancelAudioStream, { once: true });
+
+        while (true) {
+          if (input.abortSignal?.aborted) {
+            await reader.cancel().catch(() => undefined);
+            break;
+          }
+
+          const { value, done } = await reader.read();
+
+          if (done) {
+            break;
+          }
+
+          if (input.abortSignal?.aborted) {
+            break;
+          }
+
+          await input.emit({
+            type: "audio-delta",
+            value,
+          });
+        }
+      } finally {
+        input.abortSignal?.removeEventListener("abort", cancelAudioStream);
+        reader.releaseLock();
+      }
+
+      await input.emit({ type: "audio-done" });
+      await input.emit({ type: "finish" });
+      return agentResponse;
+    } catch (error) {
+      if (input.abortSignal?.aborted || isAbortError(error)) {
+        logger.info("Agent speech audio streaming aborted by the client");
+      } else {
+        logger.error(`Agent speech audio streaming failed:\n${getErrorMessage(error)}`);
+        await input.emit({
+          type: "error",
+          error: getErrorMessage(error),
+        });
+      }
+      await input.emit({ type: "finish" });
+      return agentResponse;
+    }
+  }
 }
 
 function getPartialVegaLiteFenceStartLength(text: string): number {

endpoints/context.ts

@@ -7,6 +7,7 @@ import type {
 } from "adminforth";
 import type { ZodType } from "zod";
 import type {
+  HandleSpeechTurnInput,
   HandleTurnInput,
   RunAndPersistAgentResponseInput,
   RunAndPersistAgentResponseResult,
@@ -30,6 +31,7 @@ export type AgentEndpointsContext = {
   options: PluginOptions;
   parseBody<T>(schema: ZodType<T>, body: unknown, response: EndpointResponse): T | null;
   handleTurn(input: HandleTurnInput): Promise<RunAndPersistAgentResponseResult>;
+  handleSpeechTurn(input: HandleSpeechTurnInput): Promise<RunAndPersistAgentResponseResult | null>;
   runAndPersistAgentResponse(input: RunAndPersistAgentResponseInput): Promise<RunAndPersistAgentResponseResult>;
   getSessionTurns(sessionId: string): Promise<SessionTurn[]>;
   createNewTurn(sessionId: string, prompt: string, response?: string): Promise<string>;
@@ -45,7 +47,7 @@ export type AgentEndpointsContext = {
 
 export type CoreEndpointsContext = Pick<
   AgentEndpointsContext,
-  "options" | "parseBody" | "handleTurn" | "runAndPersistAgentResponse"
+  "options" | "parseBody" | "handleTurn" | "handleSpeechTurn"
 >;
 
 export type SessionEndpointsContext = Pick<