Merge pull request #995 from hchen2020/master

Model response timeout

Author: Oceania2018

src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs

@@ -1,4 +1,5 @@
 using BotSharp.Abstraction.Utilities;
+using BotSharp.Core.Infrastructures;
 
 namespace BotSharp.Core.Realtime.Hooks;
 
@@ -29,6 +30,10 @@ public async Task OnFunctionExecuted(RoleDialogModel message)
         {
             return;
         }
+
+        // Clear cache to force to rebuild the agent instruction
+        Utilities.ClearCache();
+
         var routing = _services.GetRequiredService<IRoutingService>();
 
         message.Role = AgentRole.Function;

src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs

@@ -49,6 +49,9 @@ public async Task Listen(WebSocket userWebSocket,
             {
                 await _completer.AppenAudioBuffer(_conn.Data);
             }
+            else if (_conn.Event == "user_dtmf_receiving")
+            {
+            }
             else if (_conn.Event == "user_dtmf_received")
             {
                 await HandleUserDtmfReceived();

src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs

@@ -144,13 +144,28 @@ private async Task ReceiveMessage(RealtimeHubConnection conn,
         Action onUserInterrupted)
     {
         var buffer = new byte[1024 * 32];
-        WebSocketReceiveResult result;
+        // Model response timeout
+        var timeout = 30;
+        WebSocketReceiveResult? result = default;
 
         do
         {
             Array.Clear(buffer, 0, buffer.Length);
-            result = await _webSocket.ReceiveAsync(
-                new ArraySegment<byte>(buffer), CancellationToken.None);
+            
+            var taskWorker = _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
+            var taskTimer = Task.Delay(1000 * timeout);
+            var completedTask = await Task.WhenAny(taskWorker, taskTimer);
+
+            if (completedTask == taskWorker)
+            {
+                result = taskWorker.Result;
+            }
+            else
+            {
+                _logger.LogWarning($"Timeout {timeout} seconds waiting for Model response.");
+                await TriggerModelInference("Response user immediately");
+                continue;
+            }
 
             // Convert received data to text/audio (Twilio sends Base64-encoded audio)
             string receivedText = Encoding.UTF8.GetString(buffer, 0, result.Count);
@@ -164,6 +179,11 @@ private async Task ReceiveMessage(RealtimeHubConnection conn,
             if (response.Type == "error")
             {
                 _logger.LogError($"{response.Type}: {receivedText}");
+                var error = JsonSerializer.Deserialize<ServerEventErrorResponse>(receivedText);
+                if (error?.Body.Type == "server_error")
+                {
+                    break;
+                }
             }
             else if (response.Type == "session.created")
             {
@@ -182,7 +202,6 @@ private async Task ReceiveMessage(RealtimeHubConnection conn,
             {
                 _logger.LogInformation($"{response.Type}: {receivedText}");
                 var data = JsonSerializer.Deserialize<ResponseAudioTranscript>(receivedText);
-                await Task.Delay(1000);
                 onModelAudioTranscriptDone(data.Transcript);
             }
             else if (response.Type == "response.audio.delta")

src/Plugins/BotSharp.Plugin.Twilio/TwilioStreamMiddleware.cs

@@ -1,5 +1,4 @@
 using BotSharp.Abstraction.Realtime;
-using BotSharp.Abstraction.Realtime.Models;
 using BotSharp.Plugin.Twilio.Interfaces;
 using BotSharp.Plugin.Twilio.Models.Stream;
 using Microsoft.AspNetCore.Http;
@@ -100,6 +99,7 @@ await hub.Listen(webSocket, (receivedText) =>
                     }
                     else
                     {
+                        conn.Event = "user_dtmf_receiving";
                         conn.KeypadInputBuffer += dtmfResponse.Body.Digit;
                     }
                     break;
@@ -115,19 +115,29 @@ await hub.Listen(webSocket, (receivedText) =>
                     streamSid = response.StreamSid,
                     media = new { payload = message }
                 };
+
             conn.OnModelAudioResponseDone = () =>
                 new
                 {
                     @event = "mark",
                     streamSid = response.StreamSid,
                     mark = new { name = "responsePart" }
                 };
+
             conn.OnModelUserInterrupted = () =>
                 new
                 {
                     @event = "clear",
                     streamSid = response.StreamSid
                 };
+
+            /*if (response.Event == "dtmf")
+            {
+                // Send a Stop command to Twilio
+                string stopPlaybackCommand = "{ \"action\": \"stop_playback\" }";
+                var stopBytes = Encoding.UTF8.GetBytes(stopPlaybackCommand);
+                webSocket.SendAsync(new ArraySegment<byte>(stopBytes), WebSocketMessageType.Text, true, CancellationToken.None);
+            }*/
         });
     }
 }