Merge pull request #51 from boldare/feat/assistant-streaming

feat(openai-assistant): assistants streaming events

apps/spa/src/app/components/chat/chat-audio/chat-audio.component.ts

@@ -1,6 +1,9 @@
 import { Component, Input, OnInit } from '@angular/core';
 import { ChatClientService } from '../../../modules/+chat/shared/chat-client.service';
-import { ChatMessage, SpeechVoice } from '../../../modules/+chat/shared/chat.model';
+import {
+  ChatMessage,
+  SpeechVoice,
+} from '../../../modules/+chat/shared/chat.model';
 import { environment } from '../../../../environments/environment';
 import { MatIconModule } from '@angular/material/icon';
 import { delay } from 'rxjs';

apps/spa/src/app/components/chat/chat-message/chat-message.component.ts

@@ -1,5 +1,8 @@
 import { Component, HostBinding, Input } from '@angular/core';
-import { ChatRole, ChatMessage } from '../../../modules/+chat/shared/chat.model';
+import {
+  ChatRole,
+  ChatMessage,
+} from '../../../modules/+chat/shared/chat.model';
 import { MarkdownComponent } from 'ngx-markdown';
 import { ChatAudioComponent } from '../chat-audio/chat-audio.component';
 import { NgClass } from '@angular/common';

apps/spa/src/app/modules/+chat/containers/chat-iframe/chat-iframe.component.html

@@ -8,19 +8,19 @@
 
   <ai-spinner [isActive]="isLoading()" />
   @if (isConfigEnabled && !threadId()) {
-  <ai-configuration-form class="chat__content" />
+    <ai-configuration-form class="chat__content" />
   } @else {
-  <ai-chat-messages
-    [messages]="messages()"
-    [isTyping]="isTyping()"
-    [tips]="tips"
-    (tipSelected$)="chatService.sendMessage($event)"
-    class="chat__content" />
-  <ai-chat-footer
-    [isDisabled]="isTyping()"
-    [isTranscriptionEnabled]="isTranscriptionEnabled"
-    [isAttachmentEnabled]="isAttachmentEnabled"
-    (sendMessage$)="chatService.sendMessage($event)"
-    (sendAudio$)="chatService.sendAudio($event)" />
+    <ai-chat-messages
+      [messages]="messages()"
+      [isTyping]="isTyping()"
+      [tips]="tips"
+      (tipSelected$)="chatService.sendMessage($event)"
+      class="chat__content" />
+    <ai-chat-footer
+      [isDisabled]="isTyping()"
+      [isTranscriptionEnabled]="isTranscriptionEnabled"
+      [isAttachmentEnabled]="isAttachmentEnabled"
+      (sendMessage$)="chatService.sendMessage($event)"
+      (sendAudio$)="chatService.sendAudio($event)" />
   }
 </ai-card>

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

@@ -1,22 +1,43 @@
 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 {
-    this.socket.emit(ChatEvents.SendMessage, payload);
-  }
-
-  getMessages(): Observable<ChatCallDto> {
-    return new Observable<ChatCallDto>(observer => {
-      this.socket.on(ChatEvents.MessageReceived, data => observer.next(data));
+  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);
+  }
+
+  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

@@ -15,8 +15,21 @@ export interface ChatMessage {
 }
 
 export enum ChatEvents {
-  SendMessage = 'send_message',
-  MessageReceived = 'message_received',
+  CallStart = 'callStart',
+  CallDone = 'callDone',
+  MessageCreated = 'messageCreated',
+  MessageDelta = 'messageDelta',
+  MessageDone = 'messageDone',
+  TextCreated = 'textCreated',
+  TextDelta = 'textDelta',
+  TextDone = 'textDone',
+  ImageFileDone = 'imageFileDone',
+  ToolCallCreated = 'toolCallCreated',
+  ToolCallDelta = 'toolCallDelta',
+  ToolCallDone = 'toolCallDone',
+  RunStepCreated = 'runStepCreated',
+  RunStepDelta = 'runStepDelta',
+  RunStepDone = 'runStepDone',
 }
 
 export enum ChatMessageStatus {

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

@@ -19,7 +19,6 @@ import { OpenAiFile, GetThreadResponseDto } from '@boldare/openai-assistant';
 import { Message } from 'openai/resources/beta/threads/messages';
 import { TextContentBlock } from 'openai/resources/beta/threads/messages/messages';
 
-
 @Injectable({ providedIn: 'root' })
 export class ChatService {
   isLoading$ = new BehaviorSubject<boolean>(false);
@@ -35,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;
@@ -87,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([]);
   }
@@ -120,15 +131,42 @@ 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 => {
+    return this.chatGatewayService.callDone().subscribe(data => {
       this.addMessage({
         content: data.content,
         role: ChatRole.Assistant,

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

@@ -1,5 +1,12 @@
 import { Injectable } from '@angular/core';
-import { BehaviorSubject, catchError, Observable, Subject, take, tap } from 'rxjs';
+import {
+  BehaviorSubject,
+  catchError,
+  Observable,
+  Subject,
+  take,
+  tap,
+} from 'rxjs';
 import { environment } from '../../../../environments/environment';
 import { ThreadClientService } from './thread-client.service';
 import { ConfigurationFormService } from '../../+configuration/shared/configuration-form.service';
@@ -43,11 +50,9 @@ export class ThreadService {
   }
 
   getThread(id: string): Observable<GetThreadResponseDto> {
-    return this.threadClientService
-      .getThread(id)
-      .pipe(
-        take(1),
-        catchError(() => this.start()),
-      );
+    return this.threadClientService.getThread(id).pipe(
+      take(1),
+      catchError(() => this.start()),
+    );
   }
 }

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,
 };

libs/openai-assistant/src/lib/agent/agent.mock.ts

@@ -1,10 +1,10 @@
-import { AssistantCreateParams } from 'openai/resources/beta';
+import { FunctionTool } from 'openai/resources/beta';
 
 export const agentNameMock = 'agent-name';
 
 export const agentMock = async () => 'agent-result';
 
-export const definitionMock: AssistantCreateParams.AssistantToolsFunction = {
+export const definitionMock: FunctionTool = {
   type: 'function',
   function: { name: agentNameMock },
 };

libs/openai-assistant/src/lib/agent/agent.service.ts

@@ -7,10 +7,7 @@ export class AgentService {
   public agents: Agents = {};
   public tools: FunctionTool[] = [];
 
-  add(
-    definition: FunctionTool,
-    fn: Agent,
-  ): void {
+  add(definition: FunctionTool, fn: Agent): void {
     this.tools.push(definition);
     this.agents[definition.function.name] = fn;
   }

libs/openai-assistant/src/lib/assistant/assistant.module.ts

@@ -1,4 +1,10 @@
-import { DynamicModule, Inject, Module, OnModuleInit, Optional } from '@nestjs/common';
+import {
+  DynamicModule,
+  Inject,
+  Module,
+  OnModuleInit,
+  Optional,
+} from '@nestjs/common';
 import { HttpModule } from '@nestjs/axios';
 import {
   AssistantService,

libs/openai-assistant/src/lib/chat/chat.gateway.spec.ts

@@ -32,7 +32,7 @@ describe('ChatGateway', () => {
 
       await chatGateway.listenForMessages(request, {} as Socket);
 
-      expect(chatService.call).toHaveBeenCalledWith(request);
+      expect(chatService.call).toHaveBeenCalled();
     });
   });