From ae0abecd3b5c36ac1216ed49e5096ed8804dde7f Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Fri, 1 Aug 2025 15:47:42 -0500 Subject: [PATCH] replace conversation progress service --- .../IConversationProgressService.cs | 10 ------ .../Observers/BotSharpObserverBase.cs | 9 +++++- .../MessageHub/Observers/IBotSharpObserver.cs | 2 ++ .../MessageHub/Services/IObserverService.cs | 5 ++- .../Conversations/ConversationPlugin.cs | 1 - .../Services/ConversationProgressService.cs | 11 ------- .../Demo/Functions/GetFunEventsFn.cs | 2 +- .../Observers/ConversationObserver.cs | 5 ++- .../MessageHub/Services/ObserverService.cs | 13 +++++++- .../Controllers/ConversationController.cs | 31 +++++++++---------- .../Services/TwilioMessageQueueService.cs | 22 +++++++------ 11 files changed, 58 insertions(+), 53 deletions(-) delete mode 100644 src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationProgressService.cs delete mode 100644 src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationProgressService.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationProgressService.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationProgressService.cs deleted file mode 100644 index 8958cab5..00000000 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationProgressService.cs +++ /dev/null @@ -1,10 +0,0 @@ -namespace BotSharp.Abstraction.Conversations; - -public delegate Task FunctionExecuting(RoleDialogModel msg); -public delegate Task FunctionExecuted(RoleDialogModel msg); - -public interface IConversationProgressService -{ - FunctionExecuted OnFunctionExecuted { get; set; } - FunctionExecuting OnFunctionExecuting { get; set; } -} diff --git a/src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/BotSharpObserverBase.cs b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/BotSharpObserverBase.cs index 5afe9915..1381ea03 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/BotSharpObserverBase.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/BotSharpObserverBase.cs @@ -3,7 +3,8 @@ namespace BotSharp.Abstraction.MessageHub.Observers; public abstract class BotSharpObserverBase : IBotSharpObserver { - protected bool _active = false; + private bool _active = false; + protected Dictionary> _listeners = []; protected BotSharpObserverBase() { @@ -22,6 +23,12 @@ public abstract class BotSharpObserverBase : IBotSharpObserver public virtual void Deactivate() { _active = false; + _listeners = []; + } + + public virtual void SetEventListeners(Dictionary> listeners) + { + _listeners = listeners; } public virtual void OnCompleted() diff --git a/src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/IBotSharpObserver.cs b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/IBotSharpObserver.cs index 1c160765..158ef4a6 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/IBotSharpObserver.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/IBotSharpObserver.cs @@ -4,6 +4,8 @@ public interface IBotSharpObserver : IObserver { string Name { get; } bool Active { get; } + + void SetEventListeners(Dictionary> listeners); void Activate(); void Deactivate(); } diff --git a/src/Infrastructure/BotSharp.Abstraction/MessageHub/Services/IObserverService.cs b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Services/IObserverService.cs index bc4ad3bb..582849aa 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MessageHub/Services/IObserverService.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Services/IObserverService.cs @@ -4,7 +4,10 @@ namespace BotSharp.Abstraction.MessageHub.Services; public interface IObserverService { - IDisposable SubscribeObservers(string refId, IEnumerable? names = null) where T : ObserveDataBase; + IDisposable SubscribeObservers( + string refId, + IEnumerable? names = null, + Dictionary>? listeners = null) where T : ObserveDataBase; void UnSubscribeObservers(IEnumerable? names = null) where T : ObserveDataBase; } diff --git a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs b/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs index 511a27f7..61c23d1b 100644 --- a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs +++ b/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs @@ -56,7 +56,6 @@ public class ConversationPlugin : IBotSharpPlugin services.AddScoped(); services.AddScoped(); - services.AddScoped(); services.AddScoped(); services.AddScoped(); diff --git a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationProgressService.cs b/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationProgressService.cs deleted file mode 100644 index d7e6440c..00000000 --- a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationProgressService.cs +++ /dev/null @@ -1,11 +0,0 @@ -namespace BotSharp.Core.Conversations.Services -{ - public class ConversationProgressService : IConversationProgressService - { - - public FunctionExecuting OnFunctionExecuting { get; set; } - - - public FunctionExecuted OnFunctionExecuted { get; set; } - } -} diff --git a/src/Infrastructure/BotSharp.Core/Demo/Functions/GetFunEventsFn.cs b/src/Infrastructure/BotSharp.Core/Demo/Functions/GetFunEventsFn.cs index beae84c3..87ddb1ee 100644 --- a/src/Infrastructure/BotSharp.Core/Demo/Functions/GetFunEventsFn.cs +++ b/src/Infrastructure/BotSharp.Core/Demo/Functions/GetFunEventsFn.cs @@ -34,7 +34,7 @@ public class GetFunEventsFn : IFunctionCallback await Task.Delay(1500); - message.Indication = $"Still searching in {args?.City}"; + message.Indication = $"Still searching events in {args?.City}"; messageHub.Push(new() { EventName = ChatEvent.OnIndicationReceived, diff --git a/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs b/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs index f0d564d7..e2588c6c 100644 --- a/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs +++ b/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs @@ -36,7 +36,10 @@ public class ConversationObserver : BotSharpObserverBase(string refId, IEnumerable? names = null) where T : ObserveDataBase + public IDisposable SubscribeObservers( + string refId, + IEnumerable? names = null, + Dictionary>? listeners = null) where T : ObserveDataBase { var container = _services.GetRequiredService>(); var observers = _services.GetServices>() @@ -34,6 +37,14 @@ public class ObserverService : IObserverService return container; } + if (!listeners.IsNullOrEmpty()) + { + foreach (var observer in observers) + { + observer.SetEventListeners(listeners ?? []); + } + } + #if DEBUG _logger.LogCritical($"Subscribe observers: {string.Join(",", observers.Select(x => x.Name))}"); #endif diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs index 3d80b0d5..ec7dead5 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs @@ -7,7 +7,6 @@ using BotSharp.Abstraction.Options; using BotSharp.Abstraction.Routing; using BotSharp.Abstraction.Users.Dtos; using BotSharp.Core.Infrastructures; -using BotSharp.Core.MessageHub.Observers; namespace BotSharp.OpenAPI.Controllers; @@ -386,6 +385,12 @@ public class ConversationController : ControllerBase [HttpPost("/conversation/{agentId}/{conversationId}/sse")] public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string conversationId, [FromBody] NewMessageModel input) { + var observer = _services.GetRequiredService(); + using var container = observer.SubscribeObservers>(conversationId, listeners: new() + { + { ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(conversationId, data.Data) } + }); + var conv = _services.GetRequiredService(); var inputMsg = new RoleDialogModel(AgentRole.User, input.Text) { @@ -411,7 +416,6 @@ public class ConversationController : ControllerBase Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.ContentType, "text/event-stream"); Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.CacheControl, "no-cache"); Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.Connection, "keep-alive"); - InitProgressService(conversationId); await conv.SendMessage(agentId, inputMsg, replyMessage: input.Postback, @@ -435,23 +439,18 @@ public class ConversationController : ControllerBase // await OnEventCompleted(Response); } - private void InitProgressService(string conversationId) + private async Task OnReceiveToolCallIndication(string conversationId, RoleDialogModel msg) { - var progressService = _services.GetService(); - progressService.OnFunctionExecuting = async msg => + var indicator = new ChatResponseModel { - var indicator = new ChatResponseModel - { - ConversationId = conversationId, - MessageId = msg.MessageId, - Text = msg.Indication, - Function = "indicating", - Instruction = msg.Instruction, - States = new Dictionary() - }; - await OnChunkReceived(Response, indicator); + ConversationId = conversationId, + MessageId = msg.MessageId, + Text = msg.Indication, + Function = "indicating", + Instruction = msg.Instruction, + States = new Dictionary() }; - progressService.OnFunctionExecuted = async msg => { }; + await OnChunkReceived(Response, indicator); } #endregion diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs index 84f88b5b..28360518 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs @@ -1,4 +1,6 @@ using BotSharp.Abstraction.Files; +using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Abstraction.MessageHub.Services; using BotSharp.Abstraction.Realtime; using BotSharp.Abstraction.Routing; using BotSharp.Core.Infrastructures; @@ -82,8 +84,12 @@ public class TwilioMessageQueueService : BackgroundService var routing = sp.GetRequiredService(); var config = sp.GetRequiredService(); var sessionManager = sp.GetRequiredService(); - var progressService = sp.GetRequiredService(); - InitProgressService(message, sessionManager, progressService); + var observer = sp.GetRequiredService(); + + using var container = observer.SubscribeObservers>(message.ConversationId, listeners: new() + { + { ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(data.Data, message, sessionManager) } + }); InitConversation(message, inputMsg, conv, routing); // Need to consider Inbound and Outbound call @@ -185,15 +191,11 @@ public class TwilioMessageQueueService : BackgroundService return string.Join(", ", hints.Select(x => x.ToLower()).Distinct().Reverse()); } - private static void InitProgressService(CallerMessage message, ITwilioSessionManager sessionManager, IConversationProgressService progressService) + private static async Task OnReceiveToolCallIndication(RoleDialogModel msg, CallerMessage message, ITwilioSessionManager sessionManager) { - progressService.OnFunctionExecuting = async msg => + if (!string.IsNullOrEmpty(msg.Indication)) { - if (!string.IsNullOrEmpty(msg.Indication)) - { - await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, msg.Indication); - } - }; - progressService.OnFunctionExecuted = async msg => { }; + await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, msg.Indication); + } } }