From eeb4f06f639638c2f0ef641282f332d541a6de8a Mon Sep 17 00:00:00 2001 From: Jicheng Lu Date: Tue, 17 Jun 2025 22:23:43 -0500 Subject: [PATCH] refine stream --- .../Observers/ChatHubObserver.cs | 72 +++++++------------ .../Providers/Chat/ChatCompletionProvider.cs | 4 +- 2 files changed, 29 insertions(+), 47 deletions(-) diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs index dcf332b0..c7a852fa 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs @@ -23,7 +23,7 @@ public class ChatHubObserver : IObserver public void OnCompleted() { - _logger.LogInformation($"{nameof(ChatHubObserver)} receives complete notification."); + _logger.LogWarning($"{nameof(ChatHubObserver)} receives complete notification."); } public void OnError(Exception error) @@ -35,10 +35,11 @@ public class ChatHubObserver : IObserver { _services = value.ServiceProvider; + if (!AllowSendingMessage()) return; + var message = value.Data; var model = new ChatResponseDto(); - if (value.EventName == BEFORE_RECEIVE_LLM_STREAM_MESSAGE - || value.EventName == AFTER_RECEIVE_LLM_STREAM_MESSAGE) + if (value.EventName == BEFORE_RECEIVE_LLM_STREAM_MESSAGE) { var conv = _services.GetRequiredService(); model = new ChatResponseDto() @@ -57,37 +58,37 @@ public class ChatHubObserver : IObserver var action = new ConversationSenderActionModel { ConversationId = conv.ConversationId, - SenderAction = value.EventName == BEFORE_RECEIVE_LLM_STREAM_MESSAGE ? SenderActionEnum.TypingOn : SenderActionEnum.TypingOff + SenderAction = SenderActionEnum.TypingOn }; 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 conv = _services.GetRequiredService(); + model = new ChatResponseDto() + { + ConversationId = conv.ConversationId, + MessageId = message.MessageId, + Text = message.Content, + Sender = new() + { + FirstName = "AI", + LastName = "Assistant", + Role = AgentRole.Assistant + } + }; - //var action = new ConversationSenderActionModel - //{ - // ConversationId = conv.ConversationId, - // SenderAction = SenderActionEnum.TypingOff - //}; + var action = new ConversationSenderActionModel + { + ConversationId = conv.ConversationId, + SenderAction = SenderActionEnum.TypingOff + }; - //GenerateSenderAction(conv.ConversationId, action).ConfigureAwait(false).GetAwaiter().GetResult(); + GenerateSenderAction(conv.ConversationId, action).ConfigureAwait(false).GetAwaiter().GetResult(); - //var storage = _services.GetRequiredService(); - //storage.Append(conv.ConversationId, message); + var storage = _services.GetRequiredService(); + storage.Append(conv.ConversationId, message); } else if (value.EventName == ON_RECEIVE_LLM_STREAM_MESSAGE) { @@ -112,27 +113,6 @@ public class ChatHubObserver : IObserver OnReceiveAssistantMessage(value.EventName, model.ConversationId, model).ConfigureAwait(false).GetAwaiter().GetResult(); } - private async Task ReceiveLlmStreamResponse(RoleDialogModel message) - { - var conv = _services.GetRequiredService(); - var 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 - } - }; - await OnReceiveAssistantMessage(ON_RECEIVE_LLM_STREAM_MESSAGE, conv.ConversationId, model); - } - private async Task OnReceiveAssistantMessage(string @event, string conversationId, ChatResponseDto model) { try diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs index 83a96266..67ac7c66 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs @@ -190,6 +190,7 @@ public class ChatCompletionProvider : IChatCompletion var hub = _services.GetRequiredService(); var response = chatClient.CompleteChatStreamingAsync(messages, options); var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty; + var allText = string.Empty; hub.Push(new() { @@ -219,6 +220,7 @@ public class ChatCompletionProvider : IChatCompletion if (choice.ContentUpdate.IsNullOrEmpty()) continue; var text = choice.ContentUpdate[0]?.Text ?? string.Empty; + allText += text; _logger.LogInformation(text); var content = new RoleDialogModel(AgentRole.Assistant, text) @@ -243,7 +245,7 @@ public class ChatCompletionProvider : IChatCompletion { ServiceProvider = _services, EventName = "AfterReceiveLlmStreamMessage", - Data = new RoleDialogModel(AgentRole.Assistant, string.Empty) + Data = new RoleDialogModel(AgentRole.Assistant, allText) { CurrentAgentId = agent.Id, MessageId = messageId