diff --git a/src/Infrastructure/BotSharp.Abstraction/MLTasks/Settings/LlmModelSetting.cs b/src/Infrastructure/BotSharp.Abstraction/MLTasks/Settings/LlmModelSetting.cs index 111442367..dccd20377 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MLTasks/Settings/LlmModelSetting.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MLTasks/Settings/LlmModelSetting.cs @@ -154,6 +154,23 @@ public class ImageVariationSetting public class AudioSetting { public AudioTranscriptionSetting? Transcription { get; set; } + public AudioSynthesisSetting? Synthesis { get; set; } +} + +/// +/// Audio synthesis (text-to-speech) setting +/// +public class AudioSynthesisSetting +{ + /// + /// Timeout of a single request attempt, in seconds + /// + public int? TimeoutSeconds { get; set; } + + /// + /// Number of retries after the first attempt fails or times out + /// + public int? MaxRetries { get; set; } } /// diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Audio/AudioSynthesisProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Audio/AudioSynthesisProvider.cs index 10341abe3..6770fd037 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Audio/AudioSynthesisProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Audio/AudioSynthesisProvider.cs @@ -21,9 +21,20 @@ public void SetModelName(string model) _model = model; } + // A short phrase normally comes back in a few seconds, but some requests never finish; + // with the SDK default of 100s per attempt, four stalled attempts block the caller for ~7 minutes. + private const int DefaultTimeoutSeconds = 30; + private const int DefaultMaxRetries = 3; + public async Task GenerateAudioAsync(string text, string? voice = "alloy", string? format = "mp3", string? instructions = null) { - var audioClient = ProviderHelper.GetClient(Provider, _model, apiKey: null, _services) + var settingsService = _services.GetRequiredService(); + var synthesis = settingsService.GetSetting(Provider, _model)?.Audio?.Synthesis; + var timeout = TimeSpan.FromSeconds(synthesis?.TimeoutSeconds ?? DefaultTimeoutSeconds); + var maxRetries = synthesis?.MaxRetries ?? DefaultMaxRetries; + + var audioClient = ProviderHelper.GetClient(Provider, _model, apiKey: null, _services, + networkTimeout: timeout, maxRetries: maxRetries) .GetAudioClient(_model); var (speechVoice, options) = PrepareGenerationOptions(voice: voice, format: format); diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/ProviderHelper.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/ProviderHelper.cs index df937390d..7900afd2d 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/ProviderHelper.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/ProviderHelper.cs @@ -1,16 +1,30 @@ using OpenAI; using System.ClientModel; +using System.ClientModel.Primitives; namespace BotSharp.Plugin.OpenAI.Providers; public class ProviderHelper { - public static OpenAIClient GetClient(string provider, string model, string? apiKey, IServiceProvider services) + public static OpenAIClient GetClient(string provider, string model, string? apiKey, IServiceProvider services, + TimeSpan? networkTimeout = null, int? maxRetries = null) { var settingsService = services.GetRequiredService(); var settings = settingsService.GetSetting(provider, model); - var options = !string.IsNullOrEmpty(settings?.Endpoint) ? - new OpenAIClientOptions { Endpoint = new Uri(settings.Endpoint) } : null; + var options = new OpenAIClientOptions(); + if (!string.IsNullOrEmpty(settings?.Endpoint)) + { + options.Endpoint = new Uri(settings.Endpoint); + } + // Left unset, the SDK defaults apply: 100s per attempt and 3 retries. + if (networkTimeout.HasValue) + { + options.NetworkTimeout = networkTimeout.Value; + } + if (maxRetries.HasValue) + { + options.RetryPolicy = new ClientRetryPolicy(maxRetries.Value); + } return new OpenAIClient(new ApiKeyCredential(apiKey ?? settings!.ApiKey), options); } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs new file mode 100644 index 000000000..46a4bd17a --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs @@ -0,0 +1,89 @@ +using BotSharp.Abstraction.Infrastructures; +using BotSharp.Core.Infrastructures; +using BotSharp.Plugin.Twilio.Interfaces; +using BotSharp.Plugin.Twilio.Models.Relay; +using BotSharp.Plugin.Twilio.Services; +using Microsoft.AspNetCore.Mvc; + +namespace BotSharp.Plugin.Twilio.Controllers; + +public class TwilioConversationRelayController : TwilioController +{ + private readonly TwilioSetting _settings; + private readonly IServiceProvider _services; + private readonly ILogger _logger; + + public TwilioConversationRelayController( + TwilioSetting settings, + IServiceProvider services, + ILogger logger) + { + _settings = settings; + _services = services; + _logger = logger; + } + + /// + /// Twilio calls this when a ConversationRelay session ends: we sent "end" (handoff), the caller hung up, + /// or the session failed. + /// + [ValidateRequest] + [HttpPost("twilio/relay/action")] + public async Task SessionEnded(ConversationRelayActionRequest request) + { + _logger.LogInformation($"ConversationRelay session for conversation {request.ConversationId} ended: " + + $"{request.SessionStatus} {request.ErrorCode} {request.ErrorMessage}"); + + // Authenticate (synchronous, so identity writes survive into the async session hooks) + HookEmitter.Emit(_services, hook => hook.OnAuthenticate(request), request.AgentId); + + var twilio = _services.GetRequiredService(); + var action = GetHandoffAction(request.HandoffData); + + if (action == ConversationRelayHandoff.Transfer && !string.IsNullOrEmpty(_settings.CsrAgentNumber)) + { + await HookEmitter.Emit(_services, + async hook => await hook.OnAgentTransferring(request, _settings), request.AgentId); + return TwiML(twilio.DialCsrAgent(string.Empty)); + } + + var callStillUp = !string.Equals(request.CallStatus, "completed", StringComparison.OrdinalIgnoreCase); + var canReconnect = request.Reconnects < _settings.ConversationRelay.MaxReconnects; + if (request.SessionStatus == "failed" && callStillUp && canReconnect) + { + _logger.LogWarning($"Reconnecting ConversationRelay for conversation {request.ConversationId} " + + $"(attempt {request.Reconnects + 1})."); + var relay = twilio.ReturnConversationRelay(request.AgentId, request.ConversationId, + welcomeGreeting: null, request.Reconnects + 1); + return TwiML(relay); + } + + if (action == ConversationRelayHandoff.HangUp) + { + await HookEmitter.Emit(_services, + async hook => await hook.OnAgentHangUp(request), request.AgentId); + } + + var response = new VoiceResponse(); + response.Hangup(); + return TwiML(response); + } + + private static string? GetHandoffAction(string? handoffData) + { + if (string.IsNullOrWhiteSpace(handoffData)) + { + return null; + } + + try + { + using var doc = JsonDocument.Parse(handoffData); + return doc.RootElement.TryGetProperty("action", out var action) ? action.GetString() : null; + } + catch (JsonException) + { + return handoffData; + } + } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs index a63e776d7..7c1f083e4 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs @@ -84,8 +84,8 @@ await HookEmitter.Emit(_services, async hook => await HookEmitter.Emit(_services, async hook => await hook.OnVoicemailStarting(request), request.AgentId); - var url = twilio.GetSpeechPath(request.ConversationId, "voicemail.mp3"); - response.Play(new Uri(url)); + var voicemail = twilio.GetSavedSpeechFile(request.ConversationId, "voicemail"); + twilio.AppendSpeech(response, request.ConversationId, voicemail); } else { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs index f355d77ce..7049aeeb7 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs @@ -38,8 +38,20 @@ public async Task InitiateOutboundCall(ConversationalVoiceRequest r await HookEmitter.Emit(_services, async hook => await hook.OnVoicemailStarting(request), request.AgentId); - var url = twilio.GetSpeechPath(request.ConversationId, "voicemail.mp3"); - response.Play(new Uri(url)); + var voicemail = twilio.GetSavedSpeechFile(request.ConversationId, "voicemail"); + twilio.AppendSpeech(response, request.ConversationId, voicemail); + } + else if (_settings.ConversationRelay.IsEnabledFor(request.AgentId)) + { + // OutboundPhoneCallFn stored the initial message as reply 0 + var sessionManager = _services.GetRequiredService(); + var initial = await sessionManager.GetAssistantReplyAsync(request.ConversationId, 0); + if (string.IsNullOrWhiteSpace(initial?.Content)) + { + _logger.LogWarning($"No initial message for conversation {request.ConversationId}; " + + "starting ConversationRelay without a greeting."); + } + response = twilio.ReturnConversationRelay(request.AgentId, request.ConversationId, initial?.Content); } else { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs index 6daf19589..b56e5c7f3 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs @@ -28,8 +28,7 @@ public async Task Reconnect(ConversationalVoiceRequest request) if (!string.IsNullOrEmpty(request.InitAudioFile)) { var twilio = _services.GetRequiredService(); - var audioUrl = twilio.GetSpeechPath(request.ConversationId, request.InitAudioFile); - response.Play(new Uri(audioUrl)); + twilio.AppendSpeech(response, request.ConversationId, request.InitAudioFile); } else { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs new file mode 100644 index 000000000..f5735f168 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs @@ -0,0 +1,24 @@ +using Task = System.Threading.Tasks.Task; + +namespace BotSharp.Plugin.Twilio.Models; + +/// +/// What a phone flow wants to hear about while the agent works on one caller message. +/// +public class AgentTurnCallbacks +{ + /// + /// A tool-call indication, e.g. "#hold-on-1" or a sentence to say while the tool runs + /// + public Func? OnIndication { get; init; } + + /// + /// When set, the turn streams: called for each piece of reply text as the model writes it + /// + public Func? OnStreamText { get; init; } + + /// + /// Called when the model starts writing a new response, e.g. after a tool call + /// + public Func? OnStreamStarted { get; init; } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayActionRequest.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayActionRequest.cs new file mode 100644 index 000000000..10228123b --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayActionRequest.cs @@ -0,0 +1,33 @@ +using Microsoft.AspNetCore.Mvc; + +namespace BotSharp.Plugin.Twilio.Models.Relay; + +/// +/// What Twilio posts to the Connect action URL when a ConversationRelay session ends +/// +public class ConversationRelayActionRequest : ConversationalVoiceRequest +{ + [FromForm] + public string? SessionId { get; set; } + + /// + /// ended (we sent "end"), completed (caller hung up) or failed + /// + [FromForm] + public string? SessionStatus { get; set; } + + [FromForm] + public string? SessionDuration { get; set; } + + [FromForm] + public string? HandoffData { get; set; } + + [FromForm] + public string? ErrorCode { get; set; } + + [FromForm] + public string? ErrorMessage { get; set; } + + [FromQuery(Name = "reconnects")] + public int Reconnects { get; set; } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayHandoff.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayHandoff.cs new file mode 100644 index 000000000..1338b10df --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayHandoff.cs @@ -0,0 +1,10 @@ +namespace BotSharp.Plugin.Twilio.Models.Relay; + +/// +/// Values we put in the "end" message's handoffData, read back by the Connect action callback +/// +public static class ConversationRelayHandoff +{ + public const string HangUp = "hangup"; + public const string Transfer = "transfer"; +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs new file mode 100644 index 000000000..b8b78f40c --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs @@ -0,0 +1,9 @@ +namespace BotSharp.Plugin.Twilio.Models.Relay; + +/// +/// Custom TwiML Parameter names passed into the ConversationRelay "setup" message +/// +public static class ConversationRelayParameters +{ + public const string WelcomeGreetingLength = "welcome-greeting-length"; +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/HangupPhoneCallFn.cs b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/HangupPhoneCallFn.cs index 3bf6fb906..aea4d2df2 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/HangupPhoneCallFn.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/HangupPhoneCallFn.cs @@ -2,6 +2,7 @@ using BotSharp.Abstraction.Routing; using BotSharp.Core.Infrastructures; using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.LlmContexts; +using BotSharp.Plugin.Twilio.Services; using Twilio.Rest.Api.V2010.Account; namespace BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.Functions; @@ -28,7 +29,6 @@ public async Task Execute(RoleDialogModel message) { var args = JsonSerializer.Deserialize(message.FunctionArgs); - var fileStorage = _services.GetRequiredService(); var convService = _services.GetRequiredService(); var conversationId = convService.ConversationId; var states = _services.GetRequiredService(); @@ -44,13 +44,10 @@ public async Task Execute(RoleDialogModel message) var processUrl = $"{_twilioSetting.CallbackHost}/twilio/voice/hang-up?agent-id={message.CurrentAgentId}&conversation-id={conversationId}"; // Generate initial assistant audio - string initAudioFile = null; if (!string.IsNullOrEmpty(args.ResponseContent) && _twilioSetting.GenerateEndingAudio) { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(args.ResponseContent); - initAudioFile = "ending.mp3"; - fileStorage.SaveSpeechFile(conversationId, initAudioFile, data); + var twilio = _services.GetRequiredService(); + var initAudioFile = await twilio.SaveSpeechAsync(conversationId, "ending", args.ResponseContent); processUrl += $"&init-audio-file={initAudioFile}"; } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/LeaveVoicemailFn.cs b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/LeaveVoicemailFn.cs index 96aa55809..77a02b777 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/LeaveVoicemailFn.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/LeaveVoicemailFn.cs @@ -2,6 +2,7 @@ using BotSharp.Abstraction.Routing; using BotSharp.Core.Infrastructures; using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.LlmContexts; +using BotSharp.Plugin.Twilio.Services; namespace BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.Functions; @@ -28,7 +29,6 @@ public async Task Execute(RoleDialogModel message) { var args = JsonSerializer.Deserialize(message.FunctionArgs); - var fileStorage = _services.GetRequiredService(); var routing = _services.GetRequiredService(); var conversationId = routing.Context.ConversationId; var states = _services.GetRequiredService(); @@ -42,13 +42,10 @@ public async Task Execute(RoleDialogModel message) } // Generate voice message audio - string initAudioFile = null; if (!string.IsNullOrEmpty(args.VoicemailMessage)) { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(args.VoicemailMessage); - initAudioFile = "voicemail.mp3"; - fileStorage.SaveSpeechFile(conversationId, initAudioFile, data); + var twilio = _services.GetRequiredService(); + await twilio.SaveSpeechAsync(conversationId, "voicemail", args.VoicemailMessage); } message.Content = args.VoicemailMessage; diff --git a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/OutboundPhoneCallFn.cs b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/OutboundPhoneCallFn.cs index c0003b785..b49a10b16 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/OutboundPhoneCallFn.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/OutboundPhoneCallFn.cs @@ -9,6 +9,7 @@ using BotSharp.Plugin.Twilio.Interfaces; using BotSharp.Plugin.Twilio.Models; using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.LlmContexts; +using BotSharp.Plugin.Twilio.Services; using Twilio.Exceptions; using Twilio.Rest.Api.V2010.Account; using Twilio.TwiML.Messaging; @@ -62,7 +63,6 @@ public async Task Execute(RoleDialogModel message) return false; } - var fileStorage = _services.GetRequiredService(); var states = _services.GetRequiredService(); // Fork conversation @@ -73,24 +73,24 @@ public async Task Execute(RoleDialogModel message) var statusUrl = $"{_twilioSetting.CallbackHost}/twilio/voice/status?agent-id={message.CurrentAgentId}&conversation-id={newConversationId}"; var recordingStatusUrl = $"{_twilioSetting.CallbackHost}/twilio/recording/status?agent-id={message.CurrentAgentId}&conversation-id={newConversationId}"; - // Generate initial assistant audio + // load agent profile + var agentService = _services.GetRequiredService(); + var agent = await agentService.GetAgent(message.CurrentAgentId); + var realtime = agent.Labels.Contains("realtime"); + + // Generate initial assistant audio; on ConversationRelay Twilio speaks the initial message as the welcome greeting string initAudioFile = null; - if (!string.IsNullOrEmpty(args.InitialMessage)) + var useConversationRelay = !realtime && _twilioSetting.ConversationRelay.IsEnabledFor(message.CurrentAgentId); + if (!string.IsNullOrEmpty(args.InitialMessage) && !useConversationRelay) { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(args.InitialMessage); - initAudioFile = "intial.mp3"; - fileStorage.SaveSpeechFile(newConversationId, initAudioFile, data); + var twilio = _services.GetRequiredService(); + initAudioFile = await twilio.SaveSpeechAsync(newConversationId, "intial", args.InitialMessage); statusUrl += $"&init-audio-file={initAudioFile}"; } - // load agent profile - var agentService = _services.GetRequiredService(); - var agent = await agentService.GetAgent(message.CurrentAgentId); - // Set up process URL streaming or synchronous - if (agent.Labels.Contains("realtime")) + if (realtime) { processUrl += "/inbound"; } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/TransferPhoneCallFn.cs b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/TransferPhoneCallFn.cs index a5c644234..71aa9c5a5 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/TransferPhoneCallFn.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/TransferPhoneCallFn.cs @@ -3,6 +3,7 @@ using BotSharp.Abstraction.Routing; using BotSharp.Core.Infrastructures; using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.LlmContexts; +using BotSharp.Plugin.Twilio.Services; using Twilio.Rest.Api.V2010.Account; namespace BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.Functions; @@ -33,7 +34,6 @@ public async Task Execute(RoleDialogModel message) { var args = JsonSerializer.Deserialize(message.FunctionArgs, _options.JsonSerializerOptions); - var fileStorage = _services.GetRequiredService(); var states = _services.GetRequiredService(); var sid = states.GetState("twilio_call_sid"); if (string.IsNullOrEmpty(sid)) @@ -50,10 +50,8 @@ public async Task Execute(RoleDialogModel message) // Generate initial assistant audio if (!string.IsNullOrEmpty(args.TransitionMessage)) { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(args.TransitionMessage); - var initAudioFile = "transfer.mp3"; - fileStorage.SaveSpeechFile(conversationId, initAudioFile, data); + var twilio = _services.GetRequiredService(); + var initAudioFile = await twilio.SaveSpeechAsync(conversationId, "transfer", args.TransitionMessage); processUrl += $"&init-audio-file={initAudioFile}"; } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs new file mode 100644 index 000000000..61f204ff4 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -0,0 +1,648 @@ +using BotSharp.Plugin.Twilio.Models; +using BotSharp.Plugin.Twilio.Models.Relay; +using System.IO; +using System.Net.WebSockets; +using System.Threading; +using Task = System.Threading.Tasks.Task; + +namespace BotSharp.Plugin.Twilio.Services; + +/// +/// One phone call on Twilio ConversationRelay: Twilio sends what the caller said as text and speaks the text we send back. +/// https://www.twilio.com/docs/voice/conversationrelay/websocket-messages +/// +public class ConversationRelaySession +{ + private const string DefaultSayItAgainText = "Sorry, can you say it again?"; + private const string DefaultErrorText = "Sorry, something went wrong. Please call back later."; + private const double SpokenCharsPerSecond = 15; + + private readonly WebSocket _socket; + private readonly string _agentId; + private readonly string _conversationId; + private readonly TwilioSetting _settings; + private readonly Func> _sendCallerMessage; + private readonly ILogger _logger; + + private readonly SemaphoreSlim _sendLock = new(1, 1); + private readonly object _sync = new(); + private readonly Queue _pendingInputs = new(); + private readonly StringBuilder _digits = new(); + private readonly StringBuilder _partialPrompt = new(); + private bool _isProcessing; + private Task _processing = Task.CompletedTask; + private int _seqNum; + private int _idlePrompts; + private bool _ended; + + // Bumped on every caller input, so a decision taken during silence can tell the caller spoke since + private int _activity; + + // Caller silence is measured from this moment: the estimated end of what we last said, or the caller's last input + private DateTime _quietSinceUtc = DateTime.MaxValue; + + // Streaming of the current turn's reply (ConversationRelay.StreamReplies) + private readonly StringBuilder _streamSegment = new(); + private string _lastStreamSegment = string.Empty; + private bool _utteranceOpen; + private bool _streamedThisTurn; + private bool _dropStream; + + private string? _callSid; + private string? _from; + private string? _to; + private string? _direction; + + /// + /// Sends the caller's words to the agent and returns its reply; the callbacks receive tool-call indications + /// and the streamed reply + /// + public ConversationRelaySession( + WebSocket socket, + string agentId, + string conversationId, + TwilioSetting settings, + Func> sendCallerMessage, + ILogger logger) + { + _socket = socket; + _agentId = agentId; + _conversationId = conversationId; + _settings = settings; + _sendCallerMessage = sendCallerMessage; + _logger = logger; + } + + public async Task RunAsync(CancellationToken cancellationToken) + { + using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var idleWatch = WatchIdleAsync(cts.Token); + try + { + await ReceiveLoopAsync(cts.Token); + } + catch (Exception ex) when (ex is WebSocketException or OperationCanceledException) + { + _logger.LogInformation($"ConversationRelay socket closed for conversation {_conversationId}: {ex.Message}"); + } + finally + { + // The call is gone: stop taking turns before waiting on anything, so a turn that finishes + // meanwhile cannot start another one from the queue + int dropped; + lock (_sync) + { + _ended = true; + dropped = _pendingInputs.Count; + _pendingInputs.Clear(); + } + + cts.Cancel(); + await SwallowAsync(idleWatch); + + if (dropped > 0) + { + _logger.LogInformation($"Dropped {dropped} caller input(s) received after the agent's last turn started; " + + $"conversation {_conversationId} has ended."); + } + } + + // Let a turn already running finish so the conversation is saved + await SwallowAsync(_processing); + } + + private async Task ReceiveLoopAsync(CancellationToken cancellationToken) + { + var buffer = new byte[16 * 1024]; + using var message = new MemoryStream(); + while (_socket.State == WebSocketState.Open && !cancellationToken.IsCancellationRequested) + { + var result = await _socket.ReceiveAsync(new ArraySegment(buffer), cancellationToken); + if (result.MessageType == WebSocketMessageType.Close) + { + if (_socket.State == WebSocketState.CloseReceived) + { + await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None); + } + break; + } + + message.Write(buffer, 0, result.Count); + if (!result.EndOfMessage) + { + continue; + } + + var json = Encoding.UTF8.GetString(message.ToArray()); + message.SetLength(0); + try + { + await HandleMessageAsync(json); + } + catch (JsonException ex) + { + _logger.LogWarning($"Unreadable ConversationRelay message for conversation {_conversationId}: {ex.Message}"); + } + } + } + + private async Task HandleMessageAsync(string json) + { + using var doc = JsonDocument.Parse(json); + var root = doc.RootElement; + switch (GetString(root, "type")) + { + case "setup": + _callSid = GetString(root, "callSid"); + _from = GetString(root, "from"); + _to = GetString(root, "to"); + _direction = GetString(root, "direction"); + var greetingLength = 0; + if (root.TryGetProperty("customParameters", out var parameters) + && GetString(parameters, ConversationRelayParameters.WelcomeGreetingLength) is string length) + { + int.TryParse(length, out greetingLength); + } + lock (_sync) + { + _quietSinceUtc = DateTime.UtcNow + EstimateSpeech(greetingLength); + } + _logger.LogInformation( + $"ConversationRelay session started for conversation {_conversationId} (call {_callSid})."); + break; + + case "prompt": + var text = GetString(root, "voicePrompt") ?? string.Empty; + var last = !root.TryGetProperty("last", out var lastElement) || lastElement.ValueKind != JsonValueKind.False; + _partialPrompt.Append(text); + if (last) + { + var prompt = _partialPrompt.ToString().Trim(); + _partialPrompt.Clear(); + if (prompt.Length > 0) + { + Enqueue(prompt, digits: null); + } + } + break; + + case "dtmf": + var digit = GetString(root, "digit"); + if (digit == "#") + { + var digits = _digits.ToString(); + _digits.Clear(); + if (digits.Length > 0) + { + Enqueue(digits, digits); + } + } + else if (!string.IsNullOrEmpty(digit)) + { + _digits.Append(digit); + lock (_sync) + { + _quietSinceUtc = DateTime.UtcNow; + _activity++; + } + } + break; + + case "interrupt": + // Playback stopped where the caller cut in; the caller's words follow as a prompt + lock (_sync) + { + _quietSinceUtc = DateTime.UtcNow; + + // The caller talked over a reply still being streamed: don't send the rest of it + if (_isProcessing && _streamedThisTurn) + { + _dropStream = true; + } + } + var interruptedAfter = GetString(root, "durationUntilInterruptMs"); + _logger.LogInformation( + $"Caller interrupted the agent in conversation {_conversationId} after {interruptedAfter} ms."); + break; + + case "error": + _logger.LogWarning( + $"ConversationRelay error in conversation {_conversationId}: {GetString(root, "description")}"); + break; + } + + await Task.CompletedTask; + } + + private void Enqueue(string content, string? digits) + { + lock (_sync) + { + if (_ended) + { + return; + } + + _pendingInputs.Enqueue(new PendingInput(content, digits)); + _quietSinceUtc = DateTime.UtcNow; + _idlePrompts = 0; + _activity++; + + if (!_isProcessing) + { + _isProcessing = true; + _processing = Task.Run(ProcessPendingAsync); + } + } + } + + /// + /// Runs one agent turn at a time; what the caller says meanwhile is merged into the next turn. + /// A keypad entry ends the merge, so each entry keeps its own Digits. + /// + private async Task ProcessPendingAsync() + { + while (true) + { + CallerMessage message; + lock (_sync) + { + if (_ended || _pendingInputs.Count == 0) + { + _isProcessing = false; + return; + } + + var batch = new List(); + while (_pendingInputs.TryDequeue(out var input)) + { + batch.Add(input); + if (input.Digits != null) + { + break; + } + } + + var inbound = string.Equals(_direction, "inbound", StringComparison.OrdinalIgnoreCase); + message = new CallerMessage + { + AgentId = _agentId, + ConversationId = _conversationId, + SeqNumber = ++_seqNum, + Content = string.Join("\r\n", batch.Select(x => x.Content)), + Digits = batch[^1].Digits, + From = (inbound ? _from : _to) ?? string.Empty + }; + + _streamSegment.Clear(); + _lastStreamSegment = string.Empty; + _utteranceOpen = false; + _streamedThisTurn = false; + _dropStream = false; + } + + var stream = _settings.ConversationRelay.StreamReplies; + var callbacks = new AgentTurnCallbacks + { + OnIndication = SpeakIndicationAsync, + OnStreamText = stream ? StreamTextAsync : null, + OnStreamStarted = stream ? CloseUtteranceAsync : null + }; + + AssistantMessage? reply; + try + { + // Run the turn outside the WebSocket request's execution context; it sets up its own HttpContext + Task turn; + using (ExecutionContext.SuppressFlow()) + { + turn = Task.Run(() => _sendCallerMessage(message, callbacks)); + } + + var timeout = Task.Delay(TimeSpan.FromSeconds(_settings.ConversationRelay.AiResponseTimeoutSeconds)); + if (await Task.WhenAny(turn, timeout) != turn) + { + var timeoutSeconds = _settings.ConversationRelay.AiResponseTimeoutSeconds; + _logger.LogWarning($"Agent did not reply within {timeoutSeconds}s in conversation {_conversationId}."); + await FailAsync(); + return; + } + reply = await turn; + } + catch (Exception ex) + { + _logger.LogError(ex, $"Agent turn failed in conversation {_conversationId}."); + await FailAsync(); + return; + } + + await CloseUtteranceAsync(); + if (reply == null + || string.IsNullOrWhiteSpace(reply.Content) && !reply.ConversationEnd && !reply.HumanIntervationNeeded) + { + continue; + } + + await SpeakAsync(GetUnstreamedText(reply.Content)); + if (reply.HumanIntervationNeeded) + { + await EndAsync(ConversationRelayHandoff.Transfer); + return; + } + if (reply.ConversationEnd) + { + await EndAsync(ConversationRelayHandoff.HangUp); + return; + } + } + } + + private async Task WatchIdleAsync(CancellationToken cancellationToken) + { + var relay = _settings.ConversationRelay; + var idleTimeout = TimeSpan.FromSeconds(relay.IdleTimeoutSeconds); + while (!cancellationToken.IsCancellationRequested) + { + await Task.Delay(TimeSpan.FromMilliseconds(500), cancellationToken); + + var askAgain = false; + var hangUp = false; + var activity = 0; + lock (_sync) + { + var quietFor = DateTime.UtcNow - _quietSinceUtc; + if (_ended || _isProcessing || _quietSinceUtc == DateTime.MaxValue || quietFor < idleTimeout) + { + continue; + } + + if (_idlePrompts < relay.MaxIdlePrompts) + { + _idlePrompts++; + askAgain = true; + activity = _activity; + } + else + { + // Decide and end in one step, so caller input arriving now cannot start a turn on a call being hung up + _ended = true; + hangUp = true; + } + _quietSinceUtc = DateTime.UtcNow; + } + + if (askAgain && IsStillQuiet(activity)) + { + await SpeakPromptAsync($"say-it-again-{Random.Shared.Next(1, 5)}", DefaultSayItAgainText); + } + else if (hangUp) + { + _logger.LogInformation($"Caller stayed silent, hanging up conversation {_conversationId}."); + await SendEndAsync(ConversationRelayHandoff.HangUp); + } + } + } + + private async Task FailAsync() + { + // A turn that timed out may still be streaming; keep its words out of the apology + lock (_sync) + { + _dropStream = true; + } + await CloseUtteranceAsync(); + await SpeakPromptAsync("error", DefaultErrorText); + await EndAsync(ConversationRelayHandoff.HangUp); + } + + /// + /// Sends a piece of the reply as the model writes it; Twilio speaks it as it arrives. + /// + private async Task StreamTextAsync(string text) + { + lock (_sync) + { + if (_ended || _dropStream) + { + return; + } + _streamSegment.Append(text); + _utteranceOpen = true; + _streamedThisTurn = true; + } + + await SendAsync(new { type = "text", token = text, last = false }); + } + + /// + /// Ends the utterance being streamed, so what follows (an indication, the next response) is spoken on its own. + /// + private async Task CloseUtteranceAsync() + { + int length; + lock (_sync) + { + if (!_utteranceOpen) + { + return; + } + _utteranceOpen = false; + _lastStreamSegment = _streamSegment.ToString(); + _streamSegment.Clear(); + length = _lastStreamSegment.Length; + } + + await SendAsync(new { type = "text", token = string.Empty, last = true }); + MarkSpeaking(EstimateSpeech(length)); + } + + /// + /// The part of the final reply that was not streamed. Normally nothing; the whole reply when it is not + /// what the model streamed last, e.g. text set by a tool, or a reply rewritten after the model wrote it. + /// + private string GetUnstreamedText(string? reply) + { + var final = reply?.Trim() ?? string.Empty; + string streamed; + lock (_sync) + { + if (!_streamedThisTurn) + { + return final; + } + if (_dropStream) + { + // The caller cut in; their words come next as a new turn + return string.Empty; + } + streamed = _lastStreamSegment.Trim(); + } + + if (final == streamed) + { + return string.Empty; + } + if (streamed.Length > 0 && final.StartsWith(streamed, StringComparison.Ordinal)) + { + return final.Substring(streamed.Length); + } + + _logger.LogInformation( + $"Reply in conversation {_conversationId} differs from what was streamed; speaking it in full."); + return final; + } + + /// + /// Speaks a tool-call indication: "#name" is a prerecorded prompt, anything else is spoken as is; + /// segments are separated by "|". + /// + private async Task SpeakIndicationAsync(string indication) + { + await CloseUtteranceAsync(); + foreach (var segment in indication.Split('|')) + { + var text = segment.Trim(); + if (text.StartsWith('#')) + { + await SpeakPromptAsync(text.Substring(1), fallbackText: null); + } + else + { + await SpeakAsync(text); + } + } + } + + /// + /// Speaks a prerecorded prompt's configured text so the voice matches; without one, plays the recording. + /// + private async Task SpeakPromptAsync(string name, string? fallbackText) + { + if (_settings.PromptTexts.TryGetValue(name, out var text) && !string.IsNullOrWhiteSpace(text)) + { + await SpeakAsync(text); + } + else if (fallbackText != null) + { + await SpeakAsync(fallbackText); + } + else + { + await SendAsync(new { type = "play", source = $"{_settings.CallbackHost}/twilio/{name}.mp3" }); + MarkSpeaking(TimeSpan.FromSeconds(3)); + } + } + + private async Task SpeakAsync(string? text) + { + if (string.IsNullOrWhiteSpace(text)) + { + return; + } + + lock (_sync) + { + if (_ended) + { + return; + } + } + + await SendAsync(new { type = "text", token = text, last = true }); + MarkSpeaking(EstimateSpeech(text.Length)); + } + + private async Task EndAsync(string action) + { + lock (_sync) + { + if (_ended) + { + return; + } + _ended = true; + } + + await SendEndAsync(action); + } + + /// + /// Tells Twilio to end the session; the caller has already marked the session ended. + /// + private async Task SendEndAsync(string action) + { + await SendAsync(new { type = "end", handoffData = JsonSerializer.Serialize(new { action }) }); + } + + /// + /// True when the caller has not spoken or pressed a key since was read + /// and nothing else took over. + /// + private bool IsStillQuiet(int activity) + { + lock (_sync) + { + return !_ended && !_isProcessing && _activity == activity; + } + } + + private record PendingInput(string Content, string? Digits); + + private async Task SendAsync(object payload) + { + var bytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(payload)); + await _sendLock.WaitAsync(); + try + { + if (_socket.State == WebSocketState.Open) + { + var segment = new ArraySegment(bytes); + await _socket.SendAsync(segment, WebSocketMessageType.Text, true, CancellationToken.None); + } + } + catch (Exception ex) when (ex is WebSocketException or ObjectDisposedException) + { + // The call already moved on, e.g. a hang-up or transfer redirected it + _logger.LogInformation($"ConversationRelay socket for conversation {_conversationId} is gone: {ex.Message}"); + } + finally + { + _sendLock.Release(); + } + } + + private void MarkSpeaking(TimeSpan duration) + { + lock (_sync) + { + var now = DateTime.UtcNow; + var start = _quietSinceUtc == DateTime.MaxValue || _quietSinceUtc < now ? now : _quietSinceUtc; + _quietSinceUtc = start + duration; + } + } + + private static TimeSpan EstimateSpeech(int characters) + { + return TimeSpan.FromSeconds(characters / SpokenCharsPerSecond + 1); + } + + private static string? GetString(JsonElement element, string name) + { + if (element.ValueKind != JsonValueKind.Object || !element.TryGetProperty(name, out var value)) + { + return null; + } + return value.ValueKind == JsonValueKind.String ? value.GetString() : value.ToString(); + } + + private static async Task SwallowAsync(Task task) + { + try + { + await task; + } + catch (OperationCanceledException) + { + } + } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs new file mode 100644 index 000000000..213c024c0 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs @@ -0,0 +1,136 @@ +using BotSharp.Abstraction.Infrastructures.Enums; +using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Abstraction.MessageHub.Services; +using BotSharp.Abstraction.Routing; +using BotSharp.Plugin.Twilio.Models; +using Microsoft.AspNetCore.Http; +using System.Security.Claims; +using Task = System.Threading.Tasks.Task; + +namespace BotSharp.Plugin.Twilio.Services; + +/// +/// Sends what the caller said to the agent and returns the agent's reply. +/// Shared by the webhook flow (TwilioMessageQueueService) and the ConversationRelay WebSocket flow. +/// +public static class TwilioConversationProcessor +{ + /// A service scope created for this one caller message + /// + /// Receives tool-call indications, and the reply as it is written when OnStreamText is set + /// + public static async Task SendCallerMessageAsync( + IServiceProvider sp, + CallerMessage message, + AgentTurnCallbacks? callbacks = null) + { + // Clean static HttpContext + var httpContext = sp.GetRequiredService(); + httpContext.HttpContext = new DefaultHttpContext(); + httpContext.HttpContext.User = new ClaimsPrincipal(new ClaimsIdentity()); + foreach (var header in message.RequestHeaders ?? []) + { + httpContext.HttpContext.Request.Headers[header.Key] = header.Value; + } + httpContext.HttpContext.Request.Headers["X-Twilio-BotSharp"] = "LOST"; + + AssistantMessage? reply = null; + + var inputMsg = new RoleDialogModel(AgentRole.User, message.Content); + var conv = sp.GetRequiredService(); + var routing = sp.GetRequiredService(); + var observer = sp.GetRequiredService(); + + var streaming = callbacks?.OnStreamText != null; + var listeners = new Dictionary, Task>> + { + { + ChatEvent.OnIndicationReceived, async data => + { + if (callbacks?.OnIndication != null && !string.IsNullOrEmpty(data.Data.Indication)) + { + await callbacks.OnIndication(data.Data.Indication); + } + } + } + }; + if (streaming) + { + // Listeners run in push order on the model's stream, so the pieces arrive in order + listeners[ChatEvent.BeforeReceiveLlmStreamMessage] = async _ => + { + if (callbacks!.OnStreamStarted != null) + { + await callbacks.OnStreamStarted(); + } + }; + listeners[ChatEvent.OnReceiveLlmStreamMessage] = async data => + { + if (!string.IsNullOrEmpty(data.Data.Content)) + { + await callbacks!.OnStreamText!(data.Data.Content); + } + }; + } + + using var container = observer.SubscribeObservers>( + message.ConversationId, listeners: listeners); + await InitConversation(message, inputMsg, conv, routing); + + if (streaming) + { + // Routing streams only when both are set (RoutingService, InstructExecutor) + conv.States.SetState(StateConst.USE_STREAM_MESSAGE, true, source: StateSource.Application); + if (string.IsNullOrEmpty(conv.States.GetState(StateConst.LANGUAGE))) + { + conv.States.SetState(StateConst.LANGUAGE, LanguageType.ENGLISH, source: StateSource.Application); + } + } + + await conv.SendMessage(message.AgentId, + inputMsg, + replyMessage: await BuildPostbackMessageModel(conv, message), + async msg => + { + reply = new AssistantMessage() + { + ConversationEnd = msg.Instruction?.ConversationEnd ?? false, + HumanIntervationNeeded = string.Equals("human_intervention_needed", msg.FunctionName), + Content = msg.Content, + MessageId = msg.MessageId + }; + } + ); + + return reply; + } + + private static async Task BuildPostbackMessageModel( + IConversationService conv, CallerMessage message) + { + var messages = await conv.GetDialogHistory(1); + if (!messages.Any()) return null; + var lastMessage = messages[0]; + if (string.IsNullOrEmpty(lastMessage.PostbackFunctionName)) return null; + return new PostbackMessageModel + { + FunctionName = lastMessage.PostbackFunctionName, + ParentId = lastMessage.MessageId, + Payload = message.Digits + }; + } + + private static async Task InitConversation( + CallerMessage message, RoleDialogModel inputMsg, IConversationService conv, IRoutingService routing) + { + routing.Context.SetMessageId(message.ConversationId, inputMsg.MessageId); + var states = new List + { + new(StateConst.CHANNEL, ConversationChannel.Phone), + new("channel_id", message.From), + new("calling_phone", message.From) + }; + states.AddRange(message.States.Select(kvp => new MessageState(kvp.Key, kvp.Value))); + await conv.SetConversationId(message.ConversationId, states); + } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs index c7b7783b5..10c3376dc 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs @@ -67,96 +67,33 @@ private async Task ProcessUserMessageAsync(CallerMessage message) { using var scope = _serviceProvider.CreateScope(); var sp = scope.ServiceProvider; - - // Clean static HttpContext - var httpContext = sp.GetRequiredService(); - httpContext.HttpContext = new DefaultHttpContext(); - httpContext.HttpContext.User = new ClaimsPrincipal(new ClaimsIdentity()); - foreach (var header in message.RequestHeaders ?? []) - { - httpContext.HttpContext.Request.Headers[header.Key] = header.Value; - } - httpContext.HttpContext.Request.Headers["X-Twilio-BotSharp"] = "LOST"; - - AssistantMessage reply = default!; - - var inputMsg = new RoleDialogModel(AgentRole.User, message.Content); - var conv = sp.GetRequiredService(); - var routing = sp.GetRequiredService(); - var config = sp.GetRequiredService(); var sessionManager = sp.GetRequiredService(); - var observer = sp.GetRequiredService(); - using var container = observer.SubscribeObservers>(message.ConversationId, listeners: new() + var reply = await TwilioConversationProcessor.SendCallerMessageAsync(sp, message, new AgentTurnCallbacks { - { ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(data.Data, message, sessionManager) } + OnIndication = async indication => + await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, indication) }); - await InitConversation(message, inputMsg, conv, routing); - - // Need to consider Inbound and Outbound call - var conversation = await conv.GetConversation(message.ConversationId); - var agentId = message.AgentId; - - var result = await conv.SendMessage(agentId, - inputMsg, - replyMessage: await BuildPostbackMessageModel(conv, message), - async msg => - { - reply = new AssistantMessage() - { - ConversationEnd = msg.Instruction?.ConversationEnd ?? false, - HumanIntervationNeeded = string.Equals("human_intervention_needed", msg.FunctionName), - Content = msg.Content, - MessageId = msg.MessageId - }; - } - ); + if (reply == null) + { + _logger.LogWarning($"No reply for {message}."); + return; + } var settings = sp.GetRequiredService(); if (settings.GenerateReplyAudio) { reply.SpeechFileName = await GetReplySpeechFileName(message.ConversationId, reply, sp); } - - reply.Hints = await GetHints(agentId, reply, sp); - await sessionManager.SetAssistantReplyAsync(message.ConversationId, message.SeqNumber, reply); - } - private async Task BuildPostbackMessageModel(IConversationService conv, CallerMessage message) - { - var messages = await conv.GetDialogHistory(1); - if (!messages.Any()) return null; - var lastMessage = messages[0]; - if (string.IsNullOrEmpty(lastMessage.PostbackFunctionName)) return null; - return new PostbackMessageModel - { - FunctionName = lastMessage.PostbackFunctionName, - ParentId = lastMessage.MessageId, - Payload = message.Digits - }; - } - - private static async Task InitConversation(CallerMessage message, RoleDialogModel inputMsg, IConversationService conv, IRoutingService routing) - { - routing.Context.SetMessageId(message.ConversationId, inputMsg.MessageId); - var states = new List - { - new(StateConst.CHANNEL, ConversationChannel.Phone), - new("channel_id", message.From), - new("calling_phone", message.From) - }; - states.AddRange(message.States.Select(kvp => new MessageState(kvp.Key, kvp.Value))); - await conv.SetConversationId(message.ConversationId, states); + reply.Hints = await GetHints(message.AgentId, reply, sp); + await sessionManager.SetAssistantReplyAsync(message.ConversationId, message.SeqNumber, reply); } private static async Task GetReplySpeechFileName(string conversationId, AssistantMessage reply, IServiceProvider sp) { - var completion = CompletionProvider.GetAudioSynthesizer(sp); - var fileStorage = sp.GetRequiredService(); - var data = await completion.GenerateAudioAsync(reply.Content); - var fileName = $"reply_{reply.MessageId}.mp3"; - fileStorage.SaveSpeechFile(conversationId, fileName, data); - return fileName; + var twilio = sp.GetRequiredService(); + return await twilio.SaveSpeechAsync(conversationId, $"reply_{reply.MessageId}", reply.Content); } private static async Task GetHints(string agentId, AssistantMessage reply, IServiceProvider sp) @@ -191,12 +128,4 @@ private static async Task GetHints(string agentId, AssistantMessage repl hints.AddRange(extraWords); return string.Join(", ", hints.Select(x => x.ToLower()).Distinct().Reverse()); } - - private static async Task OnReceiveToolCallIndication(RoleDialogModel msg, CallerMessage message, ITwilioSessionManager sessionManager) - { - if (!string.IsNullOrEmpty(msg.Indication)) - { - await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, msg.Indication); - } - } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs index 1720d95d5..07c5caeac 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs @@ -5,6 +5,7 @@ using BotSharp.Core.Infrastructures; using BotSharp.Plugin.Twilio.Interfaces; using BotSharp.Plugin.Twilio.Models; +using BotSharp.Plugin.Twilio.Models.Relay; using Twilio.Jwt.AccessToken; using Twilio.Rest.Api.V2010.Account.Call; using Task = System.Threading.Tasks.Task; @@ -17,6 +18,9 @@ namespace BotSharp.Plugin.Twilio.Services; /// public class TwilioService { + private const string TextSpeechExtension = ".txt"; + private const string AudioSpeechExtension = ".mp3"; + private readonly TwilioSetting _settings; private readonly IServiceProvider _services; public readonly ILogger _logger; @@ -79,13 +83,13 @@ public VoiceResponse ReturnInstructions(ConversationalVoiceResponse conversation { foreach (var speechPath in conversationalVoiceResponse.SpeechPaths) { - gather.Play(new Uri($"{_settings.CallbackHost}/{speechPath}")); + AppendSpeech(gather, conversationalVoiceResponse.ConversationId, speechPath); } } if (!string.IsNullOrEmpty(conversationalVoiceResponse.Text)) { - gather.Say(conversationalVoiceResponse.Text); + gather.Append(CreateSay(conversationalVoiceResponse.Text)); } response.Append(gather); @@ -100,8 +104,7 @@ public VoiceResponse ReturnNoninterruptedInstructions(ConversationalVoiceRespons { foreach (var speechPath in voiceResponse.SpeechPaths) { - var uri = GetSpeechPath(conversationId, speechPath); - response.Play(new Uri(uri)); + AppendSpeech(response, conversationId, speechPath); } } @@ -145,7 +148,7 @@ public VoiceResponse HangUp(string speechPath) var response = new VoiceResponse(); if (!string.IsNullOrEmpty(speechPath)) { - response.Play(new Uri($"{_settings.CallbackHost}/{speechPath}")); + AppendSpeech(response, conversationId: null, speechPath); } response.Hangup(); return response; @@ -159,8 +162,7 @@ public VoiceResponse HangUp(ConversationalVoiceResponse voiceResponse) { foreach (var speechPath in voiceResponse.SpeechPaths) { - var uri = GetSpeechPath(conversationId, speechPath); - response.Play(new Uri(uri)); + AppendSpeech(response, conversationId, speechPath); } } @@ -173,7 +175,7 @@ public VoiceResponse DialCsrAgent(string speechPath) var response = new VoiceResponse(); if (!string.IsNullOrEmpty(speechPath)) { - response.Play(new Uri($"{_settings.CallbackHost}/{speechPath}")); + AppendSpeech(response, conversationId: null, speechPath); } response.Dial(_settings.CsrAgentNumber); return response; @@ -187,8 +189,7 @@ public VoiceResponse TransferCall(ConversationalVoiceResponse conversationalVoic { foreach (var speechPath in conversationalVoiceResponse.SpeechPaths) { - var uri = GetSpeechPath(conversationId, speechPath); - response.Play(new Uri(uri)); + AppendSpeech(response, conversationId, speechPath); } } response.Dial(conversationalVoiceResponse.TransferTo, answerOnBridge: true); @@ -214,7 +215,7 @@ public VoiceResponse HoldOn(int interval, string message = null) if (!string.IsNullOrEmpty(message)) { - gather.Say(message); + gather.Append(CreateSay(message)); } gather.Pause(interval); response.Append(gather); @@ -252,8 +253,7 @@ public VoiceResponse ReturnBidirectionalMediaStreamsInstructions(ConversationalV { foreach (var speechPath in conversationalVoiceResponse.SpeechPaths) { - var uri = GetSpeechPath(conversationId, speechPath); - response.Play(new Uri(uri)); + AppendSpeech(response, conversationId, speechPath); } } @@ -265,11 +265,59 @@ public VoiceResponse ReturnBidirectionalMediaStreamsInstructions(ConversationalV return response; } + /// + /// Hands the call to Twilio ConversationRelay. Twilio opens a WebSocket to twilio/relay/{agentId}/{conversationId} + /// and posts to twilio/relay/action when the session ends. + /// + public VoiceResponse ReturnConversationRelay( + string agentId, string conversationId, string? welcomeGreeting, int reconnects = 0) + { + var relay = _settings.ConversationRelay; + var query = $"agent-id={Uri.EscapeDataString(agentId)}" + + $"&conversation-id={Uri.EscapeDataString(conversationId)}&reconnects={reconnects}"; + var connect = new Connect(action: new Uri($"{_settings.CallbackHost}/twilio/relay/action?{query}")); + + var element = new ConversationRelay( + url: GetConversationRelayUrl(agentId, conversationId), + language: relay.Language, + ttsProvider: relay.TtsProvider, + voice: relay.Voice, + transcriptionProvider: relay.TranscriptionProvider, + speechModel: relay.SpeechModel, + dtmfDetection: relay.DtmfDetection, + welcomeGreeting: string.IsNullOrWhiteSpace(welcomeGreeting) ? null : welcomeGreeting); + + // Attributes that the SDK's typed parameters don't cover (or only cover as booleans) + element.SetOption("interruptible", relay.Interruptible); + element.SetOption("welcomeGreetingInterruptible", relay.WelcomeGreetingInterruptible); + element.SetOption("interruptSensitivity", relay.InterruptSensitivity); + element.SetOption("ignoreBackchannel", relay.IgnoreBackchannel ? "true" : "false"); + if (!string.IsNullOrWhiteSpace(relay.Hints)) + { + element.SetOption("hints", relay.Hints); + } + + // Lets the session time its idle check from the end of the greeting + var greetingLength = (welcomeGreeting?.Length ?? 0).ToString(); + element.Parameter(name: ConversationRelayParameters.WelcomeGreetingLength, value: greetingLength); + + connect.Append(element); + var response = new VoiceResponse(); + response.Append(connect); + return response; + } + + public string GetConversationRelayUrl(string agentId, string conversationId) + { + // Each id is one path segment, so a "/" or "?" in it cannot change the route + var host = _settings.CallbackHost.Split("://").Last(); + return $"wss://{host}/twilio/relay/{Uri.EscapeDataString(agentId)}/{Uri.EscapeDataString(conversationId)}"; + } + public async Task WaitingForAiResponse(ConversationalVoiceRequest request) { VoiceResponse response; var sessionManager = _services.GetRequiredService(); - var fileStorage = _services.GetRequiredService(); var indication = await sessionManager.GetReplyIndicationAsync(request.ConversationId, request.SeqNum); if (indication != null) @@ -286,16 +334,8 @@ public async Task WaitingForAiResponse(ConversationalVoiceRequest else { var hash = Utilities.HashTextMd5(seg); - var fileName = $"indication_{hash}.mp3"; - - var existing = fileStorage.GetSpeechFile(request.ConversationId, fileName); - if (existing == BinaryData.Empty) - { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(seg); - fileStorage.SaveSpeechFile(request.ConversationId, fileName, data); - } - + var fileName = await SaveSpeechAsync(request.ConversationId, $"indication_{hash}", seg, + reuseExisting: true); speechPaths.Add($"twilio/voice/speeches/{request.ConversationId}/{fileName}"); } } @@ -363,6 +403,123 @@ public string GetSpeechPath(string conversationId, string speechPath) } } + /// + /// Saves the speech of a text in the conversation and returns its file name: a text file spoken later with Say + /// when SayGeneratedSpeech is on, otherwise an mp3 from the audio synthesizer. + /// + public async Task SaveSpeechAsync(string conversationId, string name, string text, bool reuseExisting = false) + { + var fileStorage = _services.GetRequiredService(); + var fileName = name + (_settings.SayGeneratedSpeech ? TextSpeechExtension : AudioSpeechExtension); + if (reuseExisting && !fileStorage.GetSpeechFile(conversationId, fileName).ToMemory().IsEmpty) + { + return fileName; + } + + var data = _settings.SayGeneratedSpeech + ? BinaryData.FromString(text) + : await CompletionProvider.GetAudioSynthesizer(_services).GenerateAudioAsync(text); + fileStorage.SaveSpeechFile(conversationId, fileName, data); + return fileName; + } + + /// + /// Returns the file name of a speech saved by SaveSpeechAsync, whichever mode it was saved in. + /// + public string GetSavedSpeechFile(string conversationId, string name) + { + var fileStorage = _services.GetRequiredService(); + var textFile = name + TextSpeechExtension; + return fileStorage.GetSpeechFile(conversationId, textFile).ToMemory().IsEmpty + ? name + AudioSpeechExtension + : textFile; + } + + /// + /// Appends a speech to the TwiML: Say for a text file saved by SaveSpeechAsync, Play for an audio file. + /// + public void AppendSpeech(TwiML parent, string? conversationId, string speechPath) + { + if (_settings.SayGeneratedSpeech && TryGetPromptText(speechPath, out var promptText)) + { + parent.Append(CreateSay(promptText)); + return; + } + + if (!speechPath.EndsWith(TextSpeechExtension, StringComparison.OrdinalIgnoreCase)) + { + parent.Append(new Play(new Uri(GetSpeechPath(conversationId, speechPath)))); + return; + } + + var (convId, fileName) = SplitSpeechPath(conversationId, speechPath); + var fileStorage = _services.GetRequiredService(); + var data = fileStorage.GetSpeechFile(convId, fileName); + if (data.ToMemory().IsEmpty) + { + _logger.LogWarning($"Speech text {fileName} is missing (conversation {convId})."); + return; + } + + parent.Append(CreateSay(data.ToString())); + } + + /// + /// Looks up the configured text of a prerecorded prompt such as "twilio/say-it-again-1.mp3". + /// Files generated per conversation (twilio/voice/speeches/...) never match. + /// + private bool TryGetPromptText(string speechPath, out string text) + { + text = string.Empty; + const string promptRoute = "twilio/"; + var path = StripCallbackHost(speechPath); + if (!path.StartsWith(promptRoute)) + { + return false; + } + + var fileName = path.Substring(promptRoute.Length); + if (fileName.Contains('/')) + { + return false; + } + + return _settings.PromptTexts.TryGetValue(System.IO.Path.GetFileNameWithoutExtension(fileName), out text!) + && !string.IsNullOrWhiteSpace(text); + } + + private string StripCallbackHost(string speechPath) + { + return speechPath.StartsWith(_settings.CallbackHost) + ? speechPath.Substring(_settings.CallbackHost.Length).TrimStart('/') + : speechPath; + } + + private (string? ConversationId, string FileName) SplitSpeechPath(string? conversationId, string speechPath) + { + var path = StripCallbackHost(speechPath); + + // Accepts "twilio/voice/speeches/{conversationId}/{fileName}" or a bare file name in the given conversation + const string speechRoute = "twilio/voice/speeches/"; + if (path.StartsWith(speechRoute)) + { + var parts = path.Substring(speechRoute.Length).Split('/'); + if (parts.Length == 2) + { + return (parts[0], parts[1]); + } + } + + return (conversationId, path); + } + + private Say CreateSay(string text) + { + return string.IsNullOrEmpty(_settings.SayVoice) + ? new Say(text) + : new Say(text, voice: _settings.SayVoice); + } + public string GenerateStatesParameter(List states) { if (states is null || states.Count == 0) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs new file mode 100644 index 000000000..0fbc419d8 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs @@ -0,0 +1,76 @@ +namespace BotSharp.Plugin.Twilio.Settings; + +/// +/// Twilio ConversationRelay: Twilio transcribes the caller and speaks our replies, exchanging text with us over a WebSocket. +/// https://www.twilio.com/docs/voice/twiml/connect/conversationrelay +/// +public class ConversationRelaySetting +{ + public bool Enabled { get; set; } + + /// + /// Agents that use ConversationRelay for outbound calls. Empty means every agent. + /// + public List AgentIds { get; set; } = []; + + public string Language { get; set; } = "en-US"; + public string? TtsProvider { get; set; } + public string? Voice { get; set; } + public string? TranscriptionProvider { get; set; } + public string? SpeechModel { get; set; } + public string? Hints { get; set; } + + /// + /// none, dtmf, speech or any + /// + public string WelcomeGreetingInterruptible { get; set; } = "none"; + + /// + /// none, dtmf, speech or any + /// + public string Interruptible { get; set; } = "any"; + + /// + /// low, medium or high + /// + public string InterruptSensitivity { get; set; } = "medium"; + + public bool IgnoreBackchannel { get; set; } = true; + public bool DtmfDetection { get; set; } = true; + + /// + /// Seconds of caller silence, after the reply is expected to finish playing, before asking the caller to repeat + /// + public int IdleTimeoutSeconds { get; set; } = 8; + + /// + /// How many times to ask the caller to repeat before hanging up + /// + public int MaxIdlePrompts { get; set; } = 2; + + /// + /// How long to wait for the agent's reply before apologizing and hanging up + /// + public int AiResponseTimeoutSeconds { get; set; } = 30; + + /// + /// Send the reply to Twilio while the model is still writing it, so speaking starts with the first words. + /// Needs a chat provider that supports streaming, and only applies to English conversations. + /// + public bool StreamReplies { get; set; } + + /// + /// How many times to reconnect a session that failed (e.g. the pod holding the WebSocket went away) + /// + public int MaxReconnects { get; set; } = 2; + + /// + /// Verify the X-Twilio-Signature header of the WebSocket handshake + /// + public bool ValidateSignature { get; set; } = true; + + public bool IsEnabledFor(string? agentId) + { + return Enabled && (AgentIds.Count == 0 || (agentId != null && AgentIds.Contains(agentId))); + } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Settings/TwilioSetting.cs b/src/Plugins/BotSharp.Plugin.Twilio/Settings/TwilioSetting.cs index 94132241d..af884284e 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Settings/TwilioSetting.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Settings/TwilioSetting.cs @@ -33,4 +33,22 @@ public class TwilioSetting public bool GenerateReplyAudio { get; set; } = true; public bool GenerateEndingAudio { get; set; } = true; + + /// + /// Speak generated text with TwiML Say instead of pre-generating audio with the audio synthesizer + /// + public bool SayGeneratedSpeech { get; set; } + + /// + /// Voice for TwiML Say, e.g. "Polly.Joanna-Neural" or "Google.en-US-Neural2-F". Twilio's default voice when empty. + /// + public string? SayVoice { get; set; } + + /// + /// Text of the prerecorded prompts under twilio/, keyed by file name without extension (e.g. "say-it-again-1"). + /// Spoken with Say when SayGeneratedSpeech is on; prompts not listed keep playing their audio file. + /// + public Dictionary PromptTexts { get; set; } = new(); + + public ConversationRelaySetting ConversationRelay { get; set; } = new(); } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs new file mode 100644 index 000000000..eb2e863a9 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs @@ -0,0 +1,87 @@ +using BotSharp.Plugin.Twilio.Models; +using BotSharp.Plugin.Twilio.Services; +using Microsoft.AspNetCore.Http; +using Twilio.Security; +using Task = System.Threading.Tasks.Task; + +namespace BotSharp.Plugin.Twilio; + +/// +/// Accepts Twilio ConversationRelay WebSockets at /twilio/relay/{agentId}/{conversationId} +/// +public class TwilioConversationRelayMiddleware +{ + private readonly RequestDelegate _next; + private readonly ILogger _logger; + + public TwilioConversationRelayMiddleware(RequestDelegate next, ILogger logger) + { + _next = next; + _logger = logger; + } + + public async Task Invoke(HttpContext httpContext) + { + if (!httpContext.Request.Path.StartsWithSegments("/twilio/relay", out var rest) + || !httpContext.WebSockets.IsWebSocketRequest) + { + await _next(httpContext); + return; + } + + var parts = (rest.Value ?? string.Empty).Trim('/').Split('/'); + if (parts.Length != 2 || parts.Any(string.IsNullOrWhiteSpace)) + { + httpContext.Response.StatusCode = StatusCodes.Status404NotFound; + return; + } + + var (agentId, conversationId) = (parts[0], parts[1]); + var services = httpContext.RequestServices; + var settings = services.GetRequiredService(); + if (settings.ConversationRelay.ValidateSignature && !IsSignatureValid(httpContext, agentId, conversationId)) + { + httpContext.Response.StatusCode = StatusCodes.Status403Forbidden; + return; + } + + using var socket = await httpContext.WebSockets.AcceptWebSocketAsync(); + var scopeFactory = services.GetRequiredService(); + var session = new ConversationRelaySession(socket, agentId, conversationId, settings, + (message, callbacks) => SendCallerMessageAsync(scopeFactory, message, callbacks), + _logger); + await session.RunAsync(httpContext.RequestAborted); + } + + private static async Task SendCallerMessageAsync( + IServiceScopeFactory scopeFactory, CallerMessage message, AgentTurnCallbacks callbacks) + { + using var scope = scopeFactory.CreateScope(); + return await TwilioConversationProcessor.SendCallerMessageAsync(scope.ServiceProvider, message, callbacks); + } + + /// + /// Twilio signs the WebSocket handshake with X-Twilio-Signature over the wss URL from the TwiML. + /// + private bool IsSignatureValid(HttpContext httpContext, string agentId, string conversationId) + { + var services = httpContext.RequestServices; + var authToken = services.GetRequiredService()["Twilio:RequestValidation:AuthToken"]; + if (string.IsNullOrEmpty(authToken)) + { + _logger.LogWarning("Twilio auth token is not configured; skipping ConversationRelay signature check."); + return true; + } + + var signature = httpContext.Request.Headers["X-Twilio-Signature"].ToString(); + var url = services.GetRequiredService().GetConversationRelayUrl(agentId, conversationId); + var valid = !string.IsNullOrEmpty(signature) + && new RequestValidator(authToken).Validate(url, new Dictionary(), signature); + if (!valid) + { + _logger.LogWarning($"Rejected ConversationRelay connection for conversation {conversationId}: " + + "invalid or missing X-Twilio-Signature."); + } + return valid; + } +}