diff --git a/lib/tasks/llm/llms/elevenlabs_s2s.js b/lib/tasks/llm/llms/elevenlabs_s2s.js index 58e2a0fb..fe82ed48 100644 --- a/lib/tasks/llm/llms/elevenlabs_s2s.js +++ b/lib/tasks/llm/llms/elevenlabs_s2s.js @@ -2,8 +2,10 @@ const Task = require('../../task'); const TaskName = 'Llm_Elevenlabs_s2s'; const {LlmEvents_Elevenlabs} = require('../../../utils/constants'); const {request} = require('undici'); +const DtmfInput = require('../../../utils/dtmf-input'); const ClientEvent = 'client.event'; const SessionDelete = 'session.delete'; +const llmUpdateTypes = ['user_message', 'user_activity', 'contextual_update']; const elevenlabs_server_events = [ 'conversation_initiation_metadata', @@ -47,11 +49,13 @@ class TaskLlmElevenlabs_S2S extends Task { const { conversation_initiation_client_data, input_sample_rate = 16000, - output_sample_rate = 16000 + output_sample_rate = 16000, + dtmf } = this.data.llmOptions; this.conversation_initiation_client_data = conversation_initiation_client_data; this.input_sample_rate = input_sample_rate; this.output_sample_rate = output_sample_rate; + if (dtmf?.enable) this.dtmfOpts = dtmf; this.results = { completionReason: 'normal conversation end' }; @@ -105,7 +109,7 @@ class TaskLlmElevenlabs_S2S extends Task { async _api(ep, args) { const res = await ep.api('uuid_elevenlabs_s2s', `^^|${args.join('|')}`); if (!res.body?.startsWith('+OK')) { - throw new Error({args}, `Error calling uuid_elevenlabs_s2s: ${res.body}`); + throw new Error(`Error calling uuid_elevenlabs_s2s: ${res.body}`); } } @@ -114,6 +118,7 @@ class TaskLlmElevenlabs_S2S extends Task { await this._startListening(cs, ep); await this.awaitTaskDone(); + this.dtmfInput?.stop(); /* note: the parent llm verb started the span, which is why this is necessary */ await this.parent.performAction(this.results); @@ -123,6 +128,7 @@ class TaskLlmElevenlabs_S2S extends Task { async kill(cs) { super.kill(cs); + this.dtmfInput?.stop(); this._api(cs.ep, [cs.ep.uuid, SessionDelete]) .catch((err) => this.logger.info({err}, 'TaskLlmElevenlabs_S2S:kill - error deleting session')); @@ -144,7 +150,7 @@ class TaskLlmElevenlabs_S2S extends Task { 'TaskLlmElevenlabs_S2S:processToolOutput - invalid tool output, must be client_tool_result'); } else { - await this._api(ep, [ep.uuid, ClientEvent, JSON.stringify(data)]); + await this._sendClientEvent(ep, data); } } catch (err) { this.logger.info({err}, 'TaskLlmElevenlabs_S2S:processToolOutput'); @@ -152,11 +158,16 @@ class TaskLlmElevenlabs_S2S extends Task { } /** - * Send a session.update to the Elevenlabs server - * Note: creating and deleting conversation items also supported as well as interrupting the assistant + * Forward a user_message, user_activity or contextual_update client event to the agent */ async processLlmUpdate(ep, data, _callSid) { - this.logger.debug({data, _callSid}, 'TaskLlmElevenlabs_S2S:processLlmUpdate, ignored'); + this.logger.debug({data, _callSid}, 'TaskLlmElevenlabs_S2S:processLlmUpdate'); + // an ElevenLabs error event ends the conversation, so reject a malformed text message here + if (!llmUpdateTypes.includes(data?.type) || (data.type !== 'user_activity' && typeof data.text !== 'string')) { + this.logger.info({data}, 'TaskLlmElevenlabs_S2S:processLlmUpdate - invalid mid-call request'); + return; + } + await this._sendClientEvent(ep, data); } async _startListening(cs, ep) { @@ -178,7 +189,8 @@ class TaskLlmElevenlabs_S2S extends Task { let ok = true; this.logger.debug({obj}, 'TaskLlmElevenlabs_S2S:_sendClientEvent'); try { - const args = [ep.uuid, ClientEvent, JSON.stringify(obj)]; + // '|' is the api argument delimiter + const args = [ep.uuid, ClientEvent, JSON.stringify(obj).replace(/\|/g, '\\u007c')]; await this._api(ep, args); } catch (err) { ok = false; @@ -216,6 +228,14 @@ class TaskLlmElevenlabs_S2S extends Task { _onConnect(ep) { this.logger.debug('TaskLlmElevenlabs_S2S:_onConnect'); + if (this.dtmfOpts && !this.dtmfInput && !this.killed) { + this.dtmfInput = new DtmfInput(this.logger, this.dtmfOpts, { + // keeps the agent's turn timeout from firing while the caller is typing + onKey: () => this._sendClientEvent(ep, {type: 'user_activity'}), + onEntry: (text) => this._sendClientEvent(ep, {type: 'user_message', text, source_medium: 'dtmf'}) + }); + this.dtmfInput.start(ep); + } this._sendInitialMessage(ep); } _onConnectFailure(_ep, evt) { diff --git a/lib/utils/dtmf-input.js b/lib/utils/dtmf-input.js new file mode 100644 index 00000000..7213cf40 --- /dev/null +++ b/lib/utils/dtmf-input.js @@ -0,0 +1,91 @@ +/** + * Collects caller DTMF into whole entries for vendors that take keypad input as text. + * Options and termination rules follow the gather verb (seconds; timeout 15 by default, 0 = off). + */ +const DEFAULT_TIMEOUT = 15; + +const seconds = (v) => (typeof v === 'number' && v >= 0 ? v : undefined); + +class DtmfInput { + /** + * @param {object} logger - a pino logger + * @param {object} opts - gather-style dtmf options + * @param {object} handlers - onKey(digit) for every key press, onEntry(digits) for each completed entry + */ + constructor(logger, opts = {}, {onKey, onEntry}) { + this.logger = logger; + this.finishOnKey = opts.finishOnKey; + this.numDigits = opts.numDigits; + this.maxDigits = opts.maxDigits; + // gather never arms its inter-digit timer when minDigits is unset; 1 makes interDigitTimeout work on its own + this.minDigits = opts.minDigits || 1; + this.interDigitTimeout = (seconds(opts.interDigitTimeout) || 0) * 1000; + const timeout = seconds(opts.timeout); + this.timeout = (timeout === undefined ? DEFAULT_TIMEOUT : timeout) * 1000; + this.onKey = onKey; + this.onEntry = onEntry; + this.buffer = ''; + if (!this.finishOnKey && !this.numDigits && !this.maxDigits && !this.interDigitTimeout && !this.timeout) { + this.logger.warn({opts}, 'DtmfInput: no option ever completes an entry, digits will never be sent'); + } + } + + start(ep) { + if (this._onDtmf) return; + this.ep = ep; + this._onDtmf = (evt) => this.process(evt.dtmf); + ep.on('dtmf', this._onDtmf); + } + + stop() { + if (this._onDtmf) this.ep.removeListener('dtmf', this._onDtmf); + this._onDtmf = null; + this._clearTimers(); + if (this.buffer) this.logger.info({digits: this.buffer}, 'DtmfInput: discarding an incomplete entry'); + this.buffer = ''; + } + + process(digit) { + this.onKey?.(digit); + clearTimeout(this.interDigitTimer); + if (!this.timer && this.timeout > 0) { + this.timer = setTimeout(this._onTimeout.bind(this), this.timeout); + } + if (digit === this.finishOnKey) return this._complete(); + + this.buffer += digit; + const len = this.buffer.length; + if (len === this.numDigits || len === this.maxDigits) return this._complete(); + if (this.interDigitTimeout > 0 && len >= this.minDigits) { + // as in gather, the inter-digit timer replaces the overall timeout while the caller keeps typing + clearTimeout(this.timer); + this.timer = null; + this.interDigitTimer = setTimeout(this._complete.bind(this), this.interDigitTimeout); + } + } + + _onTimeout() { + if (this.buffer.length < this.minDigits) { + this.logger.info({digits: this.buffer}, 'DtmfInput: timeout before minDigits, discarding entry'); + this._clearTimers(); + this.buffer = ''; + return; + } + this._complete(); + } + + _complete() { + this._clearTimers(); + const digits = this.buffer; + this.buffer = ''; + if (digits) this.onEntry(digits); + } + + _clearTimers() { + clearTimeout(this.interDigitTimer); + clearTimeout(this.timer); + this.interDigitTimer = this.timer = null; + } +} + +module.exports = DtmfInput; diff --git a/test/unit/dtmf-input.test.js b/test/unit/dtmf-input.test.js new file mode 100644 index 00000000..766ee633 --- /dev/null +++ b/test/unit/dtmf-input.test.js @@ -0,0 +1,122 @@ +const test = require('node:test'); +const assert = require('node:assert'); +const Emitter = require('events'); + +const DtmfInput = require('../../lib/utils/dtmf-input'); + +const noop = () => {}; + +const make = (t, opts) => { + t.mock.timers.enable({apis: ['setTimeout']}); + const keys = []; + const entries = []; + const warnings = []; + const logger = {error: noop, info: noop, debug: noop, warn: (...args) => warnings.push(args)}; + const ep = new Emitter(); + const input = new DtmfInput(logger, opts, { + onKey: (d) => keys.push(d), + onEntry: (digits) => entries.push(digits) + }); + input.start(ep); + const press = (digits) => [...digits].forEach((dtmf) => ep.emit('dtmf', {dtmf})); + const tick = (secs) => t.mock.timers.tick(secs * 1000); + return {ep, input, keys, entries, warnings, press, tick}; +}; + +test('finishOnKey completes the entry and is not included', (t) => { + const {keys, entries, press} = make(t, {finishOnKey: '#'}); + press('1234#'); + assert.deepStrictEqual(entries, ['1234']); + assert.deepStrictEqual(keys, ['1', '2', '3', '4', '#']); +}); + +test('finishOnKey on an empty buffer sends nothing', (t) => { + const {entries, press} = make(t, {finishOnKey: '#'}); + press('#'); + assert.deepStrictEqual(entries, []); +}); + +test('no finishOnKey by default: # is an ordinary digit', (t) => { + const {entries, press} = make(t, {numDigits: 3}); + press('1#2'); + assert.deepStrictEqual(entries, ['1#2']); +}); + +test('numDigits completes the entry', (t) => { + const {entries, press} = make(t, {numDigits: 2}); + press('1234'); + assert.deepStrictEqual(entries, ['12', '34']); +}); + +test('maxDigits completes the entry', (t) => { + const {entries, press} = make(t, {maxDigits: 3}); + press('12345'); + assert.deepStrictEqual(entries, ['123']); +}); + +test('each key restarts the inter-digit timer', (t) => { + const {entries, press, tick} = make(t, {interDigitTimeout: 2}); + press('56'); + tick(1.5); + press('7'); + tick(1.5); + assert.deepStrictEqual(entries, []); + tick(0.5); + assert.deepStrictEqual(entries, ['567']); +}); + +test('the inter-digit timer replaces the overall timeout while typing', (t) => { + const {entries, press, tick} = make(t, {interDigitTimeout: 5, timeout: 15}); + for (let i = 0; i < 20; i++) { + press(String(i % 10)); + tick(1); + } + assert.deepStrictEqual(entries, []); + tick(4); + assert.deepStrictEqual(entries, ['01234567890123456789']); +}); + +test('interDigitTimeout waits for minDigits', (t) => { + const {entries, press, tick} = make(t, {interDigitTimeout: 1, minDigits: 3, timeout: 0}); + press('12'); + tick(5); + assert.deepStrictEqual(entries, []); + press('3'); + tick(1); + assert.deepStrictEqual(entries, ['123']); +}); + +test('timeout defaults to 15s and completes the entry', (t) => { + const {entries, press, tick} = make(t, {}); + press('42'); + tick(14.9); + assert.deepStrictEqual(entries, []); + tick(0.1); + assert.deepStrictEqual(entries, ['42']); +}); + +test('timeout below minDigits drops the entry', (t) => { + const {entries, press, tick} = make(t, {timeout: 1, minDigits: 3}); + press('42'); + tick(1); + assert.deepStrictEqual(entries, []); +}); + +test('warns when no option can ever complete an entry', (t) => { + assert.strictEqual(make(t, {timeout: 0}).warnings.length, 1); +}); + +test('no warning when one option completes an entry', (t) => { + assert.strictEqual(make(t, {timeout: 0, finishOnKey: '#'}).warnings.length, 0); +}); + +test('stop removes only its own listener and discards the buffer', (t) => { + const {ep, input, entries, press, tick} = make(t, {interDigitTimeout: 1}); + ep.on('dtmf', noop); + press('9'); + input.stop(); + tick(2); + press('8'); + assert.deepStrictEqual(entries, []); + assert.deepStrictEqual(ep.listeners('dtmf'), [noop]); +});