From f4a74258c23edc595a81cabe185ea0252e8e3e8b Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Tue, 17 Jun 2025 17:57:57 -0500 Subject: [PATCH] temp save --- .../Conversations/IConversationService.cs | 7 +- .../Observables/Models/HubObserveData.cs | 1 + .../Routing/IRoutingService.cs | 1 + .../Services/ConversationService.Stream.cs | 86 ++++++++++++ .../Observables/Queues/MessageHub.cs | 1 - .../Routing/RoutingService.InstructStream.cs | 49 +++++++ .../Controllers/ConversationController.cs | 24 ++++ .../Observers/ChatHubObserver.cs | 125 ++++++++++++++---- .../Providers/Chat/ChatCompletionProvider.cs | 53 +++++++- 9 files changed, 309 insertions(+), 38 deletions(-) create mode 100644 src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Stream.cs create mode 100644 src/Infrastructure/BotSharp.Core/Routing/RoutingService.InstructStream.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs index 656b397d..fccab0fa 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs @@ -36,10 +36,15 @@ public interface IConversationService /// Received the response from AI Agent /// Task SendMessage(string agentId, - RoleDialogModel lastDialog, + RoleDialogModel message, PostbackMessageModel? replyMessage, Func onResponseReceived); + + Task StreamMessage(string agentId, + RoleDialogModel lastDialog, + PostbackMessageModel? replyMessage); + List GetDialogHistory(int lastCount = 100, bool fromBreakpoint = true, IEnumerable? includeMessageTypes = null); Task CleanHistory(string agentId); diff --git a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs b/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs index a6d10cce..bf771cb4 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs @@ -2,5 +2,6 @@ namespace BotSharp.Abstraction.Observables.Models; public class HubObserveData : ObserveDataBase { + public string EventName { get; set; } = null!; public RoleDialogModel Data { get; set; } = null!; } diff --git a/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs b/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs index 97be8874..5dbf8d3f 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs @@ -41,6 +41,7 @@ public interface IRoutingService /// /// Task InstructDirect(Agent agent, RoleDialogModel message, List dialogs); + Task InstructStream(Agent agent, RoleDialogModel message, List dialogs); Task GetConversationContent(List dialogs, int maxDialogCount = 100); diff --git a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Stream.cs b/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Stream.cs new file mode 100644 index 00000000..f650a28f --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Stream.cs @@ -0,0 +1,86 @@ +using BotSharp.Abstraction.Hooks; +using BotSharp.Abstraction.Infrastructures.Enums; +using BotSharp.Abstraction.Routing.Enums; +using BotSharp.Abstraction.Routing.Settings; + +namespace BotSharp.Core.Conversations.Services; + +public partial class ConversationService +{ + public async Task StreamMessage(string agentId, + RoleDialogModel message, + PostbackMessageModel? replyMessage) + { + var conversation = await GetConversationRecordOrCreateNew(agentId); + var agentService = _services.GetRequiredService(); + Agent agent = await agentService.LoadAgent(agentId); + + var content = $"Received [{agent.Name}] {message.Role}: {message.Content}"; + _logger.LogInformation(content); + + message.CurrentAgentId = agent.Id; + if (string.IsNullOrEmpty(message.SenderId)) + { + message.SenderId = _user.Id; + } + + var conv = _services.GetRequiredService(); + var dialogs = conv.GetDialogHistory(); + + var statistics = _services.GetRequiredService(); + + RoleDialogModel response = message; + bool stopCompletion = false; + + // Enqueue receiving agent first in case it stop completion by OnMessageReceived + var routing = _services.GetRequiredService(); + routing.Context.SetMessageId(_conversationId, message.MessageId); + + // Save payload in order to assign the payload before hook is invoked + if (replyMessage != null && !string.IsNullOrEmpty(replyMessage.Payload)) + { + message.Payload = replyMessage.Payload; + } + + var hooks = _services.GetHooksOrderByPriority(message.CurrentAgentId); + foreach (var hook in hooks) + { + hook.SetAgent(agent) + .SetConversation(conversation); + + if (replyMessage == null || string.IsNullOrEmpty(replyMessage.FunctionName)) + { + await hook.OnMessageReceived(message); + } + else + { + await hook.OnPostbackMessageReceived(message, replyMessage); + } + + // Interrupted by hook + if (message.StopCompletion) + { + stopCompletion = true; + routing.Context.Pop(); + break; + } + } + + if (!stopCompletion) + { + // Routing with reasoning + var settings = _services.GetRequiredService(); + + // reload agent in case it has been changed by hook + if (message.CurrentAgentId != agent.Id) + { + agent = await agentService.LoadAgent(message.CurrentAgentId); + } + + await routing.InstructStream(agent, message, dialogs); + routing.Context.ResetRecursiveCounter(); + } + + return true; + } +} diff --git a/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs b/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs index ed6772b4..9950a613 100644 --- a/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs +++ b/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs @@ -20,7 +20,6 @@ public class MessageHub /// public void Push(HubObserveData item) { - _logger.LogInformation($"Pushing item to observers: {item.Data.Content}"); _observable.OnNext(item); } diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InstructStream.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InstructStream.cs new file mode 100644 index 00000000..08c72be1 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InstructStream.cs @@ -0,0 +1,49 @@ +namespace BotSharp.Core.Routing; + +public partial class RoutingService +{ + public async Task InstructStream(Agent agent, RoleDialogModel message, List dialogs) + { + var conv = _services.GetRequiredService(); + var storage = _services.GetRequiredService(); + storage.Append(conv.ConversationId, message); + + dialogs.Add(message); + Context.SetDialogs(dialogs); + + var routing = _services.GetRequiredService(); + routing.Context.Push(agent.Id, "instruct directly"); + var agentId = routing.Context.GetCurrentAgentId(); + + // Update next action agent's name + var agentService = _services.GetRequiredService(); + + if (agent.Disabled) + { + var content = $"This agent ({agent.Name}) is disabled, please install the corresponding plugin ({agent.Plugin.Name}) to activate this agent."; + + message = RoleDialogModel.From(message, role: AgentRole.Assistant, content: content); + dialogs.Add(message); + } + else + { + var provider = agent.LlmConfig.Provider; + var model = agent.LlmConfig.Model; + + if (provider == null || model == null) + { + var agentSettings = _services.GetRequiredService(); + provider = agentSettings.LlmConfig.Provider; + model = agentSettings.LlmConfig.Model; + } + + var chatCompletion = CompletionProvider.GetChatCompletion(_services, + provider: provider, + model: model); + + await chatCompletion.GetChatCompletionsStreamingAsync(agent, dialogs, async data => { }); + } + + return true; + } +} diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs index d3da550f..a3cedc1e 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs @@ -377,6 +377,30 @@ public class ConversationController : ControllerBase return response; } + + [HttpPost("/conversation/{agentId}/{conversationId}/stream")] + public async Task StreamMessage( + [FromRoute] string agentId, + [FromRoute] string conversationId, + [FromBody] NewMessageModel input) + { + var conv = _services.GetRequiredService(); + var inputMsg = new RoleDialogModel(AgentRole.User, input.Text) + { + MessageId = !string.IsNullOrWhiteSpace(input.InputMessageId) ? input.InputMessageId : Guid.NewGuid().ToString(), + CreatedAt = DateTime.UtcNow + }; + + var routing = _services.GetRequiredService(); + routing.Context.SetMessageId(conversationId, inputMsg.MessageId); + + conv.SetConversationId(conversationId, input.States); + SetStates(conv, input); + + await conv.StreamMessage(agentId, inputMsg, replyMessage: input.Postback); + } + + [HttpPost("/conversation/{agentId}/{conversationId}/sse")] public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string conversationId, [FromBody] NewMessageModel input) { diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs index 5699945f..dcf332b0 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs @@ -1,7 +1,6 @@ using BotSharp.Abstraction.Conversations.Dtos; using BotSharp.Abstraction.Observables.Models; using BotSharp.Abstraction.SideCar; -using BotSharp.Abstraction.Users.Dtos; using BotSharp.Plugin.ChatHub.Hooks; using Microsoft.AspNetCore.SignalR; @@ -11,9 +10,10 @@ public class ChatHubObserver : IObserver { private readonly ILogger _logger; private IServiceProvider _services; - private IUserIdentity _user; - private const string RECEIVE_CLIENT_MESSAGE = "OnMessageReceivedFromClient"; + 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) @@ -34,41 +34,106 @@ public class ChatHubObserver : IObserver public void OnNext(HubObserveData value) { _services = value.ServiceProvider; - _user = _services.GetRequiredService(); - - ReceiveMessage(value.Data).ConfigureAwait(false).GetAwaiter().GetResult(); + + var message = value.Data; + var model = new ChatResponseDto(); + if (value.EventName == BEFORE_RECEIVE_LLM_STREAM_MESSAGE + || value.EventName == AFTER_RECEIVE_LLM_STREAM_MESSAGE) + { + var conv = _services.GetRequiredService(); + model = new ChatResponseDto() + { + 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 = value.EventName == BEFORE_RECEIVE_LLM_STREAM_MESSAGE ? SenderActionEnum.TypingOn : SenderActionEnum.TypingOff + }; + + GenerateSenderAction(conv.ConversationId, action).ConfigureAwait(false).GetAwaiter().GetResult(); + } + else if (value.EventName == AFTER_RECEIVE_LLM_STREAM_MESSAGE) + { + //var conv = _services.GetRequiredService(); + //model = new ChatResponseDto() + //{ + // 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.TypingOff + //}; + + //GenerateSenderAction(conv.ConversationId, action).ConfigureAwait(false).GetAwaiter().GetResult(); + + //var storage = _services.GetRequiredService(); + //storage.Append(conv.ConversationId, message); + } + 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() + { + FirstName = "AI", + LastName = "Assistant", + Role = AgentRole.Assistant + } + }; + } + + OnReceiveAssistantMessage(value.EventName, model.ConversationId, model).ConfigureAwait(false).GetAwaiter().GetResult(); } - private async Task ReceiveMessage(RoleDialogModel message) + private async Task ReceiveLlmStreamResponse(RoleDialogModel message) { - if (!AllowSendingMessage()) return; - var conv = _services.GetRequiredService(); - var userService = _services.GetRequiredService(); - var sender = await userService.GetMyProfile(); - - // Update console conversation UI for CSR var model = new ChatResponseDto() { ConversationId = conv.ConversationId, MessageId = message.MessageId, - Payload = message.Payload, Text = !string.IsNullOrEmpty(message.SecondaryContent) ? message.SecondaryContent : message.Content, - Sender = UserDto.FromUser(sender) + Function = message.FunctionName, + RichContent = message.SecondaryRichContent ?? message.RichContent, + Data = message.Data, + Sender = new() + { + FirstName = "AI", + LastName = "Assistant", + Role = AgentRole.Assistant + } }; - await ReceiveClientMessage(conv.ConversationId, model); - - // Send typing-on to client - var action = new ConversationSenderActionModel - { - ConversationId = conv.ConversationId, - SenderAction = SenderActionEnum.TypingOn - }; - - await GenerateSenderAction(conv.ConversationId, action); + await OnReceiveAssistantMessage(ON_RECEIVE_LLM_STREAM_MESSAGE, conv.ConversationId, model); } - private async Task ReceiveClientMessage(string conversationId, ChatResponseDto model) + private async Task OnReceiveAssistantMessage(string @event, string conversationId, ChatResponseDto model) { try { @@ -77,11 +142,12 @@ public class ChatHubObserver : IObserver if (settings.EventDispatchBy == EventDispatchType.Group) { - await chatHub.Clients.Group(conversationId).SendAsync(RECEIVE_CLIENT_MESSAGE, model); + await chatHub.Clients.Group(conversationId).SendAsync(@event, model); } else { - await chatHub.Clients.User(_user.Id).SendAsync(RECEIVE_CLIENT_MESSAGE, model); + var user = _services.GetRequiredService(); + await chatHub.Clients.User(user.Id).SendAsync(@event, model); } } catch (Exception ex) @@ -108,7 +174,8 @@ public class ChatHubObserver : IObserver } else { - await chatHub.Clients.User(_user.Id).SendAsync(GENERATE_SENDER_ACTION, action); + var user = _services.GetRequiredService(); + await chatHub.Clients.User(user.Id).SendAsync(GENERATE_SENDER_ACTION, action); } } catch (Exception ex) diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs index 17ee1012..83a96266 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs @@ -1,4 +1,6 @@ using BotSharp.Abstraction.Hooks; +using BotSharp.Core.Observables.Queues; +using ModelContextProtocol.Protocol.Types; using OpenAI.Chat; namespace BotSharp.Plugin.OpenAI.Providers.Chat; @@ -185,7 +187,20 @@ public class ChatCompletionProvider : IChatCompletion var chatClient = client.GetChatClient(_model); var (prompt, messages, options) = PrepareOptions(agent, conversations); + var hub = _services.GetRequiredService(); var response = chatClient.CompleteChatStreamingAsync(messages, options); + var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty; + + hub.Push(new() + { + ServiceProvider = _services, + EventName = "BeforeReceiveLlmStreamMessage", + Data = new RoleDialogModel(AgentRole.Assistant, string.Empty) + { + CurrentAgentId = agent.Id, + MessageId = messageId + } + }); await foreach (var choice in response) { @@ -194,23 +209,47 @@ public class ChatCompletionProvider : IChatCompletion var update = choice.ToolCallUpdates?.FirstOrDefault()?.FunctionArgumentsUpdate?.ToString() ?? string.Empty; _logger.LogInformation(update); - await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update) - { - RenderedInstruction = string.Join("\r\n", renderedInstructions) - }); + //await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update) + //{ + // //RenderedInstruction = string.Join("\r\n", renderedInstructions) + //}); continue; } if (choice.ContentUpdate.IsNullOrEmpty()) continue; - _logger.LogInformation(choice.ContentUpdate[0]?.Text); + var text = choice.ContentUpdate[0]?.Text ?? string.Empty; + _logger.LogInformation(text); - await onMessageReceived(new RoleDialogModel(choice.Role?.ToString() ?? ChatMessageRole.Assistant.ToString(), choice.ContentUpdate[0]?.Text ?? string.Empty) + var content = new RoleDialogModel(AgentRole.Assistant, text) { - RenderedInstruction = string.Join("\r\n", renderedInstructions) + CurrentAgentId = agent.Id, + MessageId = messageId + }; + hub.Push(new() + { + ServiceProvider = _services, + EventName = "OnReceiveLlmStreamMessage", + Data = content }); + + //await onMessageReceived(new RoleDialogModel(choice.Role?.ToString() ?? ChatMessageRole.Assistant.ToString(), choice.ContentUpdate[0]?.Text ?? string.Empty) + //{ + // RenderedInstruction = string.Join("\r\n", renderedInstructions) + //}); } + hub.Push(new() + { + ServiceProvider = _services, + EventName = "AfterReceiveLlmStreamMessage", + Data = new RoleDialogModel(AgentRole.Assistant, string.Empty) + { + CurrentAgentId = agent.Id, + MessageId = messageId + } + }); + return true; }