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

temp save

Browse source
This commit is contained in:
Jicheng Lu 2025-04-14 01:25:16 -05:00
parent d88e762fdb
commit 7cbce42119
11 changed files with 755 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
src/Infrastructure/BotSharp.Core.Realtime/Services/WaveStreamChannel.cs
View file

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

11
src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/SessionConversationUpdate.cs Normal file
View file

@ -0,0 +1,11 @@
namespace BotSharp.Plugin.OpenAI.Models.Realtime;
public class SessionConversationUpdate
{
public string RawResponse { get; set; }
public SessionConversationUpdate()
{
}
}

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

@ -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<RealtimeModelSettings>();
_model = realtimeModelSettings.Model;
var settingsService = _services.GetRequiredService<ILlmProviderService>();
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<ILlmProviderService>();
//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<string,string> onModelAudioDeltaReceived,
Action<string, string> onModelAudioDeltaReceived,
Action onModelAudioResponseDone,
Action<string> onModelAudioTranscriptDone,
Action<List<RoleDialogModel>> onModelResponseDone,
@ -145,33 +165,10 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
Action<RoleDialogModel> onUserAudioTranscriptionCompleted,
Action onInterruptionDetected)
{
var buffer = new byte[1024 * 1024 * 32];
// Model response timeout
var settings = _services.GetRequiredService<RealtimeModelSettings>();
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<byte>(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<ServerEventResponse>(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<string,string> onModelAudioDeltaReceived,
// Action onModelAudioResponseDone,
// Action<string> onModelAudioTranscriptDone,
// Action<List<RoleDialogModel>> onModelResponseDone,
// Action<string> onConversationItemCreated,
// Action<RoleDialogModel> onUserAudioTranscriptionCompleted,
// Action onInterruptionDetected)
//{
// var buffer = new byte[1024 * 1024 * 32];
// // Model response timeout
// var settings = _services.GetRequiredService<RealtimeModelSettings>();
// var timeout = settings.ModelResponseTimeout;
// WebSocketReceiveResult? result = default;
// do
// {
// Array.Clear(buffer, 0, buffer.Length);
// var taskWorker = _webSocket.ReceiveAsync(new ArraySegment<byte>(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<ServerEventResponse>(receivedText);
// _logger.LogDebug($"{nameof(RealTimeCompletionProvider)} received: {response.Type} {receivedText.Length}");
// if (response.Type == "error")
// {
// _logger.LogError($"{response.Type}: {receivedText}");
// var error = JsonSerializer.Deserialize<ServerEventErrorResponse>(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<ResponseAudioTranscript>(receivedText);
// onModelAudioTranscriptDone(data.Transcript);
// }
// else if (response.Type == "response.audio.delta")
// {
// var audio = JsonSerializer.Deserialize<ResponseAudioDelta>(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<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
//await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
}
public async Task<string> 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<string>();
HookEmitter.Emit<IRealtimeHook>(_services, hook => words.AddRange(hook.OnModelTranscriptPrompt(agent)));
//if (realtimeModelSettings.InputAudioTranscribe)
//{
// var words = new List<string>();
// HookEmitter.Emit<IRealtimeHook>(_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<IContentGeneratingHook>(_services, async hook =>
{

116
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AiWebsocketPipelineResponse.cs Normal file
View file

@ -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<BinaryData> BufferContentAsync(CancellationToken cancellationToken = default)
{
return new ValueTask<BinaryData>(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;
}
}

35
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataCollectionResult.cs Normal file
View file

@ -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<ClientResult>
{
private readonly WebSocket _webSocket;
public AsyncWebsocketDataCollectionResult(WebSocket webSocket)
{
_webSocket = webSocket;
}
public override ContinuationToken? GetContinuationToken(ClientResult page)
{
return null;
}
public override async IAsyncEnumerable<ClientResult> GetRawPagesAsync()
{
await using var enumerator = new AsyncWebsocketDataResultEnumerator(_webSocket);
while (await enumerator.MoveNextAsync().ConfigureAwait(false))
{
yield return enumerator.Current;
}
}
protected override async IAsyncEnumerable<ClientResult> GetValuesFromPageAsync(ClientResult page)
{
await Task.CompletedTask;
yield return page;
}
}

49
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/AsyncWebsocketDataResultEnumerator.cs Normal file
View file

@ -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<ClientResult>
{
private readonly WebSocket _webSocket;
private readonly byte[] _buffer;
public AsyncWebsocketDataResultEnumerator(
WebSocket webSocket)
{
_webSocket = webSocket;
_buffer = ArrayPool<byte>.Shared.Rent(1024 * 32);
}
public ClientResult Current { get; private set; }
public ValueTask DisposeAsync()
{
_webSocket?.Dispose();
return new ValueTask(Task.CompletedTask);
}
public async ValueTask<bool> 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;
}
}

111
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/Session/RealtimeChatSession.cs Normal file
View file

@ -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<ILlmProviderService>();
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<SessionConversationUpdate> ReceiveUpdatesAsync()
{
await foreach (ClientResult result in ReceiveInnerUpdatesAsync())
{
var update = HandleSessionResult(result);
yield return update;
}
}
public async IAsyncEnumerable<ClientResult> 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<byte>(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();
}
}

34
tests/BotSharp.Test.RealtimeVoice/LocalSession.cs Normal file
View file

@ -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<byte>.Shared.Rent(1024 * 16);
while (true)
{
int bytesRead = await audio.ReadAsync(buffer, 0, buffer.Length).ConfigureAwait(false);
if (bytesRead == 0)
{
break;
}
ReadOnlyMemory<byte> audioMemory = buffer.AsMemory(0, bytesRead);
BinaryData audioData = BinaryData.FromBytes(audioMemory);;
await _completion.AppenAudioBuffer(audioData.ToArray(), audioData.Length);
}
}
}

119
tests/BotSharp.Test.RealtimeVoice/MicrophoneAudioStream.cs Normal file
View file

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

42
tests/BotSharp.Test.RealtimeVoice/Program.cs
View file

@ -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<IStreamChannel>();
@ -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<IRealtimeHub>();
var conn = hub.SetHubConnection(conv.Id);
@ -52,34 +54,50 @@ conn.OnModelUserInterrupted = () =>
@event = "clear"
});
var completer = services.GetServices<IRealTimeCompletion>().First(x => x.Provider == "openai");
LocalSession session = new(completer);
SpeakerOutput speakerOutput = new();
await hub.ConnectToModel(async data =>
{
var response = JsonSerializer.Deserialize<ModelResponseEvent>(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<ModelResponseMediaEvent>(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<byte>(buffer);
result = await channel.ReceiveAsync(seg, CancellationToken.None);
//do
//{
// var seg = new ArraySegment<byte>(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)
{

41
tests/BotSharp.Test.RealtimeVoice/SpeakOutput.cs Normal file
View file

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