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

Hang-up if waiting AI Response Timeout

Browse source 
Hang-up if waiting AI Response Timeout
This commit is contained in:
Haiping 2025-03-10 15:11:26 -05:00 committed by GitHub
parent f41820f830 c6fc411dd5
commit 1d2f92ad15
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 185 additions and 166 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs
View file

@ -32,13 +32,12 @@ public interface IConversationService
/// Send message to LLM
/// </summary>
/// <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="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>
Task<bool> SendMessage(string agentId,
RoleDialogModel lastDalog,
RoleDialogModel lastDialog,
PostbackMessageModel? replyMessage,
Func<RoleDialogModel, Task> onResponseReceived);

3
src/Plugins/BotSharp.Plugin.Twilio/BotSharp.Plugin.Twilio.csproj
View file

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

22
src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioVoiceController.cs
View file

@ -142,7 +142,7 @@ public class TwilioVoiceController : TwilioController
await messageQueue.EnqueueAsync(callerMessage);
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 =>
{
@ -217,8 +217,22 @@ public class TwilioVoiceController : TwilioController
var reply = await sessionManager.GetAssistantReplyAsync(request.ConversationId, request.SeqNum);
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);
if (indication != null)
@ -262,7 +276,7 @@ public class TwilioVoiceController : TwilioController
var instruction = new ConversationalVoiceResponse
{
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
};
@ -301,7 +315,7 @@ public class TwilioVoiceController : TwilioController
var instruction = new ConversationalVoiceResponse
{
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
};

8
src/Plugins/BotSharp.Plugin.Twilio/Models/ConversationalVoiceRequest.cs
View file

@ -5,17 +5,19 @@ namespace BotSharp.Plugin.Twilio.Models;
public class ConversationalVoiceRequest : VoiceRequest
{
[FromQuery(Name = "agent-id")]
public string AgentId { get; set; }
public string AgentId { get; set; } = string.Empty;
[FromRoute]
public string ConversationId { get; set; }
public string ConversationId { get; set; } = string.Empty;
[FromRoute]
public int SeqNum { get; set; }
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; } = [];
}

311
src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs
View file

@ -9,175 +9,178 @@ using System.Security.Claims;
using System.Threading;
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;
private readonly IServiceProvider _serviceProvider;
private readonly SemaphoreSlim _throttler;
_queue = queue;
_serviceProvider = serviceProvider;
_throttler = new SemaphoreSlim(10, 10);
_logger = logger;
}
public TwilioMessageQueueService(
TwilioMessageQueue queue,
IServiceProvider serviceProvider)
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await foreach (var message in _queue.Reader.ReadAllAsync(stoppingToken))
{
_queue = queue;
_serviceProvider = serviceProvider;
_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);
_ = Task.Run(async () =>
{
await _throttler.WaitAsync(stoppingToken);
_ = Task.Run(async () =>
try
{
try
{
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
};
_logger.LogInformation($"Start processing {message}.");
await ProcessUserMessageAsync(message);
}
);
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--)
catch (Exception ex)
{
hints.Add(words[j]);
if (hints.Count >= capcity)
{
break;
}
_logger.LogError($"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
};
}
);
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)
{
break;
}
}
// 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 (hints.Count >= capcity)
{
if (!string.IsNullOrEmpty(msg.Indication))
{
await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, msg.Indication);
}
};
progressService.OnFunctionExecuted = async msg => { };
break;
}
}
// 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 => { };
}
}