diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Attributes/BotSharpAuthAttribute.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Attributes/BotSharpAuthAttribute.cs new file mode 100644 index 00000000..25193cb3 --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Attributes/BotSharpAuthAttribute.cs @@ -0,0 +1,32 @@ +using BotSharp.Abstraction.Users; +using Microsoft.AspNetCore.Mvc; +using Microsoft.AspNetCore.Mvc.Filters; +using Microsoft.Extensions.DependencyInjection; + +namespace BotSharp.Abstraction.Infrastructures.Attributes; + +/// +/// BotSharp authorization: check whether the request user is admin or root role. +/// +[AttributeUsage(AttributeTargets.Class | AttributeTargets.Method)] +public class BotSharpAuthAttribute : Attribute, IAsyncAuthorizationFilter +{ + public BotSharpAuthAttribute() + { + + } + + public async Task OnAuthorizationAsync(AuthorizationFilterContext context) + { + var services = context.HttpContext.RequestServices; + + var userIdentity = services.GetRequiredService(); + var userService = services.GetRequiredService(); + + var (isAdmin, user) = await userService.IsAdminUser(userIdentity.Id); + if (!isAdmin || user == null) + { + context.Result = new BadRequestResult(); + } + } +} diff --git a/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs index d6057859..733eef92 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs @@ -14,11 +14,13 @@ public interface IRealTimeCompletion Func onModelAudioDeltaReceived, Func onModelAudioResponseDone, Func onModelAudioTranscriptDone, - Func, Task> onModelResponseDone, + Func, Task> onModelResponseDone, Func onConversationItemCreated, Func onInputAudioTranscriptionDone, Func onInterruptionDetected); + Task Reconnect(RealtimeHubConnection conn); + Task AppenAudioBuffer(string message); Task AppenAudioBuffer(ArraySegment data, int length); diff --git a/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHook.cs b/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHook.cs index bd131b93..51d7383a 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHook.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHook.cs @@ -1,5 +1,6 @@ using BotSharp.Abstraction.Hooks; using BotSharp.Abstraction.MLTasks; +using BotSharp.Abstraction.Realtime.Models; namespace BotSharp.Abstraction.Realtime; @@ -8,4 +9,5 @@ public interface IRealtimeHook : IHookBase Task OnModelReady(Agent agent, IRealTimeCompletion completer); string[] OnModelTranscriptPrompt(Agent agent); Task OnTranscribeCompleted(RoleDialogModel message, TranscriptionData data); + Task ShouldReconnect(RealtimeHubConnection conn) => Task.FromResult(false); } diff --git a/src/Infrastructure/BotSharp.Abstraction/Utilities/StringExtensions.cs b/src/Infrastructure/BotSharp.Abstraction/Utilities/StringExtensions.cs index 94bda0cd..ea6bb8ec 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Utilities/StringExtensions.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Utilities/StringExtensions.cs @@ -5,7 +5,7 @@ namespace BotSharp.Abstraction.Utilities; public static class StringExtensions { - public static string IfNullOrEmptyAs(this string str, string defaultValue) + public static string IfNullOrEmptyAs(this string? str, string defaultValue) => string.IsNullOrEmpty(str) ? defaultValue : str; public static string SubstringMax(this string str, int maxLength) diff --git a/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs b/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs index 070bcf26..53318a5f 100644 --- a/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs +++ b/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs @@ -98,8 +98,6 @@ public class RealtimeHub : IRealtimeHub } await routing.InvokeFunction(message.FunctionName, message); - dialogs.Add(message); - storage.Append(_conn.ConversationId, message); } else { @@ -107,8 +105,8 @@ public class RealtimeHub : IRealtimeHub dialogs.Add(message); storage.Append(_conn.ConversationId, message); - var hooks = _services.GetHooksOrderByPriority(_conn.CurrentAgentId); - foreach (var hook in hooks) + var convHooks = _services.GetHooksOrderByPriority(_conn.CurrentAgentId); + foreach (var hook in convHooks) { hook.SetAgent(agent) .SetConversation(conversation); @@ -117,6 +115,16 @@ public class RealtimeHub : IRealtimeHub } } } + + var isReconnect = false; + var realtimeHooks = _services.GetHooks(_conn.CurrentAgentId); + foreach (var hook in realtimeHooks) + { + isReconnect = await hook.ShouldReconnect(_conn); + if (isReconnect) break; + } + + return isReconnect; }, onConversationItemCreated: async response => { diff --git a/src/Infrastructure/BotSharp.Core/Agents/Services/AgentService.RefreshAgents.cs b/src/Infrastructure/BotSharp.Core/Agents/Services/AgentService.RefreshAgents.cs index c9b3c90c..452f54c9 100644 --- a/src/Infrastructure/BotSharp.Core/Agents/Services/AgentService.RefreshAgents.cs +++ b/src/Infrastructure/BotSharp.Core/Agents/Services/AgentService.RefreshAgents.cs @@ -16,14 +16,6 @@ public partial class AgentService return refreshResult; } - var userIdentity = _services.GetRequiredService(); - var userService = _services.GetRequiredService(); - var (isValid, _) = await userService.IsAdminUser(userIdentity.Id); - if (!isValid) - { - return "Unauthorized user."; - } - var agentDir = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, dbSettings.FileRepository, _agentSettings.DataDir); diff --git a/src/Infrastructure/BotSharp.Core/Session/BotSharpRealtimeSession.cs b/src/Infrastructure/BotSharp.Core/Session/BotSharpRealtimeSession.cs index 7f5f6c15..8dc5b08d 100644 --- a/src/Infrastructure/BotSharp.Core/Session/BotSharpRealtimeSession.cs +++ b/src/Infrastructure/BotSharp.Core/Session/BotSharpRealtimeSession.cs @@ -11,6 +11,7 @@ public class BotSharpRealtimeSession : IDisposable private readonly ChatSessionOptions? _sessionOptions; private readonly object _singleReceiveLock = new(); private AsyncWebsocketDataCollectionResult _receivedCollectionResult; + private bool _disposed = false; public BotSharpRealtimeSession( IServiceProvider services, @@ -57,23 +58,30 @@ public class BotSharpRealtimeSession : IDisposable public async Task SendEventAsync(string message) { - if (_websocket.State == WebSocketState.Open) + if (_disposed || _websocket.State != WebSocketState.Open) { - var buffer = Encoding.UTF8.GetBytes(message); - await _websocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); + return; } + + var buffer = Encoding.UTF8.GetBytes(message); + await _websocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); } public async Task DisconnectAsync() { - if (_websocket.State == WebSocketState.Open) + if (_disposed || _websocket.State != WebSocketState.Open) { - await _websocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None); + return; } + + await _websocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None); } public void Dispose() { + if (_disposed) return; + + _disposed = true; _websocket.Dispose(); } } diff --git a/src/Infrastructure/BotSharp.Core/Session/LlmRealtimeSession.cs b/src/Infrastructure/BotSharp.Core/Session/LlmRealtimeSession.cs index 60ecee04..911338b5 100644 --- a/src/Infrastructure/BotSharp.Core/Session/LlmRealtimeSession.cs +++ b/src/Infrastructure/BotSharp.Core/Session/LlmRealtimeSession.cs @@ -13,6 +13,7 @@ public class LlmRealtimeSession : IDisposable private readonly object _singleReceiveLock = new(); private readonly SemaphoreSlim _clientEventSemaphore = new(initialCount: 1, maxCount: 1); private AsyncWebsocketDataCollectionResult _receivedCollectionResult; + private bool _disposed = false; public LlmRealtimeSession( IServiceProvider services, @@ -24,6 +25,7 @@ public class LlmRealtimeSession : IDisposable public async Task ConnectAsync(Uri uri, Dictionary? headers = null, CancellationToken cancellationToken = default) { + _disposed = false; _webSocket?.Dispose(); _webSocket = new ClientWebSocket(); @@ -73,31 +75,43 @@ public class LlmRealtimeSession : IDisposable public async Task SendEventToModelAsync(object message) { - if (_webSocket.State != WebSocketState.Open) - { - return; - } - - await _clientEventSemaphore.WaitAsync(); - try { + if (_disposed) + { + return; + } + + await _clientEventSemaphore.WaitAsync(); + + if (_webSocket.State != WebSocketState.Open) + { + return; + } + if (message is not string data) { data = JsonSerializer.Serialize(message, _sessionOptions?.JsonOptions); } + //Console.WriteLine($"Sending event to model {data.Substring(0, 20)}"); + var buffer = Encoding.UTF8.GetBytes(data); await _webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); } finally { - _clientEventSemaphore.Release(); + if (!_disposed) + { + _clientEventSemaphore.Release(); + } } } public async Task DisconnectAsync() { + if (_disposed) return; + if (_webSocket.State == WebSocketState.Open) { await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None); @@ -106,6 +120,9 @@ public class LlmRealtimeSession : IDisposable public void Dispose() { + if (_disposed) return; + + _disposed = true; _clientEventSemaphore?.Dispose(); _webSocket?.Dispose(); } diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/AgentController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/AgentController.cs index 401a693f..52983506 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/AgentController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/AgentController.cs @@ -1,4 +1,5 @@ using BotSharp.Abstraction.Agents.Models; +using BotSharp.Abstraction.Infrastructures.Attributes; namespace BotSharp.OpenAPI.Controllers; @@ -108,6 +109,7 @@ public class AgentController : ControllerBase return AgentViewModel.FromAgent(createdAgent); } + [BotSharpAuth] [HttpPost("/refresh-agents")] public async Task RefreshAgents() { diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/PluginController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/PluginController.cs index 4f6d3f2b..b74cc68a 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/PluginController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/PluginController.cs @@ -1,3 +1,4 @@ +using BotSharp.Abstraction.Infrastructures.Attributes; using BotSharp.Abstraction.Plugins.Models; using BotSharp.Abstraction.Users.Enums; using BotSharp.Core.Plugins; @@ -10,15 +11,10 @@ public class PluginController(IServiceProvider services, IUserIdentity user, Plu { private readonly IUserIdentity _user = user; + [BotSharpAuth] [HttpGet("/plugins")] public async Task> GetPlugins([FromQuery] PluginFilter filter) { - var isValid = await IsValidUser(); - if (!isValid) - { - return new PagedItems(); - } - var loader = services.GetRequiredService(); return loader.GetPagedPlugins(services, filter); } @@ -72,6 +68,7 @@ public class PluginController(IServiceProvider services, IUserIdentity user, Plu return menu; } + [BotSharpAuth] [HttpPost("/plugin/{id}/install")] public PluginDef InstallPlugin([FromRoute] string id) { @@ -79,17 +76,11 @@ public class PluginController(IServiceProvider services, IUserIdentity user, Plu return loader.UpdatePluginStatus(services, id, true); } + [BotSharpAuth] [HttpPost("/plugin/{id}/remove")] public PluginDef RemovePluginStats([FromRoute] string id) { var loader = services.GetRequiredService(); return loader.UpdatePluginStatus(services, id, false); } - - private async Task IsValidUser() - { - var userService = services.GetRequiredService(); - var (isAdmin, _) = await userService.IsAdminUser(_user.Id); - return isAdmin; - } } diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/RoleController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/RoleController.cs index ee3ac863..7b356f93 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/RoleController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/RoleController.cs @@ -1,3 +1,4 @@ +using BotSharp.Abstraction.Infrastructures.Attributes; using BotSharp.Abstraction.Roles; namespace BotSharp.OpenAPI.Controllers; @@ -20,15 +21,10 @@ public class RoleController : ControllerBase _user = user; } + [BotSharpAuth] [HttpPost("/role/refresh")] public async Task RefreshRoles() { - var isValid = await IsValidUser(); - if (!isValid) - { - return false; - } - return await _roleService.RefreshRoles(); } @@ -39,6 +35,7 @@ public class RoleController : ControllerBase return await _roleService.GetRoleOptions(); } + [BotSharpAuth] [HttpPost("/roles")] public async Task> GetRoles([FromBody] RoleFilter? filter = null) { @@ -47,12 +44,6 @@ public class RoleController : ControllerBase filter = RoleFilter.Empty(); } - var isValid = await IsValidUser(); - if (!isValid) - { - return Enumerable.Empty(); - } - var roles = await _roleService.GetRoles(filter); return roles.Select(x => RoleViewModel.FromRole(x)).ToList(); } @@ -64,25 +55,13 @@ public class RoleController : ControllerBase return RoleViewModel.FromRole(role); } + [BotSharpAuth] [HttpPut("/role")] public async Task UpdateRole([FromBody] RoleUpdateModel model) { if (model == null) return false; - var isValid = await IsValidUser(); - if (!isValid) - { - return false; - } - var role = RoleUpdateModel.ToRole(model); return await _roleService.UpdateRole(role, isUpdateRoleAgents: true); } - - private async Task IsValidUser() - { - var userService = _services.GetRequiredService(); - var (isAdmin, _) = await userService.IsAdminUser(_user.Id); - return isAdmin; - } } diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/UserController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/UserController.cs index 7cb1aa2b..88a10838 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/UserController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/UserController.cs @@ -1,3 +1,4 @@ +using BotSharp.Abstraction.Infrastructures.Attributes; using BotSharp.Abstraction.Users.Settings; using Microsoft.AspNetCore.Authentication; using Microsoft.AspNetCore.Authentication.Cookies; @@ -181,12 +182,6 @@ public class UserController : ControllerBase public async Task> GetUsers([FromBody] UserFilter filter) { var userService = _services.GetRequiredService(); - var isValid = await IsValidUser(); - if (!isValid) - { - return new PagedItems(); - } - var users = await userService.GetUsers(filter); var views = users.Items.Select(x => UserViewModel.FromUser(x)).ToList(); @@ -197,6 +192,7 @@ public class UserController : ControllerBase }; } + [BotSharpAuth] [HttpGet("/user/{id}/details")] public async Task GetUserDetails(string id) { @@ -205,17 +201,12 @@ public class UserController : ControllerBase return UserViewModel.FromUser(user); } + [BotSharpAuth] [HttpPut("/user")] public async Task UpdateUser([FromBody] UserUpdateModel model) { if (model == null) return false; - var isValid = await IsValidUser(); - if (!isValid) - { - return false; - } - var userService = _services.GetRequiredService(); var updated = await userService.UpdateUser(UserUpdateModel.ToUser(model), isUpdateUserAgents: true); return updated; @@ -251,13 +242,6 @@ public class UserController : ControllerBase #region Private methods - private async Task IsValidUser() - { - var userService = _services.GetRequiredService(); - var (isAdmin, _) = await userService.IsAdminUser(_user.Id); - return isAdmin; - } - private FileContentResult BuildFileResult(string file) { var fileStorage = _services.GetRequiredService(); diff --git a/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs index e2d9a161..114ea69d 100644 --- a/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs @@ -305,7 +305,7 @@ public class ChatCompletionProvider : IChatCompletion { messages.Add(new AssistantChatMessage(new List { - ChatToolCall.CreateFunctionToolCall(message.ToolCallId ?? message.FunctionName, message.FunctionName, BinaryData.FromString(message.FunctionArgs ?? "{}")) + ChatToolCall.CreateFunctionToolCall(message.ToolCallId.IfNullOrEmptyAs(message.FunctionName), message.FunctionName, BinaryData.FromString(message.FunctionArgs ?? "{}")) })); messages.Add(new ToolChatMessage(message.ToolCallId ?? message.FunctionName, message.Content)); diff --git a/src/Plugins/BotSharp.Plugin.GoogleAI/Models/Realtime/RealtimeTranscriptionResponse.cs b/src/Plugins/BotSharp.Plugin.GoogleAI/Models/Realtime/RealtimeTranscriptionResponse.cs index 189252fa..0a383c80 100644 --- a/src/Plugins/BotSharp.Plugin.GoogleAI/Models/Realtime/RealtimeTranscriptionResponse.cs +++ b/src/Plugins/BotSharp.Plugin.GoogleAI/Models/Realtime/RealtimeTranscriptionResponse.cs @@ -9,6 +9,8 @@ internal class RealtimeTranscriptionResponse : IDisposable } + private bool _disposed = false; + private MemoryStream _contentStream = new(); public Stream? ContentStream { @@ -20,6 +22,8 @@ internal class RealtimeTranscriptionResponse : IDisposable public void Collect(string text) { + if (_disposed) return; + var binary = BinaryData.FromString(text); var bytes = binary.ToArray(); @@ -30,7 +34,7 @@ internal class RealtimeTranscriptionResponse : IDisposable public string GetText() { - if (_contentStream.Length == 0) + if (_disposed || _contentStream.Length == 0) { return string.Empty; } @@ -42,12 +46,21 @@ internal class RealtimeTranscriptionResponse : IDisposable public void Clear() { - _contentStream.Position = 0; - _contentStream.SetLength(0); + try + { + if (_disposed) return; + + _contentStream.Position = 0; + _contentStream.SetLength(0); + } + catch { } } public void Dispose() { + if (_disposed) return; + + _disposed = true; _contentStream?.Dispose(); } } diff --git a/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Realtime/RealTimeCompletionProvider.cs index 5df8ee13..6f3d9d26 100644 --- a/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Realtime/RealTimeCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Realtime/RealTimeCompletionProvider.cs @@ -14,7 +14,7 @@ public class GoogleRealTimeProvider : IRealTimeCompletion public string Provider => "google-ai"; public string Model => _model; - private string _model = GoogleAIModels.Gemini2FlashExp; + private string _model = GoogleAIModels.Gemini2FlashLive001; private readonly IServiceProvider _services; private readonly ILogger _logger; @@ -35,14 +35,14 @@ public class GoogleRealTimeProvider : IRealTimeCompletion private RealtimeTranscriptionResponse _inputStream = new(); private RealtimeTranscriptionResponse _outputStream = new(); - + private bool _isBlocking = false; private RealtimeHubConnection _conn; private Func _onModelReady; private Func _onModelAudioDeltaReceived; private Func _onModelAudioResponseDone; private Func _onModelAudioTranscriptDone; - private Func, Task> _onModelResponseDone; + private Func, Task> _onModelResponseDone; private Func _onConversationItemCreated; private Func _onInputAudioTranscriptionDone; private Func _onInterruptionDetected; @@ -68,7 +68,7 @@ public class GoogleRealTimeProvider : IRealTimeCompletion Func onModelAudioDeltaReceived, Func onModelAudioResponseDone, Func onModelAudioTranscriptDone, - Func, Task> onModelResponseDone, + Func, Task> onModelResponseDone, Func onConversationItemCreated, Func onInputAudioTranscriptionDone, Func onInterruptionDetected) @@ -90,6 +90,7 @@ public class GoogleRealTimeProvider : IRealTimeCompletion var modelSettings = settingsService.GetSetting(Provider, _model); Reset(); + _isBlocking = true; _inputStream = new(); _outputStream = new(); _session = new LlmRealtimeSession(_services, new ChatSessionOptions @@ -105,6 +106,8 @@ public class GoogleRealTimeProvider : IRealTimeCompletion private async Task ReceiveMessage() { + var isReconnect = false; + await foreach (ChatSessionUpdate update in _session.ReceiveUpdatesAsync(CancellationToken.None)) { var receivedText = update?.RawResponse; @@ -124,6 +127,7 @@ public class GoogleRealTimeProvider : IRealTimeCompletion if (response.SetupComplete != null) { _logger.LogInformation($"Session setup completed."); + _isBlocking = false; } else if (response.SessionResumptionUpdate != null) { @@ -138,7 +142,7 @@ public class GoogleRealTimeProvider : IRealTimeCompletion if (functionCall != null) { var messages = OnFunctionCall(_conn, functionCall); - await _onModelResponseDone(messages); + isReconnect = await _onModelResponseDone(messages); } } else if (response.ServerContent != null) @@ -155,8 +159,6 @@ public class GoogleRealTimeProvider : IRealTimeCompletion if (response.ServerContent.ModelTurn != null) { - _logger.LogInformation($"Model audio delta received."); - // Handle input transcription var inputTranscription = _inputStream.GetText(); if (!string.IsNullOrEmpty(inputTranscription)) @@ -188,31 +190,62 @@ public class GoogleRealTimeProvider : IRealTimeCompletion // Handle output transcription var outputTranscription = _outputStream.GetText(); - if (!string.IsNullOrEmpty(outputTranscription)) - { - var messages = await OnResponseDone(_conn, outputTranscription, response.UsageMetaData); - await _onModelResponseDone(messages); - } + var messages = await OnResponseDone(_conn, outputTranscription ?? string.Empty, response.UsageMetaData); + isReconnect = await _onModelResponseDone(messages); _inputStream.Clear(); _outputStream.Clear(); } } + + if (isReconnect) + { + break; + } } catch (Exception ex) { - _logger.LogError(ex, $"Error when deserializing server response. {ex.Message}"); + _logger.LogError(ex, $"Error when handling server response. {ex.Message}"); break; } } - _inputStream.Dispose(); - _outputStream.Dispose(); - _session.Dispose(); + if (isReconnect) + { + await Reconnect(_conn); + } + else + { + _inputStream.Dispose(); + _outputStream.Dispose(); + _session.Dispose(); + } } + public async Task Reconnect(RealtimeHubConnection conn) + { + _logger.LogInformation($"Reconnecting {Provider} realtime server..."); + + _isBlocking = true; + _conn = conn; + await Disconnect(); + await Task.Delay(500); + await Connect( + _conn, + _onModelReady, + _onModelAudioDeltaReceived, + _onModelAudioResponseDone, + _onModelAudioTranscriptDone, + _onModelResponseDone, + _onConversationItemCreated, + _onInputAudioTranscriptionDone, + _onInterruptionDetected); + } + public async Task Disconnect() { + _logger.LogInformation($"Disconnecting {Provider} realtime server..."); + if (_session != null) { _inputStream?.Dispose(); @@ -224,6 +257,8 @@ public class GoogleRealTimeProvider : IRealTimeCompletion public async Task AppenAudioBuffer(string message) { + if (_isBlocking) return; + await SendEventToModel(new RealtimeClientPayload { RealtimeInput = new() @@ -235,6 +270,8 @@ public class GoogleRealTimeProvider : IRealTimeCompletion public async Task AppenAudioBuffer(ArraySegment data, int length) { + if (_isBlocking) return; + var buffer = data.AsSpan(0, length).ToArray(); await SendEventToModel(new RealtimeClientPayload { diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs index c0a9c0d4..074c9b48 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs @@ -272,7 +272,7 @@ public class ChatCompletionProvider : IChatCompletion { messages.Add(new AssistantChatMessage(new List { - ChatToolCall.CreateFunctionToolCall(message.ToolCallId ?? message.FunctionName, message.FunctionName, BinaryData.FromString(message.FunctionArgs ?? "{}")) + ChatToolCall.CreateFunctionToolCall(message.ToolCallId.IfNullOrEmptyAs(message.FunctionName), message.FunctionName, BinaryData.FromString(message.FunctionArgs ?? "{}")) })); messages.Add(new ToolChatMessage(message.ToolCallId ?? message.FunctionName, message.Content)); diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs index 52f63447..a2a4dba3 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs @@ -16,8 +16,19 @@ public class RealTimeCompletionProvider : IRealTimeCompletion private readonly ILogger _logger; private readonly BotSharpOptions _botsharpOptions; - protected string _model = "gpt-4o-mini-realtime-preview"; + private string _model = "gpt-4o-mini-realtime-preview"; private LlmRealtimeSession _session; + private bool _isBlocking = false; + + private RealtimeHubConnection _conn; + private Func _onModelReady; + private Func _onModelAudioDeltaReceived; + private Func _onModelAudioResponseDone; + private Func _onModelAudioTranscriptDone; + private Func, Task> _onModelResponseDone; + private Func _onConversationItemCreated; + private Func _onInputAudioTranscriptionDone; + private Func _onInterruptionDetected; public RealTimeCompletionProvider( IServiceProvider services, @@ -35,11 +46,21 @@ public class RealTimeCompletionProvider : IRealTimeCompletion Func onModelAudioDeltaReceived, Func onModelAudioResponseDone, Func onModelAudioTranscriptDone, - Func, Task> onModelResponseDone, + Func, Task> onModelResponseDone, Func onConversationItemCreated, Func onInputAudioTranscriptionDone, Func onInterruptionDetected) { + _conn = conn; + _onModelReady = onModelReady; + _onModelAudioDeltaReceived = onModelAudioDeltaReceived; + _onModelAudioResponseDone = onModelAudioResponseDone; + _onModelAudioTranscriptDone = onModelAudioTranscriptDone; + _onModelResponseDone = onModelResponseDone; + _onConversationItemCreated = onConversationItemCreated; + _onInputAudioTranscriptionDone = onInputAudioTranscriptionDone; + _onInterruptionDetected = onInterruptionDetected; + var settingsService = _services.GetRequiredService(); var realtimeSettings = _services.GetRequiredService(); @@ -61,31 +82,12 @@ public class RealTimeCompletionProvider : IRealTimeCompletion }, cancellationToken: CancellationToken.None); - _ = ReceiveMessage( - realtimeSettings, - conn, - onModelReady, - onModelAudioDeltaReceived, - onModelAudioResponseDone, - onModelAudioTranscriptDone, - onModelResponseDone, - onConversationItemCreated, - onInputAudioTranscriptionDone, - onInterruptionDetected); + _ = ReceiveMessage(realtimeSettings); } - private async Task ReceiveMessage( - RealtimeModelSettings realtimeSettings, - RealtimeHubConnection conn, - Func onModelReady, - Func onModelAudioDeltaReceived, - Func onModelAudioResponseDone, - Func onModelAudioTranscriptDone, - Func, Task> onModelResponseDone, - Func onConversationItemCreated, - Func onInputAudioTranscriptionDone, - Func onInterruptionDetected) + private async Task ReceiveMessage(RealtimeModelSettings realtimeSettings) { + var isReconnect = false; DateTime? startTime = null; await foreach (ChatSessionUpdate update in _session.ReceiveUpdatesAsync(CancellationToken.None)) @@ -121,7 +123,8 @@ public class RealTimeCompletionProvider : IRealTimeCompletion else if (response.Type == "session.created") { _logger.LogInformation($"{response.Type}: {receivedText}"); - await onModelReady(); + _isBlocking = false; + await _onModelReady(); } else if (response.Type == "session.updated") { @@ -135,7 +138,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion { _logger.LogInformation($"{response.Type}: {receivedText}"); var data = JsonSerializer.Deserialize(receivedText); - await onModelAudioTranscriptDone(data.Transcript); + await _onModelAudioTranscriptDone(data.Transcript); } else if (response.Type == "response.audio.delta") { @@ -143,13 +146,13 @@ public class RealTimeCompletionProvider : IRealTimeCompletion if (audio?.Delta != null) { _logger.LogDebug($"{response.Type}: {receivedText}"); - await onModelAudioDeltaReceived(audio.Delta, audio.ItemId); + await _onModelAudioDeltaReceived(audio.Delta, audio.ItemId); } } else if (response.Type == "response.audio.done") { _logger.LogInformation($"{response.Type}: {receivedText}"); - await onModelAudioResponseDone(); + await _onModelAudioResponseDone(); } else if (response.Type == "response.done") { @@ -159,14 +162,14 @@ public class RealTimeCompletionProvider : IRealTimeCompletion { if (data.StatusDetails.Type == "incomplete" && data.StatusDetails.Reason == "max_output_tokens") { - await onInterruptionDetected(); + await _onInterruptionDetected(); await TriggerModelInference("Response user concisely"); } } else { - var messages = await OnResponsedDone(conn, receivedText); - await onModelResponseDone(messages); + var messages = await OnResponsedDone(_conn, receivedText); + isReconnect = await _onModelResponseDone(messages); } } else if (response.Type == "conversation.item.created") @@ -179,23 +182,23 @@ public class RealTimeCompletionProvider : IRealTimeCompletion startTime = DateTime.UtcNow; } - await onConversationItemCreated(receivedText); + await _onConversationItemCreated(receivedText); } else if (response.Type == "conversation.item.input_audio_transcription.completed") { _logger.LogInformation($"{response.Type}: {receivedText}"); - var message = await OnUserAudioTranscriptionCompleted(conn, receivedText); + var message = await OnUserAudioTranscriptionCompleted(_conn, receivedText); if (!string.IsNullOrEmpty(message.Content)) { - await onInputAudioTranscriptionDone(message); + await _onInputAudioTranscriptionDone(message); } } else if (response.Type == "input_audio_buffer.speech_started") { _logger.LogInformation($"{response.Type}: {receivedText}"); // Handle user interuption - await onInterruptionDetected(); + await _onInterruptionDetected(); } else if (response.Type == "input_audio_buffer.speech_stopped") { @@ -205,13 +208,48 @@ public class RealTimeCompletionProvider : IRealTimeCompletion { _logger.LogInformation($"{response.Type}: {receivedText}"); } + + if (isReconnect) + { + break; + } } - _session.Dispose(); + if (isReconnect) + { + await Reconnect(_conn); + } + else + { + _session.Dispose(); + } + } + + + public async Task Reconnect(RealtimeHubConnection conn) + { + _logger.LogInformation($"Reconnecting {Provider} realtime server..."); + + _isBlocking = true; + _conn = conn; + await Disconnect(); + await Task.Delay(500); + await Connect( + _conn, + _onModelReady, + _onModelAudioDeltaReceived, + _onModelAudioResponseDone, + _onModelAudioTranscriptDone, + _onModelResponseDone, + _onConversationItemCreated, + _onInputAudioTranscriptionDone, + _onInterruptionDetected); } public async Task Disconnect() { + _logger.LogInformation($"Disconnecting {Provider} realtime server..."); + if (_session != null) { await _session.DisconnectAsync(); @@ -221,6 +259,8 @@ public class RealTimeCompletionProvider : IRealTimeCompletion public async Task AppenAudioBuffer(string message) { + if (_isBlocking) return; + var audioAppend = new { type = "input_audio_buffer.append", @@ -232,6 +272,8 @@ public class RealTimeCompletionProvider : IRealTimeCompletion public async Task AppenAudioBuffer(ArraySegment data, int length) { + if (_isBlocking) return; + var message = Convert.ToBase64String(data.AsSpan(0, length).ToArray()); await AppenAudioBuffer(message); } diff --git a/tests/BotSharp.LLM.Tests/GoogleRealTimeTests.cs b/tests/BotSharp.LLM.Tests/GoogleRealTimeTests.cs index 061f0545..8e4202a7 100644 --- a/tests/BotSharp.LLM.Tests/GoogleRealTimeTests.cs +++ b/tests/BotSharp.LLM.Tests/GoogleRealTimeTests.cs @@ -46,7 +46,7 @@ namespace BotSharp.Plugin.Google.Core async (s, s1) => { Console.WriteLine(s); }, async () => { }, async (s) => { Console.WriteLine(s); }, - async list => { Console.WriteLine(list); }, + async list => { Console.WriteLine(list); return false; }, async s => { Console.WriteLine(s); }, async model => { Console.WriteLine(model); }, async () => { Console.WriteLine("UserInterrupted"); });