From 7cbce421195060fe1265336b7b811d3b5eeacd6a Mon Sep 17 00:00:00 2001 From: Jicheng Lu Date: Mon, 14 Apr 2025 01:25:16 -0500 Subject: [PATCH] temp save --- .../Services/WaveStreamChannel.cs | 14 +- .../Realtime/SessionConversationUpdate.cs | 11 + .../Realtime/RealTimeCompletionProvider.cs | 275 +++++++++++++----- .../Session/AiWebsocketPipelineResponse.cs | 116 ++++++++ .../AsyncWebsocketDataCollectionResult.cs | 35 +++ .../AsyncWebsocketDataResultEnumerator.cs | 49 ++++ .../Realtime/Session/RealtimeChatSession.cs | 111 +++++++ .../LocalSession.cs | 34 +++ .../MicrophoneAudioStream.cs | 119 ++++++++ tests/BotSharp.Test.RealtimeVoice/Program.cs | 42 ++- .../SpeakOutput.cs | 41 +++ 11 files changed, 755 insertions(+), 92 deletions(-) create mode 100644 src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/SessionConversationUpdate.cs create mode 100644 src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AiWebsocketPipelineResponse.cs create mode 100644 src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataCollectionResult.cs create mode 100644 src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataResultEnumerator.cs create mode 100644 src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/RealtimeChatSession.cs create mode 100644 tests/BotSharp.Test.RealtimeVoice/LocalSession.cs create mode 100644 tests/BotSharp.Test.RealtimeVoice/MicrophoneAudioStream.cs create mode 100644 tests/BotSharp.Test.RealtimeVoice/SpeakOutput.cs diff --git a/src/Infrastructure/BotSharp.Core.Realtime/Services/WaveStreamChannel.cs b/src/Infrastructure/BotSharp.Core.Realtime/Services/WaveStreamChannel.cs index 4b194efb..06ceeb42 100644 --- a/src/Infrastructure/BotSharp.Core.Realtime/Services/WaveStreamChannel.cs +++ b/src/Infrastructure/BotSharp.Core.Realtime/Services/WaveStreamChannel.cs @@ -26,7 +26,7 @@ public class WaveStreamChannel : IStreamChannel { DeviceNumber = 0, // Default recording device WaveFormat = new WaveFormat(24000, 16, 1), // 24000 Hz, 16-bit PCM, Mono - BufferMilliseconds = 100 + //BufferMilliseconds = 100 }; // Set up the DataAvailable event handler @@ -38,10 +38,14 @@ public class WaveStreamChannel : IStreamChannel // Initialize audio output for streaming var waveFormat = new WaveFormat(24000, 16, 1); // 24000 Hz, 16-bit PCM, Mono _bufferedWaveProvider = new BufferedWaveProvider(waveFormat); - _bufferedWaveProvider.BufferLength = 1024 * 1024; // Buffer length - _bufferedWaveProvider.DiscardOnBufferOverflow = true; - - _waveOut = new WaveOutEvent(); + _bufferedWaveProvider.BufferDuration = TimeSpan.FromMinutes(10); + //_bufferedWaveProvider.BufferLength = 1024 * 32; // Buffer length + _bufferedWaveProvider.DiscardOnBufferOverflow = false; + + _waveOut = new WaveOutEvent() + { + DeviceNumber = 0 + }; _waveOut.Init(_bufferedWaveProvider); _waveOut.Play(); } diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/SessionConversationUpdate.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/SessionConversationUpdate.cs new file mode 100644 index 00000000..e2b12f57 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/SessionConversationUpdate.cs @@ -0,0 +1,11 @@ +namespace BotSharp.Plugin.OpenAI.Models.Realtime; + +public class SessionConversationUpdate +{ + public string RawResponse { get; set; } + + public SessionConversationUpdate() + { + + } +} diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs index f4f3120f..0ca754eb 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs @@ -1,5 +1,7 @@ using BotSharp.Plugin.OpenAI.Models.Realtime; +using BotSharp.Plugin.OpenAI.Providers.Realtime.Session; using OpenAI.Chat; +using OpenAI.RealtimeConversation; using System.Net.WebSockets; namespace BotSharp.Plugin.OpenAI.Providers.Realtime; @@ -18,7 +20,8 @@ public class RealTimeCompletionProvider : IRealTimeCompletion private readonly BotSharpOptions _options; protected string _model = "gpt-4o-mini-realtime-preview"; - private ClientWebSocket _webSocket; + //private ClientWebSocket _webSocket; + private RealtimeChatSession _session; public RealTimeCompletionProvider( OpenAiSettings settings, @@ -45,20 +48,11 @@ public class RealTimeCompletionProvider : IRealTimeCompletion var realtimeModelSettings = _services.GetRequiredService(); _model = realtimeModelSettings.Model; - var settingsService = _services.GetRequiredService(); - var settings = settingsService.GetSetting(Provider, _model); + _session?.Dispose(); + _session = new RealtimeChatSession(_services, _options); + await _session.StartAsync(Provider, _model); - _webSocket?.Dispose(); - _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(conn, + _ = ReceiveMessage(conn, onModelReady, onModelAudioDeltaReceived, onModelAudioResponseDone, @@ -67,15 +61,41 @@ public class RealTimeCompletionProvider : IRealTimeCompletion onConversationItemCreated, onInputAudioTranscriptionCompleted, onInterruptionDetected); - } + + + //var settingsService = _services.GetRequiredService(); + //var settings = settingsService.GetSetting(Provider, _model); + + //_webSocket?.Dispose(); + //_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(conn, + // onModelReady, + // onModelAudioDeltaReceived, + // onModelAudioResponseDone, + // onModelAudioTranscriptDone, + // onModelResponseDone, + // onConversationItemCreated, + // onInputAudioTranscriptionCompleted, + // onInterruptionDetected); + //} } public async Task Disconnect() { - if (_webSocket.State == WebSocketState.Open) - { - await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None); - } + _session?.Disconnect(); + + //if (_webSocket.State == WebSocketState.Open) + //{ + // await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None); + //} } public async Task AppenAudioBuffer(string message) @@ -137,7 +157,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion private async Task ReceiveMessage(RealtimeHubConnection conn, Action onModelReady, - Action onModelAudioDeltaReceived, + Action onModelAudioDeltaReceived, Action onModelAudioResponseDone, Action onModelAudioTranscriptDone, Action> onModelResponseDone, @@ -145,33 +165,10 @@ public class RealTimeCompletionProvider : IRealTimeCompletion Action onUserAudioTranscriptionCompleted, Action onInterruptionDetected) { - var buffer = new byte[1024 * 1024 * 32]; - // Model response timeout - var settings = _services.GetRequiredService(); - var timeout = settings.ModelResponseTimeout; - WebSocketReceiveResult? result = default; - - do + await foreach (SessionConversationUpdate update in _session.ReceiveUpdatesAsync()) { - Array.Clear(buffer, 0, buffer.Length); - - var taskWorker = _webSocket.ReceiveAsync(new ArraySegment(buffer), CancellationToken.None); - var taskTimer = Task.Delay(1000 * timeout); - var completedTask = await Task.WhenAny(taskWorker, taskTimer); - - if (completedTask == taskWorker) - { - result = taskWorker.Result; - } - else - { - _logger.LogWarning($"Timeout {timeout} seconds waiting for Model response."); - await TriggerModelInference("Response user immediately"); - continue; - } - - // Convert received data to text/audio (Twilio sends Base64-encoded audio) - string receivedText = Encoding.UTF8.GetString(buffer, 0, result.Count); + var receivedText = update.RawResponse; + //Console.WriteLine($"\r\n{receivedText?.Substring(0, 30)}\r\n"); if (string.IsNullOrEmpty(receivedText)) { continue; @@ -179,8 +176,6 @@ public class RealTimeCompletionProvider : IRealTimeCompletion var response = JsonSerializer.Deserialize(receivedText); - _logger.LogDebug($"{nameof(RealTimeCompletionProvider)} received: {response.Type} {receivedText.Length}"); - if (response.Type == "error") { _logger.LogError($"{response.Type}: {receivedText}"); @@ -217,6 +212,11 @@ public class RealTimeCompletionProvider : IRealTimeCompletion _logger.LogDebug($"{response.Type}: {receivedText}"); onModelAudioDeltaReceived(audio.Delta, audio.ItemId); } + else + { + _logger.LogDebug($"{response.Type}: {receivedText}"); + onModelAudioDeltaReceived(audio.Delta, audio.ItemId); + } } else if (response.Type == "response.audio.done") { @@ -248,27 +248,148 @@ public class RealTimeCompletionProvider : IRealTimeCompletion // Handle user interuption onInterruptionDetected(); } - - } while (!result.CloseStatus.HasValue); - - await _webSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None); + } } + + //private async Task ReceiveMessage(RealtimeHubConnection conn, + // Action onModelReady, + // Action onModelAudioDeltaReceived, + // Action onModelAudioResponseDone, + // Action onModelAudioTranscriptDone, + // Action> onModelResponseDone, + // Action onConversationItemCreated, + // Action onUserAudioTranscriptionCompleted, + // Action onInterruptionDetected) + //{ + // var buffer = new byte[1024 * 1024 * 32]; + // // Model response timeout + // var settings = _services.GetRequiredService(); + // var timeout = settings.ModelResponseTimeout; + // WebSocketReceiveResult? result = default; + + // do + // { + // Array.Clear(buffer, 0, buffer.Length); + + // var taskWorker = _webSocket.ReceiveAsync(new ArraySegment(buffer), CancellationToken.None); + // var taskTimer = Task.Delay(1000 * timeout); + // var completedTask = await Task.WhenAny(taskWorker, taskTimer); + + // if (completedTask == taskWorker) + // { + // result = taskWorker.Result; + // } + // else + // { + // _logger.LogWarning($"Timeout {timeout} seconds waiting for Model response."); + // await TriggerModelInference("Response user immediately"); + // continue; + // } + + // // 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; + // } + + // var response = JsonSerializer.Deserialize(receivedText); + + // _logger.LogDebug($"{nameof(RealTimeCompletionProvider)} received: {response.Type} {receivedText.Length}"); + + // if (response.Type == "error") + // { + // _logger.LogError($"{response.Type}: {receivedText}"); + // var error = JsonSerializer.Deserialize(receivedText); + // if (error?.Body.Type == "server_error") + // { + // break; + // } + // } + // else if (response.Type == "session.created") + // { + // _logger.LogInformation($"{response.Type}: {receivedText}"); + // onModelReady(); + // } + // else if (response.Type == "session.updated") + // { + // _logger.LogInformation($"{response.Type}: {receivedText}"); + // } + // else if (response.Type == "response.audio_transcript.delta") + // { + + // } + // else if (response.Type == "response.audio_transcript.done") + // { + // _logger.LogInformation($"{response.Type}: {receivedText}"); + // var data = JsonSerializer.Deserialize(receivedText); + // onModelAudioTranscriptDone(data.Transcript); + // } + // else if (response.Type == "response.audio.delta") + // { + // var audio = JsonSerializer.Deserialize(receivedText); + // if (audio?.Delta != null) + // { + // _logger.LogDebug($"{response.Type}: {receivedText}"); + // onModelAudioDeltaReceived(audio.Delta, audio.ItemId); + // } + // } + // else if (response.Type == "response.audio.done") + // { + // _logger.LogInformation($"{response.Type}: {receivedText}"); + // onModelAudioResponseDone(); + // } + // else if (response.Type == "response.done") + // { + // _logger.LogInformation($"{response.Type}: {receivedText}"); + // var messages = await OnResponsedDone(conn, receivedText); + // onModelResponseDone(messages); + // } + // else if (response.Type == "conversation.item.created") + // { + // _logger.LogInformation($"{response.Type}: {receivedText}"); + // onConversationItemCreated(receivedText); + // } + // else if (response.Type == "conversation.item.input_audio_transcription.completed") + // { + // _logger.LogInformation($"{response.Type}: {receivedText}"); + // var message = await OnUserAudioTranscriptionCompleted(conn, receivedText); + // if (!string.IsNullOrEmpty(message.Content)) + // { + // onUserAudioTranscriptionCompleted(message); + // } + // } + // else if (response.Type == "input_audio_buffer.speech_started") + // { + // // Handle user interuption + // onInterruptionDetected(); + // } + + // } while (!result.CloseStatus.HasValue); + + // await _webSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None); + //} + public async Task SendEventToModel(object message) { - if (_webSocket.State != WebSocketState.Open) - { - return; - } + if (_session == null) return; - if (message is not string data) - { - data = JsonSerializer.Serialize(message, _options.JsonSerializerOptions); - } + await _session.SendEventToModel(message); - var buffer = Encoding.UTF8.GetBytes(data); + //if (_webSocket.State != WebSocketState.Open) + //{ + // return; + //} + + //if (message is not string data) + //{ + // data = JsonSerializer.Serialize(message, _options.JsonSerializerOptions); + //} + + //var buffer = Encoding.UTF8.GetBytes(data); - await _webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); + //await _webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); } public async Task UpdateSession(RealtimeHubConnection conn) @@ -307,10 +428,10 @@ public class RealTimeCompletionProvider : IRealTimeCompletion Tools = functions, Modalities = [ "text", "audio" ], Temperature = Math.Max(options.Temperature ?? realtimeModelSettings.Temperature, 0.6f), - MaxResponseOutputTokens = realtimeModelSettings.MaxResponseOutputTokens, + MaxResponseOutputTokens = 4096, TurnDetection = new RealtimeSessionTurnDetection { - InterruptResponse = realtimeModelSettings.InterruptResponse/*, + InterruptResponse = false/*, Threshold = realtimeModelSettings.TurnDetection.Threshold, PrefixPadding = realtimeModelSettings.TurnDetection.PrefixPadding, SilenceDuration = realtimeModelSettings.TurnDetection.SilenceDuration*/ @@ -318,22 +439,26 @@ public class RealTimeCompletionProvider : IRealTimeCompletion InputAudioNoiseReduction = new InputAudioNoiseReduction { Type = "near_field" + }, + InputAudioTranscription = new() + { + Model = "whisper-1" } } }; - if (realtimeModelSettings.InputAudioTranscribe) - { - var words = new List(); - HookEmitter.Emit(_services, hook => words.AddRange(hook.OnModelTranscriptPrompt(agent))); + //if (realtimeModelSettings.InputAudioTranscribe) + //{ + // var words = new List(); + // HookEmitter.Emit(_services, hook => words.AddRange(hook.OnModelTranscriptPrompt(agent))); - sessionUpdate.session.InputAudioTranscription = new InputAudioTranscription - { - Model = realtimeModelSettings.InputAudioTranscription.Model, - Language = realtimeModelSettings.InputAudioTranscription.Language, - Prompt = string.Join(", ", words.Select(x => x.ToLower().Trim()).Distinct()).SubstringMax(1024) - }; - } + // sessionUpdate.session.InputAudioTranscription = new InputAudioTranscription + // { + // Model = realtimeModelSettings.InputAudioTranscription.Model, + // Language = realtimeModelSettings.InputAudioTranscription.Language, + // Prompt = string.Join(", ", words.Select(x => x.ToLower().Trim()).Distinct()).SubstringMax(1024) + // }; + //} await HookEmitter.Emit(_services, async hook => { diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AiWebsocketPipelineResponse.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AiWebsocketPipelineResponse.cs new file mode 100644 index 00000000..4f179d5d --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AiWebsocketPipelineResponse.cs @@ -0,0 +1,116 @@ +using System.ClientModel.Primitives; +using System.Net; +using System.Net.WebSockets; + +namespace BotSharp.Plugin.OpenAI.Providers.Realtime.Session; + +public class AiWebsocketPipelineResponse : PipelineResponse +{ + + public AiWebsocketPipelineResponse() + { + + } + + private int _status; + public override int Status => _status; + + private string _reasonPhrase; + public override string ReasonPhrase => _reasonPhrase; + + private MemoryStream _contentStream = new(); + public override Stream? ContentStream + { + get + { + return _contentStream != null ? _contentStream : new MemoryStream(); + } + set => throw new NotImplementedException(); + } + + private BinaryData _content; + public override BinaryData Content + { + get + { + if (_content == null) + { + _content = new(_contentStream.ToArray()); + } + return _content; + } + } + + protected override PipelineResponseHeaders HeadersCore => throw new NotImplementedException(); + + public bool IsComplete { get; private set; } = false; + + + public void HandleReceivedResult(WebSocketReceiveResult receivedResult, BinaryData receivedBytes) + { + if (ContentStream.Length == 0) + { + _status = ConvertWebsocketCloseStatusToHttpStatus(receivedResult.CloseStatus ?? WebSocketCloseStatus.Empty); + _reasonPhrase = receivedResult.CloseStatusDescription?? (receivedResult.CloseStatus ?? WebSocketCloseStatus.Empty).ToString(); + } + else if (receivedResult.MessageType != WebSocketMessageType.Text) + { + throw new NotImplementedException($"{nameof(AiWebsocketPipelineResponse)} currently supports only text messages."); + } + + var rawBytes = receivedBytes.ToArray(); + _contentStream.Position = _contentStream.Length; + _contentStream.Write(rawBytes, 0, rawBytes.Length); + _contentStream.Position = 0; + IsComplete = receivedResult.EndOfMessage; + } + + public override BinaryData BufferContent(CancellationToken cancellationToken = default) + { + return Content; + } + + public override ValueTask BufferContentAsync(CancellationToken cancellationToken = default) + { + return new ValueTask(Task.FromResult(Content)); + } + + public override void Dispose() + { + ContentStream?.Dispose(); + } + + private static int ConvertWebsocketCloseStatusToHttpStatus(WebSocketCloseStatus status) + { + int res; + + switch (status) + { + case WebSocketCloseStatus.Empty: + case WebSocketCloseStatus.NormalClosure: + res = (int)HttpStatusCode.OK; + break; + case WebSocketCloseStatus.EndpointUnavailable: + case WebSocketCloseStatus.ProtocolError: + case WebSocketCloseStatus.InvalidMessageType: + case WebSocketCloseStatus.InvalidPayloadData: + case WebSocketCloseStatus.PolicyViolation: + res = (int)HttpStatusCode.BadRequest; + break; + case WebSocketCloseStatus.MessageTooBig: + res = (int)HttpStatusCode.RequestEntityTooLarge; + break; + case WebSocketCloseStatus.MandatoryExtension: + res = 418; + break; + case WebSocketCloseStatus.InternalServerError: + res = (int)HttpStatusCode.InternalServerError; + break; + default: + res = (int)HttpStatusCode.InternalServerError; + break; + } + + return res; + } +} diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataCollectionResult.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataCollectionResult.cs new file mode 100644 index 00000000..c1196450 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataCollectionResult.cs @@ -0,0 +1,35 @@ +using BotSharp.Plugin.OpenAI.Models.Realtime; +using System.ClientModel; +using System.Net.WebSockets; + +namespace BotSharp.Plugin.OpenAI.Providers.Realtime.Session; + +public class AsyncWebsocketDataCollectionResult : AsyncCollectionResult +{ + private readonly WebSocket _webSocket; + + public AsyncWebsocketDataCollectionResult(WebSocket webSocket) + { + _webSocket = webSocket; + } + + public override ContinuationToken? GetContinuationToken(ClientResult page) + { + return null; + } + + public override async IAsyncEnumerable GetRawPagesAsync() + { + await using var enumerator = new AsyncWebsocketDataResultEnumerator(_webSocket); + while (await enumerator.MoveNextAsync().ConfigureAwait(false)) + { + yield return enumerator.Current; + } + } + + protected override async IAsyncEnumerable GetValuesFromPageAsync(ClientResult page) + { + await Task.CompletedTask; + yield return page; + } +} diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataResultEnumerator.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataResultEnumerator.cs new file mode 100644 index 00000000..aa30cb5f --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataResultEnumerator.cs @@ -0,0 +1,49 @@ +using System; +using System.Buffers; +using System.ClientModel; +using System.Net.WebSockets; + +namespace BotSharp.Plugin.OpenAI.Providers.Realtime.Session; + +public class AsyncWebsocketDataResultEnumerator : IAsyncEnumerator +{ + private readonly WebSocket _webSocket; + private readonly byte[] _buffer; + + public AsyncWebsocketDataResultEnumerator( + WebSocket webSocket) + { + _webSocket = webSocket; + _buffer = ArrayPool.Shared.Rent(1024 * 32); + } + + public ClientResult Current { get; private set; } + + public ValueTask DisposeAsync() + { + _webSocket?.Dispose(); + return new ValueTask(Task.CompletedTask); + } + + public async ValueTask MoveNextAsync() + { + var response = new AiWebsocketPipelineResponse(); + while (!response.IsComplete) + { + var receivedResult = await _webSocket.ReceiveAsync(new(_buffer), CancellationToken.None); + + if (receivedResult.CloseStatus.HasValue) + { + Current = null; + return false; + } + + var receivedBytes = _buffer.AsMemory(0, receivedResult.Count); + var receivedData = BinaryData.FromBytes(receivedBytes); + response.HandleReceivedResult(receivedResult, receivedData); + } + + Current = ClientResult.FromResponse(response); + return true; + } +} diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/RealtimeChatSession.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/RealtimeChatSession.cs new file mode 100644 index 00000000..33c933d9 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/RealtimeChatSession.cs @@ -0,0 +1,111 @@ +using BotSharp.Plugin.OpenAI.Models.Realtime; +using System; +using System.ClientModel; +using System.Net.WebSockets; +using System.Text; +using System.Threading; + +namespace BotSharp.Plugin.OpenAI.Providers.Realtime.Session; + +public class RealtimeChatSession : IDisposable +{ + private readonly IServiceProvider _services; + private readonly BotSharpOptions _options; + + private ClientWebSocket _webSocket; + private readonly object _singleReceiveLock = new(); + private readonly SemaphoreSlim _clientSendSemaphore = new(initialCount: 1, maxCount: 1); + private AsyncWebsocketDataCollectionResult _receivedCollectionResult; + + public RealtimeChatSession( + IServiceProvider services, + BotSharpOptions options) + { + _services = services; + _options = options; + } + + public async Task StartAsync(string provider, string model) + { + var settingsService = _services.GetRequiredService(); + var settings = settingsService.GetSetting(provider, model); + + _webSocket?.Dispose(); + _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); + } + + public async IAsyncEnumerable ReceiveUpdatesAsync() + { + await foreach (ClientResult result in ReceiveInnerUpdatesAsync()) + { + var update = HandleSessionResult(result); + yield return update; + } + } + + public async IAsyncEnumerable ReceiveInnerUpdatesAsync() + { + lock (_singleReceiveLock) + { + _receivedCollectionResult ??= new(_webSocket); + } + + await foreach (var result in _receivedCollectionResult) + { + yield return result; + } + } + + private SessionConversationUpdate HandleSessionResult(ClientResult result) + { + using var response = result.GetRawResponse(); + var bytes = response.Content.ToArray(); + var text = Encoding.UTF8.GetString(bytes, 0, bytes.Length); + return new SessionConversationUpdate + { + RawResponse = text + }; + } + + public async Task SendEventToModel(object message) + { + if (_webSocket.State != WebSocketState.Open) + { + return; + } + + await _clientSendSemaphore.WaitAsync().ConfigureAwait(false); + + try + { + if (message is not string data) + { + data = JsonSerializer.Serialize(message, _options.JsonSerializerOptions); + } + + var buffer = Encoding.UTF8.GetBytes(data); + await _webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); + } + finally + { + _clientSendSemaphore.Release(); + } + } + + public async Task Disconnect() + { + if (_webSocket.State == WebSocketState.Open) + { + await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None); + } + } + + public void Dispose() + { + _webSocket?.Dispose(); + } +} diff --git a/tests/BotSharp.Test.RealtimeVoice/LocalSession.cs b/tests/BotSharp.Test.RealtimeVoice/LocalSession.cs new file mode 100644 index 00000000..3977ba4d --- /dev/null +++ b/tests/BotSharp.Test.RealtimeVoice/LocalSession.cs @@ -0,0 +1,34 @@ +using BotSharp.Abstraction.MLTasks; +using System.Buffers; +using System.ClientModel.Primitives; +using System.Threading; + +namespace BotSharp.Test.RealtimeVoice; + +public class LocalSession +{ + private readonly IRealTimeCompletion _completion; + + public LocalSession( + IRealTimeCompletion completion) + { + _completion = completion; + } + + public async Task SendInputAudioAsync(Stream audio) + { + byte[] buffer = ArrayPool.Shared.Rent(1024 * 16); + while (true) + { + int bytesRead = await audio.ReadAsync(buffer, 0, buffer.Length).ConfigureAwait(false); + if (bytesRead == 0) + { + break; + } + + ReadOnlyMemory audioMemory = buffer.AsMemory(0, bytesRead); + BinaryData audioData = BinaryData.FromBytes(audioMemory);; + await _completion.AppenAudioBuffer(audioData.ToArray(), audioData.Length); + } + } +} diff --git a/tests/BotSharp.Test.RealtimeVoice/MicrophoneAudioStream.cs b/tests/BotSharp.Test.RealtimeVoice/MicrophoneAudioStream.cs new file mode 100644 index 00000000..e9a6dbe2 --- /dev/null +++ b/tests/BotSharp.Test.RealtimeVoice/MicrophoneAudioStream.cs @@ -0,0 +1,119 @@ +using NAudio.Wave; + +namespace BotSharp.Test.RealtimeVoice; + +public class MicrophoneAudioStream : Stream, IDisposable +{ + private const int SAMPLES_PER_SECOND = 24000; + private const int BYTES_PER_SAMPLE = 2; + private const int CHANNELS = 1; + + // For simplicity, this is configured to use a static 10-second ring buffer. + private readonly byte[] _buffer = new byte[BYTES_PER_SAMPLE * SAMPLES_PER_SECOND * CHANNELS * 10]; + private readonly object _bufferLock = new(); + private int _bufferReadPos = 0; + private int _bufferWritePos = 0; + + private readonly WaveInEvent _waveInEvent; + + private MicrophoneAudioStream() + { + _waveInEvent = new() + { + WaveFormat = new WaveFormat(SAMPLES_PER_SECOND, BYTES_PER_SAMPLE * 8, CHANNELS), + DeviceNumber = 0 + }; + _waveInEvent.DataAvailable += (_, e) => + { + lock (_bufferLock) + { + int bytesToCopy = e.BytesRecorded; + if (_bufferWritePos + bytesToCopy >= _buffer.Length) + { + int bytesToCopyBeforeWrap = _buffer.Length - _bufferWritePos; + Array.Copy(e.Buffer, 0, _buffer, _bufferWritePos, bytesToCopyBeforeWrap); + bytesToCopy -= bytesToCopyBeforeWrap; + _bufferWritePos = 0; + } + Array.Copy(e.Buffer, e.BytesRecorded - bytesToCopy, _buffer, _bufferWritePos, bytesToCopy); + _bufferWritePos += bytesToCopy; + } + }; + _waveInEvent.StartRecording(); + } + + public static MicrophoneAudioStream Start() => new(); + + public override bool CanRead => true; + + public override bool CanSeek => false; + + public override bool CanWrite => false; + + public override long Length => throw new NotImplementedException(); + + public override long Position { get => throw new NotImplementedException(); set => throw new NotImplementedException(); } + + public override void Flush() + { + throw new NotImplementedException(); + } + + public override int Read(byte[] buffer, int offset, int count) + { + int totalCount = count; + + int GetBytesAvailable() => _bufferWritePos < _bufferReadPos + ? _bufferWritePos + (_buffer.Length - _bufferReadPos) + : _bufferWritePos - _bufferReadPos; + + // For simplicity, we'll block until all requested data is available and not perform partial reads. + while (GetBytesAvailable() < count) + { + Thread.Sleep(100); + } + + lock (_bufferLock) + { + if (_bufferReadPos + count >= _buffer.Length) + { + int bytesBeforeWrap = _buffer.Length - _bufferReadPos; + Array.Copy( + sourceArray: _buffer, + sourceIndex: _bufferReadPos, + destinationArray: buffer, + destinationIndex: offset, + length: bytesBeforeWrap); + _bufferReadPos = 0; + count -= bytesBeforeWrap; + offset += bytesBeforeWrap; + } + + Array.Copy(_buffer, _bufferReadPos, buffer, offset, count); + _bufferReadPos += count; + } + + return totalCount; + } + + public override long Seek(long offset, SeekOrigin origin) + { + throw new NotImplementedException(); + } + + public override void SetLength(long value) + { + throw new NotImplementedException(); + } + + public override void Write(byte[] buffer, int offset, int count) + { + throw new NotImplementedException(); + } + + protected override void Dispose(bool disposing) + { + _waveInEvent?.Dispose(); + base.Dispose(disposing); + } +} \ No newline at end of file diff --git a/tests/BotSharp.Test.RealtimeVoice/Program.cs b/tests/BotSharp.Test.RealtimeVoice/Program.cs index bb628512..da8b0965 100644 --- a/tests/BotSharp.Test.RealtimeVoice/Program.cs +++ b/tests/BotSharp.Test.RealtimeVoice/Program.cs @@ -4,6 +4,8 @@ using BotSharp.Abstraction.Conversations; using BotSharp.OpenAPI; using System.Text.Json; using System.Reflection; +using BotSharp.Test.RealtimeVoice; +using BotSharp.Abstraction.MLTasks; var services = ServiceBuilder.CreateHostBuilder(Assembly.GetExecutingAssembly()); var channel = services.GetRequiredService(); @@ -21,7 +23,7 @@ var conv = new Conversation }; conv = await convService.NewConversation(conv); -await channel.ConnectAsync(conv.Id); +//await channel.ConnectAsync(conv.Id); var hub = services.GetRequiredService(); var conn = hub.SetHubConnection(conv.Id); @@ -52,34 +54,50 @@ conn.OnModelUserInterrupted = () => @event = "clear" }); +var completer = services.GetServices().First(x => x.Provider == "openai"); +LocalSession session = new(completer); +SpeakerOutput speakerOutput = new(); + await hub.ConnectToModel(async data => { var response = JsonSerializer.Deserialize(data); if (response.Event == "clear") { - channel.ClearBuffer(); + //channel.ClearBuffer(); + Console.WriteLine("Before clearing audio buffer..."); + speakerOutput.ClearPlayback(); } else if (response.Event == "media") { var message = JsonSerializer.Deserialize(data); - await channel.SendAsync(Convert.FromBase64String(message.Media), CancellationToken.None); + //await channel.SendAsync(Convert.FromBase64String(message.Media), CancellationToken.None); + speakerOutput.EnqueueForPlayback(Convert.FromBase64String(message.Media)); } +}, init: async data => +{ + _ = Task.Run(async () => + { + using MicrophoneAudioStream microphoneInput = MicrophoneAudioStream.Start(); + await session.SendInputAudioAsync(microphoneInput); + }); }); StreamReceiveResult result; var buffer = new byte[1024 * 8]; -do -{ - var seg = new ArraySegment(buffer); - result = await channel.ReceiveAsync(seg, CancellationToken.None); +//do +//{ +// var seg = new ArraySegment(buffer); +// result = await channel.ReceiveAsync(seg, CancellationToken.None); - await hub.Completer.AppenAudioBuffer(seg, result.Count); +// await hub.Completer.AppenAudioBuffer(seg, result.Count); - // Display the audio level - int audioLevel = CalculateAudioLevel(buffer, result.Count); - DisplayAudioLevel(audioLevel); -} while (result.Status == StreamChannelStatus.Open); +// // Display the audio level +// int audioLevel = CalculateAudioLevel(buffer, result.Count); +// DisplayAudioLevel(audioLevel); +//} while (result.Status == StreamChannelStatus.Open); + +while (true) { } int CalculateAudioLevel(byte[] buffer, int bytesRecorded) { diff --git a/tests/BotSharp.Test.RealtimeVoice/SpeakOutput.cs b/tests/BotSharp.Test.RealtimeVoice/SpeakOutput.cs new file mode 100644 index 00000000..4a7a4d37 --- /dev/null +++ b/tests/BotSharp.Test.RealtimeVoice/SpeakOutput.cs @@ -0,0 +1,41 @@ +using NAudio.Wave; + +namespace BotSharp.Test.RealtimeVoice; + +public class SpeakerOutput : IDisposable +{ + BufferedWaveProvider _waveProvider; + WaveOutEvent _waveOutEvent; + + public SpeakerOutput() + { + WaveFormat outputAudioFormat = new( + rate: 24000, + bits: 16, + channels: 1); + _waveProvider = new(outputAudioFormat) + { + BufferDuration = TimeSpan.FromMinutes(5), + DiscardOnBufferOverflow = false + }; + _waveOutEvent = new(); + _waveOutEvent.Init(_waveProvider); + _waveOutEvent.Play(); + } + + public void EnqueueForPlayback(byte[] audioData) + { + byte[] buffer = audioData?.ToArray() ?? []; + _waveProvider.AddSamples(buffer, 0, buffer.Length); + } + + public void ClearPlayback() + { + _waveProvider.ClearBuffer(); + } + + public void Dispose() + { + _waveOutEvent?.Dispose(); + } +} \ No newline at end of file