Send lk.segment_id attribute in transcription events so non final transcriptions can be deduplicated client side (#580)

Author: 1egoman

.changeset/heavy-cherries-yell.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+add `lk.segment_id` to all generated trancription events

agents/src/constants.ts

@@ -4,4 +4,5 @@
 export const ATTRIBUTE_TRANSCRIPTION_TRACK_ID = 'lk.transcribed_track_id';
 export const ATTRIBUTE_TRANSCRIPTION_FINAL = 'lk.transcription_final';
 export const TOPIC_TRANSCRIPTION = 'lk.transcription';
+export const ATTRIBUTE_SEGMENT_ID = 'lk.segment_id';
 export const TOPIC_CHAT = 'lk.chat';

agents/src/multimodal/multimodal_agent.ts

@@ -18,9 +18,11 @@ import {
   TrackPublishOptions,
   TrackSource,
 } from '@livekit/rtc-node';
+import { randomUUID } from 'node:crypto';
 import { EventEmitter } from 'node:events';
 import { AudioByteStream } from '../audio.js';
 import {
+  ATTRIBUTE_SEGMENT_ID,
   ATTRIBUTE_TRANSCRIPTION_FINAL,
   ATTRIBUTE_TRANSCRIPTION_TRACK_ID,
   TOPIC_TRANSCRIPTION,
@@ -72,6 +74,7 @@ export class MultimodalAgent extends EventEmitter {
 
   #textResponseRetries = 0;
   #maxTextResponseRetries: number;
+  #transcriptionId?: string;
 
   constructor({
     model,
@@ -257,13 +260,20 @@ export class MultimodalAgent extends EventEmitter {
 
         const synchronizer = new TextAudioSynchronizer(defaultTextSyncOptions);
         synchronizer.on('textUpdated', async (text) => {
+          if (!this.#transcriptionId) {
+            this.#transcriptionId = randomUUID();
+          }
           await this.#publishTranscription(
             this.room!.localParticipant!.identity!,
             this.#getLocalTrackSid()!,
             text.text,
             text.final,
             text.id,
+            this.#transcriptionId,
           );
+          if (text.final) {
+            this.#transcriptionId = undefined;
+          }
         });
 
         const handle = this.#agentPlayout?.play(
@@ -312,7 +322,17 @@ export class MultimodalAgent extends EventEmitter {
         const participantIdentity = this.linkedParticipant?.identity;
         const trackSid = this.subscribedTrack?.sid;
         if (participantIdentity && trackSid) {
-          await this.#publishTranscription(participantIdentity, trackSid, '…', false, ev.itemId);
+          if (!this.#transcriptionId) {
+            this.#transcriptionId = randomUUID();
+          }
+          await this.#publishTranscription(
+            participantIdentity,
+            trackSid,
+            '…',
+            false,
+            ev.itemId,
+            this.#transcriptionId,
+          );
         } else {
           this.#logger.error('Participant or track not set');
         }
@@ -325,13 +345,18 @@ export class MultimodalAgent extends EventEmitter {
         const participantIdentity = this.linkedParticipant?.identity;
         const trackSid = this.subscribedTrack?.sid;
         if (participantIdentity && trackSid) {
+          if (!this.#transcriptionId) {
+            this.#transcriptionId = randomUUID();
+          }
           await this.#publishTranscription(
             participantIdentity,
             trackSid,
             transcription,
             true,
             ev.itemId,
+            this.#transcriptionId,
           );
+          this.#transcriptionId = undefined;
         } else {
           this.#logger.error('Participant or track not set');
         }
@@ -360,7 +385,17 @@ export class MultimodalAgent extends EventEmitter {
         const participantIdentity = this.linkedParticipant?.identity;
         const trackSid = this.subscribedTrack?.sid;
         if (participantIdentity && trackSid) {
-          await this.#publishTranscription(participantIdentity, trackSid, '…', false, ev.itemId);
+          if (!this.#transcriptionId) {
+            this.#transcriptionId = randomUUID();
+          }
+          await this.#publishTranscription(
+            participantIdentity,
+            trackSid,
+            '…',
+            false,
+            ev.itemId,
+            this.#transcriptionId,
+          );
         }
       });
 
@@ -492,6 +527,7 @@ export class MultimodalAgent extends EventEmitter {
     text: string,
     isFinal: boolean,
     id: string,
+    segmentId: string,
   ): Promise<void> {
     this.#logger.debug(
       `Publishing transcription ${participantIdentity} ${trackSid} ${text} ${isFinal} ${id}`,
@@ -522,6 +558,7 @@ export class MultimodalAgent extends EventEmitter {
       attributes: {
         [ATTRIBUTE_TRANSCRIPTION_TRACK_ID]: trackSid,
         [ATTRIBUTE_TRANSCRIPTION_FINAL]: isFinal.toString(),
+        [ATTRIBUTE_SEGMENT_ID]: segmentId,
       },
     });
     await stream.write(text);

agents/src/pipeline/pipeline_agent.ts

@@ -18,6 +18,7 @@ import type { TypedEventEmitter as TypedEmitter } from '@livekit/typed-emitter';
 import { randomUUID } from 'node:crypto';
 import EventEmitter from 'node:events';
 import {
+  ATTRIBUTE_SEGMENT_ID,
   ATTRIBUTE_TRANSCRIPTION_FINAL,
   ATTRIBUTE_TRANSCRIPTION_TRACK_ID,
   TOPIC_TRANSCRIPTION,
@@ -537,6 +538,7 @@ export class VoicePipelineAgent extends (EventEmitter as new () => TypedEmitter<
         this.#transcribedInterimText,
         false,
         this.#transcriptionId,
+        this.#transcriptionId,
       );
     });
     this.#humanInput.on(HumanInputEvent.FINAL_TRANSCRIPT, async (event) => {
@@ -556,6 +558,7 @@ export class VoicePipelineAgent extends (EventEmitter as new () => TypedEmitter<
         this.transcribedText,
         true,
         this.#transcriptionId,
+        this.#transcriptionId,
       );
 
       this.#transcriptionId = undefined;
@@ -895,6 +898,7 @@ export class VoicePipelineAgent extends (EventEmitter as new () => TypedEmitter<
     text: string,
     isFinal: boolean,
     id: string,
+    segmentId: string,
   ) {
     this.#room!.localParticipant!.publishTranscription({
       participantIdentity: participantIdentity,
@@ -916,6 +920,7 @@ export class VoicePipelineAgent extends (EventEmitter as new () => TypedEmitter<
       attributes: {
         [ATTRIBUTE_TRANSCRIPTION_TRACK_ID]: trackSid,
         [ATTRIBUTE_TRANSCRIPTION_FINAL]: isFinal.toString(),
+        [ATTRIBUTE_SEGMENT_ID]: segmentId,
       },
     });
     await stream.write(text);
@@ -930,13 +935,20 @@ export class VoicePipelineAgent extends (EventEmitter as new () => TypedEmitter<
     // TODO: where possible we would want to use deltas instead of full text segments, esp for LLM streams over the streamText API
     synchronizer.on('textUpdated', async (text) => {
       this.#agentTranscribedText = text.text;
+      if (!this.#transcriptionId) {
+        this.#transcriptionId = randomUUID();
+      }
       await this.#publishTranscription(
         this.#room!.localParticipant!.identity!,
         this.#agentPublication?.sid ?? '',
         text.text,
         text.final,
         text.id,
+        this.#transcriptionId,
       );
+      if (text.final) {
+        this.#transcriptionId = undefined;
+      }
     });
 
     if (!this.#agentOutput) {