Merge branch 'SciSharp:master' into master

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" />
@@ -44,8 +45,8 @@
     <PackageVersion Include="Whisper.net" Version="1.8.1" />
     <PackageVersion Include="Whisper.net.Runtime" Version="1.8.1" />
     <PackageVersion Include="NCrontab" Version="3.3.3" />
-    <PackageVersion Include="Azure.AI.OpenAI" Version="2.2.0-beta.4" />
-    <PackageVersion Include="OpenAI" Version="2.2.0-beta.4" />
+    <PackageVersion Include="Azure.AI.OpenAI" Version="2.2.0-beta.5" />
+    <PackageVersion Include="OpenAI" Version="2.2.0" />
     <PackageVersion Include="MailKit" Version="4.11.0" />
     <PackageVersion Include="Microsoft.Data.Sqlite" Version="8.0.8" />
     <PackageVersion Include="MySql.Data" Version="9.0.0" />
@@ -147,4 +148,4 @@
     <PackageVersion Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="6.0.25" />
     <PackageVersion Include="Microsoft.AspNetCore.SpaServices.Extensions" Version="6.0.26" />
   </ItemGroup>
-</Project>
+</Project>

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/ConversationHookBase.cs

@@ -49,10 +49,10 @@ public abstract class ConversationHookBase : IConversationHook
     public virtual Task OnHumanInterventionNeeded(RoleDialogModel message)
         => Task.CompletedTask;
 
-    public virtual Task OnFunctionExecuting(RoleDialogModel message)
+    public virtual Task OnFunctionExecuting(RoleDialogModel message, string from = InvokeSource.Manual)
         => Task.CompletedTask;
 
-    public virtual Task OnFunctionExecuted(RoleDialogModel message)
+    public virtual Task OnFunctionExecuted(RoleDialogModel message, string from = InvokeSource.Manual)
         => Task.CompletedTask;
 
     public virtual Task OnMessageReceived(RoleDialogModel message)

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/IConversationHook.cs

@@ -59,15 +59,17 @@ public interface IConversationHook : IHookBase
     /// Triggered before LLM calls function.
     /// </summary>
     /// <param name="message"></param>
+    /// <param name="from"></param>
     /// <returns></returns>
-    Task OnFunctionExecuting(RoleDialogModel message);
+    Task OnFunctionExecuting(RoleDialogModel message, string from = InvokeSource.Manual);
 
     /// <summary>
     /// Triggered when the function calling completed.
     /// </summary>
     /// <param name="message"></param>
+    /// <param name="from"></param>
     /// <returns></returns>
-    Task OnFunctionExecuted(RoleDialogModel message);
+    Task OnFunctionExecuted(RoleDialogModel message, string from = InvokeSource.Manual);
 
     Task OnResponseGenerated(RoleDialogModel message);
 

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

@@ -16,14 +16,13 @@ public interface IChatCompletion
     void SetModelName(string model);
 
     Task<RoleDialogModel> GetChatCompletions(Agent agent,
-        List<RoleDialogModel> conversations);
+        List<RoleDialogModel> conversations) => throw new NotImplementedException();
 
     Task<bool> GetChatCompletionsAsync(Agent agent, 
         List<RoleDialogModel> conversations, 
         Func<RoleDialogModel, Task> onMessageReceived,
-        Func<RoleDialogModel, Task> onFunctionExecuting);
+        Func<RoleDialogModel, Task> onFunctionExecuting) => throw new NotImplementedException();
 
-    Task<bool> GetChatCompletionsStreamingAsync(Agent agent, 
-        List<RoleDialogModel> conversations, 
-        Func<RoleDialogModel, Task> onMessageReceived);
+    Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, 
+        List<RoleDialogModel> conversations) => throw new NotImplementedException();
 }

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/Enums/InvokeSource.cs

@@ -0,0 +1,19 @@
+namespace BotSharp.Abstraction.Routing.Enums;
+
+public static class InvokeSource
+{
+    /// <summary>
+    /// Invoke manually
+    /// </summary>
+    public const string Manual = "manual";
+
+    /// <summary>
+    /// Invoke by LLM directly
+    /// </summary>
+    public const string Llm = "llm";
+
+    /// <summary>
+    /// Invoke by agent routing
+    /// </summary>
+    public const string Routing = "routing";
+}

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

@@ -26,12 +26,8 @@ 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);
-    Task<bool> InvokeFunction(string name, RoleDialogModel messages);
+    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);
 
     /// <summary>

src/Infrastructure/BotSharp.Abstraction/Using.cs

@@ -12,6 +12,7 @@ global using BotSharp.Abstraction.Agents.Enums;
 global using BotSharp.Abstraction.Infrastructures.Enums;
 global using BotSharp.Abstraction.Models;
 global using BotSharp.Abstraction.Routing.Models;
+global using BotSharp.Abstraction.Routing.Enums;
 global using BotSharp.Abstraction.Templating;
 global using BotSharp.Abstraction.Translation.Attributes;
 global using BotSharp.Abstraction.Messaging.Enums;

src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs

@@ -1,3 +1,4 @@
+using BotSharp.Abstraction.Routing.Enums;
 using BotSharp.Abstraction.Utilities;
 
 namespace BotSharp.Core.Realtime.Hooks;
@@ -10,7 +11,7 @@ public class RealtimeConversationHook : ConversationHookBase, IConversationHook
         _services = services;
     }
 
-    public async Task OnFunctionExecuting(RoleDialogModel message)
+    public async Task OnFunctionExecuting(RoleDialogModel message, string from = InvokeSource.Manual)
     {
         var hub = _services.GetRequiredService<IRealtimeHub>();
         if (hub.HubConn == null)
@@ -31,10 +32,10 @@ public class RealtimeConversationHook : ConversationHookBase, IConversationHook
         }
     }
 
-    public async Task OnFunctionExecuted(RoleDialogModel message)
+    public async Task OnFunctionExecuted(RoleDialogModel message, string from = InvokeSource.Manual)
     {
         var hub = _services.GetRequiredService<IRealtimeHub>();
-        if (hub.HubConn == null)
+        if (from != InvokeSource.Llm || hub.HubConn == null)
         {
             return;
         }

src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs

@@ -2,6 +2,7 @@ using BotSharp.Abstraction.Functions.Models;
 using BotSharp.Abstraction.Hooks;
 using BotSharp.Abstraction.Models;
 using BotSharp.Abstraction.Options;
+using BotSharp.Abstraction.Routing.Enums;
 using BotSharp.Core.Infrastructures;
 
 namespace BotSharp.Core.Realtime.Services;
@@ -98,7 +99,7 @@ public class RealtimeHub : IRealtimeHub
                                 agent.Id);
                         }
 
-                        await routing.InvokeFunction(message.FunctionName, message);
+                        await routing.InvokeFunction(message.FunctionName, message, from: InvokeSource.Llm);
                     }
                     else
                     {

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/Conversations/Services/ConversationService.SendMessage.cs

@@ -2,7 +2,6 @@ using BotSharp.Abstraction.Hooks;
 using BotSharp.Abstraction.Infrastructures.Enums;
 using BotSharp.Abstraction.Messaging;
 using BotSharp.Abstraction.Messaging.Models.RichContent;
-using BotSharp.Abstraction.Routing.Enums;
 using BotSharp.Abstraction.Routing.Settings;
 
 namespace BotSharp.Core.Conversations.Services;

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/Evaluations/EvaluationConversationHook.cs

@@ -22,13 +22,13 @@ public class EvaluationConversationHook : ConversationHookBase
         return base.OnMessageReceived(message);
     }
 
-    public override Task OnFunctionExecuted(RoleDialogModel message)
+    public override Task OnFunctionExecuted(RoleDialogModel message, string from = InvokeSource.Manual)
     {
         if (Conversation != null && _convSettings.EnableExecutionLog)
         {
             _logger.Append(Conversation.Id, $"[{DateTime.Now}] {message.Role}: {message.FunctionName}({message.FunctionArgs}) => {message.Content}");
         } 
-        return base.OnFunctionExecuted(message);
+        return base.OnFunctionExecuted(message, from: from);
     }
 
     public override Task OnResponseGenerated(RoleDialogModel message)

src/Infrastructure/BotSharp.Core/Infrastructures/CompletionProvider.cs

@@ -213,7 +213,7 @@ public class CompletionProvider
             model = agentConfig?.Model ?? agentSetting.LlmConfig?.Model;
             if (state.ContainsState("model"))
             {
-                model = state.GetState("model", model ?? "dall-e-3");
+                model = state.GetState("model", model ?? "gpt-image-1");
             }
             else if (state.ContainsState("model_id") || !string.IsNullOrEmpty(modelId))
             {

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/Repository/FileRepository/FileRepository.Agent.cs

@@ -475,7 +475,7 @@ namespace BotSharp.Core.Repository
 
             if (!filter.AgentNames.IsNullOrEmpty())
             {
-                query = query.Where(x => filter.AgentNames.Contains(x.Name));
+                query = query.Where(x => filter.AgentNames.Contains(x.Name, StringComparer.OrdinalIgnoreCase));
             }
 
             if (!string.IsNullOrEmpty(filter.SimilarName))

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

@@ -33,7 +33,7 @@ public class InstructExecutor : IExecutor
         if (message.FunctionName != null)
         {
             var msg = RoleDialogModel.From(message, role: AgentRole.Function);
-            await routing.InvokeFunction(message.FunctionName, msg);
+            await routing.InvokeFunction(message.FunctionName, msg, from: InvokeSource.Llm);
         }
 
         var agentId = routing.Context.GetCurrentAgentId();
@@ -57,7 +57,13 @@ public class InstructExecutor : IExecutor
         }
         else
         {
-            var ret = await routing.InvokeAgent(agentId, dialogs);
+            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)
+    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);
+            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)
+    private async Task<bool> InvokeFunction(
+        RoleDialogModel message,
+        List<RoleDialogModel> dialogs,
+        string from,
+        bool useStream)
     {
         // execute function
         // Save states
@@ -75,7 +93,7 @@ public partial class RoutingService
 
         var routing = _services.GetRequiredService<IRoutingService>();
         // Call functions
-        await routing.InvokeFunction(message.FunctionName, message);
+        await routing.InvokeFunction(message.FunctionName, message, from: from);
 
         // Pass execution result to LLM to get response
         if (!message.StopCompletion)
@@ -102,7 +120,7 @@ public partial class RoutingService
 
                 // Send to Next LLM
                 var curAgentId = routing.Context.GetCurrentAgentId();
-                await InvokeAgent(curAgentId, dialogs);
+                await InvokeAgent(curAgentId, dialogs, from, useStream);
             }
         }
         else

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

@@ -5,7 +5,7 @@ namespace BotSharp.Core.Routing;
 
 public partial class RoutingService
 {
-    public async Task<bool> InvokeFunction(string name, RoleDialogModel message)
+    public async Task<bool> InvokeFunction(string name, RoleDialogModel message, string from = InvokeSource.Manual)
     {
         var currentAgentId = message.CurrentAgentId;
         var agentService = _services.GetRequiredService<IAgentService>();
@@ -36,7 +36,7 @@ public partial class RoutingService
         foreach (var hook in hooks)
         {
             hook.SetAgent(agent);
-            await hook.OnFunctionExecuting(clonedMessage);
+            await hook.OnFunctionExecuting(clonedMessage, from: from);
         }
 
         bool result = false;
@@ -48,7 +48,7 @@ public partial class RoutingService
             // After functions have been executed
             foreach (var hook in hooks)
             {
-                await hook.OnFunctionExecuted(clonedMessage);
+                await hook.OnFunctionExecuted(clonedMessage, from: from);
             }
 
             // Set result to original message

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

@@ -51,7 +51,13 @@ public partial class RoutingService : IRoutingService
         }
         else
         {
-            var ret = await routing.InvokeAgent(agentId, dialogs);
+            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/Templating/TemplateRender.cs

@@ -4,10 +4,12 @@ using BotSharp.Abstraction.Templating;
 using BotSharp.Abstraction.Translation.Models;
 using Fluid;
 using Fluid.Ast;
+using Fluid.Values;
 using System.Collections;
 using System.IO;
 using System.Reflection;
 using System.Text.Encodings.Web;
+using System.Text.RegularExpressions;
 
 namespace BotSharp.Core.Templating;
 
@@ -36,9 +38,11 @@ public class TemplateRender : ITemplateRender
         _options.MemberAccessStrategy.Register<UserIdentity>();
         _options.MemberAccessStrategy.Register<TranslationInput>();
 
-        _parser.RegisterIdentifierTag("link", (string identifier, TextWriter writer, TextEncoder encoder, TemplateContext context) =>
+        _options.Filters.AddFilter("from_agent", FromAgentFilter);
+
+        _parser.RegisterExpressionTag("link", (Expression expression, TextWriter writer, TextEncoder encoder, TemplateContext context) =>
         {
-            return RenderIdentifierTag("link", identifier, writer, encoder, context);
+            return RenderTag("link", expression, writer, encoder, context, services);
         });
     }
 
@@ -82,20 +86,46 @@ public class TemplateRender : ITemplateRender
 
 
     #region Private methods
-    private static async ValueTask<Completion> RenderIdentifierTag(string tag, string identifier, TextWriter writer, TextEncoder encoder, TemplateContext context)
+    private static async ValueTask<Completion> RenderTag(
+        string tag,
+        Expression expression,
+        TextWriter writer,
+        TextEncoder encoder,
+        TemplateContext context,
+        IServiceProvider services)
     {
         try
         {
-            var value = await context.Model.GetValueAsync(TemplateRenderConstant.RENDER_AGENT, context);
-            var agent = value?.ToObjectValue() as Agent;
-            var found = agent?.Templates?.FirstOrDefault(x => x.Name.IsEqualTo(identifier));
-            var key = $"{agent?.Id} | {tag} | {identifier}";
+            var value = await expression.EvaluateAsync(context);
+            var expStr = value?.ToStringValue() ?? string.Empty;
 
-            if (found == null || (context.AmbientValues.TryGetValue(key, out var visited) && (bool)visited))
+            value = await context.Model.GetValueAsync(TemplateRenderConstant.RENDER_AGENT, context);
+            var agent = value?.ToObjectValue() as Agent;
+
+            var splited = Regex.Split(expStr, @"\s*from_agent\s*", RegexOptions.IgnoreCase | RegexOptions.CultureInvariant)
+                               .Where(x => !string.IsNullOrWhiteSpace(x))
+                               .Select(x => x.Trim())
+                               .ToArray();
+
+            var templateName = splited.ElementAtOrDefault(0);
+            var agentName = splited.ElementAtOrDefault(1);
+
+            if (splited.Length > 1 && !agentName.IsEqualTo(agent?.Name))
+            {
+                using var scope = services.CreateScope();
+                var agentService = scope.ServiceProvider.GetRequiredService<IAgentService>();
+                var result = await agentService.GetAgents(new() { SimilarName = agentName });
+                agent = result?.Items?.FirstOrDefault();
+            }
+
+            var template = agent?.Templates?.FirstOrDefault(x => x.Name.IsEqualTo(templateName));
+            var key = $"{tag} | {agent?.Id} | {templateName}";
+
+            if (template == null || (context.AmbientValues.TryGetValue(key, out var visited) && (bool)visited))
             {
                 writer.Write(string.Empty);
             }
-            else if (_parser.TryParse(found.Content, out var t, out _))
+            else if (_parser.TryParse(template.Content, out var t, out _))
             {
                 context.AmbientValues[key] = true;
                 var rendered = t.Render(context);
@@ -115,6 +145,16 @@ public class TemplateRender : ITemplateRender
         return Completion.Normal;
     }
 
+    private static ValueTask<FluidValue> FromAgentFilter(
+        FluidValue input,
+        FilterArguments arguments,
+        TemplateContext context)
+    {
+        var inputStr = input?.ToStringValue() ?? string.Empty;
+        var fromAgent = arguments.At(0).ToStringValue();
+        return new StringValue($"{inputStr} from_agent {fromAgent}");
+    }
+
     private static bool IsStringType(Type type)
     {
         return type == typeof(string);

src/Infrastructure/BotSharp.Core/Using.cs

@@ -22,6 +22,7 @@ global using BotSharp.Abstraction.Repositories;
 global using BotSharp.Abstraction.Repositories.Filters;
 global using BotSharp.Abstraction.Roles;
 global using BotSharp.Abstraction.Roles.Models;
+global using BotSharp.Abstraction.Routing.Enums;
 global using BotSharp.Abstraction.Routing;
 global using BotSharp.Abstraction.SideCar.Attributes;
 global using BotSharp.Abstraction.Statistics.Enums;

src/Infrastructure/BotSharp.Core/data/agents/01fcc3e5-9af7-49e6-ad7a-a760bd12dc4a/instructions/instruction.liquid

@@ -5,7 +5,8 @@ Follow these steps to handle user request:
 2. Determine which agent is suitable to handle this conversation. Try to minimize the routing of human service.
 3. Extract and populate agent required arguments, think carefully, leave it as blank object if user didn't provide the specific arguments.
 4. You must include all required args for the selected agent, but you must not make up any parameters when there is no exact value provided, those parameters must set value as null if not declared.
-5. If user is greeting, you can call function response_to_user with a greeting message.
+5. Call function route_to_agent if user have specific requests and available agent to proceed. Do not ask user to provide any required args by yourself. The requested agent will handle and fill the required args internally.
+6. If user is greeting or do not have specific request, then you can call function response_to_user with a greeting message.
 
 {% if routing_requirements and routing_requirements != empty %}
 [REQUIREMENTS]

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/Infrastructure/BotSharp.OpenAPI/Controllers/InstructModeController.cs

@@ -30,7 +30,8 @@ public class InstructModeController : ControllerBase
             .SetState("model_id", input.ModelId, source: StateSource.External)
             .SetState("instruction", input.Instruction, source: StateSource.External)
             .SetState("input_text", input.Text, source: StateSource.External)
-            .SetState("template_name", input.Template, source: StateSource.External);
+            .SetState("template_name", input.Template, source: StateSource.External)
+            .SetState("channel", input.Channel, source: StateSource.External);
 
         var instructor = _services.GetRequiredService<IInstructService>();
         var result = await instructor.Execute(agentId,

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

@@ -1,4 +1,5 @@
 using BotSharp.Abstraction.Routing;
+using BotSharp.Abstraction.Routing.Enums;
 
 namespace BotSharp.OpenAPI.Controllers;
 
@@ -25,7 +26,7 @@ public class RealtimeController : ControllerBase
             FunctionName = functionName,
             FunctionArgs = JsonSerializer.Serialize(args)
         };
-        await routing.InvokeFunction(functionName, message);
+        await routing.InvokeFunction(functionName, message, from: InvokeSource.Llm);
         return message.Content;
     }
 }

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

@@ -1,4 +1,5 @@
 using BotSharp.Abstraction.Conversations.Dtos;
+using BotSharp.Abstraction.Routing.Enums;
 using BotSharp.Abstraction.SideCar;
 using BotSharp.Abstraction.Users.Dtos;
 using Microsoft.AspNetCore.SignalR;
@@ -84,7 +85,7 @@ public class ChatHubConversationHook : ConversationHookBase
         await base.OnMessageReceived(message);
     }
 
-    public override async Task OnFunctionExecuting(RoleDialogModel message)
+    public override async Task OnFunctionExecuting(RoleDialogModel message, string from = InvokeSource.Manual)
     {
         var conv = _services.GetRequiredService<IConversationService>();
         var action = new ConversationSenderActionModel
@@ -95,7 +96,7 @@ public class ChatHubConversationHook : ConversationHookBase
         };
 
         await GenerateSenderAction(conv.ConversationId, action);
-        await base.OnFunctionExecuting(message);
+        await base.OnFunctionExecuting(message, from: from);
     }
 
     public override async Task OnPostbackMessageReceived(RoleDialogModel message, PostbackMessageModel replyMsg)
@@ -118,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",
@@ -133,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

@@ -1,3 +1,4 @@
+using BotSharp.Abstraction.Routing.Enums;
 using Microsoft.AspNetCore.SignalR;
 using System.Text.Encodings.Web;
 using System.Text.Unicode;
@@ -74,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)
         {
@@ -141,7 +142,7 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR
         if (!_convSettings.ShowVerboseLog) return;
     }
 
-    public override async Task OnFunctionExecuting(RoleDialogModel message)
+    public override async Task OnFunctionExecuting(RoleDialogModel message, string from = InvokeSource.Manual)
     {
         var conversationId = _state.GetConversationId();
         if (string.IsNullOrEmpty(conversationId)) return;
@@ -164,7 +165,7 @@ public class StreamingLogHook : ConversationHookBase, IContentGeneratingHook, IR
         await SendContentLog(conversationId, input);
     }
 
-    public override async Task OnFunctionExecuted(RoleDialogModel message)
+    public override async Task OnFunctionExecuted(RoleDialogModel message, string from = InvokeSource.Manual)
     {
         var conversationId = _state.GetConversationId();
         if (string.IsNullOrEmpty(conversationId)) return;
@@ -233,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.OpenAI/Providers/Image/ImageCompletionProvider.Generation.cs

@@ -38,10 +38,8 @@ public partial class ImageCompletionProvider
 
         var options = new ImageGenerationOptions
         {
-            Size = size,
-            Quality = quality,
-            Style = style,
-            ResponseFormat = format
+            Quality = new GeneratedImageQuality("medium"),
+            Size = GeneratedImageSize.W1024xH1024
         };
         return (prompt, count, options);
     }

src/Plugins/BotSharp.Plugin.OpenAI/Providers/Image/ImageCompletionProvider.cs

@@ -44,7 +44,7 @@ public partial class ImageCompletionProvider : IImageCompletion
             {
                 generatedImage.ImageUrl = image?.ImageUri?.AbsoluteUri ?? string.Empty;
             }
-            else if (format == GeneratedImageFormat.Bytes)
+            else
             {
                 var base64Str = string.Empty;
                 var bytes = image?.ImageBytes?.ToArray();
@@ -125,16 +125,16 @@ public partial class ImageCompletionProvider : IImageCompletion
 
     private GeneratedImageFormat GetImageFormat(string format)
     {
-        var value = !string.IsNullOrEmpty(format) ? format : "uri";
+        var value = !string.IsNullOrEmpty(format) ? format : "bytes";
 
         GeneratedImageFormat retFormat;
         switch (value)
         {
-            case "bytes":
-                retFormat = GeneratedImageFormat.Bytes;
+            case "url":
+                retFormat = GeneratedImageFormat.Uri;
                 break;
             default:
-                retFormat = GeneratedImageFormat.Uri;
+                retFormat = GeneratedImageFormat.Bytes;
                 break;
         }
 

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)

src/Plugins/BotSharp.Plugin.Twilio/Hooks/TwilioConversationHook.cs

@@ -1,9 +1,10 @@
+using BotSharp.Abstraction.Hooks;
 using BotSharp.Abstraction.Routing;
-using Task = System.Threading.Tasks.Task;
-using Twilio.Rest.Api.V2010.Account;
+using BotSharp.Abstraction.Routing.Enums;
 using BotSharp.Plugin.Twilio.Interfaces;
 using BotSharp.Plugin.Twilio.Models;
-using BotSharp.Abstraction.Hooks;
+using Twilio.Rest.Api.V2010.Account;
+using Task = System.Threading.Tasks.Task;
 
 namespace BotSharp.Plugin.Twilio.Hooks;
 
@@ -22,7 +23,7 @@ public class TwilioConversationHook : ConversationHookBase, IConversationHook
         _logger = logger;
     }
 
-    public override async Task OnFunctionExecuted(RoleDialogModel message)
+    public override async Task OnFunctionExecuted(RoleDialogModel message, string from = InvokeSource.Manual)
     {
         var hooks = _services.GetHooks<ITwilioSessionHook>(message.CurrentAgentId);
 

src/Plugins/BotSharp.Plugin.Twilio/TwilioStreamMiddleware.cs

@@ -3,6 +3,7 @@ using BotSharp.Abstraction.MLTasks;
 using BotSharp.Abstraction.Realtime;
 using BotSharp.Abstraction.Realtime.Models;
 using BotSharp.Abstraction.Routing;
+using BotSharp.Abstraction.Utilities;
 using BotSharp.Plugin.Twilio.Interfaces;
 using BotSharp.Plugin.Twilio.Models.Stream;
 using Microsoft.AspNetCore.Http;
@@ -88,10 +89,15 @@ public class TwilioStreamMiddleware
                 continue;
             }
 
-            var (eventType, data) = MapEvents(conn, receivedText);
+            var (eventType, data) = MapEvents(conn, receivedText, conversationId);
 
             if (eventType == "user_connected")
             {
+#if DEBUG
+                _logger.LogCritical($"Start twilio stream connection for conversation ({conversationId})");
+#else
+                _logger.LogInformation($"Start twilio stream connection for conversation ({conversationId})");
+#endif
                 // Connect to model
                 await ConnectToModel(hub, webSocket);
             }
@@ -115,6 +121,11 @@ public class TwilioStreamMiddleware
             }
             else if (eventType == "user_disconnected")
             {
+#if DEBUG
+                _logger.LogCritical($"Disconnecting twilio stream connection for conversation ({conversationId})");
+#else
+                _logger.LogInformation($"Disconnecting twilio stream connection for conversation ({conversationId})");
+#endif
                 await hub.Completer.Disconnect();
                 await HandleUserDisconnected();
             }
@@ -131,9 +142,19 @@ public class TwilioStreamMiddleware
         });
     }
 
-    private (string, string) MapEvents(RealtimeHubConnection conn, string receivedText)
+    private (string, string) MapEvents(RealtimeHubConnection conn, string receivedText, string conversationId)
     {
-        var response = JsonSerializer.Deserialize<StreamEventResponse>(receivedText);
+        StreamEventResponse? response = new();
+
+        try
+        {
+            response = JsonSerializer.Deserialize<StreamEventResponse>(receivedText);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, $"Error when deserializing twilio stream event response for conversation ({conversationId}) (response: {receivedText?.SubstringMax(30)})");
+        }
+
         conn.StreamId = response.StreamSid;
         string eventType = response.Event;
         string data = string.Empty;

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();
         }