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

refactor realtime code.

Browse source
This commit is contained in:
Haiping Chen 2025-03-05 12:29:28 -06:00
parent a205b9b407
commit d69f19bd41
7 changed files with 160 additions and 187 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs
View file

@ -1,9 +1,15 @@
using System.Collections.Concurrent;
namespace BotSharp.Abstraction.Realtime.Models;
public class RealtimeHubConnection
{
public string Event { get; set; } = null!;
public string StreamId { get; set; } = null!;
public string? LastAssistantItem { get; set; } = null!;
public long LatestMediaTimestamp { get; set; }
public long? ResponseStartTimestamp { get; set; }
public ConcurrentQueue<string> MarkQueue { get; set; } = new();
public string CurrentAgentId { get; set; } = null!;
public string ConversationId { get; set; } = null!;
public string Data { get; set; } = string.Empty;

44
src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs
View file

@ -112,8 +112,28 @@ public class RealtimeHub : IRealtimeHub
},
onModelAudioDeltaReceived: async audioDeltaData =>
{
// If this is the first delta of a new response, set the start timestamp
if (!conn.ResponseStartTimestamp.HasValue)
{
conn.ResponseStartTimestamp = conn.LatestMediaTimestamp;
_logger.LogDebug($"Setting start timestamp for new response: {conn.ResponseStartTimestamp}ms");
}
var data = conn.OnModelMessageReceived(audioDeltaData);
await SendEventToUser(userWebSocket, data);
// Send mark messages to Media Streams so we know if and when AI response playback is finished
if (!string.IsNullOrEmpty(conn.StreamId))
{
var markEvent = new
{
@event = "mark",
streamSid = conn.StreamId,
mark = new { name = "responsePart" }
};
await SendEventToUser(userWebSocket, markEvent);
conn.MarkQueue.Enqueue("responsePart");
}
},
onModelAudioResponseDone: async () =>
{
@ -160,16 +180,19 @@ public class RealtimeHub : IRealtimeHub
await completer.TriggerModelInference("Reply based on the function's output.");
}
}
// append output audio transcript to conversation
storage.Append(conn.ConversationId, message);
dialogs.Add(message);
foreach (var hook in hookProvider.HooksOrderByPriority)
else
{
hook.SetAgent(agent)
.SetConversation(conversation);
// append output audio transcript to conversation
storage.Append(conn.ConversationId, message);
dialogs.Add(message);
await hook.OnResponseGenerated(message);
foreach (var hook in hookProvider.HooksOrderByPriority)
{
hook.SetAgent(agent)
.SetConversation(conversation);
await hook.OnResponseGenerated(message);
}
}
}
},
@ -193,6 +216,11 @@ public class RealtimeHub : IRealtimeHub
},
onUserInterrupted: async () =>
{
// Reset states
conn.MarkQueue.Clear();
conn.LastAssistantItem = null;
conn.ResponseStartTimestamp = null;
var data = conn.OnModelUserInterrupted();
await SendEventToUser(userWebSocket, data);
});

6
src/Infrastructure/BotSharp.Core/Translation/TranslationResponseHook.cs
View file

@ -1,4 +1,5 @@
using BotSharp.Abstraction.Agents;
using BotSharp.Abstraction.Conversations.Enums;
using BotSharp.Abstraction.Infrastructures.Enums;
using BotSharp.Abstraction.Translation;
using System;
@ -26,6 +27,11 @@ namespace BotSharp.Logger.Hooks
{
return;
}
if (_states.GetState("channel") == ConversationChannel.Phone)
{
return;
}
// Handle multi-language for output
var agentService = _services.GetRequiredService<IAgentService>();

2
src/Infrastructure/BotSharp.Logger/Hooks/VerboseLogHook.cs
View file

@ -35,7 +35,7 @@ public class VerboseLogHook : IContentGeneratingHook
public async Task AfterGenerated(RoleDialogModel message, TokenStatsModel tokenStats)
{
if (!_convSettings.ShowVerboseLog) return;
if (!_convSettings.ShowVerboseLog || string.IsNullOrEmpty(tokenStats.Prompt)) return;
var agentService = _services.GetRequiredService<IAgentService>();
var agent = await agentService.LoadAgent(message.CurrentAgentId);

21
src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs
View file

@ -1,8 +1,27 @@
using BotSharp.Abstraction.Functions.Models;
namespace BotSharp.Plugin.OpenAI.Models.Realtime;
public class RealtimeSessionCreationRequest : RealtimeSessionBody
public class RealtimeSessionCreationRequest
{
[JsonPropertyName("model")]
[JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)]
public string Model { get; set; } = null!;
[JsonPropertyName("modalities")]
public string[] Modalities { get; set; } = ["audio", "text"];
[JsonPropertyName("instructions")]
public string Instructions { get; set; } = null!;
[JsonPropertyName("tool_choice")]
public string ToolChoice { get; set; } = "auto";
[JsonPropertyName("tools")]
public FunctionDef[] Tools { get; set; } = [];
[JsonPropertyName("turn_detection")]
public RealtimeSessionTurnDetection TurnDetection { get; set; } = new();
}
/// <summary>

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

@ -128,9 +128,8 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
Action<RoleDialogModel> onInputAudioTranscriptionCompleted,
Action onUserInterrupted)
{
var buffer = new byte[1024 * 256];
var buffer = new byte[1024 * 16];
WebSocketReceiveResult result;
string? lastAssistantItem = null;
do
{
@ -173,7 +172,16 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
else if (response.Type == "response.audio.delta")
{
var audio = JsonSerializer.Deserialize<ResponseAudioDelta>(receivedText);
lastAssistantItem = audio?.ItemId;
// Record last assistant item ID for interruption handling
if (conn.ResponseStartTimestamp.HasValue)
{
conn.ResponseStartTimestamp = conn.LatestMediaTimestamp;
}
if (!string.IsNullOrEmpty(conn.StreamId))
{
conn.LastAssistantItem = audio?.ItemId;
}
if (audio != null && audio.Delta != null)
{
@ -205,19 +213,24 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
}
else if (response.Type == "input_audio_buffer.speech_started")
{
// var elapsedTime = latestMediaTimestamp - responseStartTimestampTwilio;
// handle use interuption
if (!string.IsNullOrEmpty(lastAssistantItem))
// Handle user interuption
if (conn.MarkQueue.Count > 0 && conn.ResponseStartTimestamp != null)
{
var truncateEvent = new
{
type = "conversation.item.truncate",
item_id = lastAssistantItem,
content_index = 0,
audio_end_ms = 300
};
var elapsedTime = conn.LatestMediaTimestamp - conn.ResponseStartTimestamp;
if (!string.IsNullOrEmpty(conn.LastAssistantItem))
{
var truncateEvent = new
{
type = "conversation.item.truncate",
item_id = conn.LastAssistantItem,
content_index = 0,
audio_end_ms = elapsedTime
};
await SendEventToModel(truncateEvent);
}
await SendEventToModel(truncateEvent);
onUserInterrupted();
}
}
@ -256,6 +269,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
var args = new RealtimeSessionCreationRequest
{
Model = _model,
Instructions = instruction,
ToolChoice = "auto",
Tools = options.Tools.Select(x =>
@ -271,7 +285,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
};
var settingsService = _services.GetRequiredService<ILlmProviderService>();
var settings = settingsService.GetSetting(Provider, args.Model);
var settings = settingsService.GetSetting(Provider, args.Model ?? _model);
var api = _services.GetRequiredService<IOpenAiRealtimeApi>();
var session = await api.GetSessionAsync(args, settings.ApiKey);
@ -318,12 +332,13 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
ToolChoice = "auto",
Tools = functions,
Modalities = [ "text", "audio" ],
Temperature = Math.Max(options.Temperature ?? 0f, 0.6f),
Temperature = Math.Max(options.Temperature ?? 0f, 0.8f),
MaxResponseOutputTokens = 512,
TurnDetection = new RealtimeSessionTurnDetection
{
Threshold = 0.8f,
SilenceDuration = 800
Threshold = 0.5f,
PrefixPadding = 300,
SilenceDuration = 500
}
}
};

217
src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs
View file

@ -37,7 +37,14 @@ public class TwilioStreamMiddleware
var services = httpContext.RequestServices;
var conversationId = request.Path.Value.Split("/").Last();
using WebSocket webSocket = await httpContext.WebSockets.AcceptWebSocketAsync();
await HandleWebSocket(services, conversationId, webSocket);
try
{
await HandleWebSocket(services, conversationId, webSocket);
}
catch (Exception ex)
{
_logger.LogError(ex, $"Error in WebSocket communication: {ex.Message} for conversation {conversationId}");
}
return;
}
}
@ -48,23 +55,15 @@ public class TwilioStreamMiddleware
private async Task HandleWebSocket(IServiceProvider services, string conversationId, WebSocket webSocket)
{
var hub = services.GetRequiredService<IRealtimeHub>();
var convService = services.GetRequiredService<IConversationService>();
// Session state
var conn = new RealtimeHubConnection
{
ConversationId = conversationId
};
// Variables for timestamp and interruption handling
string streamSid = null;
long latestMediaTimestamp = 0;
string lastAssistantItem = null;
var markQueue = new ConcurrentQueue<string>();
long? responseStartTimestampTwilio = null;
// Load session and state
convService.SetConversationId(conversationId, new List<MessageState>());
// load conversation and state
var convService = services.GetRequiredService<IConversationService>();
convService.SetConversationId(conversationId, []);
var hooks = services.GetServices<ITwilioSessionHook>();
foreach (var hook in hooks)
{
@ -72,159 +71,59 @@ public class TwilioStreamMiddleware
}
convService.States.Save();
// Set up event handlers
conn.OnModelMessageReceived = message =>
await hub.Listen(webSocket, (receivedText) =>
{
// Record last assistant item ID for interruption handling
if (!string.IsNullOrEmpty(conn.StreamId))
var response = JsonSerializer.Deserialize<StreamEventResponse>(receivedText);
conn.StreamId = response.StreamSid;
conn.Event = response.Event switch
{
lastAssistantItem = conn.StreamId;
}
// If this is the first delta of a new response, set the start timestamp
if (!responseStartTimestampTwilio.HasValue)
{
responseStartTimestampTwilio = latestMediaTimestamp;
_logger.LogDebug($"Setting start timestamp for new response: {responseStartTimestampTwilio}ms");
}
// Add mark to queue
markQueue.Enqueue("responsePart");
return new
{
@event = "media",
streamSid = conn.StreamId,
media = new { payload = message }
"start" => "user_connected",
"media" => "user_data_received",
"stop" => "user_disconnected",
_ => response.Event
};
};
conn.OnModelAudioResponseDone = () =>
{
return new
if (string.IsNullOrEmpty(conn.Event))
{
@event = "mark",
streamSid = conn.StreamId,
mark = new { name = "responsePart" }
};
};
conn.OnModelUserInterrupted = () =>
{
// Reset states
markQueue.Clear();
lastAssistantItem = null;
responseStartTimestampTwilio = null;
return new
{
@event = "clear",
streamSid = conn.StreamId
};
};
try
{
await hub.Listen(webSocket, receivedText =>
{
var response = JsonSerializer.Deserialize<StreamEventResponse>(receivedText);
if (response == null)
{
_logger.LogWarning("Failed to parse received WebSocket message");
return conn;
}
conn.StreamId = response.StreamSid;
switch (response.Event)
{
case "start":
conn.Event = "user_connected";
streamSid = response.StreamSid;
_logger.LogInformation($"Incoming stream started: {streamSid}");
// Reset start and media timestamps
responseStartTimestampTwilio = null;
latestMediaTimestamp = 0;
var startResponse = JsonSerializer.Deserialize<StreamEventStartResponse>(receivedText);
if (startResponse?.Body?.CustomParameters != null)
{
conn.Data = JsonSerializer.Serialize(startResponse.Body.CustomParameters);
}
break;
case "media":
conn.Event = "user_data_received";
var mediaResponse = JsonSerializer.Deserialize<StreamEventMediaResponse>(receivedText);
if (mediaResponse?.Body != null)
{
conn.Data = mediaResponse.Body.Payload;
// Update latest media timestamp
if (long.TryParse(mediaResponse.Body.Timestamp, out latestMediaTimestamp))
{
_logger.LogDebug($"Received media message with timestamp: {latestMediaTimestamp}ms");
}
// Check if user started speaking (interruption handling)
if (markQueue.Count > 0 && responseStartTimestampTwilio.HasValue &&
!string.IsNullOrEmpty(lastAssistantItem))
{
// Detect voice activity - more complex logic can be added here
// e.g., check audio energy levels or use VAD (Voice Activity Detection)
// If voice activity detected, handle interruption
if (ShouldHandleInterruption(mediaResponse.Body.Payload))
{
conn.Event = "user_interrupted";
long elapsedTime = latestMediaTimestamp - responseStartTimestampTwilio.Value;
_logger.LogDebug($"Calculating elapsed time for truncation: {latestMediaTimestamp} - {responseStartTimestampTwilio} = {elapsedTime}ms");
}
}
}
break;
case "mark":
// Handle mark event
if (markQueue.TryDequeue(out _))
{
_logger.LogDebug("Processing mark event, removing one mark from queue");
}
break;
case "stop":
conn.Event = "user_disconnected";
break;
default:
_logger.LogInformation($"Received non-media event: {response.Event}");
break;
}
return conn;
});
}
catch (Exception ex)
{
_logger.LogError(ex, "Error in WebSocket communication");
}
}
}
// Simple interruption detection logic - can be extended as needed
private bool ShouldHandleInterruption(string audioPayload)
{
// Here should implement actual voice activity detection logic
// e.g., analyze audio energy levels or use VAD algorithm
// Simple example - should be replaced with real detection logic in production
if (!string.IsNullOrEmpty(audioPayload))
{
// Check if audio payload contains sufficient energy
// This is just a placeholder - needs actual VAD implementation
return false; // Default to false to avoid false interruptions
}
return false;
conn.OnModelMessageReceived = message =>
new
{
@event = "media",
streamSid = response.StreamSid,
media = new { payload = message }
};
conn.OnModelAudioResponseDone = () =>
new
{
@event = "mark",
streamSid = response.StreamSid,
mark = new { name = "responsePart" }
};
conn.OnModelUserInterrupted = () =>
new
{
@event = "clear",
streamSid = response.StreamSid
};
if (response.Event == "start")
{
var startResponse = JsonSerializer.Deserialize<StreamEventStartResponse>(receivedText);
conn.LatestMediaTimestamp = 0;
conn.ResponseStartTimestamp = null;
conn.Data = JsonSerializer.Serialize(startResponse.Body.CustomParameters);
}
else if (response.Event == "media")
{
var mediaResponse = JsonSerializer.Deserialize<StreamEventMediaResponse>(receivedText);
conn.LatestMediaTimestamp = long.Parse(mediaResponse.Body.Timestamp);
conn.Data = mediaResponse.Body.Payload;
}
return conn;
});
}
}