From f9cc755444fbfbc2367574dd7432174383f52734 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Thu, 6 Mar 2025 15:10:17 -0600 Subject: [PATCH] RealtimeConversationHook --- .../Realtime/IRealtimeHub.cs | 9 +- .../Hooks/RealtimeConversationHook.cs | 64 ++++++++ .../BotSharp.Core.Realtime/RealtimePlugin.cs | 2 + .../Services/RealtimeHub.cs | 140 +++++++++--------- .../Realtime/RealTimeCompletionProvider.cs | 7 +- .../Services/Stream/TwilioStreamMiddleware.cs | 11 +- 6 files changed, 148 insertions(+), 85 deletions(-) create mode 100644 src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHub.cs b/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHub.cs index 67d0f18c..23ecb654 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHub.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeHub.cs @@ -1,3 +1,4 @@ +using BotSharp.Abstraction.MLTasks; using BotSharp.Abstraction.Realtime.Models; using System.Net.WebSockets; @@ -8,5 +9,11 @@ namespace BotSharp.Abstraction.Realtime; /// public interface IRealtimeHub { - Task Listen(WebSocket userWebSocket, Func onUserMessageReceived); + RealtimeHubConnection HubConn { get; } + RealtimeHubConnection SetHubConnection(string conversationId); + + IRealTimeCompletion Completer { get; } + IRealTimeCompletion SetCompleter(string provider); + + Task Listen(WebSocket userWebSocket, Action onUserMessageReceived); } diff --git a/src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs b/src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs new file mode 100644 index 00000000..e7478e29 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs @@ -0,0 +1,64 @@ +using BotSharp.Abstraction.Utilities; + +namespace BotSharp.Core.Realtime.Hooks; + +public class RealtimeConversationHook : ConversationHookBase, IConversationHook +{ + private readonly IServiceProvider _services; + public RealtimeConversationHook(IServiceProvider services) + { + _services = services; + } + + public async Task OnFunctionExecuting(RoleDialogModel message) + { + var hub = _services.GetRequiredService(); + if (hub.HubConn == null) + { + return; + } + // Save states + var states = _services.GetRequiredService(); + states.SaveStateByArgs(message.FunctionArgs?.JsonContent() ?? JsonDocument.Parse("{}")); + } + + public async Task OnFunctionExecuted(RoleDialogModel message) + { + var hub = _services.GetRequiredService(); + if (hub.HubConn == null) + { + return; + } + var routing = _services.GetRequiredService(); + + message.Role = AgentRole.Function; + + if (message.FunctionName == "route_to_agent") + { + var inst = JsonSerializer.Deserialize(message.FunctionArgs ?? "{}") ?? new(); + message.Content = $"Connected to agent of {inst.AgentName}"; + hub.HubConn.CurrentAgentId = routing.Context.GetCurrentAgentId(); + + await hub.Completer.UpdateSession(hub.HubConn); + await hub.Completer.InsertConversationItem(message); + await hub.Completer.TriggerModelInference($"Guide the user through the next steps of the process as this Agent ({inst.AgentName}), following its instructions and operational procedures."); + } + else if (message.FunctionName == "util-routing-fallback_to_router") + { + var inst = JsonSerializer.Deserialize(message.FunctionArgs ?? "{}") ?? new(); + message.Content = $"Returned to Router due to {inst.Reason}"; + hub.HubConn.CurrentAgentId = routing.Context.GetCurrentAgentId(); + + await hub.Completer.UpdateSession(hub.HubConn); + await hub.Completer.InsertConversationItem(message); + await hub.Completer.TriggerModelInference($"Check with user whether to proceed the new request: {inst.Reason}"); + } + else + { + // Update session for changed states + await hub.Completer.UpdateSession(hub.HubConn); + await hub.Completer.InsertConversationItem(message); + await hub.Completer.TriggerModelInference("Reply based on the function's output."); + } + } +} diff --git a/src/Infrastructure/BotSharp.Core.Realtime/RealtimePlugin.cs b/src/Infrastructure/BotSharp.Core.Realtime/RealtimePlugin.cs index e632dcd3..a45325de 100644 --- a/src/Infrastructure/BotSharp.Core.Realtime/RealtimePlugin.cs +++ b/src/Infrastructure/BotSharp.Core.Realtime/RealtimePlugin.cs @@ -1,4 +1,5 @@ using BotSharp.Abstraction.Plugins; +using BotSharp.Core.Realtime.Hooks; using BotSharp.Core.Realtime.Services; using Microsoft.Extensions.Configuration; @@ -14,5 +15,6 @@ public class RealtimePlugin : IBotSharpPlugin public void RegisterDI(IServiceCollection services, IConfiguration config) { services.AddScoped(); + services.AddScoped(); } } diff --git a/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs b/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs index 924a0e64..68636bd1 100644 --- a/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs +++ b/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs @@ -1,3 +1,5 @@ +using BotSharp.Abstraction.Utilities; + namespace BotSharp.Core.Realtime.Services; public class RealtimeHub : IRealtimeHub @@ -5,6 +7,12 @@ public class RealtimeHub : IRealtimeHub private readonly IServiceProvider _services; private readonly ILogger _logger; + private RealtimeHubConnection _conn; + public RealtimeHubConnection HubConn => _conn; + + private IRealTimeCompletion _completer; + public IRealTimeCompletion Completer => _completer; + public RealtimeHub(IServiceProvider services, ILogger logger) { _services = services; @@ -12,12 +20,12 @@ public class RealtimeHub : IRealtimeHub } public async Task Listen(WebSocket userWebSocket, - Func onUserMessageReceived) + Action onUserMessageReceived) { var buffer = new byte[1024 * 16]; WebSocketReceiveResult result; - var completer = _services.GetServices().First(x => x.Provider == "openai"); + do { @@ -29,40 +37,40 @@ public class RealtimeHub : IRealtimeHub continue; } - var conn = onUserMessageReceived(receivedText); + onUserMessageReceived(receivedText); - if (conn.Event == "user_connected") + if (_conn.Event == "user_connected") { - await ConnectToModel(completer, userWebSocket, conn); + await ConnectToModel(userWebSocket); } - else if (conn.Event == "user_data_received") + else if (_conn.Event == "user_data_received") { - await completer.AppenAudioBuffer(conn.Data); + await _completer.AppenAudioBuffer(_conn.Data); } - else if (conn.Event == "user_dtmf_received") + else if (_conn.Event == "user_dtmf_received") { - await HandleUserDtmfReceived(completer, conn); + await HandleUserDtmfReceived(); } - else if (conn.Event == "user_disconnected") + else if (_conn.Event == "user_disconnected") { - await completer.Disconnect(); - await HandleUserDisconnected(conn); + await _completer.Disconnect(); + await HandleUserDisconnected(); } } while (!result.CloseStatus.HasValue); await userWebSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None); } - private async Task ConnectToModel(IRealTimeCompletion completer, WebSocket userWebSocket, RealtimeHubConnection conn) + private async Task ConnectToModel(WebSocket userWebSocket) { var hookProvider = _services.GetRequiredService(); var convService = _services.GetRequiredService(); - convService.SetConversationId(conn.ConversationId, []); - var conversation = await convService.GetConversation(conn.ConversationId); + convService.SetConversationId(_conn.ConversationId, []); + var conversation = await convService.GetConversation(_conn.ConversationId); var agentService = _services.GetRequiredService(); var agent = await agentService.LoadAgent(conversation.AgentId); - conn.CurrentAgentId = agent.Id; + _conn.CurrentAgentId = agent.Id; // Set model var model = agent.LlmConfig.Model; @@ -72,8 +80,8 @@ public class RealtimeHub : IRealtimeHub model = llmProviderService.GetProviderModel("openai", "gpt-4", realTime: true).Name; } - completer.SetModelName(model); - conn.Model = model; + _completer.SetModelName(model); + _conn.Model = model; var routing = _services.GetRequiredService(); routing.Context.Push(agent.Id); @@ -85,54 +93,48 @@ public class RealtimeHub : IRealtimeHub } routing.Context.SetDialogs(dialogs); - await completer.Connect(conn, + await _completer.Connect(_conn, onModelReady: async () => { // Control initial session, prevent initial response interruption - await completer.UpdateSession(conn, turnDetection: false); - - // Add dialog history - //foreach (var item in dialogs) - //{ - // await completer.InsertConversationItem(item); - //} + await _completer.UpdateSession(_conn, turnDetection: false); if (dialogs.LastOrDefault()?.Role == AgentRole.Assistant) { - await completer.TriggerModelInference($"Rephase your last response:\r\n{dialogs.LastOrDefault()?.Content}"); + await _completer.TriggerModelInference($"Rephase your last response:\r\n{dialogs.LastOrDefault()?.Content}"); } else { - await completer.TriggerModelInference("Reply based on the conversation context."); + await _completer.TriggerModelInference("Reply based on the conversation context."); } // Start turn detection await Task.Delay(1000 * 8); - await completer.UpdateSession(conn, turnDetection: true); + await _completer.UpdateSession(_conn, turnDetection: true); }, onModelAudioDeltaReceived: async (audioDeltaData, itemId) => { - var data = conn.OnModelMessageReceived(audioDeltaData); + var data = _conn.OnModelMessageReceived(audioDeltaData); await SendEventToUser(userWebSocket, data); // If this is the first delta of a new response, set the start timestamp - if (!conn.ResponseStartTimestamp.HasValue) + if (!_conn.ResponseStartTimestamp.HasValue) { - conn.ResponseStartTimestamp = conn.LatestMediaTimestamp; - _logger.LogDebug($"Setting start timestamp for new response: {conn.ResponseStartTimestamp}ms"); + _conn.ResponseStartTimestamp = _conn.LatestMediaTimestamp; + _logger.LogDebug($"Setting start timestamp for new response: {_conn.ResponseStartTimestamp}ms"); } // Record last assistant item ID for interruption handling if (!string.IsNullOrEmpty(itemId)) { - conn.LastAssistantItemId = itemId; + _conn.LastAssistantItemId = itemId; } // Send mark messages to Media Streams so we know if and when AI response playback is finished - await SendMark(userWebSocket, conn); + await SendMark(userWebSocket, _conn); }, onModelAudioResponseDone: async () => { - var data = conn.OnModelAudioResponseDone(); + var data = _conn.OnModelAudioResponseDone(); await SendEventToUser(userWebSocket, data); }, onAudioTranscriptDone: async transcript => @@ -144,36 +146,10 @@ public class RealtimeHub : IRealtimeHub foreach (var message in messages) { // Invoke function - if (message.MessageType == MessageTypeName.FunctionCall) + if (message.MessageType == MessageTypeName.FunctionCall && + !string.IsNullOrEmpty(message.FunctionName)) { await routing.InvokeFunction(message.FunctionName, message); - message.Role = AgentRole.Function; - - if (message.FunctionName == "route_to_agent") - { - var inst = JsonSerializer.Deserialize(message.FunctionArgs ?? "{}"); - message.Content = $"Connected to agent of {inst.AgentName}"; - conn.CurrentAgentId = routing.Context.GetCurrentAgentId(); - - await completer.UpdateSession(conn); - await completer.InsertConversationItem(message); - await completer.TriggerModelInference($"Guide the user through the next steps of the process as this Agent ({inst.AgentName}), following its instructions and operational procedures."); - } - else if (message.FunctionName == "util-routing-fallback_to_router") - { - var inst = JsonSerializer.Deserialize(message.FunctionArgs ?? "{}"); - message.Content = $"Returned to Router due to {inst.Reason}"; - conn.CurrentAgentId = routing.Context.GetCurrentAgentId(); - - await completer.UpdateSession(conn); - await completer.InsertConversationItem(message); - await completer.TriggerModelInference($"Check with user whether to proceed the new request: {inst.Reason}"); - } - else - { - await completer.InsertConversationItem(message); - await completer.TriggerModelInference("Reply based on the function's output."); - } } else { @@ -210,9 +186,9 @@ public class RealtimeHub : IRealtimeHub onUserInterrupted: async () => { // Reset states - conn.ResetResponseState(); + _conn.ResetResponseState(); - var data = conn.OnModelUserInterrupted(); + var data = _conn.OnModelUserInterrupted(); await SendEventToUser(userWebSocket, data); }); } @@ -232,17 +208,17 @@ public class RealtimeHub : IRealtimeHub } } - private async Task HandleUserDtmfReceived(IRealTimeCompletion completer, RealtimeHubConnection conn) + private async Task HandleUserDtmfReceived() { var routing = _services.GetRequiredService(); var hookProvider = _services.GetRequiredService(); var agentService = _services.GetRequiredService(); - var agent = await agentService.LoadAgent(conn.CurrentAgentId); + var agent = await agentService.LoadAgent(_conn.CurrentAgentId); var dialogs = routing.Context.GetDialogs(); var convService = _services.GetRequiredService(); - var conversation = await convService.GetConversation(conn.ConversationId); + var conversation = await convService.GetConversation(_conn.ConversationId); - var message = new RoleDialogModel(AgentRole.User, conn.Data) + var message = new RoleDialogModel(AgentRole.User, _conn.Data) { CurrentAgentId = routing.Context.GetCurrentAgentId() }; @@ -256,11 +232,11 @@ public class RealtimeHub : IRealtimeHub await hook.OnMessageReceived(message); } - await completer.InsertConversationItem(message); - await completer.TriggerModelInference("Reply based on the user input"); + await _completer.InsertConversationItem(message); + await _completer.TriggerModelInference("Reply based on the user input"); } - private async Task HandleUserDisconnected(RealtimeHubConnection conn) + private async Task HandleUserDisconnected() { // Save dialog history var routing = _services.GetRequiredService(); @@ -268,7 +244,7 @@ public class RealtimeHub : IRealtimeHub var dialogs = routing.Context.GetDialogs(); foreach (var item in dialogs) { - storage.Append(conn.ConversationId, item); + storage.Append(_conn.ConversationId, item); } } @@ -278,4 +254,20 @@ public class RealtimeHub : IRealtimeHub var buffer = Encoding.UTF8.GetBytes(data); await webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Text, true, CancellationToken.None); } + + public RealtimeHubConnection SetHubConnection(string conversationId) + { + _conn = new RealtimeHubConnection + { + ConversationId = conversationId + }; + + return _conn; + } + + public IRealTimeCompletion SetCompleter(string provider) + { + _completer = _services.GetServices().First(x => x.Provider == provider); + return _completer; + } } diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs index 6e314a99..323d0b62 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs @@ -550,7 +550,6 @@ public class RealTimeCompletionProvider : IRealTimeCompletion })); prompt += $"{verbose}\r\n"; - prompt += "\r\n[CONVERSATION]"; verbose = string.Join("\r\n", messages .Where(x => x as SystemChatMessage == null) .Select(x => @@ -581,7 +580,11 @@ public class RealTimeCompletionProvider : IRealTimeCompletion return string.Empty; })); - prompt += $"\r\n{verbose}\r\n"; + + if (!string.IsNullOrEmpty(verbose)) + { + prompt += $"\r\n[CONVERSATION]\r\n{verbose}\r\n"; + } } if (!options.Tools.IsNullOrEmpty()) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs index 0e3a68c0..35118ab3 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs @@ -51,12 +51,9 @@ public class TwilioStreamMiddleware private async Task HandleWebSocket(IServiceProvider services, string conversationId, WebSocket webSocket) { var hub = services.GetRequiredService(); - - var conn = new RealtimeHubConnection - { - ConversationId = conversationId - }; - + var conn = hub.SetHubConnection(conversationId); + var completer = hub.SetCompleter("openai"); + // load conversation and state var convService = services.GetRequiredService(); convService.SetConversationId(conversationId, []); @@ -131,8 +128,6 @@ public class TwilioStreamMiddleware @event = "clear", streamSid = response.StreamSid }; - - return conn; }); } }