Merge pull request #173 from anam-org/sam/thinking_models

feat: handle webrtc reasoning events

Author: sr-anam

src/AnamClient.ts

@@ -18,6 +18,7 @@ import {
   MessageHistoryClient,
   PublicEventEmitter,
   StreamingClient,
+  ReasoningHistoryClient,
 } from './modules';
 import {
   AnamClientOptions,
@@ -38,6 +39,7 @@ export default class AnamClient {
   private internalEventEmitter: InternalEventEmitter;
 
   private readonly messageHistoryClient: MessageHistoryClient;
+  private readonly reasoningHistoryClient: ReasoningHistoryClient;
 
   private personaConfig: PersonaConfig | undefined;
   private clientOptions: AnamClientOptions | undefined;
@@ -104,6 +106,11 @@ export default class AnamClient {
       this.publicEventEmitter,
       this.internalEventEmitter,
     );
+
+    this.reasoningHistoryClient = new ReasoningHistoryClient(
+      this.publicEventEmitter,
+      this.internalEventEmitter,
+    );
   }
 
   private decodeJwt(token: string): any {

src/modules/ReasoningHistoryClient.ts

@@ -0,0 +1,76 @@
+import { PublicEventEmitter, InternalEventEmitter } from '.';
+import {
+  AnamEvent,
+  InternalEvent,
+  ReasoningMessage,
+  ReasoningStreamEvent,
+  WebRtcReasoningTextMessageEvent,
+} from '../types';
+
+export class ReasoningHistoryClient {
+  private publicEventEmitter: PublicEventEmitter;
+  private internalEventEmitter: InternalEventEmitter;
+
+  private reasoning_messages: ReasoningMessage[] = [];
+  constructor(
+    publicEventEmitter: PublicEventEmitter,
+    internalEventEmitter: InternalEventEmitter,
+  ) {
+    this.publicEventEmitter = publicEventEmitter;
+    this.internalEventEmitter = internalEventEmitter;
+    // register for events
+    this.internalEventEmitter.addListener(
+      InternalEvent.WEBRTC_REASONING_TEXT_MESSAGE_RECEIVED,
+      this.processWebRtcReasoningTextMessageEvent.bind(this),
+    );
+  }
+
+  private webRtcTextMessageEventToReasoningStreamEvent(
+    event: WebRtcReasoningTextMessageEvent,
+  ): ReasoningStreamEvent {
+    return {
+      id: `${event.role}::${event.message_id}`,
+      content: event.content,
+      endOfThought: event.end_of_thought,
+      role: event.role,
+    };
+  }
+
+  private processWebRtcReasoningTextMessageEvent(
+    event: WebRtcReasoningTextMessageEvent,
+  ): void {
+    const ReasoningStreamEvent: ReasoningStreamEvent =
+      this.webRtcTextMessageEventToReasoningStreamEvent(event);
+
+    this.publicEventEmitter.emit(
+      AnamEvent.REASONING_STREAM_EVENT_RECEIVED,
+      ReasoningStreamEvent,
+    );
+
+    const message: ReasoningMessage = {
+      id: ReasoningStreamEvent.id,
+      content: ReasoningStreamEvent.content,
+      role: ReasoningStreamEvent.role,
+    };
+
+    const existingMessageIndex = this.reasoning_messages.findIndex(
+      (m) => m.id === message.id,
+    );
+    if (existingMessageIndex !== -1) {
+      // update existing message
+      const existingMessage = this.reasoning_messages[existingMessageIndex];
+      existingMessage.content += message.content;
+      this.reasoning_messages[existingMessageIndex] = existingMessage;
+    } else {
+      // new message
+      this.reasoning_messages.push(message);
+    }
+
+    if (ReasoningStreamEvent.endOfThought) {
+      this.publicEventEmitter.emit(
+        AnamEvent.REASONING_HISTORY_UPDATED,
+        this.reasoning_messages,
+      );
+    }
+  }
+}

src/modules/StreamingClient.ts

@@ -25,6 +25,7 @@ import {
   StreamingClientOptions,
   WebRtcClientToolEvent,
   WebRtcTextMessageEvent,
+  WebRtcReasoningTextMessageEvent,
 } from '../types';
 import { AgentAudioInputStream } from '../types/AgentAudioInputStream';
 import { TalkMessageStream } from '../types/TalkMessageStream';
@@ -693,6 +694,7 @@ export class StreamingClient {
               message.data as WebRtcTextMessageEvent,
             );
             break;
+
           case DataChannelMessage.CLIENT_TOOL_EVENT:
             const webRtcToolEvent = message.data as WebRtcClientToolEvent;
 
@@ -709,6 +711,12 @@ export class StreamingClient {
               clientToolEvent,
             );
             break;
+          case DataChannelMessage.REASONING_TEXT:
+            this.internalEventEmitter.emit(
+              InternalEvent.WEBRTC_REASONING_TEXT_MESSAGE_RECEIVED,
+              message.data as WebRtcReasoningTextMessageEvent,
+            );
+            break;
           // Unknown message types are silently ignored to maintain forward compatibility
           default:
             break;

src/modules/index.ts

@@ -6,3 +6,4 @@ export { MessageHistoryClient } from './MessageHistoryClient';
 export { PublicEventEmitter } from './PublicEventEmitter';
 export { StreamingClient } from './StreamingClient';
 export { ToolCallManager } from './ToolCallManager';
+export { ReasoningHistoryClient } from './ReasoningHistoryClient';

src/types/AgentAudioInputStream.ts

@@ -62,7 +62,9 @@ export class AgentAudioInputStream {
   private arrayBufferToBase64(buffer: ArrayBuffer | Uint8Array): string {
     const bytes =
       buffer instanceof Uint8Array ? buffer : new Uint8Array(buffer);
-    const binary = Array.from(bytes, byte => String.fromCharCode(byte)).join('');
+    const binary = Array.from(bytes, (byte) => String.fromCharCode(byte)).join(
+      '',
+    );
     return btoa(binary);
   }
 }

src/types/events/internal/InternalEvent.ts

@@ -3,4 +3,5 @@ export enum InternalEvent {
   SIGNAL_MESSAGE_RECEIVED = 'SIGNAL_MESSAGE_RECEIVED',
   WEBRTC_CHAT_MESSAGE_RECEIVED = 'WEBRTC_CHAT_MESSAGE_RECEIVED',
   WEBRTC_CLIENT_TOOL_EVENT_RECEIVED = 'WEBRTC_CLIENT_TOOL_EVENT_RECEIVED',
+  WEBRTC_REASONING_TEXT_MESSAGE_RECEIVED = 'WEBRTC_REASONING_TEXT_MESSAGE_RECEIVED',
 }

src/types/events/internal/InternalEventCallbacks.ts

@@ -3,6 +3,7 @@ import {
   SignalMessage,
   WebRtcTextMessageEvent,
   WebRtcClientToolEvent,
+  WebRtcReasoningTextMessageEvent,
 } from '../../index';
 
 export type InternalEventCallbacks = {
@@ -16,4 +17,7 @@ export type InternalEventCallbacks = {
   [InternalEvent.WEBRTC_CLIENT_TOOL_EVENT_RECEIVED]: (
     webRtcToolEvent: WebRtcClientToolEvent,
   ) => void;
+  [InternalEvent.WEBRTC_REASONING_TEXT_MESSAGE_RECEIVED]: (
+    message: WebRtcReasoningTextMessageEvent,
+  ) => void;
 };

src/types/events/public/AnamEvent.ts

@@ -15,4 +15,6 @@ export enum AnamEvent {
   MIC_PERMISSION_DENIED = 'MIC_PERMISSION_DENIED',
   INPUT_AUDIO_DEVICE_CHANGED = 'INPUT_AUDIO_DEVICE_CHANGED',
   CLIENT_TOOL_EVENT_RECEIVED = 'CLIENT_TOOL_EVENT_RECEIVED',
+  REASONING_HISTORY_UPDATED = 'REASONING_HISTORY_UPDATED',
+  REASONING_STREAM_EVENT_RECEIVED = 'REASONING_STREAM_EVENT_RECEIVED',
 }

src/types/events/public/EventCallbacks.ts

@@ -4,6 +4,8 @@ import {
   MessageStreamEvent,
   AnamEvent,
   ClientToolEvent,
+  ReasoningMessage,
+  ReasoningStreamEvent,
 } from '../../index';
 
 export type EventCallbacks = {
@@ -30,4 +32,10 @@ export type EventCallbacks = {
   [AnamEvent.CLIENT_TOOL_EVENT_RECEIVED]: (
     clientToolEvent: ClientToolEvent,
   ) => void;
+  [AnamEvent.REASONING_HISTORY_UPDATED]: (
+    thoughtMessages: ReasoningMessage[],
+  ) => void;
+  [AnamEvent.REASONING_STREAM_EVENT_RECEIVED]: (
+    thoughtEvent: ReasoningStreamEvent,
+  ) => void;
 };

src/types/messageHistory/ReasoningMessage.ts

@@ -0,0 +1,5 @@
+export interface ReasoningMessage {
+  id: string;
+  content: string;
+  role: string;
+}