support invoke function

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

@@ -40,11 +40,6 @@ public interface IConversationService
         PostbackMessageModel? replyMessage,
         Func<RoleDialogModel, Task> onResponseReceived);
 
-
-    Task<bool> StreamMessage(string agentId,
-        RoleDialogModel lastDialog,
-        PostbackMessageModel? replyMessage);
-
     List<RoleDialogModel> GetDialogHistory(int lastCount = 100, bool fromBreakpoint = true, IEnumerable<string>? includeMessageTypes = null);
     Task CleanHistory(string agentId);
 

src/Infrastructure/BotSharp.Abstraction/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/Conversations/Settings/ConversationSetting.cs

@@ -12,6 +12,7 @@ public class ConversationSetting
     public bool EnableContentLog { get; set; }
     public bool EnableStateLog { get; set; }
     public bool EnableTranslationMemory { get; set; }
+    public bool EnableStreaming { get; set; }
     public CleanConversationSetting CleanSetting { get; set; } = new();
     public RateLimitSetting RateLimit { get; set; } = new();
 }

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, 
+    Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, 
-        List<RoleDialogModel> conversations, 
+        List<RoleDialogModel> conversations) => Task.FromResult(new RoleDialogModel(AgentRole.Assistant, string.Empty));
-        Func<RoleDialogModel, Task> onMessageReceived);
 }

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

@@ -30,7 +30,7 @@ public interface IRoutingService
     //int GetRecursiveCounter();
     //void SetRecursiveCounter(int counter);
 
-    Task<bool> InvokeAgent(string agentId, List<RoleDialogModel> dialogs);
+    Task<bool> InvokeAgent(string agentId, List<RoleDialogModel> dialogs, bool useStream = false);
     Task<bool> InvokeFunction(string name, RoleDialogModel messages);
     Task<RoleDialogModel> InstructLoop(Agent agent, RoleDialogModel message, List<RoleDialogModel> dialogs);
 
@@ -41,7 +41,6 @@ public interface IRoutingService
     /// <param name="message"></param>
     /// <returns></returns>
     Task<RoleDialogModel> InstructDirect(Agent agent, RoleDialogModel message, List<RoleDialogModel> dialogs);
-    Task<bool> InstructStream(Agent agent, RoleDialogModel message, List<RoleDialogModel> dialogs);
 
     Task<string> GetConversationContent(List<RoleDialogModel> dialogs, int maxDialogCount = 100);
 

src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Stream.cs

@@ -1,86 +0,0 @@
-using BotSharp.Abstraction.Hooks;
-using BotSharp.Abstraction.Infrastructures.Enums;
-using BotSharp.Abstraction.Routing.Enums;
-using BotSharp.Abstraction.Routing.Settings;
-
-namespace BotSharp.Core.Conversations.Services;
-
-public partial class ConversationService
-{
-    public async Task<bool> StreamMessage(string agentId,
-        RoleDialogModel message,
-        PostbackMessageModel? replyMessage)
-    {
-        var conversation = await GetConversationRecordOrCreateNew(agentId);
-        var agentService = _services.GetRequiredService<IAgentService>();
-        Agent agent = await agentService.LoadAgent(agentId);
-
-        var content = $"Received [{agent.Name}] {message.Role}: {message.Content}";
-        _logger.LogInformation(content);
-
-        message.CurrentAgentId = agent.Id;
-        if (string.IsNullOrEmpty(message.SenderId))
-        {
-            message.SenderId = _user.Id;
-        }
-
-        var conv = _services.GetRequiredService<IConversationService>();
-        var dialogs = conv.GetDialogHistory();
-
-        var statistics = _services.GetRequiredService<ITokenStatistics>();
-
-        RoleDialogModel response = message;
-        bool stopCompletion = false;
-
-        // Enqueue receiving agent first in case it stop completion by OnMessageReceived
-        var routing = _services.GetRequiredService<IRoutingService>();
-        routing.Context.SetMessageId(_conversationId, message.MessageId);
-
-        // Save payload in order to assign the payload before hook is invoked
-        if (replyMessage != null && !string.IsNullOrEmpty(replyMessage.Payload))
-        {
-            message.Payload = replyMessage.Payload;
-        }
-
-        var hooks = _services.GetHooksOrderByPriority<IConversationHook>(message.CurrentAgentId);
-        foreach (var hook in hooks)
-        {
-            hook.SetAgent(agent)
-                .SetConversation(conversation);
-
-            if (replyMessage == null || string.IsNullOrEmpty(replyMessage.FunctionName))
-            {
-                await hook.OnMessageReceived(message);
-            }
-            else
-            {
-                await hook.OnPostbackMessageReceived(message, replyMessage);
-            }
-
-            // Interrupted by hook
-            if (message.StopCompletion)
-            {
-                stopCompletion = true;
-                routing.Context.Pop();
-                break;
-            }
-        }
-
-        if (!stopCompletion)
-        {
-            // Routing with reasoning
-            var settings = _services.GetRequiredService<RoutingSettings>();
-
-            // reload agent in case it has been changed by hook
-            if (message.CurrentAgentId != agent.Id)
-            {
-                agent = await agentService.LoadAgent(message.CurrentAgentId);
-            }
-
-            await routing.InstructStream(agent, message, dialogs);
-            routing.Context.ResetRecursiveCounter();
-        }
-
-        return true;
-    }
-}

src/Infrastructure/BotSharp.Core/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/Routing/Reasoning/InstructExecutor.cs

@@ -57,7 +57,8 @@ public class InstructExecutor : IExecutor
         }
         else
         {
-            var ret = await routing.InvokeAgent(agentId, dialogs);
+            var convSettings = _services.GetRequiredService<ConversationSetting>();
+            var ret = await routing.InvokeAgent(agentId, dialogs, convSettings.EnableStreaming);
         }
 
         var response = dialogs.Last();

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

@@ -1,49 +0,0 @@
-namespace BotSharp.Core.Routing;
-
-public partial class RoutingService
-{
-    public async Task<bool> InstructStream(Agent agent, RoleDialogModel message, List<RoleDialogModel> dialogs)
-    {
-        var conv = _services.GetRequiredService<IConversationService>();
-        var storage = _services.GetRequiredService<IConversationStorage>();
-        storage.Append(conv.ConversationId, message);
-
-        dialogs.Add(message);
-        Context.SetDialogs(dialogs);
-
-        var routing = _services.GetRequiredService<IRoutingService>();
-        routing.Context.Push(agent.Id, "instruct directly");
-        var agentId = routing.Context.GetCurrentAgentId();
-
-        // Update next action agent's name
-        var agentService = _services.GetRequiredService<IAgentService>();
-
-        if (agent.Disabled)
-        {
-            var content = $"This agent ({agent.Name}) is disabled, please install the corresponding plugin ({agent.Plugin.Name}) to activate this agent.";
-
-            message = RoleDialogModel.From(message, role: AgentRole.Assistant, content: content);
-            dialogs.Add(message);
-        }
-        else
-        {
-            var provider = agent.LlmConfig.Provider;
-            var model = agent.LlmConfig.Model;
-
-            if (provider == null || model == null)
-            {
-                var agentSettings = _services.GetRequiredService<AgentSettings>();
-                provider = agentSettings.LlmConfig.Provider;
-                model = agentSettings.LlmConfig.Model;
-            }
-
-            var chatCompletion = CompletionProvider.GetChatCompletion(_services,
-                provider: provider,
-                model: model);
-
-            await chatCompletion.GetChatCompletionsStreamingAsync(agent, dialogs, async data => { });
-        }
-
-        return true;
-    }
-}

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

@@ -4,7 +4,7 @@ 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, bool useStream = false)
     {
         var agentService = _services.GetRequiredService<IAgentService>();
         var agent = await agentService.LoadAgent(agentId);
@@ -30,8 +30,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 +53,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, useStream);
         }
         else
         {
@@ -59,6 +68,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 +76,7 @@ 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, bool useStream = false)
     {
         // execute function
         // Save states
@@ -102,7 +112,7 @@ public partial class RoutingService
 
                 // Send to Next LLM
                 var curAgentId = routing.Context.GetCurrentAgentId();
-                await InvokeAgent(curAgentId, dialogs);
+                await InvokeAgent(curAgentId, dialogs, useStream);
             }
         }
         else

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

@@ -51,7 +51,8 @@ public partial class RoutingService : IRoutingService
         }
         else
         {
-            var ret = await routing.InvokeAgent(agentId, dialogs);
+            var convSettings = _services.GetRequiredService<ConversationSetting>();
+            var ret = await routing.InvokeAgent(agentId, dialogs, convSettings.EnableStreaming);
         }
 
         var response = dialogs.Last();

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

@@ -378,29 +378,6 @@ public class ConversationController : ControllerBase
     }
 
 
-    [HttpPost("/conversation/{agentId}/{conversationId}/stream")]
-    public async Task SendMessageStream(
-        [FromRoute] string agentId,
-        [FromRoute] string conversationId,
-        [FromBody] NewMessageModel input)
-    {
-        var conv = _services.GetRequiredService<IConversationService>();
-        var inputMsg = new RoleDialogModel(AgentRole.User, input.Text)
-        {
-            MessageId = !string.IsNullOrWhiteSpace(input.InputMessageId) ? input.InputMessageId : Guid.NewGuid().ToString(),
-            CreatedAt = DateTime.UtcNow
-        };
-
-        var routing = _services.GetRequiredService<IRoutingService>();
-        routing.Context.SetMessageId(conversationId, inputMsg.MessageId);
-
-        conv.SetConversationId(conversationId, input.States);
-        SetStates(conv, input);
-
-        await conv.StreamMessage(agentId, inputMsg, replyMessage: input.Postback);
-    }
-
-
     [HttpPost("/conversation/{agentId}/{conversationId}/sse")]
     public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string conversationId, [FromBody] NewMessageModel input)
     {

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

@@ -105,7 +105,7 @@ public class ChatHubConversationHook : ConversationHookBase
 
     public override async Task OnResponseGenerated(RoleDialogModel message)
     {
-        if (!AllowSendingMessage()) return;
+        if (!AllowSendingMessage() || message.IsStreaming) return;
 
         var conv = _services.GetRequiredService<IConversationService>();
         var state = _services.GetRequiredService<IConversationStateService>();

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

@@ -64,6 +64,8 @@ public class ChatHubObserver : IObserver<HubObserveData>
             GenerateSenderAction(conv.ConversationId, action).ConfigureAwait(false).GetAwaiter().GetResult();
         }
         else if (value.EventName == AFTER_RECEIVE_LLM_STREAM_MESSAGE)
+        {
+            if (message.IsStreaming)
             {
                 var conv = _services.GetRequiredService<IConversationService>();
                 model = new ChatResponseDto()
@@ -86,9 +88,7 @@ public class ChatHubObserver : IObserver<HubObserveData>
                 };
 
                 GenerateSenderAction(conv.ConversationId, action).ConfigureAwait(false).GetAwaiter().GetResult();
-
+            }
-            var storage = _services.GetRequiredService<IConversationStorage>();
-            storage.Append(conv.ConversationId, message);
         }
         else if (value.EventName == ON_RECEIVE_LLM_STREAM_MESSAGE)
         {

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

@@ -1,13 +1,7 @@
 using BotSharp.Abstraction.Hooks;
 using BotSharp.Core.Infrastructures.Streams;
 using BotSharp.Core.Observables.Queues;
-using EntityFrameworkCore.BootKit;
-using Fluid;
-using ModelContextProtocol.Protocol.Types;
 using OpenAI.Chat;
-using System.Xml;
-using static Microsoft.EntityFrameworkCore.DbLoggerCategory;
-using static System.Net.Mime.MediaTypeNames;
 
 namespace BotSharp.Plugin.OpenAI.Providers.Chat;
 
@@ -187,7 +181,7 @@ 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);
@@ -227,7 +221,9 @@ public class ChatCompletionProvider : IChatCompletion
                 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)
                 {
@@ -250,7 +246,9 @@ public class ChatCompletionProvider : IChatCompletion
                 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)
                 {
@@ -270,7 +268,8 @@ public class ChatCompletionProvider : IChatCompletion
                 responseMessage = new RoleDialogModel(AgentRole.Assistant, allText)
                 {
                     CurrentAgentId = agent.Id,
-                    MessageId = messageId
+                    MessageId = messageId,
+                    IsStreaming = true
                 };
             }
         }
@@ -282,7 +281,7 @@ public class ChatCompletionProvider : IChatCompletion
             Data = responseMessage
         });
 
-        return true;
+        return responseMessage;
     }
 
 

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;
+                new RoleDialogModel(AgentRole.User, "write a poem about stars")
-            });
+            };
-            result.ShouldBeTrue();
+            var result = await chatCompletion.GetChatCompletionsStreamingAsync(agent, messages);
+
+            result.ShouldNotBeNull();
             reply.ShouldNotBeNull();
             reply.Content.ShouldNotBeNullOrEmpty();
         }