From 7008667a1ae7042eb919406480b1f4c97ccdde7e Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Mon, 23 Jun 2025 21:43:25 -0500 Subject: [PATCH] support invoke function --- .../Conversations/IConversationService.cs | 5 -- .../Conversations/Models/RoleDialogModel.cs | 6 +- .../Settings/ConversationSetting.cs | 1 + .../MLTasks/IChatCompletion.cs | 5 +- .../Routing/IRoutingService.cs | 3 +- .../Services/ConversationService.Stream.cs | 86 ------------------- .../Demo/Functions/GetWeatherFn.cs | 2 +- .../Routing/Reasoning/InstructExecutor.cs | 3 +- .../Routing/RoutingService.InstructStream.cs | 49 ----------- .../Routing/RoutingService.InvokeAgent.cs | 20 +++-- .../BotSharp.Core/Routing/RoutingService.cs | 3 +- .../Controllers/ConversationController.cs | 23 ----- .../Hooks/ChatHubConversationHook.cs | 2 +- .../Observers/ChatHubObserver.cs | 40 ++++----- .../Providers/Chat/ChatCompletionProvider.cs | 17 ++-- .../BotSharp.LLM.Tests/ChatCompletionTests.cs | 14 +-- 16 files changed, 66 insertions(+), 213 deletions(-) delete mode 100644 src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Stream.cs delete mode 100644 src/Infrastructure/BotSharp.Core/Routing/RoutingService.InstructStream.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs index fccab0fa..322d86b6 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs @@ -40,11 +40,6 @@ public interface IConversationService PostbackMessageModel? replyMessage, Func onResponseReceived); - - Task StreamMessage(string agentId, - RoleDialogModel lastDialog, - PostbackMessageModel? replyMessage); - List GetDialogHistory(int lastCount = 100, bool fromBreakpoint = true, IEnumerable? includeMessageTypes = null); Task CleanHistory(string agentId); diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/Models/RoleDialogModel.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/Models/RoleDialogModel.cs index de269759..6a8af2d6 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/Models/RoleDialogModel.cs +++ b/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 }; } } diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/Settings/ConversationSetting.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/Settings/ConversationSetting.cs index c4f131dc..980569dc 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/Settings/ConversationSetting.cs +++ b/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(); } diff --git a/src/Infrastructure/BotSharp.Abstraction/MLTasks/IChatCompletion.cs b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IChatCompletion.cs index 7cf52fe4..4d93f307 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MLTasks/IChatCompletion.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IChatCompletion.cs @@ -23,7 +23,6 @@ public interface IChatCompletion Func onMessageReceived, Func onFunctionExecuting); - Task GetChatCompletionsStreamingAsync(Agent agent, - List conversations, - Func onMessageReceived); + Task GetChatCompletionsStreamingAsync(Agent agent, + List conversations) => Task.FromResult(new RoleDialogModel(AgentRole.Assistant, string.Empty)); } diff --git a/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs b/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs index 5dbf8d3f..fb542d5f 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs @@ -30,7 +30,7 @@ public interface IRoutingService //int GetRecursiveCounter(); //void SetRecursiveCounter(int counter); - Task InvokeAgent(string agentId, List dialogs); + Task InvokeAgent(string agentId, List dialogs, bool useStream = false); Task InvokeFunction(string name, RoleDialogModel messages); Task InstructLoop(Agent agent, RoleDialogModel message, List dialogs); @@ -41,7 +41,6 @@ public interface IRoutingService /// /// Task InstructDirect(Agent agent, RoleDialogModel message, List dialogs); - Task InstructStream(Agent agent, RoleDialogModel message, List dialogs); Task GetConversationContent(List dialogs, int maxDialogCount = 100); diff --git a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Stream.cs b/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Stream.cs deleted file mode 100644 index f650a28f..00000000 --- a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Stream.cs +++ /dev/null @@ -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 StreamMessage(string agentId, - RoleDialogModel message, - PostbackMessageModel? replyMessage) - { - var conversation = await GetConversationRecordOrCreateNew(agentId); - var agentService = _services.GetRequiredService(); - 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(); - var dialogs = conv.GetDialogHistory(); - - var statistics = _services.GetRequiredService(); - - RoleDialogModel response = message; - bool stopCompletion = false; - - // Enqueue receiving agent first in case it stop completion by OnMessageReceived - var routing = _services.GetRequiredService(); - 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(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(); - - // 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; - } -} diff --git a/src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs b/src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs index a78ad4ec..cb6bcd88 100644 --- a/src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs +++ b/src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs @@ -17,7 +17,7 @@ public class GetWeatherFn : IFunctionCallback public async Task Execute(RoleDialogModel message) { message.Content = $"It is a sunny day!"; - //message.StopCompletion = true; + message.StopCompletion = false; return true; } } \ No newline at end of file diff --git a/src/Infrastructure/BotSharp.Core/Routing/Reasoning/InstructExecutor.cs b/src/Infrastructure/BotSharp.Core/Routing/Reasoning/InstructExecutor.cs index c93e133c..42a3d652 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/Reasoning/InstructExecutor.cs +++ b/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(); + var ret = await routing.InvokeAgent(agentId, dialogs, convSettings.EnableStreaming); } var response = dialogs.Last(); diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InstructStream.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InstructStream.cs deleted file mode 100644 index 08c72be1..00000000 --- a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InstructStream.cs +++ /dev/null @@ -1,49 +0,0 @@ -namespace BotSharp.Core.Routing; - -public partial class RoutingService -{ - public async Task InstructStream(Agent agent, RoleDialogModel message, List dialogs) - { - var conv = _services.GetRequiredService(); - var storage = _services.GetRequiredService(); - storage.Append(conv.ConversationId, message); - - dialogs.Add(message); - Context.SetDialogs(dialogs); - - var routing = _services.GetRequiredService(); - routing.Context.Push(agent.Id, "instruct directly"); - var agentId = routing.Context.GetCurrentAgentId(); - - // Update next action agent's name - var agentService = _services.GetRequiredService(); - - 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(); - 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; - } -} diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeAgent.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeAgent.cs index 7bfd9352..77505718 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeAgent.cs +++ b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeAgent.cs @@ -4,7 +4,7 @@ namespace BotSharp.Core.Routing; public partial class RoutingService { - public async Task InvokeAgent(string agentId, List dialogs) + public async Task InvokeAgent(string agentId, List dialogs, bool useStream = false) { var agentService = _services.GetRequiredService(); 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 InvokeFunction(RoleDialogModel message, List dialogs) + private async Task InvokeFunction(RoleDialogModel message, List 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 diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.cs index 3e45fd63..bc9e1da2 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.cs +++ b/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(); + var ret = await routing.InvokeAgent(agentId, dialogs, convSettings.EnableStreaming); } var response = dialogs.Last(); diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs index 0552b19f..bd131ca1 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs +++ b/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(); - 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(); - 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) { diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs index a41abe87..79604c98 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs +++ b/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(); var state = _services.GetRequiredService(); diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs index c7a852fa..7da61c76 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs @@ -65,30 +65,30 @@ public class ChatHubObserver : IObserver } else if (value.EventName == AFTER_RECEIVE_LLM_STREAM_MESSAGE) { - var conv = _services.GetRequiredService(); - model = new ChatResponseDto() + if (message.IsStreaming) { - ConversationId = conv.ConversationId, - MessageId = message.MessageId, - Text = message.Content, - Sender = new() + var conv = _services.GetRequiredService(); + model = new ChatResponseDto() { - FirstName = "AI", - LastName = "Assistant", - Role = AgentRole.Assistant - } - }; + ConversationId = conv.ConversationId, + MessageId = message.MessageId, + Text = message.Content, + Sender = new() + { + FirstName = "AI", + LastName = "Assistant", + Role = AgentRole.Assistant + } + }; - var action = new ConversationSenderActionModel - { - ConversationId = conv.ConversationId, - SenderAction = SenderActionEnum.TypingOff - }; + var action = new ConversationSenderActionModel + { + ConversationId = conv.ConversationId, + SenderAction = SenderActionEnum.TypingOff + }; - GenerateSenderAction(conv.ConversationId, action).ConfigureAwait(false).GetAwaiter().GetResult(); - - var storage = _services.GetRequiredService(); - storage.Append(conv.ConversationId, message); + GenerateSenderAction(conv.ConversationId, action).ConfigureAwait(false).GetAwaiter().GetResult(); + } } else if (value.EventName == ON_RECEIVE_LLM_STREAM_MESSAGE) { diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs index 38d1455e..bb8c484a 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/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 GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) + public async Task GetChatCompletionsStreamingAsync(Agent agent, List 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; } diff --git a/tests/BotSharp.LLM.Tests/ChatCompletionTests.cs b/tests/BotSharp.LLM.Tests/ChatCompletionTests.cs index ee3e132a..f6c06dc4 100644 --- a/tests/BotSharp.LLM.Tests/ChatCompletionTests.cs +++ b/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([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 { - 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(); }