Merge pull request #1029 from hchen2020/master

Oceania2018 · web-flow · commit a74a8dfe5905 · 2025-04-23T15:45:55.000-05:00
Support reconnect in Twilio
diff --git a/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs b/src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs
@@ -3,6 +3,7 @@ namespace BotSharp.Abstraction.Realtime.Models;
 public class RealtimeHubConnection
 {
     public string StreamId { get; set; } = null!;
+    public string UserSessionId {get;set;} = null!; 
     public string? LastAssistantItemId { get; set; } = null!;
     public long LatestMediaTimestamp { get; set; }
     public long? ResponseStartTimestamp { get; set; }
diff --git a/src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs b/src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs
@@ -57,7 +57,11 @@ public async Task OnFunctionExecuted(RoleDialogModel message)
             var instruction = await hub.Completer.UpdateSession(hub.HubConn);
             await hub.Completer.InsertConversationItem(message);
 
-            if (message.StopCompletion)
+            if (string.IsNullOrEmpty(message.Content))
+            {
+                return;
+            }
+            else if (message.StopCompletion)
             {
                 await hub.Completer.TriggerModelInference($"Say to user: \"{message.Content}\"");
             }
diff --git a/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs b/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs
@@ -28,23 +28,14 @@ public async Task ConnectToModel(Func<string, Task>? responseToUser = null, Func
         convService.SetConversationId(_conn.ConversationId, []);
         var conversation = await convService.GetConversation(_conn.ConversationId);
 
-        var agentService = _services.GetRequiredService<IAgentService>();
-        var agent = await agentService.LoadAgent(conversation.AgentId);
-        _conn.CurrentAgentId = agent.Id;
-
         var routing = _services.GetRequiredService<IRoutingService>();
-        routing.Context.Push(agent.Id);
+        var agentService = _services.GetRequiredService<IAgentService>();
+        var agent = await agentService.LoadAgent(_conn.CurrentAgentId);
 
         var storage = _services.GetRequiredService<IConversationStorage>();
         var dialogs = convService.GetDialogHistory();
-        if (dialogs.Count == 0)
-        {
-            dialogs.Add(new RoleDialogModel(AgentRole.User, "Hi"));
-            storage.Append(_conn.ConversationId, dialogs.First());
-        }
-
         routing.Context.SetDialogs(dialogs);
-        routing.Context.SetMessageId(_conn.ConversationId, dialogs.Last().MessageId);
+        routing.Context.SetMessageId(_conn.ConversationId, Guid.Empty.ToString());
 
         var states = _services.GetRequiredService<IConversationStateService>();
         var settings = _services.GetRequiredService<RealtimeModelSettings>();
diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs
@@ -187,8 +187,20 @@ private async Task ReceiveMessage(RealtimeHubConnection conn,
             else if (response.Type == "response.done")
             {
                 _logger.LogInformation($"{response.Type}: {receivedText}");
-                var messages = await OnResponsedDone(conn, receivedText);
-                onModelResponseDone(messages);
+                var data = JsonSerializer.Deserialize<ResponseDone>(receivedText).Body;
+                if (data.Status != "completed")
+                {
+                    if (data.StatusDetails.Type == "incomplete" && data.StatusDetails.Reason == "max_output_tokens")
+                    {
+                        onInterruptionDetected();
+                        await TriggerModelInference("Response user concisely");
+                    }
+                }
+                else
+                {
+                    var messages = await OnResponsedDone(conn, receivedText);
+                    onModelResponseDone(messages);
+                }
             }
             else if (response.Type == "conversation.item.created")
             {
@@ -295,6 +307,8 @@ await HookEmitter.Emit<IContentGeneratingHook>(_services, async hook =>
 
         await SendEventToModel(sessionUpdate);
 
+        await Task.Delay(300);
+
         return instruction;
     }
 
@@ -561,6 +575,10 @@ public async Task<List<RoleDialogModel>> OnResponsedDone(RealtimeHubConnection c
         if (data.Status != "completed")
         {
             _logger.LogError(data.StatusDetails.ToString());
+            /*if (data.StatusDetails.Type == "incomplete" && data.StatusDetails.Reason == "max_output_tokens")
+            {
+                await TriggerModelInference("Response user concisely");
+            }*/
             return [];
         }
 
diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs
@@ -0,0 +1,31 @@
+using BotSharp.Plugin.Twilio.Models;
+using Microsoft.AspNetCore.Mvc;
+
+namespace BotSharp.Plugin.Twilio.Controllers;
+
+public class TwilioReconnectController : TwilioController
+{
+    private readonly TwilioSetting _settings;
+    private readonly IServiceProvider _services;
+    private readonly ILogger _logger;
+
+    public TwilioReconnectController(IServiceProvider services, TwilioSetting settings, ILogger<TwilioReconnectController> logger)
+    {
+        _services = services;
+        _settings = settings;
+        _logger = logger;
+    }
+
+    [ValidateRequest]
+    [HttpPost("twilio/stream/reconnect")]
+    public async Task<TwiMLResult> Reconnect(ConversationalVoiceRequest request)
+    {
+        var response = new VoiceResponse();
+        var connect = new Connect();
+        var host = _settings.CallbackHost.Split("://").Last();
+        connect.Stream(url: $"wss://{host}/twilio/stream/{request.AgentId}/{request.ConversationId}");
+        response.Pause(1);
+        response.Append(connect);
+        return TwiML(response);
+    }
+}
diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Hooks/TwilioConversationHook.cs b/src/Plugins/BotSharp.Plugin.Twilio/Hooks/TwilioConversationHook.cs
@@ -0,0 +1,48 @@
+using BotSharp.Abstraction.Routing;
+using Task = System.Threading.Tasks.Task;
+using Twilio.Rest.Api.V2010.Account;
+using BotSharp.Plugin.Twilio.Interfaces;
+
+namespace BotSharp.Plugin.Twilio.Hooks;
+
+public class TwilioConversationHook : ConversationHookBase, IConversationHook
+{
+    private readonly IServiceProvider _services;
+    private readonly TwilioSetting _setting;
+    private readonly ILogger _logger;
+
+    public TwilioConversationHook(IServiceProvider services,
+        TwilioSetting setting,
+        ILogger<TwilioConversationHook> logger)
+    {
+        _services = services;
+        _setting = setting;
+        _logger = logger;
+    }
+
+    public override async Task OnFunctionExecuted(RoleDialogModel message)
+    {
+        var hooks = _services.GetServices<ITwilioSessionHook>();
+        foreach (var hook in hooks)
+        {
+            if (await hook.ShouldReconnect(message))
+            {
+                var states = _services.GetRequiredService<IConversationStateService>();
+                var sid = states.GetState("twilio_call_sid");
+
+                var routing = _services.GetRequiredService<IRoutingService>();
+                var conversationId = routing.Context.ConversationId;
+                var processUrl = $"{_setting.CallbackHost}/twilio/stream/reconnect?agent-id={message.CurrentAgentId}&conversation-id={conversationId}";
+
+                // Save all states before reconnect
+                states.Save();
+
+                CallResource.Update(
+                    pathSid: sid,
+                    url: new Uri(processUrl));
+
+                break;
+            }
+        }
+    }
+}
diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Interfaces/ITwilioSessionHook.cs b/src/Plugins/BotSharp.Plugin.Twilio/Interfaces/ITwilioSessionHook.cs
@@ -79,5 +79,13 @@ Task OnAgentHangUp(ConversationalVoiceRequest request)
     /// <param name="response"></param>
     /// <returns></returns>
     Task OnAgentTransferring(ConversationalVoiceRequest request, TwilioSetting settings)
-        => Task.CompletedTask;
+    => Task.CompletedTask;
+
+    /// <summary>
+    /// Allow Twilio to reconnect when it's in streaming mode.
+    /// </summary>
+    /// <param name="message"></param>
+    /// <returns></returns>
+    Task<bool> ShouldReconnect(RoleDialogModel message)
+        => Task.FromResult(false);
 }
diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs
@@ -240,7 +240,7 @@ public VoiceResponse ReturnBidirectionalMediaStreamsInstructions(ConversationalV
 
         var connect = new Connect();
         var host = _settings.CallbackHost.Split("://").Last();
-        connect.Stream(url: $"wss://{host}/twilio/stream/{conversationId}");
+        connect.Stream(url: $"wss://{host}/twilio/stream/{agent.Id}/{conversationId}");
         response.Append(connect);
 
         return response;
diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioPlugin.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioPlugin.cs
@@ -1,4 +1,5 @@
 using BotSharp.Abstraction.Settings;
+using BotSharp.Plugin.Twilio.Hooks;
 using BotSharp.Plugin.Twilio.Interfaces;
 using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.Hooks;
 using BotSharp.Plugin.Twilio.Services;
@@ -32,5 +33,6 @@ public void RegisterDI(IServiceCollection services, IConfiguration config)
         services.AddHostedService<TwilioMessageQueueService>();
         services.AddTwilioRequestValidation();
         services.AddScoped<IAgentUtilityHook, OutboundPhoneCallHandlerUtilityHook>();
+        services.AddScoped<IConversationHook, TwilioConversationHook>();
     }
 }
diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioStreamMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioStreamMiddleware.cs
@@ -34,11 +34,13 @@ public async Task Invoke(HttpContext httpContext)
             if (httpContext.WebSockets.IsWebSocketRequest)
             {
                 var services = httpContext.RequestServices;
-                var conversationId = request.Path.Value.Split("/").Last();
+                var parts = request.Path.Value.Split("/");
+                var agentId = parts[3];
+                var conversationId = parts[4];
                 using WebSocket webSocket = await httpContext.WebSockets.AcceptWebSocketAsync();
                 try
                 {
-                    await HandleWebSocket(services, conversationId, webSocket);
+                    await HandleWebSocket(services, agentId, conversationId, webSocket);
                 }
                 catch (Exception ex)
                 {
@@ -51,12 +53,13 @@ public async Task Invoke(HttpContext httpContext)
         await _next(httpContext);
     }
 
-    private async Task HandleWebSocket(IServiceProvider services, string conversationId, WebSocket webSocket)
+    private async Task HandleWebSocket(IServiceProvider services, string agentId, string conversationId, WebSocket webSocket)
     {
         var settings = services.GetRequiredService<RealtimeModelSettings>();
         var hub = services.GetRequiredService<IRealtimeHub>();
         var conn = hub.SetHubConnection(conversationId);
-        
+        conn.CurrentAgentId = agentId;
+
         // load conversation and state
         var convService = services.GetRequiredService<IConversationService>();
         convService.SetConversationId(conversationId, []);
@@ -67,6 +70,9 @@ private async Task HandleWebSocket(IServiceProvider services, string conversatio
         }
         convService.States.Save();
 
+        var routing = services.GetRequiredService<IRoutingService>();
+        routing.Context.Push(agentId);
+
         var buffer = new byte[1024 * 32];
         WebSocketReceiveResult result;
 
@@ -136,6 +142,7 @@ await hub.ConnectToModel(async data =>
             case "start":
                 eventType = "user_connected";
                 var startResponse = JsonSerializer.Deserialize<StreamEventStartResponse>(receivedText);
+                conn.UserSessionId = startResponse.Body.CallSid;
                 data = JsonSerializer.Serialize(startResponse.Body.CustomParameters);
                 conn.ResetStreamState();
                 break;

Original file line number	Diff line number	Diff line change
`@@ -3,6 +3,7 @@ namespace BotSharp.Abstraction.Realtime.Models;`
`3`	`3`	`public class RealtimeHubConnection`
`4`	`4`	`{`
`5`	`5`	`public string StreamId { get; set; } = null!;`
	`6`	`+ public string UserSessionId {get;set;} = null!;`
`6`	`7`	`public string? LastAssistantItemId { get; set; } = null!;`
`7`	`8`	`public long LatestMediaTimestamp { get; set; }`
`8`	`9`	`public long? ResponseStartTimestamp { get; set; }`
Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,11 @@ public async Task OnFunctionExecuted(RoleDialogModel message)`
`57`	`57`	`var instruction = await hub.Completer.UpdateSession(hub.HubConn);`
`58`	`58`	`await hub.Completer.InsertConversationItem(message);`
`59`	`59`
`60`		`- if (message.StopCompletion)`
	`60`	`+ if (string.IsNullOrEmpty(message.Content))`
	`61`	`+ {`
	`62`	`+ return;`
	`63`	`+ }`
	`64`	`+ else if (message.StopCompletion)`
`61`	`65`	`{`
`62`	`66`	`await hub.Completer.TriggerModelInference($"Say to user: \"{message.Content}\"");`
`63`	`67`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`using BotSharp.Abstraction.Settings;`
	`2`	`+using BotSharp.Plugin.Twilio.Hooks;`
`2`	`3`	`using BotSharp.Plugin.Twilio.Interfaces;`
`3`	`4`	`using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.Hooks;`
`4`	`5`	`using BotSharp.Plugin.Twilio.Services;`
`@@ -32,5 +33,6 @@ public void RegisterDI(IServiceCollection services, IConfiguration config)`
`32`	`33`	`services.AddHostedService<TwilioMessageQueueService>();`
`33`	`34`	`services.AddTwilioRequestValidation();`
`34`	`35`	`services.AddScoped<IAgentUtilityHook, OutboundPhoneCallHandlerUtilityHook>();`
	`36`	`+ services.AddScoped<IConversationHook, TwilioConversationHook>();`
`35`	`37`	`}`
`36`	`38`	`}`