diff --git a/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs b/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs index 0c36eee3..a1e3edbe 100644 --- a/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs +++ b/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs @@ -39,12 +39,12 @@ public class RealtimeHub : IRealtimeHub var dialogs = convService.GetDialogHistory(); if (dialogs.Count == 0) { - dialogs.Add(new RoleDialogModel(AgentRole.User, "Hi")); - storage.Append(_conn.ConversationId, dialogs.First()); + //dialogs.Add(new RoleDialogModel(AgentRole.User, "Hi")); + //storage.Append(_conn.ConversationId, dialogs.First()); } routing.Context.SetDialogs(dialogs); - routing.Context.SetMessageId(_conn.ConversationId, dialogs.Last().MessageId); + routing.Context.SetMessageId(_conn.ConversationId, dialogs.LastOrDefault()?.MessageId ?? Guid.NewGuid().ToString()); var states = _services.GetRequiredService(); var settings = _services.GetRequiredService(); diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingContext.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingContext.cs index 4cba2e2d..33d90d66 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/RoutingContext.cs +++ b/src/Infrastructure/BotSharp.Core/Routing/RoutingContext.cs @@ -86,7 +86,7 @@ public class RoutingContext : IRoutingContext if (!Guid.TryParse(agentId, out _)) { var agentService = _services.GetRequiredService(); - var agents = agentService.GetAgentOptions([agentId], byName: true).Result; + var agents = agentService.GetAgentOptions([agentId]).Result; if (agents.Count > 0) { diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/ChatStreamMiddleware.cs b/src/Plugins/BotSharp.Plugin.ChatHub/ChatStreamMiddleware.cs index a9411ed3..78e6d9e5 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/ChatStreamMiddleware.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/ChatStreamMiddleware.cs @@ -58,12 +58,13 @@ public class ChatStreamMiddleware convService.SetConversationId(conversationId, []); await convService.GetConversationRecordOrCreateNew(agentId); - var buffer = new byte[1024 * 1024 * 8]; + var buffer = new byte[1024 * 32]; WebSocketReceiveResult result; do { result = await webSocket.ReceiveAsync(new(buffer), CancellationToken.None); + if (result.MessageType != WebSocketMessageType.Text) { continue; @@ -109,7 +110,10 @@ public class ChatStreamMiddleware private async Task SendEventToUser(WebSocket webSocket, string message) { var buffer = Encoding.UTF8.GetBytes(message); - await webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); + if (!webSocket.CloseStatus.HasValue) + { + await webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); + } } private (string, string) MapEvents(RealtimeHubConnection conn, string receivedText) diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ConversationItemCreated.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ConversationItemCreated.cs index b46d8cc7..921de626 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ConversationItemCreated.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ConversationItemCreated.cs @@ -10,6 +10,7 @@ public class ConversationItemBody { [JsonPropertyName("id")] public string Id { get; set; } = null!; + [JsonPropertyName("type")] public string Type { get; set; } = null!; diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs index 245eeafb..4addcac9 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs @@ -1,6 +1,10 @@ using BotSharp.Plugin.OpenAI.Models.Realtime; using BotSharp.Plugin.OpenAI.Providers.Realtime.Session; +using Newtonsoft.Json.Linq; using OpenAI.Chat; +using System.Collections.Concurrent; +using System.Threading.Channels; +using static System.Runtime.InteropServices.JavaScript.JSType; namespace BotSharp.Plugin.OpenAI.Providers.Realtime; @@ -20,6 +24,9 @@ public class RealTimeCompletionProvider : IRealTimeCompletion protected string _model = "gpt-4o-mini-realtime-preview"; private RealtimeChatSession _session; + //private Channel _messageChannel; + //private ConcurrentDictionary _messageDic; + public RealTimeCompletionProvider( OpenAiSettings settings, ILogger logger, @@ -45,6 +52,13 @@ public class RealTimeCompletionProvider : IRealTimeCompletion var realtimeModelSettings = _services.GetRequiredService(); _model = realtimeModelSettings.Model; + //_messageDic = new(); + //_messageChannel = Channel.CreateUnbounded(new UnboundedChannelOptions + //{ + // SingleReader = true, + // SingleWriter = true + //}); + _session?.Dispose(); _session = new RealtimeChatSession(_services, _options); await _session.ConnectAsync(Provider, _model, CancellationToken.None); @@ -58,10 +72,43 @@ public class RealTimeCompletionProvider : IRealTimeCompletion onConversationItemCreated, onInputAudioTranscriptionCompleted, onInterruptionDetected); + + //_ = Task.Run(async () => + //{ + // await foreach (var item in _messageChannel.Reader.ReadAllAsync()) + // { + // var start = DateTime.UtcNow; + // while (_messageDic.TryGetValue(item.ItemId, out var found) && !string.IsNullOrEmpty(found.Transcript)) + // { + // if (found.Event == "conversation.item.input_audio_transcription.completed") + // { + // var message = await OnUserAudioTranscriptionCompleted(conn, found.ReceivedText); + // if (!string.IsNullOrEmpty(message.Content)) + // { + // onInputAudioTranscriptionCompleted(message); + // } + // _messageDic.TryRemove(found.ItemId, out _); + // } + // else if (found.Event == "response.done") + // { + // var messages = await OnResponsedDone(conn, found.ReceivedText); + // onModelResponseDone(messages); + // _messageDic.TryRemove(found.ItemId, out _); + // } + + // if ((DateTime.UtcNow - start).TotalSeconds > 2) + // { + // _messageDic.TryRemove(found.ItemId, out _); + // start = DateTime.UtcNow; + // } + // } + // } + //}); } public async Task Disconnect() { + _session?.Disconnect(); } @@ -187,17 +234,51 @@ public class RealTimeCompletionProvider : IRealTimeCompletion else if (response.Type == "response.done") { _logger.LogInformation($"{response.Type}: {receivedText}"); + + //var data = JsonSerializer.Deserialize(receivedText); + //var output = data.Body.Outputs.FirstOrDefault(); + + //if (output != null && _messageDic.TryGetValue(output.Id, out var item)) + //{ + // item.Event = response.Type; + // item.ReceivedText = receivedText; + // item.Transcript = output.Content.FirstOrDefault()?.Transcript; + //} + var messages = await OnResponsedDone(conn, receivedText); onModelResponseDone(messages); } else if (response.Type == "conversation.item.created") { _logger.LogInformation($"{response.Type}: {receivedText}"); + + var data = JsonSerializer.Deserialize(receivedText); + //_messageDic.TryAdd(data.Item.Id, new AudioMessage + //{ + // ItemId = data.Item.Id, + // ReceivedText = receivedText + //}); + //await _messageChannel.Writer.WriteAsync(new AudioMessage + //{ + // ItemId = data.Item.Id, + // ReceivedText = receivedText + //}); + + await Task.Delay(500); onConversationItemCreated(receivedText); } else if (response.Type == "conversation.item.input_audio_transcription.completed") { _logger.LogInformation($"{response.Type}: {receivedText}"); + + //var data = JsonSerializer.Deserialize(receivedText); + //if (_messageDic.TryGetValue(data.ItemId, out var item)) + //{ + // item.Event = response.Type; + // item.ReceivedText = receivedText; + // item.Transcript = data.Transcript; + //} + var message = await OnUserAudioTranscriptionCompleted(conn, receivedText); if (!string.IsNullOrEmpty(message.Content)) { @@ -213,8 +294,18 @@ public class RealTimeCompletionProvider : IRealTimeCompletion else if (response.Type == "input_audio_buffer.speech_stopped") { _logger.LogInformation($"{response.Type}: {receivedText}"); + await Task.Delay(500); + } + else if (response.Type == "input_audio_buffer.committed") + { + _logger.LogInformation($"{response.Type}: {receivedText}"); + await Task.Delay(500); } } + + //_messageChannel?.Writer.TryComplete(); + //_messageChannel = null; + _session.Dispose(); } public async Task SendEventToModel(object message) @@ -658,3 +749,11 @@ public class RealTimeCompletionProvider : IRealTimeCompletion return message; } } + +class AudioMessage +{ + public string ItemId { get; set; } + public string Event { get; set; } + public string ReceivedText { get; set; } + public string? Transcript { get; set; } +} \ No newline at end of file diff --git a/src/WebStarter/Program.cs b/src/WebStarter/Program.cs index 390c7e5d..814aa7e1 100644 --- a/src/WebStarter/Program.cs +++ b/src/WebStarter/Program.cs @@ -42,9 +42,12 @@ builder.Services.AddSignalR() var app = builder.Build(); +app.UseWebSockets(); + // Enable SignalR app.MapHub("/chatHub"); app.UseMiddleware(); +app.UseMiddleware(); // Use BotSharp app.UseBotSharp() diff --git a/tests/BotSharp.Test.RealtimeVoice/Program.cs b/tests/BotSharp.Test.RealtimeVoice/Program.cs index 1e6c94f5..9a48e444 100644 --- a/tests/BotSharp.Test.RealtimeVoice/Program.cs +++ b/tests/BotSharp.Test.RealtimeVoice/Program.cs @@ -74,7 +74,7 @@ await hub.ConnectToModel(async data => }); StreamReceiveResult result; -var buffer = new byte[1024 * 8]; +var buffer = new byte[1024 * 32]; do {