From 0a6066c5b276e9db2eface9f37f0ee5b3cd77cbe Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Mon, 28 Jul 2025 17:41:28 -0500 Subject: [PATCH] refine indication --- .../Conversations/Dtos/ChatResponseDto.cs | 4 + .../Conversations/Enums/ChatEvent.cs | 22 +++ .../Models/ConversationSenderActionModel.cs | 2 - .../Models/HubObserveData.cs | 2 +- .../Models/ObserveDataBase.cs | 2 +- .../Conversations/ConversationPlugin.cs | 4 +- .../Demo/Functions/GetWeatherFn.cs | 26 +++ .../Queues => MessageHub}/MessageHub.cs | 4 +- .../Routing/RoutingService.InvokeFunction.cs | 21 ++- src/Infrastructure/BotSharp.Core/Using.cs | 3 + .../Providers/Chat/ChatCompletionProvider.cs | 4 +- .../BotSharp.Plugin.ChatHub/ChatHubPlugin.cs | 4 +- .../Helpers/ChatHubHelper.cs | 38 ++++ .../Hooks/ChatHubConversationHook.cs | 170 +++-------------- .../Hooks/ChatHubCrontabHook.cs | 18 +- .../Hooks/StreamingLogHook.cs | 174 ++++++------------ .../Hooks/WelcomeHook.cs | 36 +--- .../Observers/ChatHubObserver.cs | 146 ++++++++------- src/Plugins/BotSharp.Plugin.ChatHub/Using.cs | 3 +- .../Providers/Chat/ChatCompletionProvider.cs | 4 +- .../Providers/ChatCompletionProvider.cs | 4 +- .../Providers/Chat/ChatCompletionProvider.cs | 4 +- .../Providers/ChatCompletionProvider.cs | 4 +- 23 files changed, 305 insertions(+), 394 deletions(-) create mode 100644 src/Infrastructure/BotSharp.Abstraction/Conversations/Enums/ChatEvent.cs rename src/Infrastructure/BotSharp.Abstraction/{Observables => MessageHub}/Models/HubObserveData.cs (75%) rename src/Infrastructure/BotSharp.Abstraction/{Observables => MessageHub}/Models/ObserveDataBase.cs (68%) rename src/Infrastructure/BotSharp.Core/{Observables/Queues => MessageHub}/MessageHub.cs (88%) create mode 100644 src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatHubHelper.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/Dtos/ChatResponseDto.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/Dtos/ChatResponseDto.cs index 19b5e8b2..9e4fcb41 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/Dtos/ChatResponseDto.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Conversations/Dtos/ChatResponseDto.cs @@ -32,6 +32,10 @@ public class ChatResponseDto : InstructResult [JsonPropertyName("payload")] public string? Payload { get; set; } + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] + [JsonPropertyName("indication")] + public string? Indication { get; set; } + [JsonPropertyName("has_message_files")] public bool HasMessageFiles { get; set; } diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/Enums/ChatEvent.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/Enums/ChatEvent.cs new file mode 100644 index 00000000..0d68fef7 --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Conversations/Enums/ChatEvent.cs @@ -0,0 +1,22 @@ +namespace BotSharp.Abstraction.Conversations.Enums; + +public static class ChatEvent +{ + public const string OnConversationInitFromClient = nameof(OnConversationInitFromClient); + public const string OnMessageReceivedFromClient = nameof(OnMessageReceivedFromClient); + public const string OnMessageReceivedFromAssistant = nameof(OnMessageReceivedFromAssistant); + + public const string OnMessageDeleted = nameof(OnMessageDeleted); + public const string OnNotificationGenerated = nameof(OnNotificationGenerated); + public const string OnIndicationReceived = nameof(OnIndicationReceived); + + public const string OnConversationContentLogGenerated = nameof(OnConversationContentLogGenerated); + public const string OnConversateStateLogGenerated = nameof(OnConversateStateLogGenerated); + public const string OnAgentQueueChanged = nameof(OnAgentQueueChanged); + public const string OnStateChangeGenerated = nameof(OnStateChangeGenerated); + + public const string BeforeReceiveLlmStreamMessage = nameof(BeforeReceiveLlmStreamMessage); + public const string OnReceiveLlmStreamMessage = nameof(OnReceiveLlmStreamMessage); + public const string AfterReceiveLlmStreamMessage = nameof(AfterReceiveLlmStreamMessage); + public const string OnSenderActionGenerated = nameof(OnSenderActionGenerated); +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/Models/ConversationSenderActionModel.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/Models/ConversationSenderActionModel.cs index b49fc55d..502cca4c 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/Models/ConversationSenderActionModel.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Conversations/Models/ConversationSenderActionModel.cs @@ -1,5 +1,3 @@ -using BotSharp.Abstraction.Messaging.Enums; - namespace BotSharp.Abstraction.Conversations.Models; public class ConversationSenderActionModel diff --git a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Models/HubObserveData.cs similarity index 75% rename from src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs rename to src/Infrastructure/BotSharp.Abstraction/MessageHub/Models/HubObserveData.cs index bf771cb4..403adbc6 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Models/HubObserveData.cs @@ -1,4 +1,4 @@ -namespace BotSharp.Abstraction.Observables.Models; +namespace BotSharp.Abstraction.MessageHub.Models; public class HubObserveData : ObserveDataBase { diff --git a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Models/ObserveDataBase.cs similarity index 68% rename from src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs rename to src/Infrastructure/BotSharp.Abstraction/MessageHub/Models/ObserveDataBase.cs index 17773272..62c2950d 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MessageHub/Models/ObserveDataBase.cs @@ -1,4 +1,4 @@ -namespace BotSharp.Abstraction.Observables.Models; +namespace BotSharp.Abstraction.MessageHub.Models; public abstract class ObserveDataBase { diff --git a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs b/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs index a559877f..2f0af290 100644 --- a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs +++ b/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs @@ -10,9 +10,9 @@ using BotSharp.Core.Messaging; using BotSharp.Core.Routing.Reasoning; using BotSharp.Core.Templating; using BotSharp.Core.Translation; -using BotSharp.Core.Observables.Queues; using Microsoft.Extensions.Configuration; -using BotSharp.Abstraction.Observables.Models; +using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Core.MessageHub; namespace BotSharp.Core.Conversations; diff --git a/src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs b/src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs index cb6bcd88..b8bb0f39 100644 --- a/src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs +++ b/src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs @@ -1,4 +1,5 @@ using BotSharp.Abstraction.Functions; +using BotSharp.Core.MessageHub; namespace BotSharp.Core.Demo.Functions; @@ -16,6 +17,31 @@ public class GetWeatherFn : IFunctionCallback public async Task Execute(RoleDialogModel message) { + var conv = _services.GetRequiredService(); + var messageHub = _services.GetRequiredService>(); + + await Task.Delay(1000); + + message.Indication = "Start querying weather data"; + messageHub.Push(new() + { + EventName = ChatEvent.OnIndicationReceived, + Data = message, + ServiceProvider = _services + }); + + await Task.Delay(1500); + + message.Indication = "Still working on it"; + messageHub.Push(new() + { + EventName = ChatEvent.OnIndicationReceived, + Data = message, + ServiceProvider = _services + }); + + await Task.Delay(1500); + message.Content = $"It is a sunny day!"; message.StopCompletion = false; return true; diff --git a/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs b/src/Infrastructure/BotSharp.Core/MessageHub/MessageHub.cs similarity index 88% rename from src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs rename to src/Infrastructure/BotSharp.Core/MessageHub/MessageHub.cs index affb142d..a104f4f6 100644 --- a/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs +++ b/src/Infrastructure/BotSharp.Core/MessageHub/MessageHub.cs @@ -1,11 +1,11 @@ using System.Reactive.Subjects; -namespace BotSharp.Core.Observables.Queues; +namespace BotSharp.Core.MessageHub; public class MessageHub where T : class { private readonly ILogger> _logger; - private readonly ISubject _observable = new Subject(); + private readonly ISubject _observable = Subject.Synchronize(new Subject()); public IObservable Events => _observable; public MessageHub(ILogger> logger) diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs index 1001dede..0922dcdb 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs +++ b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs @@ -1,4 +1,4 @@ -using BotSharp.Abstraction.Hooks; +using BotSharp.Core.MessageHub; using BotSharp.Core.Routing.Executor; namespace BotSharp.Core.Routing; @@ -23,15 +23,22 @@ public partial class RoutingService // Clone message var clonedMessage = RoleDialogModel.From(message); clonedMessage.FunctionName = name; - - var progressService = _services.GetService(); clonedMessage.Indication = await funcExecutor.GetIndicatorAsync(message); - if (progressService?.OnFunctionExecuting != null) + //var progressService = _services.GetService(); + //if (progressService?.OnFunctionExecuting != null) + //{ + // await progressService.OnFunctionExecuting(clonedMessage); + //} + + var messageHub = _services.GetRequiredService>(); + messageHub.Push(new() { - await progressService.OnFunctionExecuting(clonedMessage); - } - + EventName = ChatEvent.OnIndicationReceived, + Data = clonedMessage, + ServiceProvider = _services + }); + var hooks = _services.GetHooksOrderByPriority(clonedMessage.CurrentAgentId); foreach (var hook in hooks) { diff --git a/src/Infrastructure/BotSharp.Core/Using.cs b/src/Infrastructure/BotSharp.Core/Using.cs index 0e8e2762..f67baef0 100644 --- a/src/Infrastructure/BotSharp.Core/Using.cs +++ b/src/Infrastructure/BotSharp.Core/Using.cs @@ -36,6 +36,9 @@ global using BotSharp.Abstraction.Loggers.Services; global using BotSharp.Abstraction.Infrastructures.Events; global using BotSharp.Abstraction.Templating.Constants; global using BotSharp.Abstraction.Realtime.Models.Session; +global using BotSharp.Abstraction.Conversations.Enums; +global using BotSharp.Abstraction.Hooks; +global using BotSharp.Abstraction.MessageHub.Models; global using BotSharp.Core.Agents.Services; global using BotSharp.Core.Conversations.Services; global using BotSharp.Core.Infrastructures; diff --git a/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs index 605bceb8..d3c6e14a 100644 --- a/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs @@ -1,9 +1,9 @@ using Azure; using BotSharp.Abstraction.Files.Utilities; using BotSharp.Abstraction.Hooks; -using BotSharp.Abstraction.Observables.Models; +using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Core.Infrastructures.Streams; -using BotSharp.Core.Observables.Queues; +using BotSharp.Core.MessageHub; using OpenAI.Chat; using System.ClientModel; diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs b/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs index d3c3f2ac..d771d0a2 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs @@ -1,6 +1,6 @@ using BotSharp.Abstraction.Crontab; -using BotSharp.Abstraction.Observables.Models; -using BotSharp.Core.Observables.Queues; +using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Core.MessageHub; using BotSharp.Plugin.ChatHub.Hooks; using BotSharp.Plugin.ChatHub.Observers; using Microsoft.AspNetCore.Builder; diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatHubHelper.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatHubHelper.cs new file mode 100644 index 00000000..94bbc90d --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatHubHelper.cs @@ -0,0 +1,38 @@ +using Microsoft.AspNetCore.SignalR; +using System.Runtime.CompilerServices; + +namespace BotSharp.Plugin.ChatHub.Helpers; + +public class ChatHubHelper +{ + public static async Task SendChatEvent( + IServiceProvider services, + ILogger logger, + string @event, + string conversationId, + string userId, + T data, + string callerClass = "", + [CallerMemberName] string callerMethod = "", + LogLevel logLevel = LogLevel.Warning) + { + try + { + var settings = services.GetRequiredService(); + var chatHub = services.GetRequiredService>(); + + if (settings.EventDispatchBy == EventDispatchType.Group) + { + await chatHub.Clients.Group(conversationId).SendAsync(@event, data); + } + else + { + await chatHub.Clients.User(userId).SendAsync(@event, data); + } + } + catch (Exception ex) + { + logger.Log(logLevel, ex, $"Failed to send event '{@event}' in ({callerClass}-{callerMethod}) (conversation id: {conversationId})"); + } + } +} diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs index 1d641554..5cbc117e 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs @@ -1,8 +1,10 @@ using BotSharp.Abstraction.Conversations.Dtos; +using BotSharp.Abstraction.Conversations.Enums; using BotSharp.Abstraction.Routing.Enums; using BotSharp.Abstraction.SideCar; using BotSharp.Abstraction.Users.Dtos; using Microsoft.AspNetCore.SignalR; +using System.Runtime.CompilerServices; namespace BotSharp.Plugin.ChatHub.Hooks; @@ -15,15 +17,6 @@ public class ChatHubConversationHook : ConversationHookBase private readonly BotSharpOptions _options; private readonly ChatHubSettings _settings; - #region Events - private const string INIT_CLIENT_CONVERSATION = "OnConversationInitFromClient"; - private const string RECEIVE_CLIENT_MESSAGE = "OnMessageReceivedFromClient"; - private const string RECEIVE_ASSISTANT_MESSAGE = "OnMessageReceivedFromAssistant"; - private const string GENERATE_SENDER_ACTION = "OnSenderActionGenerated"; - private const string DELETE_MESSAGE = "OnMessageDeleted"; - private const string GENERATE_NOTIFICATION = "OnNotificationGenerated"; - #endregion - public ChatHubConversationHook( IServiceProvider services, IHubContext chatHub, @@ -51,7 +44,8 @@ public class ChatHubConversationHook : ConversationHookBase var user = await userService.GetUser(conv.User.Id); conv.User = UserDto.FromUser(user); - await InitClientConversation(conv.Id, conv); + //await InitClientConversation(conv.Id, conv); + await SendEvent(ChatEvent.OnConversationInitFromClient, conv.Id, conv); await base.OnConversationInitialized(conversation); } @@ -72,7 +66,7 @@ public class ChatHubConversationHook : ConversationHookBase Text = !string.IsNullOrEmpty(message.SecondaryContent) ? message.SecondaryContent : message.Content, Sender = UserDto.FromUser(sender) }; - await ReceiveClientMessage(conv.ConversationId, model); + await SendEvent(ChatEvent.OnMessageReceivedFromClient, conv.ConversationId, model); // Send typing-on to client var action = new ConversationSenderActionModel @@ -80,22 +74,12 @@ public class ChatHubConversationHook : ConversationHookBase ConversationId = conv.ConversationId, SenderAction = SenderActionEnum.TypingOn }; - - await GenerateSenderAction(conv.ConversationId, action); + await SendEvent(ChatEvent.OnSenderActionGenerated, conv.ConversationId, action); await base.OnMessageReceived(message); } public override async Task OnFunctionExecuting(RoleDialogModel message, string from = InvokeSource.Manual) { - var conv = _services.GetRequiredService(); - var action = new ConversationSenderActionModel - { - ConversationId = conv.ConversationId, - SenderAction = SenderActionEnum.TypingOn, - Indication = message.Indication - }; - - await GenerateSenderAction(conv.ConversationId, action); await base.OnFunctionExecuting(message, from: from); } @@ -110,7 +94,7 @@ public class ChatHubConversationHook : ConversationHookBase var conv = _services.GetRequiredService(); var state = _services.GetRequiredService(); - var json = JsonSerializer.Serialize(new ChatResponseDto() + var data = new ChatResponseDto() { ConversationId = conv.ConversationId, MessageId = message.MessageId, @@ -126,21 +110,20 @@ public class ChatHubConversationHook : ConversationHookBase LastName = "Assistant", Role = AgentRole.Assistant } - }, _options.JsonSerializerOptions); - - // Send typing-off to client - var action = new ConversationSenderActionModel - { - ConversationId = conv.ConversationId, - SenderAction = SenderActionEnum.TypingOff }; + // Send typing-off to client if (!message.IsStreaming) { - await GenerateSenderAction(conv.ConversationId, action); + var action = new ConversationSenderActionModel + { + ConversationId = conv.ConversationId, + SenderAction = SenderActionEnum.TypingOff + }; + await SendEvent(ChatEvent.OnSenderActionGenerated, conv.ConversationId, action); } - await ReceiveAssistantMessage(conv.ConversationId, json); + await SendEvent(ChatEvent.OnMessageReceivedFromAssistant, conv.ConversationId, data); await base.OnResponseGenerated(message); } @@ -148,7 +131,7 @@ public class ChatHubConversationHook : ConversationHookBase public override async Task OnNotificationGenerated(RoleDialogModel message) { var conv = _services.GetRequiredService(); - var json = JsonSerializer.Serialize(new ChatResponseDto() + var data = new ChatResponseDto() { ConversationId = conv.ConversationId, MessageId = message.MessageId, @@ -162,9 +145,9 @@ public class ChatHubConversationHook : ConversationHookBase LastName = "Assistant", Role = AgentRole.Assistant } - }, _options.JsonSerializerOptions); + }; - await GenerateNotification(conv.ConversationId, json); + await SendEvent(ChatEvent.OnNotificationGenerated, conv.ConversationId, data); await base.OnNotificationGenerated(message); } @@ -177,7 +160,7 @@ public class ChatHubConversationHook : ConversationHookBase MessageId = messageId }; - await DeleteMessage(conversationId, model); + await SendEvent(ChatEvent.OnMessageDeleted, conversationId, model); await base.OnMessageDeleted(conversationId, messageId); } @@ -188,119 +171,10 @@ public class ChatHubConversationHook : ConversationHookBase return sidecar == null || !sidecar.IsEnabled; } - private async Task InitClientConversation(string conversationId, ConversationDto conversation) + private async Task SendEvent(string @event, string conversationId, T data, [CallerMemberName] string callerName = "") { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(INIT_CLIENT_CONVERSATION, conversation); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(INIT_CLIENT_CONVERSATION, conversation); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to init client conversation in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})"); - } - } - - private async Task ReceiveClientMessage(string conversationId, ChatResponseDto model) - { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(RECEIVE_CLIENT_MESSAGE, model); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(RECEIVE_CLIENT_MESSAGE, model); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to receive assistant message in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})"); - } - } - - private async Task ReceiveAssistantMessage(string conversationId, string? json) - { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(RECEIVE_ASSISTANT_MESSAGE, json); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(RECEIVE_ASSISTANT_MESSAGE, json); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to receive assistant message in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})"); - } - - } - - private async Task GenerateSenderAction(string conversationId, ConversationSenderActionModel action) - { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(GENERATE_SENDER_ACTION, action); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(GENERATE_SENDER_ACTION, action); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to generate sender action in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})"); - } - } - - private async Task DeleteMessage(string conversationId, ChatResponseDto model) - { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(DELETE_MESSAGE, model); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(DELETE_MESSAGE, model); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to delete message in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})"); - } - } - - private async Task GenerateNotification(string conversationId, string? json) - { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(GENERATE_NOTIFICATION, json); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(GENERATE_NOTIFICATION, json); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to generate notification in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})"); - } + var user = _services.GetRequiredService(); + await ChatHubHelper.SendChatEvent(_services, _logger, @event, conversationId, user?.Id, data, nameof(ChatHubConversationHook), callerName); } #endregion } diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubCrontabHook.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubCrontabHook.cs index ef956670..f965d28e 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubCrontabHook.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubCrontabHook.cs @@ -1,4 +1,5 @@ using BotSharp.Abstraction.Conversations.Dtos; +using BotSharp.Abstraction.Conversations.Enums; using BotSharp.Abstraction.Crontab; using BotSharp.Abstraction.Crontab.Models; using Microsoft.AspNetCore.SignalR; @@ -14,11 +15,8 @@ public class ChatHubCrontabHook : ICrontabHook private readonly BotSharpOptions _options; private readonly ChatHubSettings _settings; - #region Events - private const string GENERATE_NOTIFICATION = "OnNotificationGenerated"; - #endregion - - public ChatHubCrontabHook(IServiceProvider services, + public ChatHubCrontabHook( + IServiceProvider services, IHubContext chatHub, ILogger logger, IUserIdentity user, @@ -35,7 +33,7 @@ public class ChatHubCrontabHook : ICrontabHook public async Task OnCronTriggered(CrontabItem item) { - var json = JsonSerializer.Serialize(new ChatResponseDto() + var data = new ChatResponseDto() { ConversationId = item.ConversationId, MessageId = Guid.NewGuid().ToString(), @@ -47,16 +45,16 @@ public class ChatHubCrontabHook : ICrontabHook LastName = "AI", Role = AgentRole.Assistant } - }, _options.JsonSerializerOptions); + }; - await SendEvent(item, json); + await SendEvent(item, data); } - private async Task SendEvent(CrontabItem item, string json) + private async Task SendEvent(CrontabItem item, ChatResponseDto data) { try { - await _chatHub.Clients.User(item.UserId).SendAsync(GENERATE_NOTIFICATION, json); + await _chatHub.Clients.User(item.UserId).SendAsync(ChatEvent.OnNotificationGenerated, data); } catch { } } diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/StreamingLogHook.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/StreamingLogHook.cs index 5dd89a2b..f036deab 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/StreamingLogHook.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/StreamingLogHook.cs @@ -1,5 +1,7 @@ +using BotSharp.Abstraction.Conversations.Enums; using BotSharp.Abstraction.Routing.Enums; using Microsoft.AspNetCore.SignalR; +using System.Runtime.CompilerServices; using System.Text.Encodings.Web; using System.Text.Unicode; @@ -19,13 +21,6 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR private readonly IAgentService _agentService; private readonly IRoutingContext _routingCtx; - #region Events - private const string CONTENT_LOG_GENERATED = "OnConversationContentLogGenerated"; - private const string STATE_LOG_GENERATED = "OnConversateStateLogGenerated"; - private const string AGENT_QUEUE_CHANGED = "OnAgentQueueChanged"; - private const string STATE_CHANGED = "OnStateChangeGenerated"; - #endregion - public StreamingLogHook( ConversationSetting convSettings, BotSharpOptions options, @@ -65,7 +60,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.UserInput, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public override async Task OnPostbackMessageReceived(RoleDialogModel message, PostbackMessageModel replyMsg) @@ -83,7 +79,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.UserInput, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public async Task OnSessionUpdated(Agent agent, string instruction, FunctionDef[] functions, bool isInit = false) @@ -112,7 +109,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.Prompt, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public async Task OnRenderingTemplate(Agent agent, string name, string content) @@ -134,7 +132,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.HardRule, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public async Task BeforeGenerating(Agent agent, List conversations) @@ -162,7 +161,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.FunctionCall, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public override async Task OnFunctionExecuted(RoleDialogModel message, string from = InvokeSource.Manual) @@ -183,7 +183,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.FunctionCall, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } /// @@ -210,7 +211,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.Prompt, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } /// @@ -225,7 +227,9 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR var conv = _services.GetRequiredService(); var routingCtx = _services.GetRequiredService(); - await SendStateLog(conv.ConversationId, routingCtx.EntryAgentId, _state.GetStates(), message); + var stateLog = BuildStateLog(conv.ConversationId, routingCtx.EntryAgentId, _state.GetStates(), message); + //await SendStateLog(conv.ConversationId, routingCtx.EntryAgentId, _state.GetStates(), message); + await SendEvent(ChatEvent.OnConversateStateLogGenerated, conv.ConversationId, stateLog); if (message.Role == AgentRole.Assistant) { @@ -244,7 +248,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.AgentResponse, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } } @@ -262,7 +267,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.FunctionCall, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public override async Task OnConversationEnding(RoleDialogModel message) @@ -279,7 +285,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.FunctionCall, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public override async Task OnBreakpointUpdated(string conversationId, bool resetStates) @@ -307,7 +314,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR }, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public override async Task OnStateChanged(StateChangeModel stateChange) @@ -317,7 +325,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR if (stateChange == null) return; - await SendStateChange(conversationId, stateChange); + //await SendStateChange(conversationId, stateChange); + await SendEvent(ChatEvent.OnStateChangeGenerated, conversationId, BuildStateChangeLog(stateChange)); } #endregion @@ -331,7 +340,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR // Agent queue log var log = $"{agent.Name} is enqueued"; - await SendAgentQueueLog(conversationId, log); + //await SendAgentQueueLog(conversationId, log); + await SendEvent(ChatEvent.OnAgentQueueChanged, conversationId, BuildAgentQueueChangedLog(conversationId, log)); // Content log log = $"{agent.Name} is enqueued{(reason != null ? $" ({reason})" : "")}"; @@ -346,7 +356,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.HardRule, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public async Task OnAgentDequeued(string agentId, string currentAgentId, string? reason = null) @@ -359,7 +370,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR // Agent queue log var log = $"{agent.Name} is dequeued"; - await SendAgentQueueLog(conversationId, log); + //await SendAgentQueueLog(conversationId, log); + await SendEvent(ChatEvent.OnAgentQueueChanged, conversationId, BuildAgentQueueChangedLog(conversationId, log)); // Content log log = $"{agent.Name} is dequeued{(reason != null ? $" ({reason})" : "")}, current agent is {currentAgent?.Name}"; @@ -374,7 +386,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.HardRule, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public async Task OnAgentReplaced(string fromAgentId, string toAgentId, string? reason = null) @@ -387,7 +400,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR // Agent queue log var log = $"Agent queue is replaced from {fromAgent.Name} to {toAgent.Name}"; - await SendAgentQueueLog(conversationId, log); + //await SendAgentQueueLog(conversationId, log); + await SendEvent(ChatEvent.OnAgentQueueChanged, conversationId, BuildAgentQueueChangedLog(conversationId, log)); // Content log log = $"{fromAgent.Name} is replaced to {toAgent.Name}{(reason != null ? $" ({reason})" : "")}"; @@ -402,7 +416,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.HardRule, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public async Task OnAgentQueueEmptied(string agentId, string? reason = null) @@ -412,7 +427,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR // Agent queue log var log = $"Agent queue is empty"; - await SendAgentQueueLog(conversationId, log); + //await SendAgentQueueLog(conversationId, log); + await SendEvent(ChatEvent.OnAgentQueueChanged, conversationId, BuildAgentQueueChangedLog(conversationId, log)); // Content log log = reason ?? "Agent queue is cleared"; @@ -427,7 +443,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.HardRule, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public async Task OnRoutingInstructionReceived(FunctionCallFromLlm instruct, RoleDialogModel message) @@ -446,7 +463,8 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.AgentResponse, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } public async Task OnRoutingInstructionRevised(FunctionCallFromLlm instruct, RoleDialogModel message) @@ -464,90 +482,20 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR Source = ContentLogSource.HardRule, Log = log }; - await SendContentLog(conversationId, input); + //await SendContentLog(conversationId, input); + await SendEvent(ChatEvent.OnConversationContentLogGenerated, conversationId, BuildContentLog(input)); } #endregion #region Private methods - private async Task SendContentLog(string conversationId, ContentLogInputModel input) + private async Task SendEvent(string @event, string conversationId, T data, [CallerMemberName] string callerName = "") { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(CONTENT_LOG_GENERATED, BuildContentLog(input)); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(CONTENT_LOG_GENERATED, BuildContentLog(input)); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to send content log in {nameof(StreamingLogHook)} (conversation id: {conversationId})."); - } + var user = _services.GetRequiredService(); + await ChatHubHelper.SendChatEvent(_services, _logger, @event, conversationId, user?.Id, data, nameof(StreamingLogHook), callerName); } - private async Task SendStateLog(string conversationId, string agentId, Dictionary states, RoleDialogModel message) - { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(STATE_LOG_GENERATED, BuildStateLog(conversationId, agentId, states, message)); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(STATE_LOG_GENERATED, BuildStateLog(conversationId, agentId, states, message)); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to send state log in {nameof(StreamingLogHook)} (conversation id: {conversationId})."); - } - } - - private async Task SendAgentQueueLog(string conversationId, string log) - { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(AGENT_QUEUE_CHANGED, BuildAgentQueueChangedLog(conversationId, log)); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(AGENT_QUEUE_CHANGED, BuildAgentQueueChangedLog(conversationId, log)); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to send agent queue log in {nameof(StreamingLogHook)} (conversation id: {conversationId})."); - } - } - - private async Task SendStateChange(string conversationId, StateChangeModel stateChange) - { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(STATE_CHANGED, BuildStateChangeLog(stateChange)); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(STATE_CHANGED, BuildStateChangeLog(stateChange)); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to send state change in {nameof(StreamingLogHook)} (conversation id: {conversationId})."); - } - } - - - private string BuildContentLog(ContentLogInputModel input) + private ContentLogOutputModel BuildContentLog(ContentLogInputModel input) { var output = new ContentLogOutputModel { @@ -561,8 +509,6 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR CreatedTime = DateTime.UtcNow }; - var json = JsonSerializer.Serialize(output, _options.JsonSerializerOptions); - var convSettings = _services.GetRequiredService(); if (convSettings.EnableContentLog) { @@ -570,10 +516,10 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR db.SaveConversationContentLog(output); } - return json; + return output; } - private string BuildStateLog(string conversationId, string agentId, Dictionary states, RoleDialogModel message) + private ConversationStateLogModel BuildStateLog(string conversationId, string agentId, Dictionary states, RoleDialogModel message) { var log = new ConversationStateLogModel { @@ -591,10 +537,10 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR db.SaveConversationStateLog(log); } - return JsonSerializer.Serialize(log, _options.JsonSerializerOptions); + return log; } - private string BuildStateChangeLog(StateChangeModel stateChange) + private StateChangeOutputModel BuildStateChangeLog(StateChangeModel stateChange) { var log = new StateChangeOutputModel { @@ -611,10 +557,10 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR CreateTime = DateTime.UtcNow }; - return JsonSerializer.Serialize(log, _options.JsonSerializerOptions); + return log; } - private string BuildAgentQueueChangedLog(string conversationId, string log) + private AgentQueueChangedLogModel BuildAgentQueueChangedLog(string conversationId, string log) { var model = new AgentQueueChangedLogModel { @@ -623,7 +569,7 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR CreatedTime = DateTime.UtcNow }; - return JsonSerializer.Serialize(model, _options.JsonSerializerOptions); + return model; } private string GetMessageContent(RoleDialogModel message) diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/WelcomeHook.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/WelcomeHook.cs index 3ae8b313..3db8017b 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/WelcomeHook.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/WelcomeHook.cs @@ -1,5 +1,7 @@ using BotSharp.Abstraction.Conversations.Dtos; +using BotSharp.Abstraction.Conversations.Enums; using Microsoft.AspNetCore.SignalR; +using System.Runtime.CompilerServices; namespace BotSharp.Plugin.ChatHub.Hooks; @@ -13,11 +15,8 @@ public class WelcomeHook : ConversationHookBase private readonly BotSharpOptions _options; private readonly ChatHubSettings _settings; - #region Events - private const string RECEIVE_ASSISTANT_MESSAGE = "OnMessageReceivedFromAssistant"; - #endregion - - public WelcomeHook(IServiceProvider services, + public WelcomeHook( + IServiceProvider services, IHubContext chatHub, ILogger logger, IUserIdentity user, @@ -64,7 +63,7 @@ public class WelcomeHook : ConversationHookBase RichContent = richContent }; - var json = JsonSerializer.Serialize(new ChatResponseDto() + var data = new ChatResponseDto() { ConversationId = conversation.Id, MessageId = dialog.MessageId, @@ -76,35 +75,20 @@ public class WelcomeHook : ConversationHookBase LastName = "", Role = AgentRole.Assistant } - }, _options.JsonSerializerOptions); + }; await Task.Delay(300); - _storage.Append(conversation.Id, dialog); - - await SendEvent(conversation.Id, json); + await SendEvent(ChatEvent.OnMessageReceivedFromAssistant, conversation.Id, data); } } await base.OnUserAgentConnectedInitially(conversation); } - private async Task SendEvent(string conversationId, string json) + private async Task SendEvent(string @event, string conversationId, T data, [CallerMemberName] string callerName = "") { - try - { - if (_settings.EventDispatchBy == EventDispatchType.Group) - { - await _chatHub.Clients.Group(conversationId).SendAsync(RECEIVE_ASSISTANT_MESSAGE, json); - } - else - { - await _chatHub.Clients.User(_user.Id).SendAsync(RECEIVE_ASSISTANT_MESSAGE, json); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, $"Failed to send event in {nameof(WelcomeHook)} (conversation id: {conversationId})."); - } + var user = _services.GetRequiredService(); + await ChatHubHelper.SendChatEvent(_services, _logger, @event, conversationId, user?.Id, data, nameof(WelcomeHook), callerName); } } diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs index c15419ed..71b07b9d 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs @@ -1,5 +1,6 @@ using BotSharp.Abstraction.Conversations.Dtos; -using BotSharp.Abstraction.Observables.Models; +using BotSharp.Abstraction.Conversations.Enums; +using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Abstraction.SideCar; using BotSharp.Plugin.ChatHub.Hooks; using Microsoft.AspNetCore.SignalR; @@ -11,11 +12,6 @@ public class ChatHubObserver : IObserver private readonly ILogger _logger; private IServiceProvider _services; - private const string BEFORE_RECEIVE_LLM_STREAM_MESSAGE = "BeforeReceiveLlmStreamMessage"; - private const string ON_RECEIVE_LLM_STREAM_MESSAGE = "OnReceiveLlmStreamMessage"; - private const string AFTER_RECEIVE_LLM_STREAM_MESSAGE = "AfterReceiveLlmStreamMessage"; - private const string GENERATE_SENDER_ACTION = "OnSenderActionGenerated"; - public ChatHubObserver(ILogger logger) { _logger = logger; @@ -39,72 +35,86 @@ public class ChatHubObserver : IObserver var message = value.Data; var model = new ChatResponseDto(); - if (value.EventName == BEFORE_RECEIVE_LLM_STREAM_MESSAGE) + var action = new ConversationSenderActionModel(); + var conv = _services.GetRequiredService(); + + switch (value.EventName) { - var conv = _services.GetRequiredService(); - model = new ChatResponseDto() - { - ConversationId = conv.ConversationId, - MessageId = message.MessageId, - Text = string.Empty, - Sender = new() + case ChatEvent.BeforeReceiveLlmStreamMessage: + model = new ChatResponseDto() { - FirstName = "AI", - LastName = "Assistant", - Role = AgentRole.Assistant - } - }; + ConversationId = conv.ConversationId, + MessageId = message.MessageId, + Text = string.Empty, + Sender = new() + { + FirstName = "AI", + LastName = "Assistant", + Role = AgentRole.Assistant + } + }; - var action = new ConversationSenderActionModel - { - ConversationId = conv.ConversationId, - SenderAction = SenderActionEnum.TypingOn - }; - - GenerateSenderAction(conv.ConversationId, action); - } - else if (value.EventName == AFTER_RECEIVE_LLM_STREAM_MESSAGE && message.IsStreaming) - { - var conv = _services.GetRequiredService(); - model = new ChatResponseDto() - { - ConversationId = conv.ConversationId, - MessageId = message.MessageId, - Text = message.Content, - Sender = new() + action = new ConversationSenderActionModel { - FirstName = "AI", - LastName = "Assistant", - Role = AgentRole.Assistant - } - }; + ConversationId = conv.ConversationId, + SenderAction = SenderActionEnum.TypingOn + }; - var action = new ConversationSenderActionModel - { - ConversationId = conv.ConversationId, - SenderAction = SenderActionEnum.TypingOff - }; - - GenerateSenderAction(conv.ConversationId, action); - } - else if (value.EventName == ON_RECEIVE_LLM_STREAM_MESSAGE) - { - var conv = _services.GetRequiredService(); - model = new ChatResponseDto() - { - ConversationId = conv.ConversationId, - MessageId = message.MessageId, - Text = !string.IsNullOrEmpty(message.SecondaryContent) ? message.SecondaryContent : message.Content, - Function = message.FunctionName, - RichContent = message.SecondaryRichContent ?? message.RichContent, - Data = message.Data, - Sender = new() + GenerateSenderAction(conv.ConversationId, action); + break; + case ChatEvent.OnReceiveLlmStreamMessage: + model = new ChatResponseDto() { - FirstName = "AI", - LastName = "Assistant", - Role = AgentRole.Assistant - } - }; + ConversationId = conv.ConversationId, + MessageId = message.MessageId, + Text = !string.IsNullOrEmpty(message.SecondaryContent) ? message.SecondaryContent : message.Content, + Function = message.FunctionName, + RichContent = message.SecondaryRichContent ?? message.RichContent, + Data = message.Data, + Sender = new() + { + FirstName = "AI", + LastName = "Assistant", + Role = AgentRole.Assistant + } + }; + break; + case ChatEvent.AfterReceiveLlmStreamMessage: + model = new ChatResponseDto() + { + ConversationId = conv.ConversationId, + MessageId = message.MessageId, + Text = message.Content, + Sender = new() + { + FirstName = "AI", + LastName = "Assistant", + Role = AgentRole.Assistant + } + }; + + action = new ConversationSenderActionModel + { + ConversationId = conv.ConversationId, + SenderAction = SenderActionEnum.TypingOff + }; + + GenerateSenderAction(conv.ConversationId, action); + break; + case ChatEvent.OnIndicationReceived: + model = new ChatResponseDto + { + ConversationId = conv.ConversationId, + MessageId = message.MessageId, + Indication = message.Indication, + Sender = new() + { + FirstName = "AI", + LastName = "Assistant", + Role = AgentRole.Assistant + } + }; + break; } OnReceiveAssistantMessage(value.EventName, model.ConversationId, model); @@ -147,12 +157,12 @@ public class ChatHubObserver : IObserver var chatHub = _services.GetRequiredService>(); if (settings.EventDispatchBy == EventDispatchType.Group) { - chatHub.Clients.Group(conversationId).SendAsync(GENERATE_SENDER_ACTION, action).ConfigureAwait(false).GetAwaiter().GetResult(); + chatHub.Clients.Group(conversationId).SendAsync(ChatEvent.OnSenderActionGenerated, action).ConfigureAwait(false).GetAwaiter().GetResult(); } else { var user = _services.GetRequiredService(); - chatHub.Clients.User(user.Id).SendAsync(GENERATE_SENDER_ACTION, action).ConfigureAwait(false).GetAwaiter().GetResult(); + chatHub.Clients.User(user.Id).SendAsync(ChatEvent.OnSenderActionGenerated, action).ConfigureAwait(false).GetAwaiter().GetResult(); } } catch (Exception ex) diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Using.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Using.cs index 4710509c..3473bb6c 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Using.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Using.cs @@ -34,4 +34,5 @@ global using BotSharp.Abstraction.Realtime; global using BotSharp.Abstraction.Realtime.Models; global using BotSharp.Plugin.ChatHub.Settings; global using BotSharp.Plugin.ChatHub.Enums; -global using BotSharp.Plugin.ChatHub.Models.Stream; \ No newline at end of file +global using BotSharp.Plugin.ChatHub.Models.Stream; +global using BotSharp.Plugin.ChatHub.Helpers; \ No newline at end of file diff --git a/src/Plugins/BotSharp.Plugin.DeepSeekAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.DeepSeekAI/Providers/Chat/ChatCompletionProvider.cs index 1335d4dc..36a5126e 100644 --- a/src/Plugins/BotSharp.Plugin.DeepSeekAI/Providers/Chat/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.DeepSeekAI/Providers/Chat/ChatCompletionProvider.cs @@ -1,8 +1,8 @@ using BotSharp.Abstraction.Files; using BotSharp.Abstraction.Hooks; -using BotSharp.Abstraction.Observables.Models; +using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Core.Infrastructures.Streams; -using BotSharp.Core.Observables.Queues; +using BotSharp.Core.MessageHub; using BotSharp.Plugin.DeepSeek.Providers; using Microsoft.Extensions.Logging; using OpenAI.Chat; diff --git a/src/Plugins/BotSharp.Plugin.LLamaSharp/Providers/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.LLamaSharp/Providers/ChatCompletionProvider.cs index 321b9aee..edeb7f58 100644 --- a/src/Plugins/BotSharp.Plugin.LLamaSharp/Providers/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.LLamaSharp/Providers/ChatCompletionProvider.cs @@ -1,9 +1,9 @@ using BotSharp.Abstraction.Agents; using BotSharp.Abstraction.Hooks; using BotSharp.Abstraction.Loggers; -using BotSharp.Abstraction.Observables.Models; +using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Core.Infrastructures.Streams; -using BotSharp.Core.Observables.Queues; +using BotSharp.Core.MessageHub; using Microsoft.AspNetCore.SignalR; using static LLama.Common.ChatHistory; using static System.Net.Mime.MediaTypeNames; diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs index 57b44252..0e39cb34 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs @@ -1,8 +1,8 @@ using Azure; using BotSharp.Abstraction.Hooks; -using BotSharp.Abstraction.Observables.Models; +using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Core.Infrastructures.Streams; -using BotSharp.Core.Observables.Queues; +using BotSharp.Core.MessageHub; using BotSharp.Plugin.OpenAI.Models.Realtime; using Fluid; using OpenAI.Chat; diff --git a/src/Plugins/BotSharp.Plugin.SparkDesk/Providers/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.SparkDesk/Providers/ChatCompletionProvider.cs index 035e03e0..6e4a58f2 100644 --- a/src/Plugins/BotSharp.Plugin.SparkDesk/Providers/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.SparkDesk/Providers/ChatCompletionProvider.cs @@ -1,9 +1,9 @@ using BotSharp.Abstraction.Agents; using BotSharp.Abstraction.Agents.Enums; using BotSharp.Abstraction.Loggers; -using BotSharp.Abstraction.Observables.Models; +using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Core.Infrastructures.Streams; -using BotSharp.Core.Observables.Queues; +using BotSharp.Core.MessageHub; using Microsoft.AspNetCore.SignalR; namespace BotSharp.Plugin.SparkDesk.Providers;