From d1b5ad592dfe2cdbaba45bb285764b3546c51db9 Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Tue, 29 Jul 2025 11:17:33 -0500 Subject: [PATCH] add conversation observer --- .../Conversations/ConversationPlugin.cs | 15 ++++++-- .../Observers/ConversationObserver.cs | 34 +++++++++++++++++++ .../Routing/RoutingService.InvokeFunction.cs | 6 ---- .../Observers/ChatHubObserver.cs | 10 +++--- .../Providers/Chat/ChatCompletionProvider.cs | 9 ++--- 5 files changed, 55 insertions(+), 19 deletions(-) create mode 100644 src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs diff --git a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs b/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs index 2f0af290..cef32cad 100644 --- a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs +++ b/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs @@ -6,17 +6,18 @@ using BotSharp.Abstraction.Plugins.Models; using BotSharp.Abstraction.Settings; using BotSharp.Abstraction.Templating; using BotSharp.Core.Instructs; +using BotSharp.Core.MessageHub; +using BotSharp.Core.MessageHub.Observers; using BotSharp.Core.Messaging; using BotSharp.Core.Routing.Reasoning; using BotSharp.Core.Templating; using BotSharp.Core.Translation; +using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.Configuration; -using BotSharp.Abstraction.MessageHub.Models; -using BotSharp.Core.MessageHub; namespace BotSharp.Core.Conversations; -public class ConversationPlugin : IBotSharpPlugin +public class ConversationPlugin : IBotSharpPlugin, IBotSharpAppPlugin { public string Id => "99e9b971-a9f1-4273-84da-876d2873d192"; public string Name => "Conversation"; @@ -67,4 +68,12 @@ public class ConversationPlugin : IBotSharpPlugin menu.Add(new PluginMenuDef("Conversation", link: "page/conversation", icon: "bx bx-conversation", weight: section.Weight + 5)); return true; } + + public void Configure(IApplicationBuilder app) + { + var services = app.ApplicationServices; + var queue = services.GetRequiredService>(); + var logger = services.GetRequiredService>>(); + queue.Events.Subscribe(new ConversationObserver(logger)); + } } diff --git a/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs b/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs new file mode 100644 index 00000000..dd4b528c --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs @@ -0,0 +1,34 @@ +namespace BotSharp.Core.MessageHub.Observers; + +public class ConversationObserver : IObserver +{ + private readonly ILogger _logger; + private IServiceProvider _services; + + public ConversationObserver(ILogger logger) + { + _logger = logger; + } + + public void OnCompleted() + { + _logger.LogWarning($"{nameof(ConversationObserver)} receives complete notification."); + } + + public void OnError(Exception error) + { + _logger.LogError(error, $"{nameof(ConversationObserver)} receives error notification: {error.Message}"); + } + + public void OnNext(HubObserveData value) + { + _services = value.ServiceProvider; + + var progress = _services.GetRequiredService(); + if (value.EventName == ChatEvent.OnIndicationReceived + && progress.OnFunctionExecuting != null) + { + progress.OnFunctionExecuting(value.Data).ConfigureAwait(false).GetAwaiter().GetResult(); + } + } +} diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs index 0922dcdb..3660f61f 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs +++ b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs @@ -25,12 +25,6 @@ public partial class RoutingService clonedMessage.FunctionName = name; clonedMessage.Indication = await funcExecutor.GetIndicatorAsync(message); - //var progressService = _services.GetService(); - //if (progressService?.OnFunctionExecuting != null) - //{ - // await progressService.OnFunctionExecuting(clonedMessage); - //} - var messageHub = _services.GetRequiredService>(); messageHub.Push(new() { diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs index 3c1799ea..8f467908 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs @@ -2,8 +2,6 @@ using BotSharp.Abstraction.Conversations.Dtos; using BotSharp.Abstraction.Conversations.Enums; using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Abstraction.SideCar; -using BotSharp.Plugin.ChatHub.Hooks; -using Microsoft.AspNetCore.SignalR; using System.Runtime.CompilerServices; namespace BotSharp.Plugin.ChatHub.Observers; @@ -32,8 +30,6 @@ public class ChatHubObserver : IObserver { _services = value.ServiceProvider; - if (!AllowSendingMessage()) return; - var message = value.Data; var model = new ChatResponseDto(); var action = new ConversationSenderActionModel(); @@ -42,6 +38,8 @@ public class ChatHubObserver : IObserver switch (value.EventName) { case ChatEvent.BeforeReceiveLlmStreamMessage: + if (!AllowSendingMessage()) return; + model = new ChatResponseDto() { ConversationId = conv.ConversationId, @@ -64,6 +62,8 @@ public class ChatHubObserver : IObserver SendEvent(ChatEvent.OnSenderActionGenerated, conv.ConversationId, action); break; case ChatEvent.OnReceiveLlmStreamMessage: + if (!AllowSendingMessage()) return; + model = new ChatResponseDto() { ConversationId = conv.ConversationId, @@ -81,6 +81,8 @@ public class ChatHubObserver : IObserver }; break; case ChatEvent.AfterReceiveLlmStreamMessage: + if (!AllowSendingMessage()) return; + model = new ChatResponseDto() { ConversationId = conv.ConversationId, diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs index 0e39cb34..e4567cfb 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs @@ -1,10 +1,7 @@ -using Azure; using BotSharp.Abstraction.Hooks; using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Core.Infrastructures.Streams; using BotSharp.Core.MessageHub; -using BotSharp.Plugin.OpenAI.Models.Realtime; -using Fluid; using OpenAI.Chat; namespace BotSharp.Plugin.OpenAI.Providers.Chat; @@ -204,7 +201,7 @@ public class ChatCompletionProvider : IChatCompletion hub.Push(new() { ServiceProvider = _services, - EventName = "BeforeReceiveLlmStreamMessage", + EventName = ChatEvent.BeforeReceiveLlmStreamMessage, Data = new RoleDialogModel(AgentRole.Assistant, string.Empty) { CurrentAgentId = agent.Id, @@ -248,7 +245,7 @@ public class ChatCompletionProvider : IChatCompletion hub.Push(new() { ServiceProvider = _services, - EventName = "OnReceiveLlmStreamMessage", + EventName = ChatEvent.OnReceiveLlmStreamMessage, Data = content }); } @@ -291,7 +288,7 @@ public class ChatCompletionProvider : IChatCompletion hub.Push(new() { ServiceProvider = _services, - EventName = "AfterReceiveLlmStreamMessage", + EventName = ChatEvent.AfterReceiveLlmStreamMessage, Data = responseMessage });