diff --git a/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs index 5eebb5b3..011dbf72 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs @@ -5,8 +5,8 @@ namespace BotSharp.Abstraction.MLTasks; public interface IRealTimeCompletion { string Provider { get; } + string Model { get; } void SetModelName(string model); - Task CreateSession(Agent agent, List conversations); } diff --git a/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHub.cs b/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHub.cs new file mode 100644 index 00000000..67d0f18c --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHub.cs @@ -0,0 +1,12 @@ +using BotSharp.Abstraction.Realtime.Models; +using System.Net.WebSockets; + +namespace BotSharp.Abstraction.Realtime; + +/// +/// Realtime hub interface. Manage the WebSocket connection include User, Agent and Model. +/// +public interface IRealtimeHub +{ + Task Listen(WebSocket userWebSocket, Func onUserMessageReceived); +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeModelConnector.cs b/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeModelConnector.cs index eaa20982..9b1e3a31 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeModelConnector.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeModelConnector.cs @@ -1,8 +1,13 @@ +using BotSharp.Abstraction.Realtime.Models; + namespace BotSharp.Abstraction.Realtime; public interface IRealtimeModelConnector { - Task Connect(Action onAudioDeltaReceived, Action onAudioResponseDone, Action onUserInterrupted); + Task Connect(RealtimeHubConnection conn, + Action onAudioDeltaReceived, + Action onAudioResponseDone, + Action onUserInterrupted); Task SendMessage(string message); Task Disconnect(); } diff --git a/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs b/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs new file mode 100644 index 00000000..c617970b --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs @@ -0,0 +1,12 @@ +namespace BotSharp.Abstraction.Realtime.Models; + +public class RealtimeHubConnection +{ + public string Event { get; set; } = null!; + public string StreamId { get; set; } = null!; + public string ConversationId { get; set; } = null!; + public string Data { get; set; } = string.Empty; + public Func OnModelMessageReceived { get; set; } = null!; + public Func OnModelAudioResponseDone { get; set; } = null!; + public Func OnModelUserInterrupted { get; set; } = null!; +} diff --git a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs index d801e943..742b39e9 100644 --- a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs +++ b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs @@ -15,6 +15,8 @@ using BotSharp.Core.Roles.Services; using BotSharp.Abstraction.Templating; using BotSharp.Core.Templating; using BotSharp.Abstraction.Infrastructures.Enums; +using BotSharp.Abstraction.Realtime; +using BotSharp.Core.Realtime; namespace BotSharp.Core; @@ -171,5 +173,7 @@ public static class BotSharpCoreExtensions }); services.AddSingleton(loader); + + services.AddScoped(); } } diff --git a/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs b/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs new file mode 100644 index 00000000..1d82ddf9 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs @@ -0,0 +1,79 @@ +using BotSharp.Abstraction.Realtime; +using System.Net.WebSockets; +using System; +using BotSharp.Abstraction.Realtime.Models; + +namespace BotSharp.Core.Realtime; + +public class RealtimeHub : IRealtimeHub +{ + private readonly IServiceProvider _services; + private readonly ILogger _logger; + public RealtimeHub(IServiceProvider services, ILogger logger) + { + _services = services; + _logger = logger; + } + + public async Task Listen(WebSocket userWebSocket, + Func onUserMessageReceived) + { + var buffer = new byte[1024 * 4]; + WebSocketReceiveResult result; + var modelConnector = _services.GetRequiredService(); + + do + { + result = await userWebSocket.ReceiveAsync(new ArraySegment(buffer), CancellationToken.None); + string receivedText = Encoding.UTF8.GetString(buffer, 0, result.Count); + _logger.LogDebug($"Received from user: {receivedText}"); + if (string.IsNullOrEmpty(receivedText)) + { + continue; + } + + var conn = onUserMessageReceived(receivedText); + if (conn.Event == "connected") + { + await ConnectToModel(modelConnector, userWebSocket, conn); + } + else if (conn.Event == "data_received") + { + await modelConnector.SendMessage(conn.Data); + } + else if (conn.Event == "disconnected") + { + await modelConnector.Disconnect(); + } + } while (!result.CloseStatus.HasValue); + + await userWebSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None); + } + + private async Task ConnectToModel(IRealtimeModelConnector modelConnector, WebSocket userWebSocket, RealtimeHubConnection conn) + { + await modelConnector.Connect(conn, onAudioDeltaReceived: async audioDeltaData => + { + var data = conn.OnModelMessageReceived(audioDeltaData); + await SendEventToWebSocket(userWebSocket, data); + }, + onAudioResponseDone: async () => + { + var data = conn.OnModelAudioResponseDone(); + await SendEventToWebSocket(userWebSocket, data); + }, + onUserInterrupted: async () => + { + var data = conn.OnModelUserInterrupted(); + await SendEventToWebSocket(userWebSocket, data); + }); + } + + private async Task SendEventToWebSocket(WebSocket webSocket, object message) + { + var data = JsonSerializer.Serialize(message); + + var buffer = Encoding.UTF8.GetBytes(data); + await webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); + } +} diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/BotSharp.Plugin.OpenAI.csproj b/src/Plugins/BotSharp.Plugin.OpenAI/BotSharp.Plugin.OpenAI.csproj index 9a9c57fb..0a509b1a 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/BotSharp.Plugin.OpenAI.csproj +++ b/src/Plugins/BotSharp.Plugin.OpenAI/BotSharp.Plugin.OpenAI.csproj @@ -18,6 +18,7 @@ + \ No newline at end of file diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/OpenAiRealtimeModelConnector.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/OpenAiRealtimeModelConnector.cs index 42d51e2c..41dbe2c5 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/OpenAiRealtimeModelConnector.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/OpenAiRealtimeModelConnector.cs @@ -1,6 +1,7 @@ using BotSharp.Abstraction.Realtime; +using BotSharp.Abstraction.Realtime.Models; +using BotSharp.Core.Infrastructures; using BotSharp.Plugin.OpenAI.Models.Realtime; -using System; using System.Net.WebSockets; using System.Text; using System.Text.Json; @@ -20,11 +21,19 @@ public class OpenAiRealtimeModelConnector : IRealtimeModelConnector _logger = logger; } - public async Task Connect(Action onAudioDeltaReceived, Action onAudioResponseDone, Action onUserInterrupted) + public async Task Connect(RealtimeHubConnection conn, Action onAudioDeltaReceived, Action onAudioResponseDone, Action onUserInterrupted) { - var model = "gpt-4o-mini-realtime-preview-2024-12-17"; + var convService = _services.GetRequiredService(); + var conv = await convService.GetConversation(conn.ConversationId); + + var agentService = _services.GetRequiredService(); + var agent = await agentService.LoadAgent(conv.AgentId); + + var completion = CompletionProvider.GetRealTimeCompletion(_services, provider: "openai", modelId: "gpt-4"); + var model = completion.Model; + var settingsService = _services.GetRequiredService(); - var settings = settingsService.GetSetting(provider: "openai", model); + var settings = settingsService.GetSetting(provider: completion.Provider, model); _webSocket = new ClientWebSocket(); _webSocket.Options.SetRequestHeader("Authorization", $"Bearer {settings.ApiKey}"); @@ -47,7 +56,7 @@ public class OpenAiRealtimeModelConnector : IRealtimeModelConnector input_audio_format = "g711_ulaw", output_audio_format = "g711_ulaw", voice = "alloy", - instructions = "You are a helpful and bubbly AI assistant who loves to chat about anything the user is interested about and is prepared to offer them facts. You have a penchant for dad jokes, owl jokes, and rickrolling – subtly. Always stay positive, but work in a joke when appropriate.", + instructions = agent.Description, modalities = new string[] { "text", "audio" }, temperature = 0.8f, } @@ -55,7 +64,7 @@ public class OpenAiRealtimeModelConnector : IRealtimeModelConnector await SendEventToWebSocket(sessionUpdate); - var initialConversationItem = new + /*var initialConversationItem = new { type = "conversation.item.create", item = new @@ -72,7 +81,7 @@ public class OpenAiRealtimeModelConnector : IRealtimeModelConnector } }; - await SendEventToWebSocket(initialConversationItem); + await SendEventToWebSocket(initialConversationItem);*/ await SendEventToWebSocket(new { type = "response.create" }); } diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs index fe52863a..96c97b73 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs @@ -10,6 +10,7 @@ namespace BotSharp.Plugin.OpenAI.Providers.Realtime; public class RealTimeCompletionProvider : IRealTimeCompletion { public string Provider => "openai"; + public string Model => _model; protected readonly OpenAiSettings _settings; protected readonly IServiceProvider _services; @@ -17,6 +18,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion protected string _model = "gpt-4o-mini-realtime-preview-2024-12-17"; + public RealTimeCompletionProvider( OpenAiSettings settings, ILogger logger, diff --git a/src/Plugins/BotSharp.Plugin.Twilio/BotSharp.Plugin.Twilio.csproj b/src/Plugins/BotSharp.Plugin.Twilio/BotSharp.Plugin.Twilio.csproj index 862a30e9..91c81e90 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/BotSharp.Plugin.Twilio.csproj +++ b/src/Plugins/BotSharp.Plugin.Twilio/BotSharp.Plugin.Twilio.csproj @@ -23,7 +23,6 @@ - diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioStreamController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioStreamController.cs index 055712a1..482489c0 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioStreamController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioStreamController.cs @@ -51,12 +51,11 @@ public class TwilioStreamController : TwilioController }); request.ConversationId = request.CallSid; + await InitConversation(request); var twilio = _services.GetRequiredService(); response = twilio.ReturnBidirectionalMediaStreamsInstructions(instruction); - - await InitConversation(request); await HookEmitter.Emit(_services, async hook => { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventMediaResponse.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventMediaResponse.cs index 86f94369..5ad291b7 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventMediaResponse.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventMediaResponse.cs @@ -4,12 +4,6 @@ namespace BotSharp.Plugin.Twilio.Models.Stream; public class StreamEventMediaResponse : StreamEventResponse { - [JsonPropertyName("sequenceNumber")] - public string SequenceNumber { get; set; } - - [JsonPropertyName("streamSid")] - public string StreamSid { get; set; } - [JsonPropertyName("media")] public StreamEventMediaBody Body { get; set; } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventResponse.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventResponse.cs index 1df41739..5be7aa60 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventResponse.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventResponse.cs @@ -9,4 +9,10 @@ public class StreamEventResponse /// [JsonPropertyName("event")] public string Event { get; set; } + + [JsonPropertyName("sequenceNumber")] + public string SequenceNumber { get; set; } + + [JsonPropertyName("streamSid")] + public string StreamSid { get; set; } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventStartResponse.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventStartResponse.cs index 6ae3e891..28b818d2 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventStartResponse.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventStartResponse.cs @@ -4,12 +4,6 @@ namespace BotSharp.Plugin.Twilio.Models.Stream; public class StreamEventStartResponse : StreamEventResponse { - [JsonPropertyName("sequenceNumber")] - public string SequenceNumber { get; set; } - - [JsonPropertyName("streamSid")] - public string StreamSid { get; set; } - [JsonPropertyName("start")] public StreamEventStartBody Body { get; set; } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamHub.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamHub.cs deleted file mode 100644 index d9945dba..00000000 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamHub.cs +++ /dev/null @@ -1,34 +0,0 @@ -using BotSharp.Plugin.Twilio.Models.Stream; -using Microsoft.AspNetCore.Http; -using Microsoft.AspNetCore.SignalR; -using Task = System.Threading.Tasks.Task; - -namespace BotSharp.Plugin.Twilio.Services.Stream; - -public class TwilioStreamHub : Hub -{ - private readonly IServiceProvider _services; - private readonly ILogger _logger; - private readonly IHttpContextAccessor _context; - - public TwilioStreamHub(IServiceProvider services, - ILogger logger, - IHttpContextAccessor context) - { - _services = services; - _logger = logger; - _context = context; - } - - public override async Task OnConnectedAsync() - { - _logger.LogInformation($"Twilio Stream Hub: {Context.ConnectionId} connected."); - - await base.OnConnectedAsync(); - } - - public async Task OnMessageReceived(StreamEventMediaResponse media) - { - return null; - } -} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs index 7ec04b6e..6c34bd2e 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs @@ -1,11 +1,8 @@ using BotSharp.Abstraction.Realtime; +using BotSharp.Abstraction.Realtime.Models; using BotSharp.Plugin.Twilio.Models.Stream; -using Microsoft.AspNetCore.Connections; using Microsoft.AspNetCore.Http; -using Microsoft.AspNetCore.SignalR; -using Microsoft.Extensions.Logging.Abstractions; using System.Net.WebSockets; -using System.Threading; using Task = System.Threading.Tasks.Task; namespace BotSharp.Plugin.Twilio.Services.Stream; @@ -41,101 +38,61 @@ public class TwilioStreamMiddleware private async Task HandleWebSocket(IServiceProvider services, WebSocket webSocket) { - var buffer = new byte[1024 * 4]; - WebSocketReceiveResult result; - var twilioHub = services.GetRequiredService(); - var modelConnector = services.GetRequiredService(); - var logger = services.GetRequiredService>(); + var hub = services.GetRequiredService(); + var conn = new RealtimeHubConnection(); - do + await hub.Listen(webSocket, (receivedText) => { - result = await webSocket.ReceiveAsync(new ArraySegment(buffer), CancellationToken.None); - - // Convert received data to text/audio (Twilio sends Base64-encoded audio) - string receivedText = Encoding.UTF8.GetString(buffer, 0, result.Count); - logger.LogDebug($"{nameof(TwilioStreamMiddleware)} received: {receivedText}"); - if (string.IsNullOrEmpty(receivedText)) - { - continue; - } var response = JsonSerializer.Deserialize(receivedText); + conn.StreamId = response.StreamSid; + conn.Event = response.Event switch + { + "connected" => string.Empty, + "start" => "connected", + "media" => "data_received", + "stop" => "disconnected", + _ => response.Event + }; + + if (string.IsNullOrEmpty(conn.Event)) + { + return conn; + } + + 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); - var hubConnectionContext = new HubConnectionContext(new DefaultConnectionContext(startResponse.StreamSid), - new HubConnectionContextOptions(), - NullLoggerFactory.Instance); - twilioHub.Context = new TwilioHubCallerContext(hubConnectionContext); - - await twilioHub.OnConnectedAsync(); - await modelConnector.Connect(onAudioDeltaReceived: async audioDeltaData => - { - var raudioDelta = new - { - @event = "media", - streamSid = startResponse.StreamSid, - media = new { payload = audioDeltaData } - }; - - await SendEventToWebSocket(webSocket, raudioDelta); - }, onAudioResponseDone: async () => - { - var mark = new - { - @event = "mark", - streamSid = startResponse.StreamSid, - mark = new { name = "responsePart" } - }; - - await SendEventToWebSocket(webSocket, mark); - }, onUserInterrupted: async () => - { - var mark = new - { - @event = "clear", - streamSid = startResponse.StreamSid - }; - - await SendEventToWebSocket(webSocket, mark); - }); + conn.Data = startResponse.Body.CallSid; + conn.ConversationId = startResponse.Body.CallSid; } else if (response.Event == "media") { var mediaResponse = JsonSerializer.Deserialize(receivedText); - var hubConnectionContext = new HubConnectionContext(new DefaultConnectionContext(mediaResponse.StreamSid), - new HubConnectionContextOptions(), - NullLoggerFactory.Instance); - twilioHub.Context = new TwilioHubCallerContext(hubConnectionContext); - - await twilioHub.OnMessageReceived(mediaResponse); - await modelConnector.SendMessage(mediaResponse.Body.Payload); - } - else if (response.Event == "mark") - { - - } - else if (response.Event == "stop") - { - var stopResponse = JsonSerializer.Deserialize(receivedText); - var hubConnectionContext = new HubConnectionContext(new DefaultConnectionContext(stopResponse.StreamSid), - new HubConnectionContextOptions(), - NullLoggerFactory.Instance); - twilioHub.Context = new TwilioHubCallerContext(hubConnectionContext); - - await twilioHub.OnDisconnectedAsync(new WebSocketException("stopped")); - await modelConnector.Disconnect(); + conn.Data = mediaResponse.Body.Payload; } - } while (!result.CloseStatus.HasValue); - - await webSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None); - } - - private async Task SendEventToWebSocket(WebSocket webSocket, object message) - { - var data = JsonSerializer.Serialize(message); - - var buffer = Encoding.UTF8.GetBytes(data); - await webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); + return conn; + }); } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioPlugin.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioPlugin.cs index 8d2d34cf..a31dbb02 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/TwilioPlugin.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/TwilioPlugin.cs @@ -1,3 +1,4 @@ +using BotSharp.Abstraction.Realtime; using BotSharp.Abstraction.Settings; using BotSharp.Plugin.Twilio.Interfaces; using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.Hooks; @@ -33,7 +34,5 @@ public class TwilioPlugin : IBotSharpPlugin services.AddHostedService(); services.AddTwilioRequestValidation(); services.AddScoped(); - - services.AddScoped(); } }