diff --git a/Directory.Packages.props b/Directory.Packages.props index cec2ac88..03798e37 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -26,6 +26,7 @@ + @@ -44,8 +45,8 @@ - - + + @@ -147,4 +148,4 @@ - + \ No newline at end of file diff --git a/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj b/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj index 97e645d0..2008c6a2 100644 --- a/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj +++ b/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj @@ -36,6 +36,7 @@ + diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/ConversationHookBase.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/ConversationHookBase.cs index da87646e..0a5244f8 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/ConversationHookBase.cs +++ b/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) diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/Dtos/ChatResponseDto.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/Dtos/ChatResponseDto.cs index b391bc7b..19b5e8b2 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/Dtos/ChatResponseDto.cs +++ b/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; } diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationHook.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationHook.cs index 6d967391..d1b6f83c 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationHook.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationHook.cs @@ -59,15 +59,17 @@ public interface IConversationHook : IHookBase /// Triggered before LLM calls function. /// /// + /// /// - Task OnFunctionExecuting(RoleDialogModel message); + Task OnFunctionExecuting(RoleDialogModel message, string from = InvokeSource.Manual); /// /// Triggered when the function calling completed. /// /// + /// /// - Task OnFunctionExecuted(RoleDialogModel message); + Task OnFunctionExecuted(RoleDialogModel message, string from = InvokeSource.Manual); Task OnResponseGenerated(RoleDialogModel message); diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs index 656b397d..322d86b6 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs @@ -36,7 +36,7 @@ public interface IConversationService /// Received the response from AI Agent /// Task SendMessage(string agentId, - RoleDialogModel lastDialog, + RoleDialogModel message, PostbackMessageModel? replyMessage, Func onResponseReceived); 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/MLTasks/IChatCompletion.cs b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IChatCompletion.cs index 7cf52fe4..4ac62841 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MLTasks/IChatCompletion.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MLTasks/IChatCompletion.cs @@ -16,14 +16,13 @@ public interface IChatCompletion void SetModelName(string model); Task GetChatCompletions(Agent agent, - List conversations); + List conversations) => throw new NotImplementedException(); Task GetChatCompletionsAsync(Agent agent, List conversations, Func onMessageReceived, - Func onFunctionExecuting); + Func onFunctionExecuting) => throw new NotImplementedException(); - Task GetChatCompletionsStreamingAsync(Agent agent, - List conversations, - Func onMessageReceived); + Task GetChatCompletionsStreamingAsync(Agent agent, + List conversations) => throw new NotImplementedException(); } diff --git a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs b/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs new file mode 100644 index 00000000..bf771cb4 --- /dev/null +++ b/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!; +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs b/src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs new file mode 100644 index 00000000..17773272 --- /dev/null +++ b/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!; +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Routing/Enums/InvokeSource.cs b/src/Infrastructure/BotSharp.Abstraction/Routing/Enums/InvokeSource.cs new file mode 100644 index 00000000..b3460bb6 --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Routing/Enums/InvokeSource.cs @@ -0,0 +1,19 @@ +namespace BotSharp.Abstraction.Routing.Enums; + +public static class InvokeSource +{ + /// + /// Invoke manually + /// + public const string Manual = "manual"; + + /// + /// Invoke by LLM directly + /// + public const string Llm = "llm"; + + /// + /// Invoke by agent routing + /// + public const string Routing = "routing"; +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs b/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs index 97be8874..56c98a82 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Routing/IRoutingService.cs @@ -26,12 +26,8 @@ public interface IRoutingService /// RoutingRule[] GetRulesByAgentId(string id); - //void ResetRecursiveCounter(); - //int GetRecursiveCounter(); - //void SetRecursiveCounter(int counter); - - Task InvokeAgent(string agentId, List dialogs); - Task InvokeFunction(string name, RoleDialogModel messages); + Task InvokeAgent(string agentId, List dialogs, string from = InvokeSource.Manual, bool useStream = false); + Task InvokeFunction(string name, RoleDialogModel messages, string from = InvokeSource.Manual); Task InstructLoop(Agent agent, RoleDialogModel message, List dialogs); /// diff --git a/src/Infrastructure/BotSharp.Abstraction/Using.cs b/src/Infrastructure/BotSharp.Abstraction/Using.cs index 8a9d30cb..38a6dca0 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Using.cs +++ b/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; diff --git a/src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs b/src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs index 4eb4c64e..095ff2b8 100644 --- a/src/Infrastructure/BotSharp.Core.Realtime/Hooks/RealtimeConversationHook.cs +++ b/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(); 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(); - if (hub.HubConn == null) + if (from != InvokeSource.Llm || hub.HubConn == null) { return; } diff --git a/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs b/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs index 32b0a112..fe544b4d 100644 --- a/src/Infrastructure/BotSharp.Core.Realtime/Services/RealtimeHub.cs +++ b/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 { diff --git a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs b/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs index 13ee1de6..a559877f 100644 --- a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs +++ b/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("GoogleApi"); }); + services.AddSingleton>(); + services.AddScoped(); services.AddScoped(); services.AddScoped(); diff --git a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.SendMessage.cs b/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.SendMessage.cs index a28ccbc7..1155dc38 100644 --- a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.SendMessage.cs +++ b/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; 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/Evaluations/EvaluationConversationHook.cs b/src/Infrastructure/BotSharp.Core/Evaluations/EvaluationConversationHook.cs index b315b683..b1b4541e 100644 --- a/src/Infrastructure/BotSharp.Core/Evaluations/EvaluationConversationHook.cs +++ b/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) diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/CompletionProvider.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/CompletionProvider.cs index 4bf45d86..4409c8eb 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/CompletionProvider.cs +++ b/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)) { diff --git a/src/Plugins/BotSharp.Plugin.GoogleAI/Models/Realtime/RealtimeTranscriptionResponse.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Streams/RealtimeTextStream.cs similarity index 81% rename from src/Plugins/BotSharp.Plugin.GoogleAI/Models/Realtime/RealtimeTranscriptionResponse.cs rename to src/Infrastructure/BotSharp.Core/Infrastructures/Streams/RealtimeTextStream.cs index 0a383c80..e041a53b 100644 --- a/src/Plugins/BotSharp.Plugin.GoogleAI/Models/Realtime/RealtimeTranscriptionResponse.cs +++ b/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; diff --git a/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs b/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs new file mode 100644 index 00000000..affb142d --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs @@ -0,0 +1,43 @@ +using System.Reactive.Subjects; + +namespace BotSharp.Core.Observables.Queues; + +public class MessageHub where T : class +{ + private readonly ILogger> _logger; + private readonly ISubject _observable = new Subject(); + public IObservable Events => _observable; + + public MessageHub(ILogger> logger) + { + _logger = logger; + } + + /// + /// Push an item to the observers. + /// + /// + public void Push(T item) + { + _observable.OnNext(item); + } + + /// + /// Send a complete notification to the observers. + /// This will stop the observers from receiving data. + /// + public void Complete() + { + _observable.OnCompleted(); + } + + /// + /// Send an error notification to the observers. + /// This will stop the observers from receiving data. + /// + /// + public void Error(Exception error) + { + _observable.OnError(error); + } +} diff --git a/src/Infrastructure/BotSharp.Core/Repository/FileRepository/FileRepository.Agent.cs b/src/Infrastructure/BotSharp.Core/Repository/FileRepository/FileRepository.Agent.cs index f7e70ca7..c5947708 100644 --- a/src/Infrastructure/BotSharp.Core/Repository/FileRepository/FileRepository.Agent.cs +++ b/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)) diff --git a/src/Infrastructure/BotSharp.Core/Routing/Reasoning/InstructExecutor.cs b/src/Infrastructure/BotSharp.Core/Routing/Reasoning/InstructExecutor.cs index c93e133c..e0ac69a2 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/Reasoning/InstructExecutor.cs +++ b/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(); + 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(); diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeAgent.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeAgent.cs index 7bfd9352..6d2813d9 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeAgent.cs +++ b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeAgent.cs @@ -4,7 +4,11 @@ namespace BotSharp.Core.Routing; public partial class RoutingService { - public async Task InvokeAgent(string agentId, List dialogs) + public async Task InvokeAgent( + string agentId, + List dialogs, + string from = InvokeSource.Manual, + bool useStream = false) { var agentService = _services.GetRequiredService(); 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 InvokeFunction(RoleDialogModel message, List dialogs) + private async Task InvokeFunction( + RoleDialogModel message, + List dialogs, + string from, + bool useStream) { // execute function // Save states @@ -75,7 +93,7 @@ public partial class RoutingService var routing = _services.GetRequiredService(); // 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 diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs index 75754939..1001dede 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs +++ b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs @@ -5,7 +5,7 @@ namespace BotSharp.Core.Routing; public partial class RoutingService { - public async Task InvokeFunction(string name, RoleDialogModel message) + public async Task InvokeFunction(string name, RoleDialogModel message, string from = InvokeSource.Manual) { var currentAgentId = message.CurrentAgentId; var agentService = _services.GetRequiredService(); @@ -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 diff --git a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.cs b/src/Infrastructure/BotSharp.Core/Routing/RoutingService.cs index 3e45fd63..c66ef4b2 100644 --- a/src/Infrastructure/BotSharp.Core/Routing/RoutingService.cs +++ b/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(); + 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(); diff --git a/src/Infrastructure/BotSharp.Core/Templating/TemplateRender.cs b/src/Infrastructure/BotSharp.Core/Templating/TemplateRender.cs index f7e5ecde..2124da16 100644 --- a/src/Infrastructure/BotSharp.Core/Templating/TemplateRender.cs +++ b/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(); _options.MemberAccessStrategy.Register(); - _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 RenderIdentifierTag(string tag, string identifier, TextWriter writer, TextEncoder encoder, TemplateContext context) + private static async ValueTask 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(); + 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 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); diff --git a/src/Infrastructure/BotSharp.Core/Using.cs b/src/Infrastructure/BotSharp.Core/Using.cs index aa5a4595..0e8e2762 100644 --- a/src/Infrastructure/BotSharp.Core/Using.cs +++ b/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; diff --git a/src/Infrastructure/BotSharp.Core/data/agents/01fcc3e5-9af7-49e6-ad7a-a760bd12dc4a/instructions/instruction.liquid b/src/Infrastructure/BotSharp.Core/data/agents/01fcc3e5-9af7-49e6-ad7a-a760bd12dc4a/instructions/instruction.liquid index 775de66c..1ada1850 100644 --- a/src/Infrastructure/BotSharp.Core/data/agents/01fcc3e5-9af7-49e6-ad7a-a760bd12dc4a/instructions/instruction.liquid +++ b/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] diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs index d3da550f..bd131ca1 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs +++ b/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) { diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/InstructModeController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/InstructModeController.cs index e8c87c2c..de67fcab 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/InstructModeController.cs +++ b/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(); var result = await instructor.Execute(agentId, diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/RealtimeController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/RealtimeController.cs index f8851cb3..61ae77a7 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/RealtimeController.cs +++ b/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; } } diff --git a/src/Plugins/BotSharp.Plugin.AnthropicAI/Providers/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.AnthropicAI/Providers/ChatCompletionProvider.cs index 285e8abd..50d67a1b 100644 --- a/src/Plugins/BotSharp.Plugin.AnthropicAI/Providers/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.AnthropicAI/Providers/ChatCompletionProvider.cs @@ -96,8 +96,7 @@ public class ChatCompletionProvider : IChatCompletion throw new NotImplementedException(); } - public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations, - Func onMessageReceived) + public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations) { throw new NotImplementedException(); } diff --git a/src/Plugins/BotSharp.Plugin.AzureOpenAI/BotSharp.Plugin.AzureOpenAI.csproj b/src/Plugins/BotSharp.Plugin.AzureOpenAI/BotSharp.Plugin.AzureOpenAI.csproj index 494326fa..372fb3de 100644 --- a/src/Plugins/BotSharp.Plugin.AzureOpenAI/BotSharp.Plugin.AzureOpenAI.csproj +++ b/src/Plugins/BotSharp.Plugin.AzureOpenAI/BotSharp.Plugin.AzureOpenAI.csproj @@ -16,7 +16,7 @@ - + diff --git a/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs index ab7135ef..605bceb8 100644 --- a/src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/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 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); var (prompt, messages, options) = PrepareOptions(agent, conversations); - var response = chatClient.CompleteChatStreamingAsync(messages, options); + var hub = _services.GetRequiredService>(); + var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty; - await foreach (var choice in response) + var contentHooks = _services.GetHooks(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(); + 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, ChatCompletionOptions) PrepareOptions(Agent agent, List conversations) diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs b/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs index 725655fc..d3c3f2ac 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs +++ b/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; /// /// The dialogue channel connects users, AI assistants and customer service representatives. /// -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(); services.AddScoped(); } + + public void Configure(IApplicationBuilder app) + { + var services = app.ApplicationServices; + var queue = services.GetRequiredService>(); + var logger = services.GetRequiredService>>(); + queue.Events.Subscribe(new ChatHubObserver(logger)); + } } diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs index 3c7a675d..1d641554 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/ChatHubConversationHook.cs +++ b/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(); 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); } diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/StreamingLogHook.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/StreamingLogHook.cs index 720f3fd0..5dd89a2b 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Hooks/StreamingLogHook.cs +++ b/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) diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs new file mode 100644 index 00000000..c15419ed --- /dev/null +++ b/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 +{ + 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(); + 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(); + 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(); + 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(); + return sidecar == null || !sidecar.IsEnabled; + } + + private void OnReceiveAssistantMessage(string @event, string conversationId, ChatResponseDto model) + { + try + { + var settings = _services.GetRequiredService(); + var chatHub = _services.GetRequiredService>(); + + if (settings.EventDispatchBy == EventDispatchType.Group) + { + chatHub.Clients.Group(conversationId).SendAsync(@event, model).ConfigureAwait(false).GetAwaiter().GetResult(); + } + else + { + var user = _services.GetRequiredService(); + 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(); + var chatHub = _services.GetRequiredService>(); + if (settings.EventDispatchBy == EventDispatchType.Group) + { + chatHub.Clients.Group(conversationId).SendAsync(GENERATE_SENDER_ACTION, action).ConfigureAwait(false).GetAwaiter().GetResult(); + } + else + { + var user = _services.GetRequiredService(); + 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})"); + } + } +} diff --git a/src/Plugins/BotSharp.Plugin.DeepSeekAI/BotSharp.Plugin.DeepSeekAI.csproj b/src/Plugins/BotSharp.Plugin.DeepSeekAI/BotSharp.Plugin.DeepSeekAI.csproj index 2f7e326a..3f9a26ce 100644 --- a/src/Plugins/BotSharp.Plugin.DeepSeekAI/BotSharp.Plugin.DeepSeekAI.csproj +++ b/src/Plugins/BotSharp.Plugin.DeepSeekAI/BotSharp.Plugin.DeepSeekAI.csproj @@ -15,7 +15,7 @@ - + diff --git a/src/Plugins/BotSharp.Plugin.DeepSeekAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.DeepSeekAI/Providers/Chat/ChatCompletionProvider.cs index eb14ac99..1335d4dc 100644 --- a/src/Plugins/BotSharp.Plugin.DeepSeekAI/Providers/Chat/ChatCompletionProvider.cs +++ b/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 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); var (prompt, messages, options) = PrepareOptions(agent, conversations); - var response = chatClient.CompleteChatStreamingAsync(messages, options); + var hub = _services.GetRequiredService>(); + var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty; - await foreach (var choice in response) + var contentHooks = _services.GetHooks(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(); + 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) diff --git a/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/GeminiChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/GeminiChatCompletionProvider.cs index 1ead28e6..1d67eac8 100644 --- a/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/GeminiChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/GeminiChatCompletionProvider.cs @@ -159,40 +159,9 @@ public class GeminiChatCompletionProvider : IChatCompletion return true; } - public async Task GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) + public Task GetChatCompletionsStreamingAsync(Agent agent, List 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) diff --git a/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/PalmChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/PalmChatCompletionProvider.cs index 72a47adc..e992fd60 100644 --- a/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/PalmChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/PalmChatCompletionProvider.cs @@ -145,7 +145,7 @@ public class PalmChatCompletionProvider : IChatCompletion throw new NotImplementedException(); } - public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) + public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations) { throw new NotImplementedException(); } diff --git a/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Realtime/RealTimeCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Realtime/RealTimeCompletionProvider.cs index 6838c284..5d2dd8ca 100644 --- a/src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Realtime/RealTimeCompletionProvider.cs +++ b/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; diff --git a/src/Plugins/BotSharp.Plugin.HuggingFace/Providers/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.HuggingFace/Providers/ChatCompletionProvider.cs index 460677ba..5a38b0a9 100644 --- a/src/Plugins/BotSharp.Plugin.HuggingFace/Providers/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.HuggingFace/Providers/ChatCompletionProvider.cs @@ -76,9 +76,9 @@ public class ChatCompletionProvider : IChatCompletion return true; } - public async Task GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) + public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations) { - return true; + throw new NotImplementedException(); } public void SetModelName(string model) diff --git a/src/Plugins/BotSharp.Plugin.LLamaSharp/BotSharp.Plugin.LLamaSharp.csproj b/src/Plugins/BotSharp.Plugin.LLamaSharp/BotSharp.Plugin.LLamaSharp.csproj index af7a6237..80807a26 100644 --- a/src/Plugins/BotSharp.Plugin.LLamaSharp/BotSharp.Plugin.LLamaSharp.csproj +++ b/src/Plugins/BotSharp.Plugin.LLamaSharp/BotSharp.Plugin.LLamaSharp.csproj @@ -15,7 +15,7 @@ - + diff --git a/src/Plugins/BotSharp.Plugin.LLamaSharp/Providers/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.LLamaSharp/Providers/ChatCompletionProvider.cs index 45f41e95..321b9aee 100644 --- a/src/Plugins/BotSharp.Plugin.LLamaSharp/Providers/ChatCompletionProvider.cs +++ b/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 GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) + public async Task GetChatCompletionsStreamingAsync(Agent agent, List 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(); 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>(); + 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) diff --git a/src/Plugins/BotSharp.Plugin.LangChain/Providers/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.LangChain/Providers/ChatCompletionProvider.cs index aa6b1175..9e2aa1d4 100644 --- a/src/Plugins/BotSharp.Plugin.LangChain/Providers/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.LangChain/Providers/ChatCompletionProvider.cs @@ -65,7 +65,7 @@ namespace BotSharp.Plugin.VertexAI.Providers throw new NotImplementedException(); } - public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) + public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations) { throw new NotImplementedException(); } diff --git a/src/Plugins/BotSharp.Plugin.MetaGLM/Providers/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.MetaGLM/Providers/ChatCompletionProvider.cs index c1d4bef1..8b3ecea3 100644 --- a/src/Plugins/BotSharp.Plugin.MetaGLM/Providers/ChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.MetaGLM/Providers/ChatCompletionProvider.cs @@ -235,7 +235,7 @@ public class ChatCompletionProvider : IChatCompletion throw new NotImplementedException(); } - public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) + public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations) { throw new NotImplementedException(); } diff --git a/src/Plugins/BotSharp.Plugin.MicrosoftExtensionsAI/MicrosoftExtensionsAIChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.MicrosoftExtensionsAI/MicrosoftExtensionsAIChatCompletionProvider.cs index d95a25ae..e510a9a2 100644 --- a/src/Plugins/BotSharp.Plugin.MicrosoftExtensionsAI/MicrosoftExtensionsAIChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.MicrosoftExtensionsAI/MicrosoftExtensionsAIChatCompletionProvider.cs @@ -169,8 +169,10 @@ public sealed class MicrosoftExtensionsAIChatCompletionProvider : IChatCompletio throw new NotImplementedException(); /// - public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) => + public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations) + { throw new NotImplementedException(); + } private sealed class NopAIFunction(string name, string description, JsonElement schema) : AIFunction { diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs index 17ee1012..57b44252 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs +++ b/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 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); var (prompt, messages, options) = PrepareOptions(agent, conversations); - var response = chatClient.CompleteChatStreamingAsync(messages, options); + var hub = _services.GetRequiredService>(); + var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty; - await foreach (var choice in response) + var contentHooks = _services.GetHooks(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(); + 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 ToolCalls { get; set; } = []; } \ No newline at end of file diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Image/ImageCompletionProvider.Generation.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Image/ImageCompletionProvider.Generation.cs index 2f55d917..5747df2b 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Image/ImageCompletionProvider.Generation.cs +++ b/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); } diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Image/ImageCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Image/ImageCompletionProvider.cs index 7cfda0f5..eaa03bb0 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Image/ImageCompletionProvider.cs +++ b/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; } diff --git a/src/Plugins/BotSharp.Plugin.SemanticKernel/SemanticKernelChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.SemanticKernel/SemanticKernelChatCompletionProvider.cs index 3f742777..57277399 100644 --- a/src/Plugins/BotSharp.Plugin.SemanticKernel/SemanticKernelChatCompletionProvider.cs +++ b/src/Plugins/BotSharp.Plugin.SemanticKernel/SemanticKernelChatCompletionProvider.cs @@ -94,7 +94,7 @@ namespace BotSharp.Plugin.SemanticKernel throw new NotImplementedException(); } /// - public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) + public Task GetChatCompletionsStreamingAsync(Agent agent, List conversations) { throw new NotImplementedException(); } diff --git a/src/Plugins/BotSharp.Plugin.SparkDesk/BotSharp.Plugin.SparkDesk.csproj b/src/Plugins/BotSharp.Plugin.SparkDesk/BotSharp.Plugin.SparkDesk.csproj index 51f26f87..d49d41df 100644 --- a/src/Plugins/BotSharp.Plugin.SparkDesk/BotSharp.Plugin.SparkDesk.csproj +++ b/src/Plugins/BotSharp.Plugin.SparkDesk/BotSharp.Plugin.SparkDesk.csproj @@ -15,7 +15,7 @@ - + diff --git a/src/Plugins/BotSharp.Plugin.SparkDesk/Providers/ChatCompletionProvider.cs b/src/Plugins/BotSharp.Plugin.SparkDesk/Providers/ChatCompletionProvider.cs index ba0aa220..035e03e0 100644 --- a/src/Plugins/BotSharp.Plugin.SparkDesk/Providers/ChatCompletionProvider.cs +++ b/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 GetChatCompletionsStreamingAsync(Agent agent, List conversations, Func onMessageReceived) + public async Task GetChatCompletionsStreamingAsync(Agent agent, List 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>(); + + 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) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Hooks/TwilioConversationHook.cs b/src/Plugins/BotSharp.Plugin.Twilio/Hooks/TwilioConversationHook.cs index bc647140..11bd8d36 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Hooks/TwilioConversationHook.cs +++ b/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(message.CurrentAgentId); diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioStreamMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioStreamMiddleware.cs index ea4d5c50..3bcaf18b 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/TwilioStreamMiddleware.cs +++ b/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(receivedText); + StreamEventResponse? response = new(); + + try + { + response = JsonSerializer.Deserialize(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; 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(); }