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

refine websocket session

Browse source
This commit is contained in:
Jicheng Lu 2025-05-01 12:47:30 -05:00
parent f35bf4dfbf
commit a26f918161
13 changed files with 160 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
src/Infrastructure/BotSharp.Core.Realtime/Models/Options/ChatSessionOptions.cs Normal file
View file

@ -0,0 +1,7 @@
namespace BotSharp.Core.Realtime.Models.Options;
public class ChatSessionOptions
{
public int? BufferSize { get; set; }
public JsonSerializerOptions? JsonOptions { get; set; }
}

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

@ -39,6 +39,7 @@ public class WaveStreamChannel : IStreamChannel
var waveFormat = new WaveFormat(24000, 16, 1); // 24000 Hz, 16-bit PCM, Mono
_bufferedWaveProvider = new BufferedWaveProvider(waveFormat);
_bufferedWaveProvider.BufferDuration = TimeSpan.FromMinutes(10);
//_bufferedWaveProvider.BufferLength = 1024;
_bufferedWaveProvider.DiscardOnBufferOverflow = true;
_waveOut = new WaveOutEvent()

4
src/Infrastructure/BotSharp.Core.Realtime/Using.cs
View file

@ -15,3 +15,7 @@ global using BotSharp.Abstraction.Agents;
global using BotSharp.Abstraction.Routing;
global using BotSharp.Abstraction.Agents.Enums;
global using BotSharp.Abstraction.Conversations.Models;
global using BotSharp.Core.Realtime.Models.Chat;
global using BotSharp.Core.Realtime.Models.Options;
global using BotSharp.Core.Realtime.Websocket.Chat;

79
src/Infrastructure/BotSharp.Core.Realtime/Websocket/Chat/BotSharpRealtimeSession.cs Normal file
View file

@ -0,0 +1,79 @@
using BotSharp.Core.Realtime.Websocket.Common;
using System.ClientModel;
using System.Runtime.CompilerServices;
namespace BotSharp.Core.Realtime.Websocket.Chat;
public class BotSharpRealtimeSession : IDisposable
{
private readonly IServiceProvider _services;
private readonly WebSocket _websocket;
private readonly ChatSessionOptions? _sessionOptions;
private readonly object _singleReceiveLock = new();
private AsyncWebsocketDataCollectionResult _receivedCollectionResult;
public BotSharpRealtimeSession(
IServiceProvider services,
WebSocket websocket,
ChatSessionOptions? sessionOptions)
{
_services = services;
_websocket = websocket;
_sessionOptions = sessionOptions;
}
public async IAsyncEnumerable<ChatSessionUpdate> ReceiveUpdatesAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
{
await foreach (ClientResult result in ReceiveInnerUpdatesAsync(cancellationToken))
{
var update = HandleSessionResult(result);
yield return update;
}
}
private async IAsyncEnumerable<ClientResult> ReceiveInnerUpdatesAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
{
lock (_singleReceiveLock)
{
_receivedCollectionResult ??= new(_websocket, _sessionOptions, cancellationToken);
}
await foreach (var result in _receivedCollectionResult)
{
yield return result;
}
}
private ChatSessionUpdate HandleSessionResult(ClientResult result)
{
using var response = result.GetRawResponse();
var bytes = response.Content.ToArray();
var text = Encoding.UTF8.GetString(bytes, 0, bytes.Length);
return new ChatSessionUpdate
{
RawResponse = text
};
}
public async Task SendEvent(string message)
{
if (_websocket.State == WebSocketState.Open)
{
var buffer = Encoding.UTF8.GetBytes(message);
await _websocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
}
}
public async Task Disconnect()
{
if (_websocket.State == WebSocketState.Open)
{
await _websocket.CloseAsync(WebSocketCloseStatus.NormalClosure, null, CancellationToken.None);
}
}
public void Dispose()
{
_websocket.Dispose();
}
}

10
src/Infrastructure/BotSharp.Core.Realtime/Websocket/Chat/AsyncWebsocketDataCollectionResult.cs → src/Infrastructure/BotSharp.Core.Realtime/Websocket/Common/AsyncWebsocketDataCollectionResult.cs
View file

@ -1,17 +1,21 @@
using BotSharp.Core.Realtime.Models.Options;
using System.ClientModel;
namespace BotSharp.Core.Realtime.Websocket.Chat;
namespace BotSharp.Core.Realtime.Websocket.Common;
public class AsyncWebsocketDataCollectionResult : AsyncCollectionResult<ClientResult>
internal class AsyncWebsocketDataCollectionResult : AsyncCollectionResult<ClientResult>
{
private readonly WebSocket _webSocket;
private readonly ChatSessionOptions? _sessionOptions;
private readonly CancellationToken _cancellationToken;
public AsyncWebsocketDataCollectionResult(
WebSocket webSocket,
ChatSessionOptions? sessionOptions,
CancellationToken cancellationToken)
{
_webSocket = webSocket;
_sessionOptions = sessionOptions;
_cancellationToken = cancellationToken;
}
@ -22,7 +26,7 @@ public class AsyncWebsocketDataCollectionResult : AsyncCollectionResult<ClientRe
public override async IAsyncEnumerable<ClientResult> GetRawPagesAsync()
{
await using var enumerator = new AsyncWebsocketDataResultEnumerator(_webSocket, _cancellationToken);
await using var enumerator = new AsyncWebsocketDataResultEnumerator(_webSocket, _sessionOptions, _cancellationToken);
while (await enumerator.MoveNextAsync().ConfigureAwait(false))
{
yield return enumerator.Current;

15
src/Infrastructure/BotSharp.Core.Realtime/Websocket/Chat/AsyncWebsocketDataResultEnumerator.cs → src/Infrastructure/BotSharp.Core.Realtime/Websocket/Common/AsyncWebsocketDataResultEnumerator.cs
View file

@ -1,21 +1,28 @@
using BotSharp.Core.Realtime.Models.Options;
using System.Buffers;
using System.ClientModel;
namespace BotSharp.Core.Realtime.Websocket.Chat;
namespace BotSharp.Core.Realtime.Websocket.Common;
public class AsyncWebsocketDataResultEnumerator : IAsyncEnumerator<ClientResult>
internal class AsyncWebsocketDataResultEnumerator : IAsyncEnumerator<ClientResult>
{
private readonly WebSocket _webSocket;
private readonly ChatSessionOptions? _sessionOptions;
private readonly CancellationToken _cancellationToken;
private readonly byte[] _buffer;
private const int DEFAULT_BUFFER_SIZE = 1024 * 32;
public AsyncWebsocketDataResultEnumerator(
WebSocket webSocket,
ChatSessionOptions? sessionOptions,
CancellationToken cancellationToken)
{
_webSocket = webSocket;
_sessionOptions = sessionOptions;
_cancellationToken = cancellationToken;
_buffer = ArrayPool<byte>.Shared.Rent(1024 * 32);
var bufferSize = sessionOptions?.BufferSize > 0 ? sessionOptions.BufferSize.Value : DEFAULT_BUFFER_SIZE;
_buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
}
public ClientResult Current { get; private set; }
@ -29,7 +36,7 @@ public class AsyncWebsocketDataResultEnumerator : IAsyncEnumerator<ClientResult>
public async ValueTask<bool> MoveNextAsync()
{
var response = new AiWebsocketPipelineResponse();
var response = new WebsocketPipelineResponse();
while (!response.IsComplete)
{
var receivedResult = await _webSocket.ReceiveAsync(new(_buffer), _cancellationToken);

8
src/Infrastructure/BotSharp.Core.Realtime/Websocket/Chat/AiWebsocketPipelineResponse.cs → src/Infrastructure/BotSharp.Core.Realtime/Websocket/Common/WebsocketPipelineResponse.cs
View file

@ -1,11 +1,11 @@
using System.ClientModel.Primitives;
using System.Net;
namespace BotSharp.Core.Realtime.Websocket.Chat;
namespace BotSharp.Core.Realtime.Websocket.Common;
public class AiWebsocketPipelineResponse : PipelineResponse
internal class WebsocketPipelineResponse : PipelineResponse
{
public AiWebsocketPipelineResponse()
public WebsocketPipelineResponse()
{
}
@ -53,7 +53,7 @@ public class AiWebsocketPipelineResponse : PipelineResponse
}
else if (receivedResult.MessageType != WebSocketMessageType.Text)
{
throw new NotImplementedException($"{nameof(AiWebsocketPipelineResponse)} currently supports only text messages.");
throw new NotImplementedException($"{nameof(WebsocketPipelineResponse)} currently supports only text messages.");
}
var rawBytes = receivedBytes.ToArray();

20
src/Infrastructure/BotSharp.Core.Realtime/Websocket/Chat/RealtimeChatSession.cs → src/Infrastructure/BotSharp.Core.Realtime/Websocket/Llm/LlmRealtimeSession.cs
View file

@ -1,25 +1,27 @@
using System.ClientModel;
using System.Runtime.CompilerServices;
using BotSharp.Core.Realtime.Models.Chat;
using BotSharp.Core.Realtime.Models.Options;
using BotSharp.Core.Realtime.Websocket.Common;
namespace BotSharp.Core.Realtime.Websocket.Chat;
namespace BotSharp.Core.Realtime.Websocket.Llm;
public class RealtimeChatSession : IDisposable
public class LlmRealtimeSession : IDisposable
{
private readonly IServiceProvider _services;
private readonly JsonSerializerOptions _jsonOptions;
private readonly ChatSessionOptions? _sessionOptions;
private ClientWebSocket _webSocket;
private readonly object _singleReceiveLock = new();
private readonly SemaphoreSlim _clientEventSemaphore = new(initialCount: 1, maxCount: 1);
private AsyncWebsocketDataCollectionResult _receivedCollectionResult;
public RealtimeChatSession(
public LlmRealtimeSession(
IServiceProvider services,
JsonSerializerOptions jsonOptions)
ChatSessionOptions? sessionOptions = null)
{
_services = services;
_jsonOptions = jsonOptions;
_sessionOptions = sessionOptions;
}
public async Task ConnectAsync(Uri uri, Dictionary<string, string> headers, CancellationToken cancellationToken = default)
@ -44,11 +46,11 @@ public class RealtimeChatSession : IDisposable
}
}
public async IAsyncEnumerable<ClientResult> ReceiveInnerUpdatesAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
private async IAsyncEnumerable<ClientResult> ReceiveInnerUpdatesAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
{
lock (_singleReceiveLock)
{
_receivedCollectionResult ??= new(_webSocket, cancellationToken);
_receivedCollectionResult ??= new(_webSocket, _sessionOptions, cancellationToken);
}
await foreach (var result in _receivedCollectionResult)
@ -81,7 +83,7 @@ public class RealtimeChatSession : IDisposable
{
if (message is not string data)
{
data = JsonSerializer.Serialize(message, _jsonOptions);
data = JsonSerializer.Serialize(message, _sessionOptions?.JsonOptions);
}
var buffer = Encoding.UTF8.GetBytes(data);

1
src/Plugins/BotSharp.Plugin.ChatHub/BotSharp.Plugin.ChatHub.csproj
View file

@ -20,6 +20,7 @@
<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App" />
<ProjectReference Include="..\..\Infrastructure\BotSharp.Core.Crontab\BotSharp.Core.Crontab.csproj" />
<ProjectReference Include="..\..\Infrastructure\BotSharp.Core.Realtime\BotSharp.Core.Realtime.csproj" />
<ProjectReference Include="..\..\Infrastructure\BotSharp.OpenAPI\BotSharp.OpenAPI.csproj" />
</ItemGroup>
</Project>

41
src/Plugins/BotSharp.Plugin.ChatHub/ChatStreamMiddleware.cs
View file

@ -1,5 +1,3 @@
using BotSharp.Abstraction.Realtime;
using BotSharp.Abstraction.Realtime.Models;
using Microsoft.AspNetCore.Http;
using System.Net.WebSockets;
@ -9,6 +7,7 @@ public class ChatStreamMiddleware
{
private readonly RequestDelegate _next;
private readonly ILogger<ChatStreamMiddleware> _logger;
private BotSharpRealtimeSession _session;
public ChatStreamMiddleware(
RequestDelegate next,
@ -38,6 +37,7 @@ public class ChatStreamMiddleware
}
catch (Exception ex)
{
_session?.Dispose();
_logger.LogError(ex, $"Error when connecting Chat stream. ({ex.Message})");
}
return;
@ -49,6 +49,13 @@ public class ChatStreamMiddleware
private async Task HandleWebSocket(IServiceProvider services, string agentId, string conversationId, WebSocket webSocket)
{
_session?.Dispose();
_session = new BotSharpRealtimeSession(services, webSocket, new ChatSessionOptions
{
BufferSize = 1024 * 16,
JsonOptions = BotSharpOptions.defaultJsonOptions
});
var hub = services.GetRequiredService<IRealtimeHub>();
var conn = hub.SetHubConnection(conversationId);
conn.CurrentAgentId = agentId;
@ -58,26 +65,15 @@ public class ChatStreamMiddleware
convService.SetConversationId(conversationId, []);
await convService.GetConversationRecordOrCreateNew(agentId);
var buffer = new byte[1024 * 32];
WebSocketReceiveResult result;
do
await foreach (ChatSessionUpdate update in _session.ReceiveUpdatesAsync(CancellationToken.None))
{
result = await webSocket.ReceiveAsync(new(buffer), CancellationToken.None);
if (result.MessageType != WebSocketMessageType.Text)
{
continue;
}
var receivedText = Encoding.UTF8.GetString(buffer, 0, result.Count);
var receivedText = update?.RawResponse;
if (string.IsNullOrEmpty(receivedText))
{
continue;
}
var (eventType, data) = MapEvents(conn, receivedText);
if (eventType == "start")
{
await ConnectToModel(hub, webSocket);
@ -95,28 +91,19 @@ public class ChatStreamMiddleware
break;
}
}
while (!webSocket.CloseStatus.HasValue);
await webSocket.CloseAsync(result?.CloseStatus ?? WebSocketCloseStatus.NormalClosure, result?.CloseStatusDescription, CancellationToken.None);
_session?.Disconnect();
_session?.Dispose();
}
private async Task ConnectToModel(IRealtimeHub hub, WebSocket webSocket)
{
await hub.ConnectToModel(async data =>
{
await SendEventToUser(webSocket, data);
await _session.SendEvent(data);
});
}
private async Task SendEventToUser(WebSocket webSocket, string message)
{
if (webSocket.State == WebSocketState.Open)
{
var buffer = Encoding.UTF8.GetBytes(message);
await webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
}
}
private (string, string) MapEvents(RealtimeHubConnection conn, string receivedText)
{
var response = JsonSerializer.Deserialize<ChatStreamEventResponse>(receivedText);

12
src/Plugins/BotSharp.Plugin.ChatHub/Using.cs
View file

@ -18,8 +18,6 @@ global using BotSharp.Abstraction.Agents.Settings;
global using BotSharp.Abstraction.Conversations.Settings;
global using BotSharp.Abstraction.Agents.Enums;
global using BotSharp.Abstraction.Conversations.Models;
global using BotSharp.OpenAPI.ViewModels.Conversations;
global using BotSharp.OpenAPI.ViewModels.Users;
global using BotSharp.Abstraction.Agents.Models;
global using BotSharp.Abstraction.Functions.Models;
global using BotSharp.Abstraction.Loggers;
@ -32,6 +30,14 @@ global using BotSharp.Abstraction.Messaging;
global using BotSharp.Abstraction.Messaging.Enums;
global using BotSharp.Abstraction.Messaging.Models.RichContent;
global using BotSharp.Abstraction.Templating;
global using BotSharp.Abstraction.Realtime;
global using BotSharp.Abstraction.Realtime.Models;
global using BotSharp.OpenAPI.ViewModels.Conversations;
global using BotSharp.OpenAPI.ViewModels.Users;
global using BotSharp.Plugin.ChatHub.Settings;
global using BotSharp.Plugin.ChatHub.Enums;
global using BotSharp.Plugin.ChatHub.Models.Stream;
global using BotSharp.Plugin.ChatHub.Models.Stream;
global using BotSharp.Core.Realtime.Models.Chat;
global using BotSharp.Core.Realtime.Models.Options;
global using BotSharp.Core.Realtime.Websocket.Chat;

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

@ -1,5 +1,3 @@
using BotSharp.Core.Realtime.Models.Chat;
using BotSharp.Core.Realtime.Websocket.Chat;
using BotSharp.Plugin.OpenAI.Models.Realtime;
using OpenAI.Chat;
@ -19,7 +17,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
private readonly BotSharpOptions _botsharpOptions;
protected string _model = "gpt-4o-mini-realtime-preview";
private RealtimeChatSession _session;
private LlmRealtimeSession _session;
public RealTimeCompletionProvider(
RealtimeModelSettings settings,
@ -54,15 +52,18 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
{
_session.Dispose();
}
_session = new RealtimeChatSession(_services, _botsharpOptions.JsonSerializerOptions);
_session = new LlmRealtimeSession(_services, new ChatSessionOptions
{
JsonOptions = _botsharpOptions.JsonSerializerOptions
});
await _session.ConnectAsync(
new Uri($"wss://api.openai.com/v1/realtime?model={_model}"),
new Dictionary<string, string>
uri: new Uri($"wss://api.openai.com/v1/realtime?model={_model}"),
headers: new Dictionary<string, string>
{
{"Authorization", $"Bearer {settings.ApiKey}"},
{"OpenAI-Beta", "realtime=v1"}
},
CancellationToken.None);
cancellationToken: CancellationToken.None);
_ = ReceiveMessage(
conn,

4
src/Plugins/BotSharp.Plugin.OpenAI/Using.cs
View file

@ -32,3 +32,7 @@ global using BotSharp.Abstraction.Realtime.Models;
global using BotSharp.Core.Infrastructures;
global using BotSharp.Plugin.OpenAI.Models;
global using BotSharp.Plugin.OpenAI.Settings;
global using BotSharp.Core.Realtime.Models.Chat;
global using BotSharp.Core.Realtime.Models.Options;
global using BotSharp.Core.Realtime.Websocket.Llm;