SemianiakaVY/BotSharp
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

add session reconnect

Browse source
This commit is contained in:
Jicheng Lu 2025-05-19 14:54:40 -05:00
parent d91a55232b
commit 82b831d406
18 changed files with 251 additions and 142 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
src/Infrastructure/BotSharp.Abstraction/Infrastructures/Attributes/BotSharpAuthAttribute.cs Normal file
View file

@ -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;
/// <summary>
/// BotSharp authorization: check whether the request user is admin or root role.
/// </summary>
[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<IUserIdentity>();
var userService = services.GetRequiredService<IUserService>();
var (isAdmin, user) = await userService.IsAdminUser(userIdentity.Id);
if (!isAdmin || user == null)
{
context.Result = new BadRequestResult();
}
}
}

4
src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs
View file

@ -14,11 +14,13 @@ public interface IRealTimeCompletion
Func<string, string, Task> onModelAudioDeltaReceived,
Func<Task> onModelAudioResponseDone,
Func<string, Task> onModelAudioTranscriptDone,
Func<List<RoleDialogModel>, Task> onModelResponseDone,
Func<List<RoleDialogModel>, Task<bool>> onModelResponseDone,
Func<string, Task> onConversationItemCreated,
Func<RoleDialogModel, Task> onInputAudioTranscriptionDone,
Func<Task> onInterruptionDetected);
Task Reconnect(RealtimeHubConnection conn);
Task AppenAudioBuffer(string message);
Task AppenAudioBuffer(ArraySegment<byte> data, int length);

2
src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHook.cs
View file

@ -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<bool> ShouldReconnect(RealtimeHubConnection conn) => Task.FromResult(false);
}

2
src/Infrastructure/BotSharp.Abstraction/Utilities/StringExtensions.cs
View file

@ -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)

16
src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs
View file

@ -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<IConversationHook>(_conn.CurrentAgentId);
foreach (var hook in hooks)
var convHooks = _services.GetHooksOrderByPriority<IConversationHook>(_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<IRealtimeHook>(_conn.CurrentAgentId);
foreach (var hook in realtimeHooks)
{
isReconnect = await hook.ShouldReconnect(_conn);
if (isReconnect) break;
}
return isReconnect;
},
onConversationItemCreated: async response =>
{

8
src/Infrastructure/BotSharp.Core/Agents/Services/AgentService.RefreshAgents.cs
View file

@ -16,14 +16,6 @@ public partial class AgentService
return refreshResult;
}
var userIdentity = _services.GetRequiredService<IUserIdentity>();
var userService = _services.GetRequiredService<IUserService>();
var (isValid, _) = await userService.IsAdminUser(userIdentity.Id);
if (!isValid)
{
return "Unauthorized user.";
}
var agentDir = Path.Combine(AppDomain.CurrentDomain.BaseDirectory,
dbSettings.FileRepository,
_agentSettings.DataDir);

18
src/Infrastructure/BotSharp.Core/Session/BotSharpRealtimeSession.cs
View file

@ -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<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
return;
}
var buffer = Encoding.UTF8.GetBytes(message);
await _websocket.SendAsync(new ArraySegment<byte>(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();
}
}

33
src/Infrastructure/BotSharp.Core/Session/LlmRealtimeSession.cs
View file

@ -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<string, string>? 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<byte>(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();
}

2
src/Infrastructure/BotSharp.OpenAPI/Controllers/AgentController.cs
View file

@ -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<string> RefreshAgents()
{

17
src/Infrastructure/BotSharp.OpenAPI/Controllers/PluginController.cs
View file

@ -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<PagedItems<PluginDef>> GetPlugins([FromQuery] PluginFilter filter)
{
var isValid = await IsValidUser();
if (!isValid)
{
return new PagedItems<PluginDef>();
}
var loader = services.GetRequiredService<PluginLoader>();
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<PluginLoader>();
return loader.UpdatePluginStatus(services, id, false);
}
private async Task<bool> IsValidUser()
{
var userService = services.GetRequiredService<IUserService>();
var (isAdmin, _) = await userService.IsAdminUser(_user.Id);
return isAdmin;
}
}

29
src/Infrastructure/BotSharp.OpenAPI/Controllers/RoleController.cs
View file

@ -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<bool> 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<IEnumerable<RoleViewModel>> 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<RoleViewModel>();
}
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<bool> 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<bool> IsValidUser()
{
var userService = _services.GetRequiredService<IUserService>();
var (isAdmin, _) = await userService.IsAdminUser(_user.Id);
return isAdmin;
}
}

22
src/Infrastructure/BotSharp.OpenAPI/Controllers/UserController.cs
View file

@ -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<PagedItems<UserViewModel>> GetUsers([FromBody] UserFilter filter)
{
var userService = _services.GetRequiredService<IUserService>();
var isValid = await IsValidUser();
if (!isValid)
{
return new PagedItems<UserViewModel>();
}
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<UserViewModel> GetUserDetails(string id)
{
@ -205,17 +201,12 @@ public class UserController : ControllerBase
return UserViewModel.FromUser(user);
}
[BotSharpAuth]
[HttpPut("/user")]
public async Task<bool> UpdateUser([FromBody] UserUpdateModel model)
{
if (model == null) return false;
var isValid = await IsValidUser();
if (!isValid)
{
return false;
}
var userService = _services.GetRequiredService<IUserService>();
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<bool> IsValidUser()
{
var userService = _services.GetRequiredService<IUserService>();
var (isAdmin, _) = await userService.IsAdminUser(_user.Id);
return isAdmin;
}
private FileContentResult BuildFileResult(string file)
{
var fileStorage = _services.GetRequiredService<IFileStorageService>();

2
src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs
View file

@ -305,7 +305,7 @@ public class ChatCompletionProvider : IChatCompletion
{
messages.Add(new AssistantChatMessage(new List<ChatToolCall>
{
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));

19
src/Plugins/BotSharp.Plugin.GoogleAI/Models/Realtime/RealtimeTranscriptionResponse.cs
View file

@ -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();
}
}

69
src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Realtime/RealTimeCompletionProvider.cs
View file

@ -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<Task> _onModelReady;
private Func<string, string, Task> _onModelAudioDeltaReceived;
private Func<Task> _onModelAudioResponseDone;
private Func<string, Task> _onModelAudioTranscriptDone;
private Func<List<RoleDialogModel>, Task> _onModelResponseDone;
private Func<List<RoleDialogModel>, Task<bool>> _onModelResponseDone;
private Func<string, Task> _onConversationItemCreated;
private Func<RoleDialogModel, Task> _onInputAudioTranscriptionDone;
private Func<Task> _onInterruptionDetected;
@ -68,7 +68,7 @@ public class GoogleRealTimeProvider : IRealTimeCompletion
Func<string, string, Task> onModelAudioDeltaReceived,
Func<Task> onModelAudioResponseDone,
Func<string, Task> onModelAudioTranscriptDone,
Func<List<RoleDialogModel>, Task> onModelResponseDone,
Func<List<RoleDialogModel>, Task<bool>> onModelResponseDone,
Func<string, Task> onConversationItemCreated,
Func<RoleDialogModel, Task> onInputAudioTranscriptionDone,
Func<Task> 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<byte> data, int length)
{
if (_isBlocking) return;
var buffer = data.AsSpan(0, length).ToArray();
await SendEventToModel(new RealtimeClientPayload
{

2
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs
View file

@ -272,7 +272,7 @@ public class ChatCompletionProvider : IChatCompletion
{
messages.Add(new AssistantChatMessage(new List<ChatToolCall>
{
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));

114
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs
View file

@ -16,8 +16,19 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
private readonly ILogger<RealTimeCompletionProvider> _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<Task> _onModelReady;
private Func<string, string, Task> _onModelAudioDeltaReceived;
private Func<Task> _onModelAudioResponseDone;
private Func<string, Task> _onModelAudioTranscriptDone;
private Func<List<RoleDialogModel>, Task<bool>> _onModelResponseDone;
private Func<string, Task> _onConversationItemCreated;
private Func<RoleDialogModel, Task> _onInputAudioTranscriptionDone;
private Func<Task> _onInterruptionDetected;
public RealTimeCompletionProvider(
IServiceProvider services,
@ -35,11 +46,21 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
Func<string, string, Task> onModelAudioDeltaReceived,
Func<Task> onModelAudioResponseDone,
Func<string, Task> onModelAudioTranscriptDone,
Func<List<RoleDialogModel>, Task> onModelResponseDone,
Func<List<RoleDialogModel>, Task<bool>> onModelResponseDone,
Func<string, Task> onConversationItemCreated,
Func<RoleDialogModel, Task> onInputAudioTranscriptionDone,
Func<Task> onInterruptionDetected)
{
_conn = conn;
_onModelReady = onModelReady;
_onModelAudioDeltaReceived = onModelAudioDeltaReceived;
_onModelAudioResponseDone = onModelAudioResponseDone;
_onModelAudioTranscriptDone = onModelAudioTranscriptDone;
_onModelResponseDone = onModelResponseDone;
_onConversationItemCreated = onConversationItemCreated;
_onInputAudioTranscriptionDone = onInputAudioTranscriptionDone;
_onInterruptionDetected = onInterruptionDetected;
var settingsService = _services.GetRequiredService<ILlmProviderService>();
var realtimeSettings = _services.GetRequiredService<RealtimeModelSettings>();
@ -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<Task> onModelReady,
Func<string, string, Task> onModelAudioDeltaReceived,
Func<Task> onModelAudioResponseDone,
Func<string, Task> onModelAudioTranscriptDone,
Func<List<RoleDialogModel>, Task> onModelResponseDone,
Func<string, Task> onConversationItemCreated,
Func<RoleDialogModel, Task> onInputAudioTranscriptionDone,
Func<Task> 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<ResponseAudioTranscript>(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<byte> data, int length)
{
if (_isBlocking) return;
var message = Convert.ToBase64String(data.AsSpan(0, length).ToArray());
await AppenAudioBuffer(message);
}

2
tests/BotSharp.LLM.Tests/GoogleRealTimeTests.cs
View file

@ -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"); });