feat(spa): displaying the bot's reply as a streaming message

Author: sebastianmusial

apps/spa/src/app/modules/+chat/shared/chat-gateway.service.ts

@@ -1,22 +1,41 @@
 import { Injectable } from '@angular/core';
 import { ChatEvents } from './chat.model';
 import io from 'socket.io-client';
-import { ChatCallDto } from '@boldare/openai-assistant';
+import {
+  ChatCallDto,
+  TextCreatedPayload, TextDeltaPayload, TextDonePayload
+} from '@boldare/openai-assistant';
 import { Observable } from 'rxjs';
 import { environment } from '../../../../environments/environment';
 
 @Injectable({ providedIn: 'root' })
 export class ChatGatewayService {
   private socket = io(environment.websocketUrl);
 
-  sendMessage(payload: ChatCallDto): void {
+  watchEvent<T>(event: ChatEvents): Observable<T> {
+    return new Observable<T>(observer => {
+      this.socket.on(event, data => observer.next(data));
+      return () => this.socket.disconnect();
+    });
+  }
+
+  callStart(payload: ChatCallDto): void {
     this.socket.emit(ChatEvents.CallStart, payload);
   }
 
-  getMessages(): Observable<ChatCallDto> {
-    return new Observable<ChatCallDto>(observer => {
-      this.socket.on(ChatEvents.CallDone, data => observer.next(data));
-      return () => this.socket.disconnect();
-    });
+  callDone(): Observable<ChatCallDto> {
+    return this.watchEvent(ChatEvents.CallDone);
+  }
+
+  textCreated(): Observable<TextCreatedPayload> {
+    return this.watchEvent(ChatEvents.TextCreated);
+  }
+
+  textDelta(): Observable<TextDeltaPayload> {
+    return this.watchEvent(ChatEvents.TextDelta);
+  }
+
+  textDone(): Observable<TextDonePayload> {
+    return this.watchEvent(ChatEvents.TextDone);
   }
 }

apps/spa/src/app/modules/+chat/shared/chat.model.ts

@@ -17,12 +17,12 @@ export interface ChatMessage {
 export enum ChatEvents {
   CallStart = 'callStart',
   CallDone = 'callDone',
-  TextDone = 'textCreated',
-  TextCreated = 'textDelta',
-  TextDelta = 'textDone',
   MessageCreated = 'messageCreated',
   MessageDelta = 'messageDelta',
   MessageDone = 'messageDone',
+  TextCreated = 'textCreated',
+  TextDelta = 'textDelta',
+  TextDone = 'textDone',
   ImageFileDone = 'imageFileDone',
   ToolCallCreated = 'toolCallCreated',
   ToolCallDelta = 'toolCallDelta',

apps/spa/src/app/modules/+chat/shared/chat.service.ts

@@ -34,11 +34,21 @@ export class ChatService {
   ) {
     document.body.classList.add('ai-chat');
 
+    this.subscribeMessages();
     this.setInitialValues();
-    this.watchMessages();
     this.watchVisibility();
   }
 
+  subscribeMessages(): void {
+    if (!environment.isStreamingEnabled) {
+      this.watchMessages();
+    } else {
+      this.watchTextCreated();
+      this.watchTextDelta();
+      this.watchTextDone();
+    }
+  }
+
   isMessageInvisible(message: Message): boolean {
     const metadata = message.metadata as Record<string, unknown>;
     return metadata?.['status'] === ChatMessageStatus.Invisible;
@@ -86,11 +96,13 @@ export class ChatService {
 
   refresh(): void {
     this.isLoading$.next(true);
+    this.isTyping$.next(false);
     this.messages$.next([]);
     this.threadService.start().subscribe();
   }
 
   clear(): void {
+    this.isTyping$.next(false);
     this.threadService.clear();
     this.messages$.next([]);
   }
@@ -119,21 +131,56 @@ export class ChatService {
     const files = await this.chatFilesService.sendFiles();
     this.addFileMessage(files);
 
-    this.chatGatewayService.sendMessage({
+    this.chatGatewayService.callStart({
       content,
       threadId: this.threadService.threadId$.value,
       file_ids: files.map(file => file.id) || [],
     });
   }
 
+  watchTextCreated(): Subscription {
+    return this.chatGatewayService
+      .textCreated()
+      .subscribe((data) => {
+        this.isTyping$.next(false)
+        this.addMessage({ content: data.text.value, role: ChatRole.Assistant })
+      });
+  }
+
+  watchTextDelta(): Subscription {
+    return this.chatGatewayService
+      .textDelta()
+      .subscribe((data) => {
+        const length = this.messages$.value.length;
+        this.messages$.value[length - 1].content = data.text.value;
+      });
+  }
+
+  watchTextDone(): Subscription {
+    return this.chatGatewayService
+      .textDone()
+      .subscribe((data) => {
+        this.isTyping$.next(false);
+        this.messages$.next([
+          ...this.messages$.value.slice(0, -1),
+          {
+            content: data.text.value,
+            role: ChatRole.Assistant,
+          },
+        ]);
+      });
+  }
+
   watchMessages(): Subscription {
-    return this.chatGatewayService.getMessages().subscribe(data => {
-      this.addMessage({
-        content: data.content,
-        role: ChatRole.Assistant,
+    return this.chatGatewayService
+      .callDone()
+      .subscribe(data => {
+        this.addMessage({
+          content: data.content,
+          role: ChatRole.Assistant,
+        });
+        this.isTyping$.next(false);
       });
-      this.isTyping$.next(false);
-    });
   }
 
   sendAudio(file: Blob): void {

apps/spa/src/environments/environment.development.ts

@@ -10,4 +10,5 @@ export const environment = {
   isRefreshEnabled: true,
   isConfigEnabled: true,
   isAutoOpen: true,
+  isStreamingEnabled: true,
 };

apps/spa/src/environments/environment.ts

@@ -10,4 +10,5 @@ export const environment = {
   isRefreshEnabled: true,
   isConfigEnabled: true,
   isAutoOpen: true,
+  isStreamingEnabled: true,
 };