temp save

src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs

@@ -36,10 +36,15 @@ public interface IConversationService
     /// <param name="onResponseReceived">Received the response from AI Agent</param>
     /// <returns></returns>
     Task<bool> SendMessage(string agentId,
-        RoleDialogModel lastDialog, 
+        RoleDialogModel message, 
         PostbackMessageModel? replyMessage,
         Func<RoleDialogModel, Task> onResponseReceived);
 
+
+    Task<bool> StreamMessage(string agentId,
+        RoleDialogModel lastDialog,
+        PostbackMessageModel? replyMessage);
+
     List<RoleDialogModel> GetDialogHistory(int lastCount = 100, bool fromBreakpoint = true, IEnumerable<string>? includeMessageTypes = null);
     Task CleanHistory(string agentId);
 

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!;
 }

src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs

@@ -41,6 +41,7 @@ public interface IRoutingService
     /// <param name="message"></param>
     /// <returns></returns>
     Task<RoleDialogModel> InstructDirect(Agent agent, RoleDialogModel message, List<RoleDialogModel> dialogs);
+    Task<bool> InstructStream(Agent agent, RoleDialogModel message, List<RoleDialogModel> dialogs);
 
     Task<string> GetConversationContent(List<RoleDialogModel> dialogs, int maxDialogCount = 100);
 

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<bool> StreamMessage(string agentId,
+        RoleDialogModel message,
+        PostbackMessageModel? replyMessage)
+    {
+        var conversation = await GetConversationRecordOrCreateNew(agentId);
+        var agentService = _services.GetRequiredService<IAgentService>();
+        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<IConversationService>();
+        var dialogs = conv.GetDialogHistory();
+
+        var statistics = _services.GetRequiredService<ITokenStatistics>();
+
+        RoleDialogModel response = message;
+        bool stopCompletion = false;
+
+        // Enqueue receiving agent first in case it stop completion by OnMessageReceived
+        var routing = _services.GetRequiredService<IRoutingService>();
+        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<IConversationHook>(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<RoutingSettings>();
+
+            // 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;
+    }
+}

src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs

@@ -20,7 +20,6 @@ public class MessageHub
     /// <param name="item"></param>
     public void Push(HubObserveData item)
     {
-        _logger.LogInformation($"Pushing item to observers: {item.Data.Content}");
         _observable.OnNext(item);
     }
 

src/Infrastructure/BotSharp.Core/Routing/RoutingService.InstructStream.cs

@@ -0,0 +1,49 @@
+namespace BotSharp.Core.Routing;
+
+public partial class RoutingService
+{
+    public async Task<bool> InstructStream(Agent agent, RoleDialogModel message, List<RoleDialogModel> dialogs)
+    {
+        var conv = _services.GetRequiredService<IConversationService>();
+        var storage = _services.GetRequiredService<IConversationStorage>();
+        storage.Append(conv.ConversationId, message);
+
+        dialogs.Add(message);
+        Context.SetDialogs(dialogs);
+
+        var routing = _services.GetRequiredService<IRoutingService>();
+        routing.Context.Push(agent.Id, "instruct directly");
+        var agentId = routing.Context.GetCurrentAgentId();
+
+        // Update next action agent's name
+        var agentService = _services.GetRequiredService<IAgentService>();
+
+        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<AgentSettings>();
+                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;
+    }
+}

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<IConversationService>();
+        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<IRoutingService>();
+        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)
     {

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<HubObserveData>
 {
     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<HubObserveData>
     public void OnNext(HubObserveData value)
     {
         _services = value.ServiceProvider;
-        _user = _services.GetRequiredService<IUserIdentity>();
+
-        
+        var message = value.Data;
-        ReceiveMessage(value.Data).ConfigureAwait(false).GetAwaiter().GetResult();
+        var model = new ChatResponseDto();
+        if (value.EventName == BEFORE_RECEIVE_LLM_STREAM_MESSAGE
+            || value.EventName == AFTER_RECEIVE_LLM_STREAM_MESSAGE)
+        {
+            var conv = _services.GetRequiredService<IConversationService>();
+            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<IConversationService>();
+            //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<IConversationStorage>();
+            //storage.Append(conv.ConversationId, message);
+        }
+        else if (value.EventName == ON_RECEIVE_LLM_STREAM_MESSAGE)
+        {
+            var conv = _services.GetRequiredService<IConversationService>();
+            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<IConversationService>();
-        var userService = _services.GetRequiredService<IUserService>();
-        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);
+        await OnReceiveAssistantMessage(ON_RECEIVE_LLM_STREAM_MESSAGE, conv.ConversationId, model);
-
-        // Send typing-on to client
-        var action = new ConversationSenderActionModel
-        {
-            ConversationId = conv.ConversationId,
-            SenderAction = SenderActionEnum.TypingOn
-        };
-
-        await GenerateSenderAction(conv.ConversationId, action);
     }
 
-    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<HubObserveData>
 
             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<IUserIdentity>();
+                await chatHub.Clients.User(user.Id).SendAsync(@event, model);
             }
         }
         catch (Exception ex)
@@ -108,7 +174,8 @@ public class ChatHubObserver : IObserver<HubObserveData>
             }
             else
             {
-                await chatHub.Clients.User(_user.Id).SendAsync(GENERATE_SENDER_ACTION, action);
+                var user = _services.GetRequiredService<IUserIdentity>();
+                await chatHub.Clients.User(user.Id).SendAsync(GENERATE_SENDER_ACTION, action);
             }
         }
         catch (Exception ex)

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<MessageHub>();
         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)
+                //await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update)
-                {
+                //{
-                    RenderedInstruction = string.Join("\r\n", renderedInstructions)
+                //    //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;
     }