From d69f19bd414b21d03ddb72979de8f46204b60f2b Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Wed, 5 Mar 2025 12:29:28 -0600 Subject: [PATCH] refactor realtime code. --- .../Realtime/Models/RealtimeHubConnection.cs | 6 + .../BotSharp.Core/Realtime/RealtimeHub.cs | 44 +++- .../Translation/TranslationResponseHook.cs | 6 + .../BotSharp.Logger/Hooks/VerboseLogHook.cs | 2 +- .../Models/Realtime/RealtimeSessionRequest.cs | 21 +- .../Realtime/RealTimeCompletionProvider.cs | 51 ++-- .../Services/Stream/TwilioStreamMiddleware.cs | 217 +++++------------- 7 files changed, 160 insertions(+), 187 deletions(-) diff --git a/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs b/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs index ec521637..61a5cfe4 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs @@ -1,9 +1,15 @@ +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 ConcurrentQueue MarkQueue { get; set; } = new(); public string CurrentAgentId { get; set; } = null!; public string ConversationId { get; set; } = null!; public string Data { get; set; } = string.Empty; diff --git a/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs b/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs index 17c5162a..6517ef3c 100644 --- a/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs +++ b/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs @@ -112,8 +112,28 @@ public class RealtimeHub : IRealtimeHub }, 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 +180,19 @@ public class RealtimeHub : IRealtimeHub 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 + storage.Append(conn.ConversationId, message); + dialogs.Add(message); - await hook.OnResponseGenerated(message); + foreach (var hook in hookProvider.HooksOrderByPriority) + { + hook.SetAgent(agent) + .SetConversation(conversation); + + await hook.OnResponseGenerated(message); + } } } }, @@ -193,6 +216,11 @@ public class RealtimeHub : IRealtimeHub }, onUserInterrupted: async () => { + // Reset states + conn.MarkQueue.Clear(); + conn.LastAssistantItem = null; + conn.ResponseStartTimestamp = null; + var data = conn.OnModelUserInterrupted(); await SendEventToUser(userWebSocket, data); }); diff --git a/src/Infrastructure/BotSharp.Core/Translation/TranslationResponseHook.cs b/src/Infrastructure/BotSharp.Core/Translation/TranslationResponseHook.cs index 3df585f2..d5e61c12 100644 --- a/src/Infrastructure/BotSharp.Core/Translation/TranslationResponseHook.cs +++ b/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 @@ namespace BotSharp.Logger.Hooks { return; } + + if (_states.GetState("channel") == ConversationChannel.Phone) + { + return; + } // Handle multi-language for output var agentService = _services.GetRequiredService(); diff --git a/src/Infrastructure/BotSharp.Logger/Hooks/VerboseLogHook.cs b/src/Infrastructure/BotSharp.Logger/Hooks/VerboseLogHook.cs index 9740bb20..7affc3a4 100644 --- a/src/Infrastructure/BotSharp.Logger/Hooks/VerboseLogHook.cs +++ b/src/Infrastructure/BotSharp.Logger/Hooks/VerboseLogHook.cs @@ -35,7 +35,7 @@ public class VerboseLogHook : IContentGeneratingHook public async Task AfterGenerated(RoleDialogModel message, TokenStatsModel tokenStats) { - if (!_convSettings.ShowVerboseLog) return; + if (!_convSettings.ShowVerboseLog || string.IsNullOrEmpty(tokenStats.Prompt)) return; var agentService = _services.GetRequiredService(); var agent = await agentService.LoadAgent(message.CurrentAgentId); diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs index b2125e99..cabf3d30 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs +++ b/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(); } /// diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs index 7abefbbb..5d7b2e80 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs @@ -128,9 +128,8 @@ public class RealTimeCompletionProvider : IRealTimeCompletion Action onInputAudioTranscriptionCompleted, Action onUserInterrupted) { - var buffer = new byte[1024 * 256]; + var buffer = new byte[1024 * 16]; WebSocketReceiveResult result; - string? lastAssistantItem = null; do { @@ -173,7 +172,16 @@ public class RealTimeCompletionProvider : IRealTimeCompletion else if (response.Type == "response.audio.delta") { var audio = JsonSerializer.Deserialize(receivedText); - lastAssistantItem = audio?.ItemId; + // Record last assistant item ID for interruption handling + if (conn.ResponseStartTimestamp.HasValue) + { + conn.ResponseStartTimestamp = conn.LatestMediaTimestamp; + } + + if (!string.IsNullOrEmpty(conn.StreamId)) + { + conn.LastAssistantItem = audio?.ItemId; + } if (audio != null && audio.Delta != null) { @@ -205,19 +213,24 @@ public class RealTimeCompletionProvider : IRealTimeCompletion } else if (response.Type == "input_audio_buffer.speech_started") { - // var elapsedTime = latestMediaTimestamp - responseStartTimestampTwilio; - // handle use interuption - if (!string.IsNullOrEmpty(lastAssistantItem)) + // Handle user interuption + if (conn.MarkQueue.Count > 0 && conn.ResponseStartTimestamp != null) { - var truncateEvent = new - { - type = "conversation.item.truncate", - item_id = lastAssistantItem, - content_index = 0, - audio_end_ms = 300 - }; + var elapsedTime = conn.LatestMediaTimestamp - conn.ResponseStartTimestamp; + + if (!string.IsNullOrEmpty(conn.LastAssistantItem)) + { + var truncateEvent = new + { + type = "conversation.item.truncate", + item_id = conn.LastAssistantItem, + content_index = 0, + audio_end_ms = elapsedTime + }; + + await SendEventToModel(truncateEvent); + } - await SendEventToModel(truncateEvent); onUserInterrupted(); } } @@ -256,6 +269,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion var args = new RealtimeSessionCreationRequest { + Model = _model, Instructions = instruction, ToolChoice = "auto", Tools = options.Tools.Select(x => @@ -271,7 +285,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion }; var settingsService = _services.GetRequiredService(); - var settings = settingsService.GetSetting(Provider, args.Model); + var settings = settingsService.GetSetting(Provider, args.Model ?? _model); var api = _services.GetRequiredService(); var session = await api.GetSessionAsync(args, settings.ApiKey); @@ -318,12 +332,13 @@ public class RealTimeCompletionProvider : IRealTimeCompletion ToolChoice = "auto", Tools = functions, Modalities = [ "text", "audio" ], - Temperature = Math.Max(options.Temperature ?? 0f, 0.6f), + Temperature = Math.Max(options.Temperature ?? 0f, 0.8f), MaxResponseOutputTokens = 512, TurnDetection = new RealtimeSessionTurnDetection { - Threshold = 0.8f, - SilenceDuration = 800 + Threshold = 0.5f, + PrefixPadding = 300, + SilenceDuration = 500 } } }; diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs index 08966396..fbb8d794 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs @@ -37,7 +37,14 @@ public class TwilioStreamMiddleware var services = httpContext.RequestServices; var conversationId = request.Path.Value.Split("/").Last(); using WebSocket webSocket = await httpContext.WebSockets.AcceptWebSocketAsync(); - await HandleWebSocket(services, conversationId, webSocket); + try + { + await HandleWebSocket(services, conversationId, webSocket); + } + catch (Exception ex) + { + _logger.LogError(ex, $"Error in WebSocket communication: {ex.Message} for conversation {conversationId}"); + } return; } } @@ -48,23 +55,15 @@ public class TwilioStreamMiddleware private async Task HandleWebSocket(IServiceProvider services, string conversationId, WebSocket webSocket) { var hub = services.GetRequiredService(); - var convService = services.GetRequiredService(); - // Session state var conn = new RealtimeHubConnection { ConversationId = conversationId }; - // Variables for timestamp and interruption handling - string streamSid = null; - long latestMediaTimestamp = 0; - string lastAssistantItem = null; - var markQueue = new ConcurrentQueue(); - long? responseStartTimestampTwilio = null; - - // Load session and state - convService.SetConversationId(conversationId, new List()); + // load conversation and state + var convService = services.GetRequiredService(); + convService.SetConversationId(conversationId, []); var hooks = services.GetServices(); foreach (var hook in hooks) { @@ -72,159 +71,59 @@ public class TwilioStreamMiddleware } convService.States.Save(); - // Set up event handlers - conn.OnModelMessageReceived = message => + await hub.Listen(webSocket, (receivedText) => { - // Record last assistant item ID for interruption handling - if (!string.IsNullOrEmpty(conn.StreamId)) + var response = JsonSerializer.Deserialize(receivedText); + conn.StreamId = response.StreamSid; + conn.Event = response.Event switch { - lastAssistantItem = conn.StreamId; - } - - // If this is the first delta of a new response, set the start timestamp - if (!responseStartTimestampTwilio.HasValue) - { - responseStartTimestampTwilio = latestMediaTimestamp; - _logger.LogDebug($"Setting start timestamp for new response: {responseStartTimestampTwilio}ms"); - } - - // Add mark to queue - markQueue.Enqueue("responsePart"); - - return new - { - @event = "media", - streamSid = conn.StreamId, - media = new { payload = message } + "start" => "user_connected", + "media" => "user_data_received", + "stop" => "user_disconnected", + _ => response.Event }; - }; - conn.OnModelAudioResponseDone = () => - { - return new + if (string.IsNullOrEmpty(conn.Event)) { - @event = "mark", - streamSid = conn.StreamId, - mark = new { name = "responsePart" } - }; - }; - - conn.OnModelUserInterrupted = () => - { - // Reset states - markQueue.Clear(); - lastAssistantItem = null; - responseStartTimestampTwilio = null; - - return new - { - @event = "clear", - streamSid = conn.StreamId - }; - }; - - try - { - await hub.Listen(webSocket, receivedText => - { - var response = JsonSerializer.Deserialize(receivedText); - if (response == null) - { - _logger.LogWarning("Failed to parse received WebSocket message"); - return conn; - } - - conn.StreamId = response.StreamSid; - - switch (response.Event) - { - case "start": - conn.Event = "user_connected"; - streamSid = response.StreamSid; - _logger.LogInformation($"Incoming stream started: {streamSid}"); - - // Reset start and media timestamps - responseStartTimestampTwilio = null; - latestMediaTimestamp = 0; - - var startResponse = JsonSerializer.Deserialize(receivedText); - if (startResponse?.Body?.CustomParameters != null) - { - conn.Data = JsonSerializer.Serialize(startResponse.Body.CustomParameters); - } - break; - - case "media": - conn.Event = "user_data_received"; - var mediaResponse = JsonSerializer.Deserialize(receivedText); - if (mediaResponse?.Body != null) - { - conn.Data = mediaResponse.Body.Payload; - - // Update latest media timestamp - if (long.TryParse(mediaResponse.Body.Timestamp, out latestMediaTimestamp)) - { - _logger.LogDebug($"Received media message with timestamp: {latestMediaTimestamp}ms"); - } - - // Check if user started speaking (interruption handling) - if (markQueue.Count > 0 && responseStartTimestampTwilio.HasValue && - !string.IsNullOrEmpty(lastAssistantItem)) - { - // Detect voice activity - more complex logic can be added here - // e.g., check audio energy levels or use VAD (Voice Activity Detection) - - // If voice activity detected, handle interruption - if (ShouldHandleInterruption(mediaResponse.Body.Payload)) - { - conn.Event = "user_interrupted"; - long elapsedTime = latestMediaTimestamp - responseStartTimestampTwilio.Value; - _logger.LogDebug($"Calculating elapsed time for truncation: {latestMediaTimestamp} - {responseStartTimestampTwilio} = {elapsedTime}ms"); - } - } - } - break; - - case "mark": - // Handle mark event - if (markQueue.TryDequeue(out _)) - { - _logger.LogDebug("Processing mark event, removing one mark from queue"); - } - break; - - case "stop": - conn.Event = "user_disconnected"; - break; - - default: - _logger.LogInformation($"Received non-media event: {response.Event}"); - break; - } - return conn; - }); - } - catch (Exception ex) - { - _logger.LogError(ex, "Error in WebSocket communication"); - } - } + } - // Simple interruption detection logic - can be extended as needed - private bool ShouldHandleInterruption(string audioPayload) - { - // Here should implement actual voice activity detection logic - // e.g., analyze audio energy levels or use VAD algorithm - - // Simple example - should be replaced with real detection logic in production - if (!string.IsNullOrEmpty(audioPayload)) - { - // Check if audio payload contains sufficient energy - // This is just a placeholder - needs actual VAD implementation - return false; // Default to false to avoid false interruptions - } - - return false; + conn.OnModelMessageReceived = message => + new + { + @event = "media", + 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 == "start") + { + var startResponse = JsonSerializer.Deserialize(receivedText); + conn.LatestMediaTimestamp = 0; + conn.ResponseStartTimestamp = null; + conn.Data = JsonSerializer.Serialize(startResponse.Body.CustomParameters); + } + else if (response.Event == "media") + { + var mediaResponse = JsonSerializer.Deserialize(receivedText); + conn.LatestMediaTimestamp = long.Parse(mediaResponse.Body.Timestamp); + conn.Data = mediaResponse.Body.Payload; + } + + return conn; + }); } }