Hang-up if waiting AI Response Timeout

This commit is contained in:
Haiping Chen 2025-03-10 10:10:22 -05:00
parent 6af7a3e6e3
commit ae7680da3d
5 changed files with 185 additions and 166 deletions

View file

@ -32,13 +32,12 @@ public interface IConversationService
/// Send message to LLM /// Send message to LLM
/// </summary> /// </summary>
/// <param name="agentId"></param> /// <param name="agentId"></param>
/// <param name="lastDalog"></param> /// <param name="lastDialog"></param>
/// <param name="replyMessage"></param>
/// <param name="onResponseReceived">Received the response from AI Agent</param> /// <param name="onResponseReceived">Received the response from AI Agent</param>
/// <param name="onFunctionExecuting">This delegate is useful when you want to report progress on UI</param>
/// <param name="onFunctionExecuted">This delegate is useful when you want to report progress on UI</param>
/// <returns></returns> /// <returns></returns>
Task<bool> SendMessage(string agentId, Task<bool> SendMessage(string agentId,
RoleDialogModel lastDalog, RoleDialogModel lastDialog,
PostbackMessageModel? replyMessage, PostbackMessageModel? replyMessage,
Func<RoleDialogModel, Task> onResponseReceived); Func<RoleDialogModel, Task> onResponseReceived);

View file

@ -1,4 +1,4 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<TargetFramework>$(TargetFramework)</TargetFramework> <TargetFramework>$(TargetFramework)</TargetFramework>
@ -6,6 +6,7 @@
<VersionPrefix>$(BotSharpVersion)</VersionPrefix> <VersionPrefix>$(BotSharpVersion)</VersionPrefix>
<GeneratePackageOnBuild>$(GeneratePackageOnBuild)</GeneratePackageOnBuild> <GeneratePackageOnBuild>$(GeneratePackageOnBuild)</GeneratePackageOnBuild>
<OutputPath>$(SolutionDir)packages</OutputPath> <OutputPath>$(SolutionDir)packages</OutputPath>
<Nullable>enable</Nullable>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>

View file

@ -142,7 +142,7 @@ public class TwilioVoiceController : TwilioController
await messageQueue.EnqueueAsync(callerMessage); await messageQueue.EnqueueAsync(callerMessage);
response = new VoiceResponse(); response = new VoiceResponse();
response.Redirect(new Uri($"{_settings.CallbackHost}/twilio/voice/{request.ConversationId}/reply/{request.SeqNum}?{GenerateStatesParameter(request.States)}"), HttpMethod.Post); response.Redirect(new Uri($"{_settings.CallbackHost}/twilio/voice/{request.ConversationId}/reply/{request.SeqNum}?{GenerateStatesParameter(request.States)}&AIResponseWaitTime=0"), HttpMethod.Post);
await HookEmitter.Emit<ITwilioSessionHook>(_services, async hook => await HookEmitter.Emit<ITwilioSessionHook>(_services, async hook =>
{ {
@ -217,8 +217,22 @@ public class TwilioVoiceController : TwilioController
var reply = await sessionManager.GetAssistantReplyAsync(request.ConversationId, request.SeqNum); var reply = await sessionManager.GetAssistantReplyAsync(request.ConversationId, request.SeqNum);
VoiceResponse response; VoiceResponse response;
if (request.AIResponseWaitTime > 5)
{
// Wait AI Response Timeout
await HookEmitter.Emit<ITwilioSessionHook>(_services, async hook =>
{
request.AIResponseErrorMessage = $"AI response timeout: AIResponseWaitTime greater than {request.AIResponseWaitTime}, please check internal error log!";
await hook.OnAgentHangUp(request);
}, new HookEmitOption
{
OnlyOnce = true
});
if (reply == null) response = twilio.HangUp($"twilio/error.mp3");
}
else if (reply == null)
{ {
var indication = await sessionManager.GetReplyIndicationAsync(request.ConversationId, request.SeqNum); var indication = await sessionManager.GetReplyIndicationAsync(request.ConversationId, request.SeqNum);
if (indication != null) if (indication != null)
@ -262,7 +276,7 @@ public class TwilioVoiceController : TwilioController
var instruction = new ConversationalVoiceResponse var instruction = new ConversationalVoiceResponse
{ {
SpeechPaths = speechPaths, SpeechPaths = speechPaths,
CallbackPath = $"twilio/voice/{request.ConversationId}/reply/{request.SeqNum}?{GenerateStatesParameter(request.States)}", CallbackPath = $"twilio/voice/{request.ConversationId}/reply/{request.SeqNum}?{GenerateStatesParameter(request.States)}&AIResponseWaitTime={++request.AIResponseWaitTime}",
ActionOnEmptyResult = true ActionOnEmptyResult = true
}; };
@ -301,7 +315,7 @@ public class TwilioVoiceController : TwilioController
var instruction = new ConversationalVoiceResponse var instruction = new ConversationalVoiceResponse
{ {
SpeechPaths = instructions, SpeechPaths = instructions,
CallbackPath = $"twilio/voice/{request.ConversationId}/reply/{request.SeqNum}?{GenerateStatesParameter(request.States)}", CallbackPath = $"twilio/voice/{request.ConversationId}/reply/{request.SeqNum}?{GenerateStatesParameter(request.States)}&AIResponseWaitTime={++request.AIResponseWaitTime}",
ActionOnEmptyResult = true ActionOnEmptyResult = true
}; };

View file

@ -5,17 +5,19 @@ namespace BotSharp.Plugin.Twilio.Models;
public class ConversationalVoiceRequest : VoiceRequest public class ConversationalVoiceRequest : VoiceRequest
{ {
[FromQuery(Name = "agent-id")] [FromQuery(Name = "agent-id")]
public string AgentId { get; set; } public string AgentId { get; set; } = string.Empty;
[FromRoute] [FromRoute]
public string ConversationId { get; set; } public string ConversationId { get; set; } = string.Empty;
[FromRoute] [FromRoute]
public int SeqNum { get; set; } public int SeqNum { get; set; }
public int Attempts { get; set; } = 1; public int Attempts { get; set; } = 1;
public int AIResponseWaitTime { get; set; } = 0;
public string? AIResponseErrorMessage { get; set; } = string.Empty;
public string Intent { get; set; } public string Intent { get; set; } = string.Empty;
public List<string> States { get; set; } = []; public List<string> States { get; set; } = [];
} }

View file

@ -9,175 +9,178 @@ using System.Security.Claims;
using System.Threading; using System.Threading;
using Task = System.Threading.Tasks.Task; using Task = System.Threading.Tasks.Task;
namespace BotSharp.Plugin.Twilio.Services namespace BotSharp.Plugin.Twilio.Services;
public class TwilioMessageQueueService : BackgroundService
{ {
public class TwilioMessageQueueService : BackgroundService private readonly TwilioMessageQueue _queue;
private readonly IServiceProvider _serviceProvider;
private readonly SemaphoreSlim _throttler;
private readonly ILogger _logger;
public TwilioMessageQueueService(
TwilioMessageQueue queue,
IServiceProvider serviceProvider,
ILogger<TwilioMessageQueueService> logger)
{ {
private readonly TwilioMessageQueue _queue; _queue = queue;
private readonly IServiceProvider _serviceProvider; _serviceProvider = serviceProvider;
private readonly SemaphoreSlim _throttler; _throttler = new SemaphoreSlim(10, 10);
_logger = logger;
}
public TwilioMessageQueueService( protected override async Task ExecuteAsync(CancellationToken stoppingToken)
TwilioMessageQueue queue, {
IServiceProvider serviceProvider) await foreach (var message in _queue.Reader.ReadAllAsync(stoppingToken))
{ {
_queue = queue; await _throttler.WaitAsync(stoppingToken);
_serviceProvider = serviceProvider; _ = Task.Run(async () =>
_throttler = new SemaphoreSlim(4, 4);
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await foreach (var message in _queue.Reader.ReadAllAsync(stoppingToken))
{ {
await _throttler.WaitAsync(stoppingToken); try
_ = Task.Run(async () =>
{ {
try _logger.LogInformation($"Start processing {message}.");
{ await ProcessUserMessageAsync(message);
Console.WriteLine($"Start processing {message}.");
await ProcessUserMessageAsync(message);
}
catch (Exception ex)
{
Console.WriteLine($"Processing {message} failed due to {ex.Message}.");
}
finally
{
_throttler.Release();
}
});
}
}
public override async Task StopAsync(CancellationToken cancellationToken)
{
_queue.Stop();
await base.StopAsync(cancellationToken);
}
private async Task ProcessUserMessageAsync(CallerMessage message)
{
using var scope = _serviceProvider.CreateScope();
var sp = scope.ServiceProvider;
// Clean static HttpContext
var httpContext = sp.GetRequiredService<IHttpContextAccessor>();
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<IConversationService>();
var routing = sp.GetRequiredService<IRoutingService>();
var config = sp.GetRequiredService<TwilioSetting>();
var sessionManager = sp.GetRequiredService<ITwilioSessionManager>();
var progressService = sp.GetRequiredService<IConversationProgressService>();
InitProgressService(message, sessionManager, progressService);
InitConversation(message, inputMsg, conv, routing);
// Need to consider Inbound and Outbound call
var conversation = await conv.GetConversation(message.ConversationId);
var agentId = string.IsNullOrWhiteSpace(conversation?.AgentId) ? config.AgentId : conversation.AgentId;
var result = await conv.SendMessage(agentId,
inputMsg,
replyMessage: 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
};
} }
); catch (Exception ex)
reply.SpeechFileName = await GetReplySpeechFileName(message.ConversationId, reply, sp);
reply.Hints = GetHints(reply);
reply.Content = null;
await sessionManager.SetAssistantReplyAsync(message.ConversationId, message.SeqNumber, reply);
}
private PostbackMessageModel BuildPostbackMessageModel(IConversationService conv, CallerMessage message)
{
var messages = 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 void InitConversation(CallerMessage message, RoleDialogModel inputMsg, IConversationService conv, IRoutingService routing)
{
routing.Context.SetMessageId(message.ConversationId, inputMsg.MessageId);
var states = new List<MessageState>
{
new("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)));
conv.SetConversationId(message.ConversationId, states);
}
private static async Task<string> GetReplySpeechFileName(string conversationId, AssistantMessage reply, IServiceProvider sp)
{
var completion = CompletionProvider.GetAudioCompletion(sp, "openai", "tts-1");
var fileStorage = sp.GetRequiredService<IFileStorageService>();
var data = await completion.GenerateAudioFromTextAsync(reply.Content);
var fileName = $"reply_{reply.MessageId}.mp3";
fileStorage.SaveSpeechFile(conversationId, fileName, data);
return fileName;
}
private static string GetHints(AssistantMessage reply)
{
var phrases = reply.Content.Split(',', StringSplitOptions.RemoveEmptyEntries);
int capcity = 100;
var hints = new List<string>(capcity);
for (int i = phrases.Length - 1; i >= 0; i--)
{
var words = phrases[i].Split(' ', StringSplitOptions.RemoveEmptyEntries);
for (int j = words.Length - 1; j >= 0; j--)
{ {
hints.Add(words[j]); _logger.LogError($"Processing {message} failed due to {ex.Message}.");
if (hints.Count >= capcity)
{
break;
}
} }
finally
{
_throttler.Release();
}
});
}
}
public override async Task StopAsync(CancellationToken cancellationToken)
{
_queue.Stop();
await base.StopAsync(cancellationToken);
}
private async Task ProcessUserMessageAsync(CallerMessage message)
{
using var scope = _serviceProvider.CreateScope();
var sp = scope.ServiceProvider;
// Clean static HttpContext
var httpContext = sp.GetRequiredService<IHttpContextAccessor>();
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<IConversationService>();
var routing = sp.GetRequiredService<IRoutingService>();
var config = sp.GetRequiredService<TwilioSetting>();
var sessionManager = sp.GetRequiredService<ITwilioSessionManager>();
var progressService = sp.GetRequiredService<IConversationProgressService>();
InitProgressService(message, sessionManager, progressService);
InitConversation(message, inputMsg, conv, routing);
// Need to consider Inbound and Outbound call
var conversation = await conv.GetConversation(message.ConversationId);
var agentId = string.IsNullOrWhiteSpace(conversation?.AgentId) ? config.AgentId : conversation.AgentId;
var result = await conv.SendMessage(agentId,
inputMsg,
replyMessage: 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
};
}
);
reply.SpeechFileName = await GetReplySpeechFileName(message.ConversationId, reply, sp);
reply.Hints = GetHints(reply);
reply.Content = null;
await sessionManager.SetAssistantReplyAsync(message.ConversationId, message.SeqNumber, reply);
}
private PostbackMessageModel BuildPostbackMessageModel(IConversationService conv, CallerMessage message)
{
var messages = 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 void InitConversation(CallerMessage message, RoleDialogModel inputMsg, IConversationService conv, IRoutingService routing)
{
routing.Context.SetMessageId(message.ConversationId, inputMsg.MessageId);
var states = new List<MessageState>
{
new("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)));
conv.SetConversationId(message.ConversationId, states);
}
private static async Task<string> GetReplySpeechFileName(string conversationId, AssistantMessage reply, IServiceProvider sp)
{
var completion = CompletionProvider.GetAudioCompletion(sp, "openai", "tts-1");
var fileStorage = sp.GetRequiredService<IFileStorageService>();
var data = await completion.GenerateAudioFromTextAsync(reply.Content);
var fileName = $"reply_{reply.MessageId}.mp3";
fileStorage.SaveSpeechFile(conversationId, fileName, data);
return fileName;
}
private static string GetHints(AssistantMessage reply)
{
var phrases = reply.Content.Split(',', StringSplitOptions.RemoveEmptyEntries);
int capcity = 100;
var hints = new List<string>(capcity);
for (int i = phrases.Length - 1; i >= 0; i--)
{
var words = phrases[i].Split(' ', StringSplitOptions.RemoveEmptyEntries);
for (int j = words.Length - 1; j >= 0; j--)
{
hints.Add(words[j]);
if (hints.Count >= capcity) if (hints.Count >= capcity)
{ {
break; break;
} }
} }
// add frequency short words if (hints.Count >= capcity)
hints.AddRange(["yes", "no", "correct", "right"]);
return string.Join(", ", hints.Select(x => x.ToLower()).Distinct().Reverse());
}
private static void InitProgressService(CallerMessage message, ITwilioSessionManager sessionManager, IConversationProgressService progressService)
{
progressService.OnFunctionExecuting = async msg =>
{ {
if (!string.IsNullOrEmpty(msg.Indication)) break;
{ }
await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, msg.Indication);
}
};
progressService.OnFunctionExecuted = async msg => { };
} }
// add frequency short words
hints.AddRange(["yes", "no", "correct", "right"]);
return string.Join(", ", hints.Select(x => x.ToLower()).Distinct().Reverse());
}
private static void InitProgressService(CallerMessage message, ITwilioSessionManager sessionManager, IConversationProgressService progressService)
{
progressService.OnFunctionExecuting = async msg =>
{
if (!string.IsNullOrEmpty(msg.Indication))
{
await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, msg.Indication);
}
};
progressService.OnFunctionExecuted = async msg => { };
} }
} }