Merge pull request #915 from hchen2020/master

refactor realtime code.

Author: Oceania2018

src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs

@@ -24,7 +24,7 @@ Task Connect(RealtimeHubConnection conn,
     Task Disconnect();
 
     Task<RealtimeSession> CreateSession(Agent agent, List<RoleDialogModel> conversations);
-    Task UpdateSession(RealtimeHubConnection conn);
+    Task UpdateSession(RealtimeHubConnection conn, bool turnDetection = true);
     Task InsertConversationItem(RoleDialogModel message);
     Task RemoveConversationItem(string itemId);
     Task TriggerModelInference(string? instructions = null);

src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs

@@ -1,9 +1,16 @@
+using System.Collections.Concurrent;
+
 namespace BotSharp.Abstraction.Realtime.Models;
 
 public class RealtimeHubConnection
 {
     public string Event { get; set; } = null!;
     public string StreamId { get; set; } = null!;
+    public string? LastAssistantItem { get; set; } = null!;
+    public long LatestMediaTimestamp { get; set; }
+    public long? ResponseStartTimestamp { get; set; }
+    public string KeypadInputBuffer { get; set; } = string.Empty;
+    public ConcurrentQueue<string> MarkQueue { get; set; } = new();
     public string CurrentAgentId { get; set; } = null!;
     public string ConversationId { get; set; } = null!;
     public string Data { get; set; } = string.Empty;

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

@@ -4,6 +4,10 @@
 using BotSharp.Abstraction.MLTasks;
 using BotSharp.Abstraction.Conversations.Enums;
 using BotSharp.Abstraction.Routing.Models;
+using NetTopologySuite.Index.HPRtree;
+using BotSharp.Abstraction.Agents.Models;
+using Microsoft.Identity.Client.Extensions.Msal;
+using Microsoft.AspNetCore.Cors.Infrastructure;
 
 namespace BotSharp.Core.Realtime;
 
@@ -46,9 +50,14 @@ public async Task Listen(WebSocket userWebSocket,
             {
                 await completer.AppenAudioBuffer(conn.Data);
             }
+            else if (conn.Event == "user_dtmf_received")
+            {
+                await HandleUserDtmfReceived(completer, conn);
+            }
             else if (conn.Event == "user_disconnected")
             {
                 await completer.Disconnect();
+                await HandleUserDisconnected(conn);
             }
         } while (!result.CloseStatus.HasValue);
 
@@ -58,8 +67,6 @@ public async Task Listen(WebSocket userWebSocket,
     private async Task ConnectToModel(IRealTimeCompletion completer, WebSocket userWebSocket, RealtimeHubConnection conn)
     {
         var hookProvider = _services.GetRequiredService<ConversationHookProvider>();
-        var storage = _services.GetRequiredService<IConversationStorage>();
-
         var convService = _services.GetRequiredService<IConversationService>();
         convService.SetConversationId(conn.ConversationId, []);
         var conversation = await convService.GetConversation(conn.ConversationId);
@@ -92,8 +99,8 @@ private async Task ConnectToModel(IRealTimeCompletion completer, WebSocket userW
         await completer.Connect(conn, 
             onModelReady: async () => 
             {
-                // Control initial session
-                await completer.UpdateSession(conn);
+                // Control initial session, prevent initial response interruption
+                await completer.UpdateSession(conn, turnDetection: false);
 
                 // Add dialog history
                 foreach (var item in dialogs)
@@ -103,17 +110,41 @@ await completer.Connect(conn,
 
                 if (dialogs.LastOrDefault()?.Role == AgentRole.Assistant)
                 {
-                    // await completer.TriggerModelInference($"Rephase your last response:\r\n{dialogs.LastOrDefault()?.Content}");
+                    await completer.TriggerModelInference($"Rephase your last response:\r\n{dialogs.LastOrDefault()?.Content}");
                 }
                 else
                 {
                     await completer.TriggerModelInference("Reply based on the conversation context.");
                 }
+
+                // Start turn detection
+                await Task.Delay(1000 * 8);
+                await completer.UpdateSession(conn, turnDetection: true);
             },
             onModelAudioDeltaReceived: async audioDeltaData =>
             {
+                // If this is the first delta of a new response, set the start timestamp
+                if (!conn.ResponseStartTimestamp.HasValue)
+                {
+                    conn.ResponseStartTimestamp = conn.LatestMediaTimestamp;
+                    _logger.LogDebug($"Setting start timestamp for new response: {conn.ResponseStartTimestamp}ms");
+                }
+
                 var data = conn.OnModelMessageReceived(audioDeltaData);
                 await SendEventToUser(userWebSocket, data);
+
+                // Send mark messages to Media Streams so we know if and when AI response playback is finished
+                if (!string.IsNullOrEmpty(conn.StreamId))
+                {
+                    var markEvent = new
+                    {
+                        @event = "mark",
+                        streamSid = conn.StreamId,
+                        mark = new { name = "responsePart" }
+                    };
+                    await SendEventToUser(userWebSocket, markEvent);
+                    conn.MarkQueue.Enqueue("responsePart");
+                }
             }, 
             onModelAudioResponseDone: async () =>
             {
@@ -160,16 +191,18 @@ await completer.Connect(conn,
                             await completer.TriggerModelInference("Reply based on the function's output.");
                         }
                     }
-                    // append output audio transcript to conversation
-                    storage.Append(conn.ConversationId, message);
-                    dialogs.Add(message);
-
-                    foreach (var hook in hookProvider.HooksOrderByPriority)
+                    else
                     {
-                        hook.SetAgent(agent)
-                            .SetConversation(conversation);
+                        // append output audio transcript to conversation
+                        dialogs.Add(message);
+
+                        foreach (var hook in hookProvider.HooksOrderByPriority)
+                        {
+                            hook.SetAgent(agent)
+                                .SetConversation(conversation);
 
-                        await hook.OnResponseGenerated(message);
+                            await hook.OnResponseGenerated(message);
+                        }
                     }
                 }
             },
@@ -180,7 +213,6 @@ await completer.Connect(conn,
             onInputAudioTranscriptionCompleted: async message =>
             {
                 // append input audio transcript to conversation
-                storage.Append(conn.ConversationId, message);
                 dialogs.Add(message);
 
                 foreach (var hook in hookProvider.HooksOrderByPriority)
@@ -193,11 +225,56 @@ await completer.Connect(conn,
             },
             onUserInterrupted: async () =>
             {
+                // Reset states
+                conn.MarkQueue.Clear();
+                conn.LastAssistantItem = null;
+                conn.ResponseStartTimestamp = null;
+
                 var data = conn.OnModelUserInterrupted();
                 await SendEventToUser(userWebSocket, data);
             });
     }
 
+    private async Task HandleUserDtmfReceived(IRealTimeCompletion completer, RealtimeHubConnection conn)
+    {
+        var routing = _services.GetRequiredService<IRoutingService>();
+        var hookProvider = _services.GetRequiredService<ConversationHookProvider>();
+        var agentService = _services.GetRequiredService<IAgentService>();
+        var agent = await agentService.LoadAgent(conn.CurrentAgentId);
+        var dialogs = routing.Context.GetDialogs();
+        var convService = _services.GetRequiredService<IConversationService>();
+        var conversation = await convService.GetConversation(conn.ConversationId);
+
+        var message = new RoleDialogModel(AgentRole.User, conn.Data)
+        {
+            CurrentAgentId = routing.Context.GetCurrentAgentId()
+        };
+        dialogs.Add(message);
+
+        foreach (var hook in hookProvider.HooksOrderByPriority)
+        {
+            hook.SetAgent(agent)
+                .SetConversation(conversation);
+
+            await hook.OnMessageReceived(message);
+        }
+
+        await completer.InsertConversationItem(message);
+        await completer.TriggerModelInference("Reply based on the user input");
+    }
+
+    private async Task HandleUserDisconnected(RealtimeHubConnection conn)
+    {
+        // Save dialog history
+        var routing = _services.GetRequiredService<IRoutingService>();
+        var storage = _services.GetRequiredService<IConversationStorage>();
+        var dialogs = routing.Context.GetDialogs();
+        foreach (var item in dialogs)
+        {
+            storage.Append(conn.ConversationId, item);
+        }
+    }
+
     private async Task SendEventToUser(WebSocket webSocket, object message)
     {
         var data = JsonSerializer.Serialize(message);

src/Infrastructure/BotSharp.Core/Translation/TranslationResponseHook.cs

@@ -1,4 +1,5 @@
 using BotSharp.Abstraction.Agents;
+using BotSharp.Abstraction.Conversations.Enums;
 using BotSharp.Abstraction.Infrastructures.Enums;
 using BotSharp.Abstraction.Translation;
 using System;
@@ -26,6 +27,11 @@ public override async Task OnResponseGenerated(RoleDialogModel message)
             {
                 return;
             }
+
+            if (_states.GetState("channel") == ConversationChannel.Phone)
+            {
+                return;
+            }
 
             // Handle multi-language for output
             var agentService = _services.GetRequiredService<IAgentService>();

src/Infrastructure/BotSharp.Logger/Hooks/VerboseLogHook.cs

@@ -35,7 +35,7 @@ public async Task BeforeGenerating(Agent agent, List<RoleDialogModel> conversati
 
     public async Task AfterGenerated(RoleDialogModel message, TokenStatsModel tokenStats)
     {
-        if (!_convSettings.ShowVerboseLog) return;
+        if (!_convSettings.ShowVerboseLog || string.IsNullOrEmpty(tokenStats.Prompt)) return;
 
         var agentService = _services.GetRequiredService<IAgentService>();
         var agent = await agentService.LoadAgent(message.CurrentAgentId);

src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionBody.cs

@@ -47,7 +47,7 @@ public class RealtimeSessionBody
     public FunctionDef[] Tools { get; set; } = [];
 
     [JsonPropertyName("turn_detection")]
-    public RealtimeSessionTurnDetection TurnDetection { get; set; } = new();
+    public RealtimeSessionTurnDetection? TurnDetection { get; set; } = new();
 }
 
 public class RealtimeSessionTurnDetection

src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs

@@ -1,8 +1,27 @@
+using BotSharp.Abstraction.Functions.Models;
+
 namespace BotSharp.Plugin.OpenAI.Models.Realtime;
 
-public class RealtimeSessionCreationRequest : RealtimeSessionBody
+public class RealtimeSessionCreationRequest
 {
+    [JsonPropertyName("model")]
+    [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)]
+    public string Model { get; set; } = null!;
+
+    [JsonPropertyName("modalities")]
+    public string[] Modalities { get; set; } = ["audio", "text"];
+
+    [JsonPropertyName("instructions")]
+    public string Instructions { get; set; } = null!;
+
+    [JsonPropertyName("tool_choice")]
+    public string ToolChoice { get; set; } = "auto";
+
+    [JsonPropertyName("tools")]
+    public FunctionDef[] Tools { get; set; } = [];
 
+    [JsonPropertyName("turn_detection")]
+    public RealtimeSessionTurnDetection TurnDetection { get; set; } = new();
 }
 
 /// <summary>