Custom transcriber

1
const { createClient, LiveTranscriptionEvents } = require("@deepgram/sdk");
2
const EventEmitter = require("events");
3
4
const PUNCTUATION_TERMINATORS = [".", "!", "?"];
5
const MAX_RETRY_ATTEMPTS = 3;
6
const DEBOUNCE_DELAY_IN_SECS = 3;
7
const DEBOUNCE_DELAY = DEBOUNCE_DELAY_IN_SECS * 1000;
8
const DEEPGRAM_API_KEY = process.env["DEEPGRAM_API_KEY"] || "";
9
10
class TranscriptionService extends EventEmitter {
11
  constructor(config, logger) {
12
    super();
13
    this.config = config;
14
    this.logger = logger;
15
    this.flowLogger = require("./fileLogger").createNamedLogger(
16
      "transcriber-flow.log"
17
    );
18
    if (!DEEPGRAM_API_KEY) {
19
      throw new Error("Missing Deepgram API Key");
20
    }
21
    this.deepgramClient = createClient(DEEPGRAM_API_KEY);
22
    this.logger.logDetailed(
23
      "INFO",
24
      "Initializing Deepgram live connection",
25
      "TranscriptionService",
26
      {
27
        model: "nova-2",
28
        sample_rate: 16000,
29
        channels: 2,
30
      }
31
    );
32
    this.deepgramLive = this.deepgramClient.listen.live({
33
      encoding: "linear16",
34
      channels: 2,
35
      sample_rate: 16000,
36
      model: "nova-2",
37
      smart_format: true,
38
      interim_results: true,
39
      endpointing: 800,
40
      language: "en",
41
      multichannel: true,
42
    });
43
    this.finalResult = { customer: "", assistant: "" };
44
    this.audioBuffer = [];
45
    this.retryAttempts = 0;
46
    this.lastTranscriptionTime = Date.now();
47
    this.pcmBuffer = Buffer.alloc(0);
48
49
    this.deepgramLive.addListener(LiveTranscriptionEvents.Open, () => {
50
      this.logger.logDetailed(
51
        "INFO",
52
        "Deepgram connection opened",
53
        "TranscriptionService"
54
      );
55
      this.deepgramLive.on(LiveTranscriptionEvents.Close, () => {
56
        this.logger.logDetailed(
57
          "INFO",
58
          "Deepgram connection closed",
59
          "TranscriptionService"
60
        );
61
        this.emitTranscription();
62
        this.audioBuffer = [];
63
      });
64
      this.deepgramLive.on(LiveTranscriptionEvents.Metadata, (data) => {
65
        this.logger.logDetailed(
66
          "DEBUG",
67
          "Deepgram metadata received",
68
          "TranscriptionService",
69
          data
70
        );
71
      });
72
      this.deepgramLive.on(LiveTranscriptionEvents.Transcript, (event) => {
73
        this.handleTranscript(event);
74
      });
75
      this.deepgramLive.on(LiveTranscriptionEvents.Error, (err) => {
76
        this.logger.logDetailed(
77
          "ERROR",
78
          "Deepgram error received",
79
          "TranscriptionService",
80
          { error: err }
81
        );
82
        this.emit("transcriptionerror", err);
83
      });
84
    });
85
  }
86
87
  send(payload) {
88
    if (payload instanceof Buffer) {
89
      this.pcmBuffer =
90
        this.pcmBuffer.length === 0
91
          ? payload
92
          : Buffer.concat([this.pcmBuffer, payload]);
93
    } else {
94
      this.logger.warn("TranscriptionService: Received non-Buffer data chunk.");
95
    }
96
    if (this.deepgramLive.getReadyState() === 1 && this.pcmBuffer.length > 0) {
97
      this.sendBufferedData(this.pcmBuffer);
98
      this.pcmBuffer = Buffer.alloc(0);
99
    }
100
  }
101
102
  sendBufferedData(bufferedData) {
103
    try {
104
      this.logger.logDetailed(
105
        "INFO",
106
        "Sending buffered data to Deepgram",
107
        "TranscriptionService",
108
        { bytes: bufferedData.length }
109
      );
110
      this.deepgramLive.send(bufferedData);
111
      this.audioBuffer = [];
112
      this.retryAttempts = 0;
113
    } catch (error) {
114
      this.logger.logDetailed(
115
        "ERROR",
116
        "Error sending buffered data",
117
        "TranscriptionService",
118
        { error }
119
      );
120
      this.retryAttempts++;
121
      if (this.retryAttempts <= MAX_RETRY_ATTEMPTS) {
122
        setTimeout(() => {
123
          this.sendBufferedData(bufferedData);
124
        }, 1000);
125
      } else {
126
        this.logger.logDetailed(
127
          "ERROR",
128
          "Max retry attempts reached, discarding data",
129
          "TranscriptionService"
130
        );
131
        this.audioBuffer = [];
132
        this.retryAttempts = 0;
133
      }
134
    }
135
  }
136
137
  handleTranscript(transcription) {
138
    if (!transcription.channel || !transcription.channel.alternatives?.[0]) {
139
      this.logger.logDetailed(
140
        "WARN",
141
        "Invalid transcript format",
142
        "TranscriptionService",
143
        { transcription }
144
      );
145
      return;
146
    }
147
    const text = transcription.channel.alternatives[0].transcript.trim();
148
    if (!text) return;
149
    const currentTime = Date.now();
150
    const channelIndex = transcription.channel_index
151
      ? transcription.channel_index[0]
152
      : 0;
153
    const channel = channelIndex === 0 ? "customer" : "assistant";
154
    this.logger.logDetailed(
155
      "INFO",
156
      "Received transcript",
157
      "TranscriptionService",
158
      { channel, text }
159
    );
160
    if (transcription.is_final || transcription.speech_final) {
161
      this.finalResult[channel] += ` ${text}`;
162
      this.emitTranscription();
163
    } else {
164
      this.finalResult[channel] += ` ${text}`;
165
      if (currentTime - this.lastTranscriptionTime >= DEBOUNCE_DELAY) {
166
        this.logger.logDetailed(
167
          "INFO",
168
          `Emitting transcript after ${DEBOUNCE_DELAY_IN_SECS}s inactivity`,
169
          "TranscriptionService"
170
        );
171
        this.emitTranscription();
172
      }
173
    }
174
    this.lastTranscriptionTime = currentTime;
175
  }
176
177
  emitTranscription() {
178
    for (const chan of ["customer", "assistant"]) {
179
      if (this.finalResult[chan].trim()) {
180
        const transcript = this.finalResult[chan].trim();
181
        this.logger.logDetailed(
182
          "INFO",
183
          "Emitting transcription",
184
          "TranscriptionService",
185
          { channel: chan, transcript }
186
        );
187
        this.emit("transcription", transcript, chan);
188
        this.finalResult[chan] = "";
189
      }
190
    }
191
  }
192
}
193
194
module.exports = TranscriptionService;

1
const express = require("express");
2
const http = require("http");
3
const TranscriptionService = require("./transcriptionService");
4
const FileLogger = require("./fileLogger");
5
require("dotenv").config();
6
7
const app = express();
8
app.use(express.json());
9
app.use(express.urlencoded({ extended: true }));
10
11
app.get("/", (req, res) => {
12
  res.send("Custom Transcriber Service is running");
13
});
14
15
const server = http.createServer(app);
16
17
const config = {
18
  DEEPGRAM_API_KEY: process.env.DEEPGRAM_API_KEY,
19
  PORT: process.env.PORT || 3001,
20
};
21
22
const logger = new FileLogger();
23
const transcriptionService = new TranscriptionService(config, logger);
24
25
transcriptionService.setupWebSocketServer = function (server) {
26
  const WebSocketServer = require("ws").Server;
27
  const wss = new WebSocketServer({ server, path: "/api/custom-transcriber" });
28
  wss.on("connection", (ws) => {
29
    logger.logDetailed(
30
      "INFO",
31
      "New WebSocket client connected on /api/custom-transcriber",
32
      "Server"
33
    );
34
    ws.on("message", (data, isBinary) => {
35
      if (!isBinary) {
36
        try {
37
          const msg = JSON.parse(data.toString());
38
          if (msg.type === "start") {
39
            logger.logDetailed(
40
              "INFO",
41
              "Received start message from client",
42
              "Server",
43
              { sampleRate: msg.sampleRate, channels: msg.channels }
44
            );
45
          }
46
        } catch (err) {
47
          logger.error("JSON parse error", err, "Server");
48
        }
49
      } else {
50
        transcriptionService.send(data);
51
      }
52
    });
53
    ws.on("close", () => {
54
      logger.logDetailed("INFO", "WebSocket client disconnected", "Server");
55
      if (
56
        transcriptionService.deepgramLive &&
57
        transcriptionService.deepgramLive.getReadyState() === 1
58
      ) {
59
        transcriptionService.deepgramLive.finish();
60
      }
61
    });
62
    ws.on("error", (error) => {
63
      logger.error("WebSocket error", error, "Server");
64
    });
65
    transcriptionService.on("transcription", (text, channel) => {
66
      const response = {
67
        type: "transcriber-response",
68
        transcription: text,
69
        channel,
70
        transcriptType: "final",
71
      };
72
      ws.send(JSON.stringify(response));
73
      logger.logDetailed("INFO", "Sent transcription to client", "Server", {
74
        channel,
75
        text,
76
      });
77
    });
78
    transcriptionService.on("transcriptionerror", (err) => {
79
      ws.send(
80
        JSON.stringify({ type: "error", error: "Transcription service error" })
81
      );
82
      logger.error("Transcription service error", err, "Server");
83
    });
84
  });
85
};
86
87
transcriptionService.setupWebSocketServer(server);
88
89
server.listen(config.PORT, () => {
90
  console.log(`Server is running on http://localhost:${config.PORT}`);
91
});