Speech vendor/cobalt (#463)

* initial changes for cobalt speech

* wip

* wip

* update to drachtio-fsmrf that supports cobalt

* update to verb-specifications with cobalt speech support

* more wip

* lint

* use node 18 with gh actions

* support for compiling cobalt hints

* fix bug in uuid_cobalt_compile_context

* update verb-specifications

* remove repeated code

* cobalt support for transcribe

* update to verb specs

.github/workflows/build.yml

@@ -9,7 +9,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-node@v3
         with:
-          node-version: 16
+          node-version: 18
       - run: npm ci
       - run: npm run jslint
       - run: docker pull drachtio/sipp

lib/session/call-session.js

@@ -731,8 +731,8 @@ class CallSession extends Emitter {
         (type === 'tts' && credential.use_for_tts) ||
         (type === 'stt' && credential.use_for_stt)
       )) {
-        this.logger.info(`Speech credential vendor: ${credential.vendor} 
+        this.logger.info(
-${credential.label ? `, label: ${credential.label}` : ''} is chosen`);
+          `Speech vendor: ${credential.vendor} ${credential.label ? `, label: ${credential.label}` : ''} selected`);
         if ('google' === vendor) {
           if (type === 'tts' && !credential.tts_tested_ok ||
             type === 'stt' && !credential.stt_tested_ok) {
@@ -817,6 +817,12 @@ ${credential.label ? `, label: ${credential.label}` : ''} is chosen`);
             riva_server_uri: credential.riva_server_uri
           };
         }
+        else if ('cobalt' === vendor) {
+          return {
+            speech_credential_sid: credential.speech_credential_sid,
+            cobalt_server_uri: credential.cobalt_server_uri
+          };
+        }
         else if (vendor.startsWith('custom:')) {
           return {
             speech_credential_sid: credential.speech_credential_sid,

lib/tasks/gather.js

@@ -6,6 +6,7 @@ const {
   AzureTranscriptionEvents,
   DeepgramTranscriptionEvents,
   SonioxTranscriptionEvents,
+  CobaltTranscriptionEvents,
   IbmTranscriptionEvents,
   NvidiaTranscriptionEvents,
   JambonzTranscriptionEvents
@@ -187,12 +188,18 @@ class TaskGather extends SttTask {
       }
     }
 
+    /* when using cobalt model is required */
+    if (this.vendor === 'cobalt' && !this.data.recognizer.model) {
+      this.notifyError({ msg: 'ASR error', details:'Cobalt requires a model to be specified'});
+      throw new Error('Cobalt requires a model to be specified');
+    }
+
     const startListening = async(cs, ep) => {
       this._startTimer();
       if (this.isContinuousAsr && 0 === this.timeout) this._startAsrTimer();
       if (this.input.includes('speech') && !this.listenDuringPrompt) {
         try {
-          await this._initSpeech(cs, ep);
+          await this._setSpeechHandlers(cs, ep);
           if (this.killed) {
             this.logger.info('Gather:exec - task was quickly killed so do not transcribe');
             return;
@@ -260,7 +267,7 @@ class TaskGather extends SttTask {
       }
 
       if (this.input.includes('speech') && this.listenDuringPrompt) {
-        await this._initSpeech(cs, ep);
+        await this._setSpeechHandlers(cs, ep);
         this._startTranscribing(ep);
         updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid)
           .catch(() => {/*already logged error */});
@@ -334,7 +341,9 @@ class TaskGather extends SttTask {
     }
   }
 
-  async _initSpeech(cs, ep) {
+  async _setSpeechHandlers(cs, ep) {
+    if (this._speechHandlersSet) return;
+    this._speechHandlersSet = true;
     const opts = this.setChannelVarsForStt(this, this.sttCredentials, this.data.recognizer);
     switch (this.vendor) {
       case 'google':
@@ -387,6 +396,28 @@ class TaskGather extends SttTask {
         ep.addCustomEventListener(SonioxTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep));
         break;
 
+      case 'cobalt':
+        this.bugname = 'cobalt_speech';
+        ep.addCustomEventListener(CobaltTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep));
+
+        /* special case: if using hints with cobalt we need to compile them */
+        if (this.vendor === 'cobalt' && opts.COBALT_SPEECH_HINTS) {
+          try {
+            const context = await this.compileHintsForCobalt(
+              ep,
+              opts.COBALT_SERVER_URI,
+              this.data.recognizer.model,
+              opts.COBALT_CONTEXT_TOKEN,
+              opts.COBALT_SPEECH_HINTS
+            );
+            if (context) opts.COBALT_COMPILED_CONTEXT_DATA = context;
+            delete opts.COBALT_SPEECH_HINTS;
+          } catch (err) {
+            this.logger.error({err}, 'Error compiling hints for cobalt');
+          }
+        }
+        break;
+
       case 'ibm':
         this.bugname = 'ibm_transcribe';
         ep.addCustomEventListener(IbmTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep));

lib/tasks/stt-task.js

@@ -1,6 +1,7 @@
 const Task = require('./task');
 const assert = require('assert');
-const { TaskPreconditions } = require('../utils/constants');
+const crypto = require('crypto');
+const { TaskPreconditions, CobaltTranscriptionEvents } = require('../utils/constants');
 
 class SttTask extends Task {
 
@@ -95,6 +96,44 @@ class SttTask extends Task {
     this.data.recognizer.label = this.label;
     this.sttCredentials = await this._initSpeechCredentials(this.cs, this.vendor, this.label);
   }
+
+  async compileHintsForCobalt(ep, hostport, model, token, hints) {
+    const {retrieveKey} = this.cs.srf.locals.dbHelpers;
+    const hash = crypto.createHash('sha1');
+    hash.update(`${model}:${hints}`);
+    const key = `cobalt:${hash.digest('hex')}`;
+    this.context = await retrieveKey(key);
+    if (this.context) {
+      this.logger.debug({model, hints}, 'found cached cobalt context for supplied hints');
+      return this.context;
+    }
+
+    this.logger.debug({model, hints}, 'compiling cobalt context for supplied hints');
+
+    return new Promise((resolve, reject) => {
+      this.cobaltCompileResolver = resolve;
+      ep.addCustomEventListener(CobaltTranscriptionEvents.CompileContext, this._onCompileContext.bind(this, ep, key));
+      ep.api('uuid_cobalt_compile_context', [ep.uuid, hostport, model, token, hints], (err, evt) => {
+        if (err || 0 !== evt.getBody().indexOf('+OK')) {
+          ep.removeCustomEventListener(CobaltTranscriptionEvents.CompileContext);
+          return reject(err);
+        }
+      });
+    });
+  }
+
+  _onCompileContext(ep, key, evt) {
+    const {addKey} = this.cs.srf.locals.dbHelpers;
+    this.logger.debug({evt}, `received cobalt compile context event, will cache under ${key}`);
+
+    this.cobaltCompileResolver(evt.compiled_context);
+    ep.removeCustomEventListener(CobaltTranscriptionEvents.CompileContext);
+    this.cobaltCompileResolver = null;
+
+    //cache the compiled context
+    addKey(key, evt.compiled_context, 3600 * 12)
+      .catch((err) => this.logger.info({err}, `Error caching cobalt context for ${key}`));
+  }
 }
 
 module.exports = SttTask;

lib/tasks/transcribe.js

@@ -7,6 +7,7 @@ const {
   AzureTranscriptionEvents,
   DeepgramTranscriptionEvents,
   SonioxTranscriptionEvents,
+  CobaltTranscriptionEvents,
   IbmTranscriptionEvents,
   NvidiaTranscriptionEvents,
   JambonzTranscriptionEvents,
@@ -103,6 +104,12 @@ class TaskTranscribe extends SttTask {
       }
     }
 
+    /* when using cobalt model is required */
+    if (this.vendor === 'cobalt' && !this.data.recognizer.model) {
+      this.notifyError({ msg: 'ASR error', details:'Cobalt requires a model to be specified'});
+      throw new Error('Cobalt requires a model to be specified');
+    }
+
     try {
       await this._startTranscribing(cs, ep, 1);
       if (this.separateRecognitionPerChannel && ep2) {
@@ -163,7 +170,9 @@ class TaskTranscribe extends SttTask {
     }
   }
 
-  async _startTranscribing(cs, ep, channel) {
+  async _setSpeechHandlers(cs, ep, channel) {
+    if (this._speechHandlersSet) return;
+    this._speechHandlersSet = true;
     const opts = this.setChannelVarsForStt(this, this.sttCredentials, this.data.recognizer);
     switch (this.vendor) {
       case 'google':
@@ -216,6 +225,29 @@ class TaskTranscribe extends SttTask {
         ep.addCustomEventListener(SonioxTranscriptionEvents.Transcription,
           this._onTranscription.bind(this, cs, ep, channel));
         break;
+      case 'cobalt':
+        this.bugname = 'cobalt_transcribe';
+        ep.addCustomEventListener(CobaltTranscriptionEvents.Transcription,
+          this._onTranscription.bind(this, cs, ep, channel));
+
+        /* special case: if using hints with cobalt we need to compile them */
+        if (this.vendor === 'cobalt' && opts.COBALT_SPEECH_HINTS) {
+          try {
+            const context = await this.compileHintsForCobalt(
+              ep,
+              opts.COBALT_SERVER_URI,
+              this.data.recognizer.model,
+              opts.COBALT_CONTEXT_TOKEN,
+              opts.COBALT_SPEECH_HINTS
+            );
+            if (context) opts.COBALT_COMPILED_CONTEXT_DATA = context;
+            delete opts.COBALT_SPEECH_HINTS;
+          } catch (err) {
+            this.logger.error({err}, 'Error compiling hints for cobalt');
+          }
+        }
+        break;
+
       case 'ibm':
         this.bugname = 'ibm_transcribe';
         ep.addCustomEventListener(IbmTranscriptionEvents.Transcription,
@@ -237,6 +269,7 @@ class TaskTranscribe extends SttTask {
         ep.addCustomEventListener(NvidiaTranscriptionEvents.VadDetected,
           this._onVadDetected.bind(this, cs, ep));
         break;
+
       default:
         if (this.vendor.startsWith('custom:')) {
           this.bugname = `${this.vendor}_transcribe`;
@@ -258,7 +291,10 @@ class TaskTranscribe extends SttTask {
     ep.addCustomEventListener(JambonzTranscriptionEvents.Error, this._onJambonzError.bind(this, cs, ep));
     await ep.set(opts)
       .catch((err) => this.logger.info(err, 'Error setting channel variables'));
+  }
 
+  async _startTranscribing(cs, ep, channel) {
+    await this._setSpeechHandlers(cs, ep, channel);
     await this._transcribe(ep);
 
     /* start child span for this channel */

lib/utils/amd-utils.js

@@ -9,6 +9,7 @@ const {
   NvidiaTranscriptionEvents,
   IbmTranscriptionEvents,
   SonioxTranscriptionEvents,
+  CobaltTranscriptionEvents,
   DeepgramTranscriptionEvents,
   JambonzTranscriptionEvents,
   AmdEvents,
@@ -314,6 +315,10 @@ module.exports = (logger) => {
         ep.addCustomEventListener(NvidiaTranscriptionEvents.Transcription, amd.transcriptionHandler);
         break;
 
+      case 'cobalt':
+        ep.addCustomEventListener(CobaltTranscriptionEvents.Transcription, amd.transcriptionHandler);
+        break;
+
       default:
         if (vendor.startsWith('custom:')) {
           ep.addCustomEventListener(JambonzTranscriptionEvents.Transcription, amd.transcriptionHandler);

lib/utils/constants.json

@@ -95,6 +95,11 @@
     "Transcription": "soniox_transcribe::transcription",
     "Error": "soniox_transcribe::error"
   },
+  "CobaltTranscriptionEvents": {
+    "Transcription": "cobalt_speech::transcription",
+    "CompileContext": "cobalt_speech::compile_context_response",
+    "Error": "cobalt_speech::error"
+  },
   "IbmTranscriptionEvents": {
     "Transcription": "ibm_transcribe::transcription",
     "ConnectFailure": "ibm_transcribe::connect_failed",

lib/utils/db-utils.js

@@ -81,6 +81,10 @@ const speechMapper = (cred) => {
       const o = JSON.parse(decrypt(credential));
       obj.riva_server_uri = o.riva_server_uri;
     }
+    else if ('cobalt' === obj.vendor) {
+      const o = JSON.parse(decrypt(credential));
+      obj.cobalt_server_uri = o.cobalt_server_uri;
+    }
     else if (obj.vendor.startsWith('custom:')) {
       const o = JSON.parse(decrypt(credential));
       obj.auth_token = o.auth_token;

lib/utils/transcription-utils.js

@@ -7,6 +7,7 @@ const {
   DeepgramTranscriptionEvents,
   SonioxTranscriptionEvents,
   NvidiaTranscriptionEvents,
+  CobaltTranscriptionEvents,
   JambonzTranscriptionEvents
 } = require('./constants');
 
@@ -92,6 +93,11 @@ const stickyVars = {
   nvidia: [
     'NVIDIA_HINTS'
   ],
+  cobalt: [
+    'COBALT_SPEECH_HINTS',
+    'COBALT_COMPILED_CONTEXT_DATA',
+    'COBALT_METADATA'
+  ],
   soniox: [
     'SONIOX_PROFANITY_FILTER',
     'SONIOX_MODEL'
@@ -225,6 +231,26 @@ const normalizeGoogle = (evt, channel, language) => {
   };
 };
 
+const normalizeCobalt = (evt, channel, language) => {
+  const copy = JSON.parse(JSON.stringify(evt));
+  const alternatives = (evt.alternatives || [])
+    .map((alt) => ({
+      confidence: alt.confidence,
+      transcript: alt.transcript_formatted,
+    }));
+
+  return {
+    language_code: language,
+    channel_tag: channel,
+    is_final: evt.is_final,
+    alternatives,
+    vendor: {
+      name: 'cobalt',
+      evt: copy
+    }
+  };
+};
+
 const normalizeCustom = (evt, channel, language, vendor) => {
   const copy = JSON.parse(JSON.stringify(evt));
   return {
@@ -317,6 +343,8 @@ module.exports = (logger) => {
         return normalizeNvidia(evt, channel, language);
       case 'soniox':
         return normalizeSoniox(evt, channel, language);
+      case 'cobalt':
+        return normalizeCobalt(evt, channel, language);
       default:
         if (vendor.startsWith('custom:')) {
           return normalizeCustom(evt, channel, language, vendor);
@@ -584,6 +612,25 @@ module.exports = (logger) => {
           {NVIDIA_CUSTOM_CONFIGURATION: JSON.stringify(nvidiaOptions.customConfiguration)}),
       };
     }
+    else if ('cobalt' === vendor) {
+      const {cobaltOptions = {}} = rOpts;
+      const cobaltUri = cobaltOptions.serverUri || sttCredentials.cobalt_server_uri;
+      opts = {
+        ...opts,
+        ...(rOpts.words && {COBALT_WORD_TIME_OFFSETS: 1}),
+        ...(!rOpts.words && {COBALT_WORD_TIME_OFFSETS: 0}),
+        ...(rOpts.model && {COBALT_MODEL: rOpts.model}),
+        ...(cobaltUri && {COBALT_SERVER_URI: cobaltUri}),
+        ...(rOpts.hints?.length > 0 && typeof rOpts.hints[0] === 'string' &&
+          {COBALT_SPEECH_HINTS: rOpts.hints.join(',')}),
+        ...(rOpts.hints?.length > 0 && typeof rOpts.hints[0] === 'object' &&
+          {COBALT_SPEECH_HINTS: JSON.stringify(rOpts.hints)}),
+        ...(rOpts.hints?.length > 0 && {COBALT_CONTEXT_TOKEN: cobaltOptions.contextToken || 'unk:default'}),
+        ...(cobaltOptions.metadata && {COBALT_METADATA: cobaltOptions.metadata}),
+        ...(cobaltOptions.enableConfusionNetwork && {COBALT_ENABLE_CONFUSION_NETWORK: 1}),
+        ...(cobaltOptions.compiledContextData && {COBALT_COMPILED_CONTEXT_DATA: cobaltOptions.compiledContextData}),
+      };
+    }
     else if (vendor.startsWith('custom:')) {
       let {options = {}} = rOpts;
       const {auth_token, custom_stt_url} = sttCredentials;
@@ -633,6 +680,9 @@ module.exports = (logger) => {
 
     ep.removeCustomEventListener(SonioxTranscriptionEvents.Transcription);
 
+    ep.removeCustomEventListener(CobaltTranscriptionEvents.Transcription);
+    ep.removeCustomEventListener(CobaltTranscriptionEvents.CompileContext);
+
     ep.removeCustomEventListener(NvidiaTranscriptionEvents.Transcription);
     ep.removeCustomEventListener(NvidiaTranscriptionEvents.TranscriptionComplete);
     ep.removeCustomEventListener(NvidiaTranscriptionEvents.StartOfSpeech);
@@ -664,6 +714,10 @@ module.exports = (logger) => {
       const {apiKey} = recognizer.sonioxOptions || {};
       if (apiKey) return {api_key: apiKey};
     }
+    else if (recognizer.vendor === 'cobalt') {
+      const {serverUri} = recognizer.cobaltOptions || {};
+      if (serverUri) return {cobalt_server_uri: serverUri};
+    }
     else if (recognizer.vendor === 'ibm') {
       const {ttsApiKey, ttsRegion, sttApiKey, sttRegion, instanceId} = recognizer.ibmOptions || {};
       if (ttsApiKey || sttApiKey) return {

package-lock.json

@@ -17,7 +17,7 @@
         "@jambonz/speech-utils": "^0.0.21",
         "@jambonz/stats-collector": "^0.1.9",
         "@jambonz/time-series": "^0.2.8",
-        "@jambonz/verb-specifications": "^0.0.30",
+        "@jambonz/verb-specifications": "^0.0.34",
         "@opentelemetry/api": "^1.4.0",
         "@opentelemetry/exporter-jaeger": "^1.9.0",
         "@opentelemetry/exporter-trace-otlp-http": "^0.35.0",
@@ -30,7 +30,7 @@
         "bent": "^7.3.12",
         "debug": "^4.3.4",
         "deepcopy": "^2.1.0",
-        "drachtio-fsmrf": "^3.0.24",
+        "drachtio-fsmrf": "^3.0.26",
         "drachtio-srf": "^4.5.26",
         "express": "^4.18.2",
         "ip": "^1.1.8",
@@ -3019,9 +3019,9 @@
       }
     },
     "node_modules/@jambonz/verb-specifications": {
-      "version": "0.0.30",
+      "version": "0.0.34",
-      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.30.tgz",
+      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.34.tgz",
-      "integrity": "sha512-IbusgtBBo2V5Tc1FvDJvkWogHOhR2tNZN6Iyb2PjUomMI48BsWKmHW/wegppKTDpBBeN3ABY+L96pvX4N0+mCw==",
+      "integrity": "sha512-6/oplhOMpxz6qbIdz0K3Y2+KXtnb/DLziYmnGUK8NALWD7rhDvp2RdTsuEjOOkOhwLyRJUhuIAM17GcpAq/lCw==",
       "dependencies": {
         "debug": "^4.3.4",
         "pino": "^8.8.0"
@@ -5152,9 +5152,9 @@
       }
     },
     "node_modules/drachtio-fsmrf": {
-      "version": "3.0.24",
+      "version": "3.0.26",
-      "resolved": "https://registry.npmjs.org/drachtio-fsmrf/-/drachtio-fsmrf-3.0.24.tgz",
+      "resolved": "https://registry.npmjs.org/drachtio-fsmrf/-/drachtio-fsmrf-3.0.26.tgz",
-      "integrity": "sha512-MsTwHDOOg5yrKCcQ2buNTVSEwoMZzTcVeU5/BZ2Km0TPxBI1lErW6IiyZxA5pyoLfSdBtbqSh9ZwXrtxWTp6kA==",
+      "integrity": "sha512-QoXa+lQ6j7NIIVhWq/oL5qHl/RokaxixJ7IhuuYmfCU+LsGdjFqPib8JEsieCXcp2qF98lOrEytJ5RESvWQGYQ==",
       "dependencies": {
         "camel-case": "^4.1.2",
         "debug": "^2.6.9",
@@ -12985,9 +12985,9 @@
       }
     },
     "@jambonz/verb-specifications": {
-      "version": "0.0.30",
+      "version": "0.0.34",
-      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.30.tgz",
+      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.34.tgz",
-      "integrity": "sha512-IbusgtBBo2V5Tc1FvDJvkWogHOhR2tNZN6Iyb2PjUomMI48BsWKmHW/wegppKTDpBBeN3ABY+L96pvX4N0+mCw==",
+      "integrity": "sha512-6/oplhOMpxz6qbIdz0K3Y2+KXtnb/DLziYmnGUK8NALWD7rhDvp2RdTsuEjOOkOhwLyRJUhuIAM17GcpAq/lCw==",
       "requires": {
         "debug": "^4.3.4",
         "pino": "^8.8.0"
@@ -14610,9 +14610,9 @@
       }
     },
     "drachtio-fsmrf": {
-      "version": "3.0.24",
+      "version": "3.0.26",
-      "resolved": "https://registry.npmjs.org/drachtio-fsmrf/-/drachtio-fsmrf-3.0.24.tgz",
+      "resolved": "https://registry.npmjs.org/drachtio-fsmrf/-/drachtio-fsmrf-3.0.26.tgz",
-      "integrity": "sha512-MsTwHDOOg5yrKCcQ2buNTVSEwoMZzTcVeU5/BZ2Km0TPxBI1lErW6IiyZxA5pyoLfSdBtbqSh9ZwXrtxWTp6kA==",
+      "integrity": "sha512-QoXa+lQ6j7NIIVhWq/oL5qHl/RokaxixJ7IhuuYmfCU+LsGdjFqPib8JEsieCXcp2qF98lOrEytJ5RESvWQGYQ==",
       "requires": {
         "camel-case": "^4.1.2",
         "debug": "^2.6.9",

package.json

@@ -33,7 +33,7 @@
     "@jambonz/speech-utils": "^0.0.21",
     "@jambonz/stats-collector": "^0.1.9",
     "@jambonz/time-series": "^0.2.8",
-    "@jambonz/verb-specifications": "^0.0.30",
+    "@jambonz/verb-specifications": "^0.0.34",
     "@opentelemetry/api": "^1.4.0",
     "@opentelemetry/exporter-jaeger": "^1.9.0",
     "@opentelemetry/exporter-trace-otlp-http": "^0.35.0",
@@ -46,7 +46,7 @@
     "bent": "^7.3.12",
     "debug": "^4.3.4",
     "deepcopy": "^2.1.0",
-    "drachtio-fsmrf": "^3.0.24",
+    "drachtio-fsmrf": "^3.0.26",
     "drachtio-srf": "^4.5.26",
     "express": "^4.18.2",
     "ip": "^1.1.8",