Merge pull request #1084 from iceljc/features/init-reactive-x

Features/init reactive x

Directory.Packages.props

@@ -26,6 +26,7 @@
     <PackageVersion Include="System.IdentityModel.Tokens.Jwt" Version="8.0.0" />
     <PackageVersion Include="System.Memory.Data" Version="8.0.0" />
     <PackageVersion Include="System.Text.Json" Version="8.0.5" />
+    <PackageVersion Include="System.Reactive" Version="6.0.1" />
     <PackageVersion Include="Serilog.Sinks.Console" Version="6.0.0" />
     <PackageVersion Include="Serilog.Extensions.Logging" Version="9.0.0" />
     <PackageVersion Include="Serilog.Sinks.File" Version="6.0.0" />

src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj

@@ -36,6 +36,7 @@
     <PackageReference Include="System.IdentityModel.Tokens.Jwt" />
     <PackageReference Include="System.Memory.Data" />
     <PackageReference Include="System.Text.Json" />
+    <PackageReference Include="System.Reactive" />
     <PackageReference Include="Serilog.Sinks.Console" />
     <PackageReference Include="Serilog.Sinks.File" />
     <PackageReference Include="Rougamo.Fody" />

src/Infrastructure/BotSharp.Abstraction/Conversations/Dtos/ChatResponseDto.cs

@@ -35,6 +35,9 @@ public class ChatResponseDto : InstructResult
     [JsonPropertyName("has_message_files")]
     public bool HasMessageFiles { get; set; }
 
+    [JsonPropertyName("is_streaming")]
+    public bool IsStreaming { get; set; }
+
     [JsonPropertyName("created_at")]
     public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
 }

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

@@ -36,7 +36,7 @@ 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);
 

src/Infrastructure/BotSharp.Abstraction/Conversations/Models/RoleDialogModel.cs

@@ -117,6 +117,9 @@ public class RoleDialogModel : ITrackableMessage
     [JsonIgnore(Condition = JsonIgnoreCondition.Always)]
     public string RenderedInstruction { get; set; } = string.Empty;
 
+    [JsonIgnore(Condition = JsonIgnoreCondition.Always)]
+    public bool IsStreaming { get; set; }
+
     private RoleDialogModel()
     {
     }
@@ -159,7 +162,8 @@ public class RoleDialogModel : ITrackableMessage
             Payload = source.Payload,
             StopCompletion = source.StopCompletion,
             Instruction = source.Instruction,
-            Data = source.Data
+            Data = source.Data,
+            IsStreaming = source.IsStreaming
         };
     }
 }

src/Infrastructure/BotSharp.Abstraction/MLTasks/IChatCompletion.cs

@@ -23,7 +23,6 @@ public interface IChatCompletion
         Func<RoleDialogModel, Task> onMessageReceived,
         Func<RoleDialogModel, Task> onFunctionExecuting);
 
-    Task<bool> GetChatCompletionsStreamingAsync(Agent agent, 
-        List<RoleDialogModel> conversations, 
-        Func<RoleDialogModel, Task> onMessageReceived);
+    Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, 
+        List<RoleDialogModel> conversations);
 }

src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs

@@ -0,0 +1,7 @@
+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/Observables/Models/ObserveDataBase.cs

@@ -0,0 +1,6 @@
+namespace BotSharp.Abstraction.Observables.Models;
+
+public abstract class ObserveDataBase
+{
+    public IServiceProvider ServiceProvider { get; set; } = null!;
+}

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

@@ -26,11 +26,7 @@ public interface IRoutingService
     /// <returns></returns>
     RoutingRule[] GetRulesByAgentId(string id);
 
-    //void ResetRecursiveCounter();
-    //int GetRecursiveCounter();
-    //void SetRecursiveCounter(int counter);
-
-    Task<bool> InvokeAgent(string agentId, List<RoleDialogModel> dialogs, string from = InvokeSource.Manual);
+    Task<bool> InvokeAgent(string agentId, List<RoleDialogModel> dialogs, string from = InvokeSource.Manual, bool useStream = false);
     Task<bool> InvokeFunction(string name, RoleDialogModel messages, string from = InvokeSource.Manual);
     Task<RoleDialogModel> InstructLoop(Agent agent, RoleDialogModel message, List<RoleDialogModel> dialogs);
 

src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs

@@ -10,7 +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;
 
 namespace BotSharp.Core.Conversations;
 
@@ -41,6 +43,8 @@ public class ConversationPlugin : IBotSharpPlugin
             return settingService.Bind<GoogleApiSettings>("GoogleApi");
         });
 
+        services.AddSingleton<MessageHub<HubObserveData>>();
+
         services.AddScoped<IConversationStorage, ConversationStorage>();
         services.AddScoped<IConversationService, ConversationService>();
         services.AddScoped<IConversationProgressService, ConversationProgressService>();

src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs

@@ -17,7 +17,7 @@ public class GetWeatherFn : IFunctionCallback
     public async Task<bool> Execute(RoleDialogModel message)
     {
         message.Content = $"It is a sunny day!";
-        //message.StopCompletion = true;
+        message.StopCompletion = false;
         return true;
     }
 }

src/Infrastructure/BotSharp.Core/Infrastructures/Streams/RealtimeTextStream.cs

@@ -1,12 +1,12 @@
 using System.IO;
 
-namespace BotSharp.Plugin.GoogleAI.Models.Realtime;
+namespace BotSharp.Core.Infrastructures.Streams;
 
-internal class RealtimeTranscriptionResponse : IDisposable
+public class RealtimeTextStream : IDisposable
 {
-    public RealtimeTranscriptionResponse()
+    public RealtimeTextStream()
     {
-        
+
     }
 
     private bool _disposed = false;
@@ -20,6 +20,13 @@ internal class RealtimeTranscriptionResponse : IDisposable
         }
     }
 
+    public long Length => _contentStream.Length;
+
+    public bool IsNullOrEmpty()
+    {
+        return _contentStream == null || Length == 0;
+    }
+
     public void Collect(string text)
     {
         if (_disposed) return;

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

@@ -0,0 +1,43 @@
+using System.Reactive.Subjects;
+
+namespace BotSharp.Core.Observables.Queues;
+
+public class MessageHub<T> where T : class
+{
+    private readonly ILogger<MessageHub<T>> _logger;
+    private readonly ISubject<T> _observable = new Subject<T>();
+    public IObservable<T> Events => _observable;
+
+    public MessageHub(ILogger<MessageHub<T>> logger)
+    {
+        _logger = logger;
+    }
+
+    /// <summary>
+    /// Push an item to the observers.
+    /// </summary>
+    /// <param name="item"></param>
+    public void Push(T item)
+    {
+        _observable.OnNext(item);
+    }
+
+    /// <summary>
+    /// Send a complete notification to the observers.
+    /// This will stop the observers from receiving data.
+    /// </summary>
+    public void Complete()
+    {
+        _observable.OnCompleted();
+    }
+
+    /// <summary>
+    /// Send an error notification to the observers.
+    /// This will stop the observers from receiving data.
+    /// </summary>
+    /// <param name="error"></param>
+    public void Error(Exception error)
+    {
+        _observable.OnError(error);
+    }
+}

src/Infrastructure/BotSharp.Core/Routing/Reasoning/InstructExecutor.cs

@@ -57,7 +57,13 @@ public class InstructExecutor : IExecutor
         }
         else
         {
-            var ret = await routing.InvokeAgent(agentId, dialogs, from: InvokeSource.Routing);
+            var state = _services.GetRequiredService<IConversationStateService>();
+            var useStreamMsg = state.GetState("use_stream_message");
+            var ret = await routing.InvokeAgent(
+                agentId,
+                dialogs,
+                from: InvokeSource.Routing,
+                useStream: bool.TryParse(useStreamMsg, out var useStream) && useStream);
         }
 
         var response = dialogs.Last();

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

@@ -4,7 +4,11 @@ namespace BotSharp.Core.Routing;
 
 public partial class RoutingService
 {
-    public async Task<bool> InvokeAgent(string agentId, List<RoleDialogModel> dialogs, string from = InvokeSource.Manual)
+    public async Task<bool> InvokeAgent(
+        string agentId,
+        List<RoleDialogModel> dialogs,
+        string from = InvokeSource.Manual,
+        bool useStream = false)
     {
         var agentService = _services.GetRequiredService<IAgentService>();
         var agent = await agentService.LoadAgent(agentId);
@@ -30,8 +34,16 @@ public partial class RoutingService
             provider: provider,
             model: model);
 
+        RoleDialogModel response;
         var message = dialogs.Last();
-        var response = await chatCompletion.GetChatCompletions(agent, dialogs);
+        if (useStream)
+        {
+            response = await chatCompletion.GetChatCompletionsStreamingAsync(agent, dialogs);
+        }
+        else
+        {
+            response = await chatCompletion.GetChatCompletions(agent, dialogs);
+        }
 
         if (response.Role == AgentRole.Function)
         {
@@ -45,8 +57,9 @@ public partial class RoutingService
             message.FunctionArgs = response.FunctionArgs;
             message.Indication = response.Indication;
             message.CurrentAgentId = agent.Id;
+            message.IsStreaming = response.IsStreaming;
 
-            await InvokeFunction(message, dialogs, from: from);
+            await InvokeFunction(message, dialogs, from: from, useStream: useStream);
         }
         else
         {
@@ -59,6 +72,7 @@ public partial class RoutingService
 
             message = RoleDialogModel.From(message, role: AgentRole.Assistant, content: response.Content);
             message.CurrentAgentId = agent.Id;
+            message.IsStreaming = response.IsStreaming;
             dialogs.Add(message);
             Context.SetDialogs(dialogs);
         }
@@ -66,7 +80,11 @@ public partial class RoutingService
         return true;
     }
 
-    private async Task<bool> InvokeFunction(RoleDialogModel message, List<RoleDialogModel> dialogs, string from)
+    private async Task<bool> InvokeFunction(
+        RoleDialogModel message,
+        List<RoleDialogModel> dialogs,
+        string from,
+        bool useStream)
     {
         // execute function
         // Save states
@@ -102,7 +120,7 @@ public partial class RoutingService
 
                 // Send to Next LLM
                 var curAgentId = routing.Context.GetCurrentAgentId();
-                await InvokeAgent(curAgentId, dialogs, from);
+                await InvokeAgent(curAgentId, dialogs, from, useStream);
             }
         }
         else

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

@@ -51,7 +51,13 @@ public partial class RoutingService : IRoutingService
         }
         else
         {
-            var ret = await routing.InvokeAgent(agentId, dialogs, from: InvokeSource.Routing);
+            var state = _services.GetRequiredService<IConversationStateService>();
+            var useStreamMsg = state.GetState("use_stream_message");
+            var ret = await routing.InvokeAgent(
+                agentId,
+                dialogs,
+                from: InvokeSource.Routing,
+                useStream: bool.TryParse(useStreamMsg, out var useStream) && useStream);
         }
 
         var response = dialogs.Last();

src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs

@@ -377,6 +377,7 @@ public class ConversationController : ControllerBase
         return response;
     }
 
+
     [HttpPost("/conversation/{agentId}/{conversationId}/sse")]
     public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string conversationId, [FromBody] NewMessageModel input)
     {

src/Plugins/BotSharp.Plugin.AnthropicAI/Providers/ChatCompletionProvider.cs

@@ -96,8 +96,7 @@ public class ChatCompletionProvider : IChatCompletion
         throw new NotImplementedException();
     }
 
-    public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations,
-        Func<RoleDialogModel, Task> onMessageReceived)
+    public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
         throw new NotImplementedException();
     }

src/Plugins/BotSharp.Plugin.AzureOpenAI/BotSharp.Plugin.AzureOpenAI.csproj

@@ -16,7 +16,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <ProjectReference Include="..\..\Infrastructure\BotSharp.Abstraction\BotSharp.Abstraction.csproj" />
+    <ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
   </ItemGroup>
 
 </Project>

src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs

@@ -1,6 +1,9 @@
 using Azure;
 using BotSharp.Abstraction.Files.Utilities;
 using BotSharp.Abstraction.Hooks;
+using BotSharp.Abstraction.Observables.Models;
+using BotSharp.Core.Infrastructures.Streams;
+using BotSharp.Core.Observables.Queues;
 using OpenAI.Chat;
 using System.ClientModel;
 
@@ -203,39 +206,133 @@ public class ChatCompletionProvider : IChatCompletion
         return true;
     }
 
-    public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+    public async Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
         var client = ProviderHelper.GetClient(Provider, _model, _services);
         var chatClient = client.GetChatClient(_model);
         var (prompt, messages, options) = PrepareOptions(agent, conversations);
 
-        var response = chatClient.CompleteChatStreamingAsync(messages, options);
+        var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
+        var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty;
 
-        await foreach (var choice in response)
+        var contentHooks = _services.GetHooks<IContentGeneratingHook>(agent.Id);
+        // Before chat completion hook
+        foreach (var hook in contentHooks)
         {
-            if (choice.FinishReason == ChatFinishReason.FunctionCall || choice.FinishReason == ChatFinishReason.ToolCalls)
-            {
-                var update = choice.ToolCallUpdates?.FirstOrDefault()?.FunctionArgumentsUpdate?.ToString() ?? string.Empty;
-                Console.Write(update);
+            await hook.BeforeGenerating(agent, conversations);
+        }
 
-                await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update)
-                {
-                    RenderedInstruction = string.Join("\r\n", renderedInstructions)
-                });
-                continue;
+        hub.Push(new()
+        {
+            ServiceProvider = _services,
+            EventName = "BeforeReceiveLlmStreamMessage",
+            Data = new RoleDialogModel(AgentRole.Assistant, string.Empty)
+            {
+                CurrentAgentId = agent.Id,
+                MessageId = messageId
+            }
+        });
+
+        using var textStream = new RealtimeTextStream();
+        var toolCalls = new List<StreamingChatToolCallUpdate>();
+        ChatTokenUsage? tokenUsage = null;
+
+        var responseMessage = new RoleDialogModel(AgentRole.Assistant, string.Empty)
+        {
+            CurrentAgentId = agent.Id,
+            MessageId = messageId
+        };
+
+        await foreach (var choice in chatClient.CompleteChatStreamingAsync(messages, options))
+        {
+            tokenUsage = choice.Usage;
+
+            if (!choice.ToolCallUpdates.IsNullOrEmpty())
+            {
+                toolCalls.AddRange(choice.ToolCallUpdates);
             }
 
-            if (choice.ContentUpdate.IsNullOrEmpty()) continue;
-
-            _logger.LogInformation(choice.ContentUpdate[0]?.Text);
-
-            await onMessageReceived(new RoleDialogModel(choice.Role?.ToString() ?? ChatMessageRole.Assistant.ToString(), choice.ContentUpdate[0]?.Text ?? string.Empty)
+            if (!choice.ContentUpdate.IsNullOrEmpty())
             {
-                RenderedInstruction = string.Join("\r\n", renderedInstructions)
+                var text = choice.ContentUpdate[0]?.Text ?? string.Empty;
+                textStream.Collect(text);
+
+#if DEBUG
+                _logger.LogCritical($"Content update: {text}");
+#endif
+
+                var content = new RoleDialogModel(AgentRole.Assistant, text)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId
+                };
+                hub.Push(new()
+                {
+                    ServiceProvider = _services,
+                    EventName = "OnReceiveLlmStreamMessage",
+                    Data = content
+                });
+            }
+
+            if (choice.FinishReason == ChatFinishReason.ToolCalls || choice.FinishReason == ChatFinishReason.FunctionCall)
+            {
+                var meta = toolCalls.FirstOrDefault(x => !string.IsNullOrEmpty(x.FunctionName));
+                var functionName = meta?.FunctionName;
+                var toolCallId = meta?.ToolCallId;
+                var args = toolCalls.Where(x => x.FunctionArgumentsUpdate != null).Select(x => x.FunctionArgumentsUpdate.ToString()).ToList();
+                var functionArgument = string.Join(string.Empty, args);
+
+#if DEBUG
+                _logger.LogCritical($"Tool Call (id: {toolCallId}) => {functionName}({functionArgument})");
+#endif
+
+                responseMessage = new RoleDialogModel(AgentRole.Function, string.Empty)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId,
+                    ToolCallId = toolCallId,
+                    FunctionName = functionName,
+                    FunctionArgs = functionArgument
+                };
+            }
+            else if (choice.FinishReason.HasValue)
+            {
+                var allText = textStream.GetText();
+                _logger.LogCritical($"Text Content: {allText}");
+
+                responseMessage = new RoleDialogModel(AgentRole.Assistant, allText)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId,
+                    IsStreaming = true
+                };
+            }
+        }
+
+        hub.Push(new()
+        {
+            ServiceProvider = _services,
+            EventName = "AfterReceiveLlmStreamMessage",
+            Data = responseMessage
+        });
+
+
+        var inputTokenDetails = tokenUsage?.InputTokenDetails;
+        // After chat completion hook
+        foreach (var hook in contentHooks)
+        {
+            await hook.AfterGenerated(responseMessage, new TokenStatsModel
+            {
+                Prompt = prompt,
+                Provider = Provider,
+                Model = _model,
+                TextInputTokens = (tokenUsage?.InputTokenCount ?? 0) - (inputTokenDetails?.CachedTokenCount ?? 0),
+                CachedTextInputTokens = inputTokenDetails?.CachedTokenCount ?? 0,
+                TextOutputTokens = tokenUsage?.OutputTokenCount ?? 0
             });
         }
 
-        return true;
+        return responseMessage;
     }
 
     protected (string, IEnumerable<ChatMessage>, ChatCompletionOptions) PrepareOptions(Agent agent, List<RoleDialogModel> conversations)

src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs

@@ -1,5 +1,9 @@
 using BotSharp.Abstraction.Crontab;
+using BotSharp.Abstraction.Observables.Models;
+using BotSharp.Core.Observables.Queues;
 using BotSharp.Plugin.ChatHub.Hooks;
+using BotSharp.Plugin.ChatHub.Observers;
+using Microsoft.AspNetCore.Builder;
 using Microsoft.Extensions.Configuration;
 
 namespace BotSharp.Plugin.ChatHub;
@@ -7,7 +11,7 @@ namespace BotSharp.Plugin.ChatHub;
 /// <summary>
 /// The dialogue channel connects users, AI assistants and customer service representatives.
 /// </summary>
-public class ChatHubPlugin : IBotSharpPlugin
+public class ChatHubPlugin : IBotSharpPlugin, IBotSharpAppPlugin
 {
     public string Id => "6e52d42d-1e23-406b-8599-36af36c83209";
     public string Name => "Chat Hub";
@@ -28,4 +32,12 @@ public class ChatHubPlugin : IBotSharpPlugin
         services.AddScoped<IContentGeneratingHook, StreamingLogHook>();
         services.AddScoped<ICrontabHook, ChatHubCrontabHook>();
     }
+
+    public void Configure(IApplicationBuilder app)
+    {
+        var services = app.ApplicationServices;
+        var queue = services.GetRequiredService<MessageHub<HubObserveData>>();
+        var logger = services.GetRequiredService<ILogger<MessageHub<HubObserveData>>>();
+        queue.Events.Subscribe(new ChatHubObserver(logger));
+    }
 }

src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs

@@ -119,6 +119,7 @@ public class ChatHubConversationHook : ConversationHookBase
             RichContent = message.SecondaryRichContent ?? message.RichContent,
             Data = message.Data,
             States = state.GetStates(),
+            IsStreaming = message.IsStreaming,
             Sender = new()
             {
                 FirstName = "AI",
@@ -134,7 +135,11 @@ public class ChatHubConversationHook : ConversationHookBase
             SenderAction = SenderActionEnum.TypingOff
         };
 
-        await GenerateSenderAction(conv.ConversationId, action);
+        if (!message.IsStreaming)
+        {
+            await GenerateSenderAction(conv.ConversationId, action);
+        }
+        
         await ReceiveAssistantMessage(conv.ConversationId, json);
         await base.OnResponseGenerated(message);
     }

src/Plugins/BotSharp.Plugin.ChatHub/Hooks/StreamingLogHook.cs

@@ -75,7 +75,7 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR
 
         var log = $"{GetMessageContent(message)}";
         var replyContent = JsonSerializer.Serialize(replyMsg, _options.JsonSerializerOptions);
-        log += $"\r\n```json\r\n{replyContent}\r\n```";
+        log += $"\r\n\r\n```json\r\n{replyContent}\r\n```";
 
         var input = new ContentLogInputModel(conversationId, message)
         {
@@ -234,7 +234,7 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR
             if (message.RichContent != null || message.SecondaryRichContent != null)
             {
                 var richContent = JsonSerializer.Serialize(message.SecondaryRichContent ?? message.RichContent, _localJsonOptions);
-                log += $"\r\n```json\r\n{richContent}\r\n```";
+                log += $"\r\n\r\n```json\r\n{richContent}\r\n```";
             }
 
             var input = new ContentLogInputModel(conv.ConversationId, message)

src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs

@@ -0,0 +1,163 @@
+using BotSharp.Abstraction.Conversations.Dtos;
+using BotSharp.Abstraction.Observables.Models;
+using BotSharp.Abstraction.SideCar;
+using BotSharp.Plugin.ChatHub.Hooks;
+using Microsoft.AspNetCore.SignalR;
+
+namespace BotSharp.Plugin.ChatHub.Observers;
+
+public class ChatHubObserver : IObserver<HubObserveData>
+{
+    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;
+    }
+
+    public void OnCompleted()
+    {
+        _logger.LogWarning($"{nameof(ChatHubObserver)} receives complete notification.");
+    }
+
+    public void OnError(Exception error)
+    {
+        _logger.LogError(error, $"{nameof(ChatHubObserver)} receives error notification: {error.Message}");
+    }
+
+    public void OnNext(HubObserveData value)
+    {
+        _services = value.ServiceProvider;
+
+        if (!AllowSendingMessage()) return;
+
+        var message = value.Data;
+        var model = new ChatResponseDto();
+        if (value.EventName == BEFORE_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.TypingOn
+            };
+
+            GenerateSenderAction(conv.ConversationId, action);
+        }
+        else if (value.EventName == AFTER_RECEIVE_LLM_STREAM_MESSAGE && message.IsStreaming)
+        {
+            var conv = _services.GetRequiredService<IConversationService>();
+            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
+            };
+
+            GenerateSenderAction(conv.ConversationId, action);
+        }
+        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);
+    }
+
+    private bool AllowSendingMessage()
+    {
+        var sidecar = _services.GetService<IConversationSideCar>();
+        return sidecar == null || !sidecar.IsEnabled;
+    }
+
+    private void OnReceiveAssistantMessage(string @event, string conversationId, ChatResponseDto model)
+    {
+        try
+        {
+            var settings = _services.GetRequiredService<ChatHubSettings>();
+            var chatHub = _services.GetRequiredService<IHubContext<SignalRHub>>();
+
+            if (settings.EventDispatchBy == EventDispatchType.Group)
+            {
+                chatHub.Clients.Group(conversationId).SendAsync(@event, model).ConfigureAwait(false).GetAwaiter().GetResult();
+            }
+            else
+            {
+                var user = _services.GetRequiredService<IUserIdentity>();
+                chatHub.Clients.User(user.Id).SendAsync(@event, model).ConfigureAwait(false).GetAwaiter().GetResult();
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, $"Failed to receive assistant message in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})");
+        }
+    }
+
+    private void GenerateSenderAction(string conversationId, ConversationSenderActionModel action)
+    {
+        try
+        {
+            var settings = _services.GetRequiredService<ChatHubSettings>();
+            var chatHub = _services.GetRequiredService<IHubContext<SignalRHub>>();
+            if (settings.EventDispatchBy == EventDispatchType.Group)
+            {
+                chatHub.Clients.Group(conversationId).SendAsync(GENERATE_SENDER_ACTION, action).ConfigureAwait(false).GetAwaiter().GetResult();
+            }
+            else
+            {
+                var user = _services.GetRequiredService<IUserIdentity>();
+                chatHub.Clients.User(user.Id).SendAsync(GENERATE_SENDER_ACTION, action).ConfigureAwait(false).GetAwaiter().GetResult();
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, $"Failed to generate sender action in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})");
+        }
+    }
+}

src/Plugins/BotSharp.Plugin.DeepSeekAI/BotSharp.Plugin.DeepSeekAI.csproj

@@ -15,7 +15,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <ProjectReference Include="..\..\Infrastructure\BotSharp.Abstraction\BotSharp.Abstraction.csproj" />
+    <ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
   </ItemGroup>
 
 </Project>

src/Plugins/BotSharp.Plugin.DeepSeekAI/Providers/Chat/ChatCompletionProvider.cs

@@ -1,8 +1,11 @@
+using BotSharp.Abstraction.Files;
+using BotSharp.Abstraction.Hooks;
+using BotSharp.Abstraction.Observables.Models;
+using BotSharp.Core.Infrastructures.Streams;
+using BotSharp.Core.Observables.Queues;
+using BotSharp.Plugin.DeepSeek.Providers;
 using Microsoft.Extensions.Logging;
 using OpenAI.Chat;
-using BotSharp.Abstraction.Files;
-using BotSharp.Plugin.DeepSeek.Providers;
-using BotSharp.Abstraction.Hooks;
 
 namespace BotSharp.Plugin.DeepSeekAI.Providers.Chat;
 
@@ -170,39 +173,133 @@ public class ChatCompletionProvider : IChatCompletion
         return true;
     }
 
-    public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+    public async Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
         var client = ProviderHelper.GetClient(Provider, _model, _services);
         var chatClient = client.GetChatClient(_model);
         var (prompt, messages, options) = PrepareOptions(agent, conversations);
 
-        var response = chatClient.CompleteChatStreamingAsync(messages, options);
+        var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
+        var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty;
 
-        await foreach (var choice in response)
+        var contentHooks = _services.GetHooks<IContentGeneratingHook>(agent.Id);
+        // Before chat completion hook
+        foreach (var hook in contentHooks)
         {
-            if (choice.FinishReason == ChatFinishReason.FunctionCall || choice.FinishReason == ChatFinishReason.ToolCalls)
-            {
-                var update = choice.ToolCallUpdates?.FirstOrDefault()?.FunctionArgumentsUpdate?.ToString() ?? string.Empty;
-                _logger.LogInformation(update);
+            await hook.BeforeGenerating(agent, conversations);
+        }
 
-                await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update)
-                {
-                    RenderedInstruction = string.Join("\r\n", renderedInstructions)
-                });
-                continue;
+        hub.Push(new()
+        {
+            ServiceProvider = _services,
+            EventName = "BeforeReceiveLlmStreamMessage",
+            Data = new RoleDialogModel(AgentRole.Assistant, string.Empty)
+            {
+                CurrentAgentId = agent.Id,
+                MessageId = messageId
+            }
+        });
+
+        using var textStream = new RealtimeTextStream();
+        var toolCalls = new List<StreamingChatToolCallUpdate>();
+        ChatTokenUsage? tokenUsage = null;
+
+        var responseMessage = new RoleDialogModel(AgentRole.Assistant, string.Empty)
+        {
+            CurrentAgentId = agent.Id,
+            MessageId = messageId
+        };
+
+        await foreach (var choice in chatClient.CompleteChatStreamingAsync(messages, options))
+        {
+            tokenUsage = choice.Usage;
+
+            if (!choice.ToolCallUpdates.IsNullOrEmpty())
+            {
+                toolCalls.AddRange(choice.ToolCallUpdates);
             }
 
-            if (choice.ContentUpdate.IsNullOrEmpty()) continue;
-
-            _logger.LogInformation(choice.ContentUpdate[0]?.Text);
-
-            await onMessageReceived(new RoleDialogModel(choice.Role?.ToString() ?? ChatMessageRole.Assistant.ToString(), choice.ContentUpdate[0]?.Text ?? string.Empty)
+            if (!choice.ContentUpdate.IsNullOrEmpty())
             {
-                RenderedInstruction = string.Join("\r\n", renderedInstructions)
+                var text = choice.ContentUpdate[0]?.Text ?? string.Empty;
+                textStream.Collect(text);
+
+#if DEBUG
+                _logger.LogCritical($"Content update: {text}");
+#endif
+
+                var content = new RoleDialogModel(AgentRole.Assistant, text)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId
+                };
+                hub.Push(new()
+                {
+                    ServiceProvider = _services,
+                    EventName = "OnReceiveLlmStreamMessage",
+                    Data = content
+                });
+            }
+
+            if (choice.FinishReason == ChatFinishReason.ToolCalls || choice.FinishReason == ChatFinishReason.FunctionCall)
+            {
+                var meta = toolCalls.FirstOrDefault(x => !string.IsNullOrEmpty(x.FunctionName));
+                var functionName = meta?.FunctionName;
+                var toolCallId = meta?.ToolCallId;
+                var args = toolCalls.Where(x => x.FunctionArgumentsUpdate != null).Select(x => x.FunctionArgumentsUpdate.ToString()).ToList();
+                var functionArgument = string.Join(string.Empty, args);
+
+#if DEBUG
+                _logger.LogCritical($"Tool Call (id: {toolCallId}) => {functionName}({functionArgument})");
+#endif
+
+                responseMessage = new RoleDialogModel(AgentRole.Function, string.Empty)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId,
+                    ToolCallId = toolCallId,
+                    FunctionName = functionName,
+                    FunctionArgs = functionArgument
+                };
+            }
+            else if (choice.FinishReason.HasValue)
+            {
+                var allText = textStream.GetText();
+                _logger.LogCritical($"Text Content: {allText}");
+
+                responseMessage = new RoleDialogModel(AgentRole.Assistant, allText)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId,
+                    IsStreaming = true
+                };
+            }
+        }
+
+        hub.Push(new()
+        {
+            ServiceProvider = _services,
+            EventName = "AfterReceiveLlmStreamMessage",
+            Data = responseMessage
+        });
+
+
+        var inputTokenDetails = tokenUsage?.InputTokenDetails;
+        // After chat completion hook
+        foreach (var hook in contentHooks)
+        {
+            await hook.AfterGenerated(responseMessage, new TokenStatsModel
+            {
+                Prompt = prompt,
+                Provider = Provider,
+                Model = _model,
+                TextInputTokens = (tokenUsage?.InputTokenCount ?? 0) - (inputTokenDetails?.CachedTokenCount ?? 0),
+                CachedTextInputTokens = inputTokenDetails?.CachedTokenCount ?? 0,
+                TextOutputTokens = tokenUsage?.OutputTokenCount ?? 0
             });
         }
 
-        return true;
+        return responseMessage;
     }
 
     public void SetModelName(string model)

src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/GeminiChatCompletionProvider.cs

@@ -159,40 +159,9 @@ public class GeminiChatCompletionProvider : IChatCompletion
         return true;
     }
 
-    public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+    public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
-        var client = ProviderHelper.GetGeminiClient(Provider, _model, _services);
-        var chatClient = client.CreateGenerativeModel(_model.ToModelId());
-        var (prompt, messages) = PrepareOptions(chatClient,agent, conversations);
-
-        var asyncEnumerable = chatClient.StreamContentAsync(messages);
-
-        await foreach (var response in asyncEnumerable)
-        {
-            if (response.GetFunction() != null)
-            {
-                var func = response.GetFunction();
-                var update = func?.Args?.ToJsonString().ToString() ?? string.Empty;
-                _logger.LogInformation(update);
-
-                await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update)
-                {
-                    RenderedInstruction = string.Join("\r\n", renderedInstructions)
-                });
-                continue;
-            }
-
-            if (response.Text().IsNullOrEmpty()) continue;
-
-            _logger.LogInformation(response.Text());
-
-            await onMessageReceived(new RoleDialogModel(response.Candidates?.LastOrDefault()?.Content?.Role?.ToString() ?? AgentRole.Assistant.ToString(), response.Text() ?? string.Empty)
-            {
-                RenderedInstruction = string.Join("\r\n", renderedInstructions)
-            });
-        }
-
-        return true;
+        throw new NotImplementedException();
     }
 
     public void SetModelName(string model)

src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/PalmChatCompletionProvider.cs

@@ -145,7 +145,7 @@ public class PalmChatCompletionProvider : IChatCompletion
         throw new NotImplementedException();
     }
 
-    public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+    public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
         throw new NotImplementedException();
     }

src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Realtime/RealTimeCompletionProvider.cs

@@ -1,11 +1,12 @@
-using System.Threading;
 using BotSharp.Abstraction.Hooks;
 using BotSharp.Abstraction.Realtime.Models.Session;
+using BotSharp.Core.Infrastructures.Streams;
 using BotSharp.Core.Session;
 using BotSharp.Plugin.GoogleAI.Models.Realtime;
 using GenerativeAI;
 using GenerativeAI.Types;
 using GenerativeAI.Types.Converters;
+using System.Threading;
 
 namespace BotSharp.Plugin.GoogleAi.Providers.Realtime;
 
@@ -33,8 +34,8 @@ public class GoogleRealTimeProvider : IRealTimeCompletion
         UnknownTypeHandling = JsonUnknownTypeHandling.JsonElement
     };
 
-    private RealtimeTranscriptionResponse _inputStream = new();
-    private RealtimeTranscriptionResponse _outputStream = new();
+    private RealtimeTextStream _inputStream = new();
+    private RealtimeTextStream _outputStream = new();
     private bool _isBlocking = false;
 
     private RealtimeHubConnection _conn;

src/Plugins/BotSharp.Plugin.HuggingFace/Providers/ChatCompletionProvider.cs

@@ -76,9 +76,9 @@ public class ChatCompletionProvider : IChatCompletion
         return true;
     }
 
-    public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+    public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
-        return true;
+        throw new NotImplementedException();
     }
 
     public void SetModelName(string model)

src/Plugins/BotSharp.Plugin.LLamaSharp/BotSharp.Plugin.LLamaSharp.csproj

@@ -15,7 +15,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <ProjectReference Include="..\..\Infrastructure\BotSharp.Abstraction\BotSharp.Abstraction.csproj" />
+    <ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
   </ItemGroup>
 
 </Project>

src/Plugins/BotSharp.Plugin.LLamaSharp/Providers/ChatCompletionProvider.cs

@@ -1,5 +1,12 @@
 using BotSharp.Abstraction.Agents;
+using BotSharp.Abstraction.Hooks;
 using BotSharp.Abstraction.Loggers;
+using BotSharp.Abstraction.Observables.Models;
+using BotSharp.Core.Infrastructures.Streams;
+using BotSharp.Core.Observables.Queues;
+using Microsoft.AspNetCore.SignalR;
+using static LLama.Common.ChatHistory;
+using static System.Net.Mime.MediaTypeNames;
 
 namespace BotSharp.Plugin.LLamaSharp.Providers;
 
@@ -159,12 +166,8 @@ public class ChatCompletionProvider : IChatCompletion
         return true;
     }
 
-    public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+    public async Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
-        string totalResponse = "";
-        var content = string.Join("\r\n", conversations.Select(x => $"{x.Role}: {x.Content}")).Trim();
-        content += $"\r\n{AgentRole.Assistant}: ";
-
         var state = _services.GetRequiredService<IConversationStateService>();
         var model = state.GetState("model", "llama-2-7b-chat.Q8_0");
 
@@ -180,13 +183,60 @@ public class ChatCompletionProvider : IChatCompletion
             _logger.LogInformation(agent.Instruction);
         }
 
+        var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
+        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
+            }
+        });
+
+        using var textStream = new RealtimeTextStream();
+        var responseMessage = new RoleDialogModel(AgentRole.Assistant, string.Empty)
+        {
+            CurrentAgentId = agent.Id,
+            MessageId = messageId
+        };
+
         await foreach (var response in executor.InferAsync(agent.Instruction, inferenceParams))
         {
             Console.Write(response);
-            totalResponse += response;
+            textStream.Collect(response);
+
+            var content = new RoleDialogModel(AgentRole.Assistant, response)
+            {
+                CurrentAgentId = agent.Id,
+                MessageId = messageId
+            };
+            hub.Push(new()
+            {
+                ServiceProvider = _services,
+                EventName = "OnReceiveLlmStreamMessage",
+                Data = content
+            });
         }
 
-        return true;
+        responseMessage = new RoleDialogModel(AgentRole.Assistant, textStream.GetText())
+        {
+            CurrentAgentId = agent.Id,
+            MessageId = messageId,
+            IsStreaming = true
+        };
+
+        hub.Push(new()
+        {
+            ServiceProvider = _services,
+            EventName = "AfterReceiveLlmStreamMessage",
+            Data = responseMessage
+        });
+
+        return responseMessage;
     }
 
     public void SetModelName(string model)

src/Plugins/BotSharp.Plugin.LangChain/Providers/ChatCompletionProvider.cs

@@ -65,7 +65,7 @@ namespace BotSharp.Plugin.VertexAI.Providers
             throw new NotImplementedException();
         }
 
-        public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+        public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
         {
             throw new NotImplementedException();
         }

src/Plugins/BotSharp.Plugin.MetaGLM/Providers/ChatCompletionProvider.cs

@@ -235,7 +235,7 @@ public class ChatCompletionProvider : IChatCompletion
         throw new NotImplementedException();
     }
 
-    public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+    public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
         throw new NotImplementedException();
     }

src/Plugins/BotSharp.Plugin.MicrosoftExtensionsAI/MicrosoftExtensionsAIChatCompletionProvider.cs

@@ -169,8 +169,10 @@ public sealed class MicrosoftExtensionsAIChatCompletionProvider : IChatCompletio
         throw new NotImplementedException();
 
     /// <inheritdoc/>
-    public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived) =>
+    public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
+    {
         throw new NotImplementedException();
+    }
 
     private sealed class NopAIFunction(string name, string description, JsonElement schema) : AIFunction
     {

src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs

@@ -1,4 +1,10 @@
+using Azure;
 using BotSharp.Abstraction.Hooks;
+using BotSharp.Abstraction.Observables.Models;
+using BotSharp.Core.Infrastructures.Streams;
+using BotSharp.Core.Observables.Queues;
+using BotSharp.Plugin.OpenAI.Models.Realtime;
+using Fluid;
 using OpenAI.Chat;
 
 namespace BotSharp.Plugin.OpenAI.Providers.Chat;
@@ -179,39 +185,133 @@ public class ChatCompletionProvider : IChatCompletion
         return true;
     }
 
-    public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+    public async Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
         var client = ProviderHelper.GetClient(Provider, _model, _services);
         var chatClient = client.GetChatClient(_model);
         var (prompt, messages, options) = PrepareOptions(agent, conversations);
 
-        var response = chatClient.CompleteChatStreamingAsync(messages, options);
+        var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
+        var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty;
 
-        await foreach (var choice in response)
+        var contentHooks = _services.GetHooks<IContentGeneratingHook>(agent.Id);
+        // Before chat completion hook
+        foreach (var hook in contentHooks)
         {
-            if (choice.FinishReason == ChatFinishReason.FunctionCall || choice.FinishReason == ChatFinishReason.ToolCalls)
-            {
-                var update = choice.ToolCallUpdates?.FirstOrDefault()?.FunctionArgumentsUpdate?.ToString() ?? string.Empty;
-                _logger.LogInformation(update);
+            await hook.BeforeGenerating(agent, conversations);
+        }
 
-                await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update)
-                {
-                    RenderedInstruction = string.Join("\r\n", renderedInstructions)
-                });
-                continue;
+        hub.Push(new()
+        {
+            ServiceProvider = _services,
+            EventName = "BeforeReceiveLlmStreamMessage",
+            Data = new RoleDialogModel(AgentRole.Assistant, string.Empty)
+            {
+                CurrentAgentId = agent.Id,
+                MessageId = messageId
+            }
+        });
+        
+        using var textStream = new RealtimeTextStream();
+        var toolCalls = new List<StreamingChatToolCallUpdate>();
+        ChatTokenUsage? tokenUsage = null;
+
+        var responseMessage = new RoleDialogModel(AgentRole.Assistant, string.Empty)
+        {
+            CurrentAgentId = agent.Id,
+            MessageId = messageId
+        };
+
+        await foreach (var choice in chatClient.CompleteChatStreamingAsync(messages, options))
+        {
+            tokenUsage = choice.Usage;
+
+            if (!choice.ToolCallUpdates.IsNullOrEmpty())
+            {
+                toolCalls.AddRange(choice.ToolCallUpdates);
             }
 
-            if (choice.ContentUpdate.IsNullOrEmpty()) continue;
-
-            _logger.LogInformation(choice.ContentUpdate[0]?.Text);
-
-            await onMessageReceived(new RoleDialogModel(choice.Role?.ToString() ?? ChatMessageRole.Assistant.ToString(), choice.ContentUpdate[0]?.Text ?? string.Empty)
+            if (!choice.ContentUpdate.IsNullOrEmpty())
             {
-                RenderedInstruction = string.Join("\r\n", renderedInstructions)
+                var text = choice.ContentUpdate[0]?.Text ?? string.Empty;
+                textStream.Collect(text);
+
+#if DEBUG
+                _logger.LogCritical($"Stream Content update: {text}");
+#endif
+
+                var content = new RoleDialogModel(AgentRole.Assistant, text)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId
+                };
+                hub.Push(new()
+                {
+                    ServiceProvider = _services,
+                    EventName = "OnReceiveLlmStreamMessage",
+                    Data = content
+                });
+            }
+
+            if (choice.FinishReason == ChatFinishReason.ToolCalls || choice.FinishReason == ChatFinishReason.FunctionCall)
+            {
+                var meta = toolCalls.FirstOrDefault(x => !string.IsNullOrEmpty(x.FunctionName));
+                var functionName = meta?.FunctionName;
+                var toolCallId = meta?.ToolCallId;
+                var args = toolCalls.Where(x => x.FunctionArgumentsUpdate != null).Select(x => x.FunctionArgumentsUpdate.ToString()).ToList();
+                var functionArguments = string.Join(string.Empty, args);
+
+#if DEBUG
+                _logger.LogCritical($"Tool Call (id: {toolCallId}) => {functionName}({functionArguments})");
+#endif
+
+                responseMessage = new RoleDialogModel(AgentRole.Function, string.Empty)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId,
+                    ToolCallId = toolCallId,
+                    FunctionName = functionName,
+                    FunctionArgs = functionArguments
+                };
+            }
+            else if (choice.FinishReason.HasValue)
+            {
+                var allText = textStream.GetText();
+                _logger.LogInformation($"Stream text Content: {allText}");
+
+                responseMessage = new RoleDialogModel(AgentRole.Assistant, allText)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId,
+                    IsStreaming = true
+                };
+            }
+        }
+
+        hub.Push(new()
+        {
+            ServiceProvider = _services,
+            EventName = "AfterReceiveLlmStreamMessage",
+            Data = responseMessage
+        });
+
+
+        var inputTokenDetails = tokenUsage?.InputTokenDetails;
+        // After chat completion hook
+        foreach (var hook in contentHooks)
+        {
+            await hook.AfterGenerated(responseMessage, new TokenStatsModel
+            {
+                Prompt = prompt,
+                Provider = Provider,
+                Model = _model,
+                TextInputTokens = (tokenUsage?.InputTokenCount ?? 0) - (inputTokenDetails?.CachedTokenCount ?? 0),
+                CachedTextInputTokens = inputTokenDetails?.CachedTokenCount ?? 0,
+                TextOutputTokens = tokenUsage?.OutputTokenCount ?? 0
             });
         }
 
-        return true;
+        return responseMessage;
     }
 
 
@@ -412,4 +512,11 @@ public class ChatCompletionProvider : IChatCompletion
     {
         _model = model;
     }
+}
+
+
+class ToolCallData
+{
+    public ChatFinishReason? Reason { get; set; }
+    public List<StreamingChatToolCallUpdate> ToolCalls { get; set; } = [];
 }

src/Plugins/BotSharp.Plugin.SemanticKernel/SemanticKernelChatCompletionProvider.cs

@@ -94,7 +94,7 @@ namespace BotSharp.Plugin.SemanticKernel
             throw new NotImplementedException();
         }
         /// <inheritdoc/>
-        public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+        public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
         {
             throw new NotImplementedException();
         }

src/Plugins/BotSharp.Plugin.SparkDesk/BotSharp.Plugin.SparkDesk.csproj

@@ -15,7 +15,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <ProjectReference Include="..\..\Infrastructure\BotSharp.Abstraction\BotSharp.Abstraction.csproj" />
+    <ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
   </ItemGroup>
 
 </Project>

src/Plugins/BotSharp.Plugin.SparkDesk/Providers/ChatCompletionProvider.cs

@@ -1,6 +1,10 @@
 using BotSharp.Abstraction.Agents;
 using BotSharp.Abstraction.Agents.Enums;
 using BotSharp.Abstraction.Loggers;
+using BotSharp.Abstraction.Observables.Models;
+using BotSharp.Core.Infrastructures.Streams;
+using BotSharp.Core.Observables.Queues;
+using Microsoft.AspNetCore.SignalR;
 
 namespace BotSharp.Plugin.SparkDesk.Providers;
 
@@ -143,34 +147,77 @@ public class ChatCompletionProvider : IChatCompletion
         return true;
     }
 
-    public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
+    public async Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
     {
         var client = new SparkDeskClient(appId: _settings.AppId, apiKey: _settings.ApiKey, apiSecret: _settings.ApiSecret);
         var (prompt, messages, funcall) = PrepareOptions(agent, conversations);
+        var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty;
+        var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
+
+        hub.Push(new()
+        {
+            ServiceProvider = _services,
+            EventName = "BeforeReceiveLlmStreamMessage",
+            Data = new RoleDialogModel(AgentRole.Assistant, string.Empty)
+            {
+                CurrentAgentId = agent.Id,
+                MessageId = messageId
+            }
+        });
+
+        var responseMessage = new RoleDialogModel(AgentRole.Assistant, string.Empty)
+        {
+            CurrentAgentId = agent.Id,
+            MessageId = messageId
+        };
+
+        using var textStream = new RealtimeTextStream();
 
         await foreach (StreamedChatResponse response in client.ChatAsStreamAsync(modelVersion: _settings.ModelVersion, messages, functions: funcall.Length == 0 ? null : funcall))
         {
-            if (response.FunctionCall !=null)
+            if (response.FunctionCall != null)
             {
-                await onMessageReceived(new RoleDialogModel(AgentRole.Function, response.Text) 
-                { 
+                responseMessage = new RoleDialogModel(AgentRole.Function, string.Empty)
+                {
                     CurrentAgentId = agent.Id,
+                    MessageId = messageId,
+                    ToolCallId = response.FunctionCall.Name,
                     FunctionName = response.FunctionCall.Name,
-                    FunctionArgs = response.FunctionCall.Arguments,
-                    RenderedInstruction = string.Join("\r\n", renderedInstructions)
-                });
-                continue;
+                    FunctionArgs = response.FunctionCall.Arguments
+                };
             }
-
-            await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, response.Text)
+            else
             {
-                CurrentAgentId = agent.Id,
-                RenderedInstruction = string.Join("\r\n", renderedInstructions)
-            });
- 
-        } 
+                textStream.Collect(response.Text);
+                responseMessage = new RoleDialogModel(AgentRole.Assistant, response.Text)
+                {
+                    CurrentAgentId = agent.Id,
+                    MessageId = messageId
+                };
 
-        return true;
+                hub.Push(new()
+                {
+                    ServiceProvider = _services,
+                    EventName = "OnReceiveLlmStreamMessage",
+                    Data = responseMessage
+                });
+            } 
+        }
+
+        if (responseMessage.Role == AgentRole.Assistant)
+        {
+            responseMessage.Content = textStream.GetText();
+            responseMessage.IsStreaming = true;
+        }
+
+        hub.Push(new()
+        {
+            ServiceProvider = _services,
+            EventName = "AfterReceiveLlmStreamMessage",
+            Data = responseMessage
+        });
+
+        return responseMessage;
     }
 
     public void SetModelName(string model)

tests/BotSharp.LLM.Tests/ChatCompletionTests.cs

@@ -1,4 +1,4 @@
-using BotSharp.Abstraction.Agents.Enums;
+using BotSharp.Abstraction.Agents.Enums;
 using BotSharp.Abstraction.Agents.Models;
 using BotSharp.Abstraction.Conversations.Models;
 using BotSharp.Abstraction.MLTasks;
@@ -96,13 +96,15 @@ namespace BotSharp.Plugin.Google.Core
         public async Task GetChatCompletionsStreamingAsync_Test(IChatCompletion chatCompletion, Agent agent, string modelName)
         {
             chatCompletion.SetModelName(modelName);
-            var conversation = new List<RoleDialogModel>([new RoleDialogModel(AgentRole.User, "write a poem about stars")]);
+
             RoleDialogModel reply = null;
-            var result = await chatCompletion.GetChatCompletionsStreamingAsync(agent,conversation, async (received) =>
+            var messages = new List<RoleDialogModel>
             {
-                reply = received;
-            });
-            result.ShouldBeTrue();
+                new RoleDialogModel(AgentRole.User, "write a poem about stars")
+            };
+            var result = await chatCompletion.GetChatCompletionsStreamingAsync(agent, messages);
+
+            result.ShouldNotBeNull();
             reply.ShouldNotBeNull();
             reply.Content.ShouldNotBeNullOrEmpty();
         }