diff --git a/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs index 011dbf72..7071bd3f 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs @@ -8,5 +8,22 @@ public interface IRealTimeCompletion string Model { get; } void SetModelName(string model); + + Task Connect(RealtimeHubConnection conn, + Action onModelReady, + Action onModelAudioDeltaReceived, + Action onModelAudioResponseDone, + Action onAudioTranscriptDone, + Action onModelResponseDone, + Action onUserInterrupted); + Task AppenAudioBuffer(string message); + + Task SendEventToModel(object message); + Task Disconnect(); + Task CreateSession(Agent agent, List conversations); + Task UpdateInitialSession(RealtimeHubConnection conn); + Task InertConversationItem(RoleDialogModel message); + + Task> OnResponsedDone(RealtimeHubConnection conn, string response); } diff --git a/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeModelConnector.cs b/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeModelConnector.cs deleted file mode 100644 index 9b1e3a31..00000000 --- a/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeModelConnector.cs +++ /dev/null @@ -1,13 +0,0 @@ -using BotSharp.Abstraction.Realtime.Models; - -namespace BotSharp.Abstraction.Realtime; - -public interface IRealtimeModelConnector -{ - 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 index c617970b..3e4a1f73 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs @@ -6,6 +6,7 @@ public class RealtimeHubConnection public string StreamId { get; set; } = null!; public string ConversationId { get; set; } = null!; public string Data { get; set; } = string.Empty; + public string Model { get; set; } = null!; 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/Realtime/RealtimeHub.cs b/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs index 1d82ddf9..a74959d5 100644 --- a/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs +++ b/src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs @@ -2,6 +2,8 @@ using BotSharp.Abstraction.Realtime; using System.Net.WebSockets; using System; using BotSharp.Abstraction.Realtime.Models; +using BotSharp.Abstraction.MLTasks; +using BotSharp.Abstraction.Agents.Models; namespace BotSharp.Core.Realtime; @@ -20,7 +22,13 @@ public class RealtimeHub : IRealtimeHub { var buffer = new byte[1024 * 4]; WebSocketReceiveResult result; - var modelConnector = _services.GetRequiredService(); + + var llmProviderService = _services.GetRequiredService(); + var model = llmProviderService.GetProviderModel("openai", "gpt-4", + realTime: true).Name; + + var completer = _services.GetServices().First(x => x.Provider == "openai"); + completer.SetModelName(model); do { @@ -33,46 +41,97 @@ public class RealtimeHub : IRealtimeHub } var conn = onUserMessageReceived(receivedText); - if (conn.Event == "connected") + conn.Model = model; + + if (conn.Event == "user_connected") { - await ConnectToModel(modelConnector, userWebSocket, conn); + await ConnectToModel(completer, userWebSocket, conn); } - else if (conn.Event == "data_received") + else if (conn.Event == "user_data_received") { - await modelConnector.SendMessage(conn.Data); + await completer.AppenAudioBuffer(conn.Data); } - else if (conn.Event == "disconnected") + else if (conn.Event == "user_disconnected") { - await modelConnector.Disconnect(); + await completer.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) + private async Task ConnectToModel(IRealTimeCompletion completer, 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); - }); + var hookProvider = _services.GetRequiredService(); + var storage = _services.GetRequiredService(); + var convService = _services.GetRequiredService(); + convService.SetConversationId(conn.ConversationId, []); + var conversation = await convService.GetConversation(conn.ConversationId); + var agentService = _services.GetRequiredService(); + var agent = await agentService.LoadAgent(conversation.AgentId); + var routing = _services.GetRequiredService(); + var dialogs = convService.GetDialogHistory(); + routing.Context.SetDialogs(dialogs); + + await completer.Connect(conn, + onModelReady: async () => + { + // Control initial session + var data = await completer.UpdateInitialSession(conn); + await completer.SendEventToModel(data); + }, + onModelAudioDeltaReceived: async audioDeltaData => + { + var data = conn.OnModelMessageReceived(audioDeltaData); + await SendEventToUser(userWebSocket, data); + }, + onModelAudioResponseDone: async () => + { + var data = conn.OnModelAudioResponseDone(); + await SendEventToUser(userWebSocket, data); + }, + onAudioTranscriptDone: async transcript => + { + var message = new RoleDialogModel(AgentRole.Assistant, transcript); + + // append transcript to conversation + storage.Append(conn.ConversationId, message); + + foreach (var hook in hookProvider.HooksOrderByPriority) + { + hook.SetAgent(agent) + .SetConversation(conversation); + + if (!string.IsNullOrEmpty(transcript)) + { + await hook.OnMessageReceived(message); + } + } + }, + onModelResponseDone: async response => + { + var messages = await completer.OnResponsedDone(conn, response); + foreach (var message in messages) + { + // Invoke function + if (message.FunctionName != null) + { + await routing.InvokeFunction(message.FunctionName, message); + var data = await completer.InertConversationItem(message); + await completer.SendEventToModel(data); + } + } + }, + onUserInterrupted: async () => + { + var data = conn.OnModelUserInterrupted(); + await SendEventToUser(userWebSocket, data); + }); } - private async Task SendEventToWebSocket(WebSocket webSocket, object message) + private async Task SendEventToUser(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 0a509b1a..9a9c57fb 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/BotSharp.Plugin.OpenAI.csproj +++ b/src/Plugins/BotSharp.Plugin.OpenAI/BotSharp.Plugin.OpenAI.csproj @@ -18,7 +18,6 @@ - \ No newline at end of file diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionBody.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionBody.cs index b4018c28..a5ede20f 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionBody.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionBody.cs @@ -5,23 +5,35 @@ namespace BotSharp.Plugin.OpenAI.Models.Realtime; public class RealtimeSessionBody { [JsonPropertyName("id")] + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] public string Id { get; set; } = null!; [JsonPropertyName("object")] + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] public string Object { get; set; } = null!; [JsonPropertyName("model")] + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] public string Model { get; set; } = null!; [JsonPropertyName("temperature")] - public float temperature { get; set; } = 0.8f; + public float Temperature { get; set; } = 0.8f; [JsonPropertyName("modalities")] public string[] Modalities { get; set; } = ["audio", "text"]; + [JsonPropertyName("input_audio_format")] + public string InputAudioFormat { get; set; } = "pcm16"; + + [JsonPropertyName("output_audio_format")] + public string OutputAudioFormat { get; set; } = "pcm16"; + [JsonPropertyName("instructions")] public string Instructions { get; set; } = "You are a friendly assistant."; + [JsonPropertyName("voice")] + public string Voice { get; set; } = "sage"; + [JsonPropertyName("max_response_output_tokens")] public int MaxResponseOutputTokens { get; set; } = 512; diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs index 41ad31f4..b2125e99 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs @@ -1,6 +1,14 @@ namespace BotSharp.Plugin.OpenAI.Models.Realtime; -public class RealtimeSessionRequest : RealtimeSessionBody +public class RealtimeSessionCreationRequest : RealtimeSessionBody +{ + +} + +/// +/// https://platform.openai.com/docs/api-reference/realtime-client-events/session/update +/// +public class RealtimeSessionUpdateRequest : RealtimeSessionBody { } \ No newline at end of file diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ResponseAudioTranscript.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ResponseAudioTranscript.cs new file mode 100644 index 00000000..3f83ef1b --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ResponseAudioTranscript.cs @@ -0,0 +1,19 @@ +namespace BotSharp.Plugin.OpenAI.Models.Realtime; + +public class ResponseAudioTranscript : ServerEventResponse +{ + [JsonPropertyName("response_id")] + public string ResponseId { get; set; } = null!; + + [JsonPropertyName("item_id")] + public string ItemId { get; set; } = null!; + + [JsonPropertyName("output_index")] + public int OutputIndex { get; set; } + + [JsonPropertyName("content_index")] + public int ContentIndex { get; set; } + + [JsonPropertyName("transcript")] + public string? Transcript { get; set; } +} diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ResponseDone.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ResponseDone.cs new file mode 100644 index 00000000..94cc16c3 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ResponseDone.cs @@ -0,0 +1,66 @@ +namespace BotSharp.Plugin.OpenAI.Models.Realtime; + +public class ResponseDone : ServerEventResponse +{ + [JsonPropertyName("response")] + public ResponseDoneBody Body { get; set; } = new(); +} + +public class ResponseDoneBody +{ + [JsonPropertyName("id")] + public string Id { get; set; } = null!; + + [JsonPropertyName("object")] + public string Object { get; set; } = null!; + + [JsonPropertyName("status")] + public string Status { get; set; } = null!; + + [JsonPropertyName("status_details")] + public string? StatusDetails { get; set; } = null!; + + [JsonPropertyName("conversation_id")] + public string ConversationId { get; set; } = null!; + + [JsonPropertyName("usage")] + public ModelTokenUsage Usage { get; set; } = new(); + + [JsonPropertyName("output")] + public ModelResponseDoneOutput[] Outputs { get; set; } = []; +} + +public class ModelTokenUsage +{ + [JsonPropertyName("total_tokens")] + public int TotalTokens { get; set; } + + [JsonPropertyName("input_tokens")] + public int InputTokens { get; set; } + + [JsonPropertyName("output_tokens")] + public int OutputTokens { get; set; } +} + +public class ModelResponseDoneOutput +{ + [JsonPropertyName("id")] + public string Id { get; set; } = null!; + [JsonPropertyName("object")] + public string Object { get; set; } = null!; + + [JsonPropertyName("type")] + public string Type { get; set; } = null!; + + [JsonPropertyName("status")] + public string Status { get; set; } = null!; + + [JsonPropertyName("name")] + public string Name { get; set; } = null!; + + [JsonPropertyName("call_id")] + public string CallId { get; set; } = null!; + + [JsonPropertyName("arguments")] + public string Arguments { get; set; } = null!; +} \ No newline at end of file diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ServerEventErrorResponse.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ServerEventErrorResponse.cs new file mode 100644 index 00000000..f14b5437 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ServerEventErrorResponse.cs @@ -0,0 +1,19 @@ +namespace BotSharp.Plugin.OpenAI.Models.Realtime; + +public class ServerEventErrorResponse : ServerEventResponse +{ + [JsonPropertyName("error")] + public ServerEventErrorBody Body { get; set; } = new(); +} + +public class ServerEventErrorBody +{ + [JsonPropertyName("type")] + public string Type { get; set; } = null!; + + [JsonPropertyName("code")] + public string Code { get; set; } = null!; + + [JsonPropertyName("message")] + public string? Message { get; set; } +} diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/OpenAiPlugin.cs b/src/Plugins/BotSharp.Plugin.OpenAI/OpenAiPlugin.cs index c35f4433..c1adbbe7 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/OpenAiPlugin.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/OpenAiPlugin.cs @@ -8,8 +8,6 @@ using BotSharp.Plugin.OpenAI.Providers.Audio; using Microsoft.Extensions.Configuration; using Refit; using BotSharp.Plugin.OpenAI.Providers.Realtime; -using BotSharp.Plugin.Twilio.Services.Stream; -using BotSharp.Abstraction.Realtime; namespace BotSharp.Plugin.OpenAI; @@ -37,7 +35,6 @@ public class OpenAiPlugin : IBotSharpPlugin services.AddScoped(); services.AddScoped(); services.AddScoped(); - services.AddScoped(); services.AddRefitClient() .ConfigureHttpClient(c => c.BaseAddress = new Uri("https://api.openai.com")); diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/IOpenAiRealtimeApi.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/IOpenAiRealtimeApi.cs index 810c730d..68382dff 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/IOpenAiRealtimeApi.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/IOpenAiRealtimeApi.cs @@ -7,5 +7,5 @@ namespace BotSharp.Plugin.OpenAI.Providers.Realtime; public interface IOpenAiRealtimeApi { [Post("/v1/realtime/sessions")] - Task GetSessionAsync(RealtimeSessionRequest model, [Authorize("Bearer")] string token); + Task GetSessionAsync(RealtimeSessionCreationRequest model, [Authorize("Bearer")] string token); } diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/OpenAiRealtimeModelConnector.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/OpenAiRealtimeModelConnector.cs deleted file mode 100644 index 41dbe2c5..00000000 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/OpenAiRealtimeModelConnector.cs +++ /dev/null @@ -1,186 +0,0 @@ -using BotSharp.Abstraction.Realtime; -using BotSharp.Abstraction.Realtime.Models; -using BotSharp.Core.Infrastructures; -using BotSharp.Plugin.OpenAI.Models.Realtime; -using System.Net.WebSockets; -using System.Text; -using System.Text.Json; -using System.Threading; -using Task = System.Threading.Tasks.Task; -namespace BotSharp.Plugin.Twilio.Services.Stream; - -public class OpenAiRealtimeModelConnector : IRealtimeModelConnector -{ - private readonly IServiceProvider _services; - private readonly ILogger _logger; - private ClientWebSocket _webSocket; - - public OpenAiRealtimeModelConnector(IServiceProvider services, ILogger logger) - { - _services = services; - _logger = logger; - } - - public async Task Connect(RealtimeHubConnection conn, Action onAudioDeltaReceived, Action onAudioResponseDone, Action onUserInterrupted) - { - 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: completion.Provider, model); - - _webSocket = new ClientWebSocket(); - _webSocket.Options.SetRequestHeader("Authorization", $"Bearer {settings.ApiKey}"); - _webSocket.Options.SetRequestHeader("OpenAI-Beta", "realtime=v1"); - - await _webSocket.ConnectAsync(new Uri($"wss://api.openai.com/v1/realtime?model={model}"), CancellationToken.None); - - if (_webSocket.State == WebSocketState.Open) - { - // Receive a message - ReceiveMessage(onAudioDeltaReceived, onAudioResponseDone, onUserInterrupted); - - // Control initial session with OpenAI - var sessionUpdate = new - { - type = "session.update", - session = new - { - turn_detection = new { type = "server_vad" }, - input_audio_format = "g711_ulaw", - output_audio_format = "g711_ulaw", - voice = "alloy", - instructions = agent.Description, - modalities = new string[] { "text", "audio" }, - temperature = 0.8f, - } - }; - - await SendEventToWebSocket(sessionUpdate); - - /*var initialConversationItem = new - { - type = "conversation.item.create", - item = new - { - type = "message", - role = "user", - content = new object[] - { - new { - type = "input_text", - text = "Greet the user with \"Hello there! I am an AI voice assistant powered by Twilio and the OpenAI Realtime API. You can ask me for facts, jokes, or anything you can imagine. How can I help you?\"" - } - } - } - }; - - await SendEventToWebSocket(initialConversationItem);*/ - - await SendEventToWebSocket(new { type = "response.create" }); - } - } - - public async Task Disconnect() - { - await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None); - } - - public async Task SendMessage(string message) - { - var audioAppend = new - { - type = "input_audio_buffer.append", - audio = message - }; - - await SendEventToWebSocket(audioAppend); - } - - private async Task ReceiveMessage(Action onAudioDeltaReceived, Action onAudioResponseDone, Action onUserInterrupted) - { - var buffer = new byte[1024 * 1024 * 1]; - WebSocketReceiveResult result; - string lastAssistantItem = ""; - do - { - 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); - if (string.IsNullOrEmpty(receivedText)) - { - continue; - } - _logger.LogDebug($"{nameof(OpenAiRealtimeModelConnector)} received: {receivedText}"); - var response = JsonSerializer.Deserialize(receivedText); - if (response.Type == "session.created") - { - - } - else if (response.Type == "session.updated") - { - - } - else if (response.Type == "response.audio_transcript.delta") - { - - } - else if (response.Type == "response.audio_transcript.done") - { - - } - else if (response.Type == "response.audio.delta") - { - var audio = JsonSerializer.Deserialize(receivedText); - lastAssistantItem = audio?.ItemId ?? ""; - - if (audio != null && audio.Delta != null) - { - onAudioDeltaReceived(audio.Delta); - } - } - else if (response.Type == "response.audio.done") - { - onAudioResponseDone(); - } - else if (response.Type == "response.done") - { - - } - else if (response.Type == "input_audio_buffer.speech_started") - { - // var elapsedTime = latestMediaTimestamp - responseStartTimestampTwilio; - // handle use interuption - var truncateEvent = new - { - type = "conversation.item.truncate", - item_id = lastAssistantItem, - content_index = 0, - audio_end_ms = 100 - }; - - await SendEventToWebSocket(truncateEvent); - onUserInterrupted(); - } - - } while (!result.CloseStatus.HasValue); - - await _webSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None); - } - - private async Task SendEventToWebSocket(object message) - { - var data = JsonSerializer.Serialize(message); - - var buffer = Encoding.UTF8.GetBytes(data); - await _webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); - } -} \ No newline at end of file diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs index 96c97b73..fc4e1882 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs @@ -3,10 +3,16 @@ using BotSharp.Abstraction.Functions.Models; using BotSharp.Abstraction.Realtime.Models; using BotSharp.Plugin.OpenAI.Models.Realtime; using OpenAI.Chat; +using System.Net.WebSockets; +using System.Text; using System.Text.Json; +using System.Threading; namespace BotSharp.Plugin.OpenAI.Providers.Realtime; +/// +/// Reference to https://platform.openai.com/docs/api-reference/realtime-server-events +/// public class RealTimeCompletionProvider : IRealTimeCompletion { public string Provider => "openai"; @@ -17,7 +23,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion protected readonly ILogger _logger; protected string _model = "gpt-4o-mini-realtime-preview-2024-12-17"; - + private ClientWebSocket _webSocket; public RealTimeCompletionProvider( OpenAiSettings settings, @@ -29,6 +35,150 @@ public class RealTimeCompletionProvider : IRealTimeCompletion _services = services; } + public async Task Connect(RealtimeHubConnection conn, + Action onModelReady, + Action onModelAudioDeltaReceived, + Action onModelAudioResponseDone, + Action onAudioTranscriptDone, + Action onModelResponseDone, + Action onUserInterrupted) + { + var settingsService = _services.GetRequiredService(); + var settings = settingsService.GetSetting(provider: "openai", conn.Model); + + _webSocket = new ClientWebSocket(); + _webSocket.Options.SetRequestHeader("Authorization", $"Bearer {settings.ApiKey}"); + _webSocket.Options.SetRequestHeader("OpenAI-Beta", "realtime=v1"); + + await _webSocket.ConnectAsync(new Uri($"wss://api.openai.com/v1/realtime?model={conn.Model}"), CancellationToken.None); + + if (_webSocket.State == WebSocketState.Open) + { + onModelReady(); + + // Receive a message + _ = ReceiveMessage(onModelAudioDeltaReceived, + onModelAudioResponseDone, + onAudioTranscriptDone, + onModelResponseDone, + onUserInterrupted); + + // Triggering model inference + await SendEventToModel(new { type = "response.create" }); + } + } + + public async Task Disconnect() + { + await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None); + } + + public async Task AppenAudioBuffer(string message) + { + var audioAppend = new + { + type = "input_audio_buffer.append", + audio = message + }; + + await SendEventToModel(audioAppend); + } + + private async Task ReceiveMessage(Action onModelAudioDeltaReceived, + Action onModelAudioResponseDone, + Action onAudioTranscriptDone, + Action onModelResponseDone, + Action onUserInterrupted) + { + var buffer = new byte[1024 * 1024 * 1]; + WebSocketReceiveResult result; + string lastAssistantItem = ""; + do + { + 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); + if (string.IsNullOrEmpty(receivedText)) + { + continue; + } + _logger.LogDebug($"{nameof(RealTimeCompletionProvider)} received: {receivedText}"); + var response = JsonSerializer.Deserialize(receivedText); + + if (response.Type == "error") + { + var error = JsonSerializer.Deserialize(receivedText); + _logger.LogError($"Error: {error.Body.Message}"); + } + else if (response.Type == "session.created") + { + + } + else if (response.Type == "session.updated") + { + + } + else if (response.Type == "response.audio_transcript.delta") + { + + } + else if (response.Type == "response.audio_transcript.done") + { + var data = JsonSerializer.Deserialize(receivedText); + onAudioTranscriptDone(data.Transcript); + } + else if (response.Type == "response.audio.delta") + { + var audio = JsonSerializer.Deserialize(receivedText); + lastAssistantItem = audio?.ItemId ?? ""; + + if (audio != null && audio.Delta != null) + { + onModelAudioDeltaReceived(audio.Delta); + } + } + else if (response.Type == "response.audio.done") + { + onModelAudioResponseDone(); + } + else if (response.Type == "response.done") + { + onModelResponseDone(receivedText); + } + else if (response.Type == "input_audio_buffer.speech_started") + { + // var elapsedTime = latestMediaTimestamp - responseStartTimestampTwilio; + // handle use interuption + var truncateEvent = new + { + type = "conversation.item.truncate", + item_id = lastAssistantItem, + content_index = 0, + audio_end_ms = 100 + }; + + await SendEventToModel(truncateEvent); + onUserInterrupted(); + } + + } while (!result.CloseStatus.HasValue); + + await _webSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None); + } + + public async Task SendEventToModel(object message) + { + if (message is not string data) + { + data = JsonSerializer.Serialize(message); + } + + var buffer = Encoding.UTF8.GetBytes(data); + await _webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); + } + public async Task CreateSession(Agent agent, List conversations) { var contentHooks = _services.GetServices().ToList(); @@ -37,9 +187,11 @@ public class RealTimeCompletionProvider : IRealTimeCompletion var chatClient = client.GetChatClient(_model); var (prompt, messages, options) = PrepareOptions(agent, conversations); - var args = new RealtimeSessionRequest + var instruction = messages.FirstOrDefault()?.Content.FirstOrDefault()?.Text ?? agent.Description; + + var args = new RealtimeSessionCreationRequest { - Instructions = prompt, + Instructions = instruction, ToolChoice = "auto", Tools = options.Tools.Select(x => { @@ -61,6 +213,71 @@ public class RealTimeCompletionProvider : IRealTimeCompletion return session; } + public async Task UpdateInitialSession(RealtimeHubConnection conn) + { + var convService = _services.GetRequiredService(); + var conv = await convService.GetConversation(conn.ConversationId); + + var agentService = _services.GetRequiredService(); + var agent = await agentService.LoadAgent(conv.AgentId); + + var client = ProviderHelper.GetClient(Provider, _model, _services); + var chatClient = client.GetChatClient(_model); + var (prompt, messages, options) = PrepareOptions(agent, []); + + var instruction = messages.FirstOrDefault()?.Content.FirstOrDefault()?.Text ?? agent.Description; + + var sessionUpdate = new + { + type = "session.update", + session = new RealtimeSessionUpdateRequest + { + InputAudioFormat = "g711_ulaw", + OutputAudioFormat = "g711_ulaw", + Voice = "alloy", + Instructions = instruction, + ToolChoice = "auto", + Tools = options.Tools.Select(x => + { + var fn = new FunctionDef + { + Name = x.FunctionName, + Description = x.FunctionDescription + }; + fn.Parameters = JsonSerializer.Deserialize(x.FunctionParameters); + return fn; + }).ToArray(), + Modalities = [ "text", "audio" ], + Temperature = Math.Max(options.Temperature ?? 0f, 0.6f) + } + }; + + return JsonSerializer.Serialize(sessionUpdate); + } + + public async Task InertConversationItem(RoleDialogModel message) + { + var conversationItem = new + { + type = "conversation.item.create", + item = new + { + type = "message", + role = message.Role, + content = new object[] + { + new + { + type = "text", + text = message.Content + } + } + } + }; + + return JsonSerializer.Serialize(conversationItem); + } + protected (string, IEnumerable, ChatCompletionOptions) PrepareOptions(Agent agent, List conversations) { var agentService = _services.GetRequiredService(); @@ -174,7 +391,6 @@ public class RealTimeCompletionProvider : IRealTimeCompletion return (prompt, messages, options); } - private string GetPrompt(IEnumerable messages, ChatCompletionOptions options) { var prompt = string.Empty; @@ -246,4 +462,36 @@ public class RealTimeCompletionProvider : IRealTimeCompletion { _model = model; } + + public async Task> OnResponsedDone(RealtimeHubConnection conn, string response) + { + var outputs = new List(); + + var data = JsonSerializer.Deserialize(response).Body; + foreach (var output in data.Outputs) + { + if (output.Type == "function_call") + { + outputs.Add(new RoleDialogModel(AgentRole.Assistant, output.Arguments) + { + FunctionName = output.Name, + FunctionArgs = output.Arguments + }); + } + else if (output.Type == "message") + { + outputs.Add(new RoleDialogModel(AgentRole.Assistant, "") + { + FunctionName = output.Name, + FunctionArgs = output.Arguments + }); + } + else + { + throw new NotImplementedException($"not implemented for output type {output.Type}"); + } + } + + return outputs; + } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs index 6c34bd2e..e7524e7e 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs @@ -30,6 +30,7 @@ public class TwilioStreamMiddleware var services = httpContext.RequestServices; using WebSocket webSocket = await httpContext.WebSockets.AcceptWebSocketAsync(); await HandleWebSocket(services, webSocket); + httpContext.Abort(); } } @@ -47,10 +48,9 @@ public class TwilioStreamMiddleware conn.StreamId = response.StreamSid; conn.Event = response.Event switch { - "connected" => string.Empty, - "start" => "connected", - "media" => "data_received", - "stop" => "disconnected", + "start" => "user_connected", + "media" => "user_data_received", + "stop" => "user_disconnected", _ => response.Event }; @@ -83,7 +83,7 @@ public class TwilioStreamMiddleware if (response.Event == "start") { var startResponse = JsonSerializer.Deserialize(receivedText); - conn.Data = startResponse.Body.CallSid; + conn.Data = JsonSerializer.Serialize(startResponse.Body.CustomParameters); conn.ConversationId = startResponse.Body.CallSid; } else if (response.Event == "media")