This commit is contained in:
Jicheng Lu 2025-04-11 16:29:44 -05:00
parent aba37501ce
commit f413daeb04
6 changed files with 63 additions and 29 deletions

View file

@ -9,5 +9,6 @@ public interface IStreamChannel
Task ConnectAsync(string conversationId);
Task<StreamReceiveResult> ReceiveAsync(ArraySegment<byte> buffer, CancellationToken cancellation);
Task SendAsync(byte[] data, CancellationToken cancellation);
void ClearBuffer();
Task CloseAsync(StreamChannelStatus status, string description, CancellationToken cancellation);
}

View file

@ -12,6 +12,7 @@ public class RealtimeHubConnection
public ConcurrentQueue<string> MarkQueue { get; set; } = new();
public string CurrentAgentId { get; set; } = null!;
public string ConversationId { get; set; } = null!;
public Func<string> OnModelReady { get; set; } = () => string.Empty;
public Func<string, string> OnModelMessageReceived { get; set; } = null!;
public Func<string> OnModelAudioResponseDone { get; set; } = null!;
public Func<string> OnModelUserInterrupted { get; set; } = null!;

View file

@ -53,8 +53,10 @@ public class RealtimeHub : IRealtimeHub
{
// Not TriggerModelInference, waiting for user utter.
var instruction = await _completer.UpdateSession(_conn);
var data = _conn.OnModelReady();
await responseToUser(data);
await HookEmitter.Emit<IRealtimeHook>(_services, async hook => await hook.OnModeReady(agent, _completer));
},
onModelAudioDeltaReceived: async (audioDeltaData, itemId) =>
{

View file

@ -1,7 +1,6 @@
using System.Collections.Concurrent;
using BotSharp.Abstraction.Realtime.Enums;
using NAudio.Wave;
using System.Collections.Concurrent;
using System.IO;
namespace BotSharp.Core.Realtime.Services;
@ -11,7 +10,7 @@ public class WaveStremChannel : IStreamChannel
private WaveInEvent _waveIn;
private WaveOutEvent _waveOut;
private BufferedWaveProvider _bufferedWaveProvider;
private readonly ConcurrentQueue<byte[]> _audioBufferQueue = new ConcurrentQueue<byte[]>();
private readonly ConcurrentQueue<byte[]> _audioBufferQueue = [];
private readonly ILogger _logger;
public WaveStremChannel(IServiceProvider services, ILogger<WaveStremChannel> logger)
@ -39,7 +38,7 @@ public class WaveStremChannel : 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 * 512; // Buffer length
_bufferedWaveProvider.BufferLength = 1024 * 1024; // Buffer length
_bufferedWaveProvider.DiscardOnBufferOverflow = true;
_waveOut = new WaveOutEvent();
@ -83,6 +82,11 @@ public class WaveStremChannel : IStreamChannel
return Task.CompletedTask;
}
public void ClearBuffer()
{
_bufferedWaveProvider?.ClearBuffer();
}
private void WaveIn_DataAvailable(object? sender, WaveInEventArgs e)
{
// Add the buffer to the queue

View file

@ -15,6 +15,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
protected readonly OpenAiSettings _settings;
protected readonly IServiceProvider _services;
protected readonly ILogger<RealTimeCompletionProvider> _logger;
private readonly BotSharpOptions _options;
protected string _model = "gpt-4o-mini-realtime-preview";
private ClientWebSocket _webSocket;
@ -22,11 +23,13 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
public RealTimeCompletionProvider(
OpenAiSettings settings,
ILogger<RealTimeCompletionProvider> logger,
IServiceProvider services)
IServiceProvider services,
BotSharpOptions options)
{
_settings = settings;
_logger = logger;
_services = services;
_options = options;
}
public async Task Connect(RealtimeHubConnection conn,
@ -45,6 +48,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
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");
@ -141,7 +145,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
Action<RoleDialogModel> onUserAudioTranscriptionCompleted,
Action onUserInterrupted)
{
var buffer = new byte[1024 * 32];
var buffer = new byte[1024 * 1024 * 32];
// Model response timeout
var timeout = 30;
WebSocketReceiveResult? result = default;
@ -276,7 +280,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
if (message is not string data)
{
data = JsonSerializer.Serialize(message, BotSharpOptions.defaultJsonOptions);
data = JsonSerializer.Serialize(message, _options.JsonSerializerOptions);
}
var buffer = Encoding.UTF8.GetBytes(data);
@ -291,12 +295,9 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
var agentService = _services.GetRequiredService<IAgentService>();
var agent = await agentService.LoadAgent(conn.CurrentAgentId);
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 instruction = messages.FirstOrDefault()?.Content.FirstOrDefault()?.Text ?? agent?.Description;
var functions = options.Tools.Select(x =>
{
var fn = new FunctionDef

View file

@ -3,7 +3,6 @@ using BotSharp.Abstraction.Conversations.Models;
using BotSharp.Abstraction.Conversations;
using BotSharp.OpenAPI;
using System.Text.Json;
using Google.Ai.Generativelanguage.V1Beta2;
var services = ServiceBuilder.CreateHostBuilder();
var channel = services.GetRequiredService<IStreamChannel>();
@ -27,18 +26,11 @@ var hub = services.GetRequiredService<IRealtimeHub>();
var conn = hub.SetHubConnection(conv.Id);
var completer = hub.SetCompleter("openai");
await hub.ConnectToModel(async data =>
{
var response = JsonSerializer.Deserialize<ModelResponseEvent>(data);
if (response.Event == "media")
conn.OnModelReady = () =>
JsonSerializer.Serialize(new
{
var message = JsonSerializer.Deserialize<ModelResponseMediaEvent>(data);
await channel.SendAsync(Convert.FromBase64String(message.Media), CancellationToken.None);
}
});
StreamReceiveResult result;
var buffer = new byte[1024 * 8];
@event = "init"
});
conn.OnModelMessageReceived = message =>
JsonSerializer.Serialize(new
@ -60,6 +52,23 @@ conn.OnModelUserInterrupted = () =>
@event = "clear"
});
await hub.ConnectToModel(async data =>
{
var response = JsonSerializer.Deserialize<ModelResponseEvent>(data);
if (response.Event == "clear")
{
channel.ClearBuffer();
}
else if (response.Event == "media")
{
var message = JsonSerializer.Deserialize<ModelResponseMediaEvent>(data);
await channel.SendAsync(Convert.FromBase64String(message.Media), CancellationToken.None);
}
});
StreamReceiveResult result;
var buffer = new byte[1024 * 8];
do
{
var seg = new ArraySegment<byte>(buffer);
@ -75,22 +84,38 @@ do
int CalculateAudioLevel(byte[] buffer, int bytesRecorded)
{
// Simple audio level calculation (RMS)
int sum = 0;
int bytesPerSample = 2; // 16-bit PCM = 2 bytes per sample
int sampleCount = bytesRecorded / bytesPerSample;
if (sampleCount == 0) return 0;
double sum = 0;
for (int i = 0; i < bytesRecorded; i += 2)
{
if (i + 1 < bytesRecorded)
{
short sample = (short)((buffer[i + 1] << 8) | buffer[i]);
sum += Math.Abs(sample);
double normalized = sample / (short.MaxValue * 1.0 + 1);
sum += normalized * normalized;
}
}
return bytesRecorded > 0 ? sum / (bytesRecorded / 2) : 0;
double rms = Math.Sqrt(sum / sampleCount);
double db = 20 * Math.Log10(rms);
if (double.IsInfinity(db) || double.IsNaN(db))
{
return 0;
}
db = Math.Clamp(db, -100, 0);
return (int)((db + 100) * 1);
}
void DisplayAudioLevel(int level)
{
const int sep = 50;
// Normalize level to 0-50 range for display
int displayLevel = Math.Min(50, level / 100);
int displayLevel = (level * sep) / 100;
// Clear the current line
Console.Write("\r" + new string(' ', 60));
@ -98,6 +123,6 @@ void DisplayAudioLevel(int level)
// Display audio level as a bar
Console.Write("\rMicrophone: [");
Console.Write(new string('#', displayLevel));
Console.Write(new string(' ', 50 - displayLevel));
Console.Write(new string(' ', sep - displayLevel));
Console.Write("]");
}