diff --git a/src/Infrastructure/BotSharp.Abstraction/Evaluations/Models/EvaluationRequest.cs b/src/Infrastructure/BotSharp.Abstraction/Evaluations/Models/EvaluationRequest.cs index 9dcad586..538b0609 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Evaluations/Models/EvaluationRequest.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Evaluations/Models/EvaluationRequest.cs @@ -10,6 +10,18 @@ public class EvaluationRequest : LlmBaseRequest [JsonPropertyName("states")] public IEnumerable States { get; set; } = []; + [JsonPropertyName("duplicate_limit")] + public int DuplicateLimit { get; set; } = 2; + [JsonPropertyName("max_rounds")] public int MaxRounds { get; set; } = 20; + + + [JsonPropertyName("additional_instruction")] + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] + public string? AdditionalInstruction { get; set; } + + [JsonPropertyName("stop_criteria")] + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] + public string? StopCriteria { get; set; } } diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs new file mode 100644 index 00000000..e242937e --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs @@ -0,0 +1,16 @@ +namespace BotSharp.Abstraction.Infrastructures.Events; + +public interface IEventPublisher +{ + /// + /// Boardcast message to all subscribers + /// + /// + /// + /// + Task BroadcastAsync(string channel, string message); + + Task PublishAsync(string channel, string message); + + Task ReDispatchAsync(string channel, int count = 10, string order = "asc"); +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs new file mode 100644 index 00000000..f295e54e --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs @@ -0,0 +1,8 @@ +namespace BotSharp.Abstraction.Infrastructures.Events; + +public interface IEventSubscriber +{ + Task SubscribeAsync(string channel, Func received); + + Task SubscribeAsync(string channel, string group, Func received); +} diff --git a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs index 4bdfe0d7..862ef899 100644 --- a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs +++ b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs @@ -9,6 +9,8 @@ using BotSharp.Abstraction.Users.Settings; using BotSharp.Abstraction.Interpreters.Settings; using BotSharp.Abstraction.Infrastructures; using BotSharp.Core.Processors; +using StackExchange.Redis; +using BotSharp.Core.Infrastructures.Events; namespace BotSharp.Core; @@ -26,18 +28,18 @@ public static class BotSharpCoreExtensions services.AddScoped(); services.AddScoped(); - services.AddSingleton(); - // Register cache service var cacheSettings = new SharpCacheSettings(); config.Bind("SharpCache", cacheSettings); services.AddSingleton(x => cacheSettings); services.AddSingleton(); + AddRedisEvents(services, config); + services.AddMemoryCache(); RegisterPlugins(services, config); - ConfigureBotSharpOptions(services, configOptions); + AddBotSharpOptions(services, configOptions); return services; } @@ -79,7 +81,7 @@ public static class BotSharpCoreExtensions return app; } - private static void ConfigureBotSharpOptions(IServiceCollection services, Action? configure) + private static void AddBotSharpOptions(IServiceCollection services, Action? configure) { var options = new BotSharpOptions(); if (configure != null) @@ -91,6 +93,17 @@ public static class BotSharpCoreExtensions services.AddSingleton(options); } + private static void AddRedisEvents(IServiceCollection services, IConfiguration config) + { + // Add Redis connection as a singleton + var dbSettings = new BotSharpDatabaseSettings(); + config.Bind("Database", dbSettings); + + services.AddSingleton(ConnectionMultiplexer.Connect(dbSettings.Redis)); + services.AddSingleton(); + services.AddSingleton(); + } + private static void AddDefaultJsonConverters(BotSharpOptions options) { options.JsonSerializerOptions.Converters.Add(new RichContentJsonConverter()); diff --git a/src/Infrastructure/BotSharp.Core/Evaluations/Services/EvaluatingService.Evaluate.cs b/src/Infrastructure/BotSharp.Core/Evaluations/Services/EvaluatingService.Evaluate.cs index a75450af..0442c857 100644 --- a/src/Infrastructure/BotSharp.Core/Evaluations/Services/EvaluatingService.Evaluate.cs +++ b/src/Infrastructure/BotSharp.Core/Evaluations/Services/EvaluatingService.Evaluate.cs @@ -42,6 +42,7 @@ public partial class EvaluatingService private async Task SimulateConversation(string initMessage, IEnumerable refDialogs, EvaluationRequest request) { var count = 0; + var duplicateCount = 0; var convId = Guid.NewGuid().ToString(); var curDialogs = new List(); var curUserMsg = initMessage; @@ -79,13 +80,30 @@ public partial class EvaluatingService { { "ref_conversation", refDialogs }, { "cur_conversation", curDialogs }, + { "additional_instruction", request.AdditionalInstruction }, + { "stop_criteria", request.StopCriteria } } }); _logger.LogInformation($"Generated message: {result?.GeneratedMessage}, stop: {result?.Stop}, reason: {result?.Reason}"); - if (curUserMsg.IsEqualTo(prevUserMsg) || curBotMsg.IsEqualTo(prevBotMsg) - || count > request.MaxRounds || (result != null && result.Stop)) + if (count > request.MaxRounds || (result != null && result.Stop)) + { + break; + } + + + if (curUserMsg.IsEqualTo(prevUserMsg) || curBotMsg.IsEqualTo(prevBotMsg)) + { + duplicateCount++; + } + else + { + duplicateCount = 0; + } + + + if (duplicateCount >= request.DuplicateLimit) { break; } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs index 25c3a047..32c7af52 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs @@ -5,65 +5,48 @@ namespace BotSharp.Core.Infrastructures; public class DistributedLocker { - private readonly BotSharpDatabaseSettings _settings; - private static ConnectionMultiplexer connection; + private readonly IConnectionMultiplexer _redis; + private readonly ILogger _logger; - public DistributedLocker(BotSharpDatabaseSettings settings) + public DistributedLocker(IConnectionMultiplexer redis, ILogger logger) { - _settings = settings; + _redis = redis; + _logger = logger; } public async Task Lock(string resource, Func> action, int timeoutInSeconds = 30) { - await ConnectToRedisAsync(); - var timeout = TimeSpan.FromSeconds(timeoutInSeconds); - var @lock = new RedisDistributedLock(resource, connection.GetDatabase()); + var @lock = new RedisDistributedLock(resource, _redis.GetDatabase()); await using (var handle = await @lock.TryAcquireAsync(timeout)) { if (handle == null) { - Serilog.Log.Logger.Error($"Acquire lock for {resource} failed due to after {timeout}s timeout."); + _logger.LogWarning($"Acquire lock for {resource} failed due to after {timeout}s timeout."); } return await action(); } } - public void Lock(string resource, Action action, int timeoutInSeconds = 30) + public bool Lock(string resource, Action action, int timeoutInSeconds = 30) { - ConnectToRedis(); - var timeout = TimeSpan.FromSeconds(timeoutInSeconds); - var @lock = new RedisDistributedLock(resource, connection.GetDatabase()); + var @lock = new RedisDistributedLock(resource, _redis.GetDatabase()); using (var handle = @lock.TryAcquire(timeout)) { if (handle == null) { - Serilog.Log.Logger.Error($"Acquire lock for {resource} failed due to after {timeout}s timeout."); + _logger.LogWarning($"Acquire lock for {resource} failed due to after {timeout}s timeout."); + return false; } else { action(); + return true; } } } - - private void ConnectToRedis() - { - if (connection == null) - { - connection = ConnectionMultiplexer.Connect(_settings.Redis); - } - } - - private async Task ConnectToRedisAsync() - { - if (connection == null) - { - connection = await ConnectionMultiplexer.ConnectAsync(_settings.Redis); - } - } } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs new file mode 100644 index 00000000..f6daa59d --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs @@ -0,0 +1,57 @@ +using StackExchange.Redis; + +namespace BotSharp.Core.Infrastructures.Events; + +public class RedisPublisher : IEventPublisher +{ + private readonly IConnectionMultiplexer _redis; + private readonly ISubscriber _subscriber; + private readonly ILogger _logger; + + public RedisPublisher(IConnectionMultiplexer redis, ILogger logger) + { + _redis = redis; + _logger = logger; + _subscriber = _redis.GetSubscriber(); + } + + public async Task BroadcastAsync(string channel, string message) + { + await _subscriber.PublishAsync(channel, message); + } + + public async Task PublishAsync(string channel, string message) + { + var db = _redis.GetDatabase(); + // Add a message to the stream, keeping only the latest 1 million messages + await db.StreamAddAsync(channel, "message", message, + maxLength: 1000 * 10000); + + _logger.LogInformation($"Published message {channel} {message}"); + } + + public async Task ReDispatchAsync(string channel, int count = 10, string order = "asc") + { + var db = _redis.GetDatabase(); + + var entries = await db.StreamRangeAsync(channel, "-", "+", count: count, messageOrder: order == "asc" ? Order.Ascending : Order.Descending); + foreach (var entry in entries) + { + _logger.LogInformation($"Fetched message: {channel} {entry.Values[0].Value} ({entry.Id})"); + + try + { + var messageId = await db.StreamAddAsync(channel, "message", entry.Values[0].Value); + + _logger.LogWarning($"ReDispatched message: {channel} {entry.Values[0].Value} ({messageId})"); + + // Optionally delete the message to save space + await db.StreamDeleteAsync(channel, [entry.Id]); + } + catch (Exception ex) + { + _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}"); + } + } + } +} diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs new file mode 100644 index 00000000..39bb1f91 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -0,0 +1,73 @@ +using StackExchange.Redis; +using System.Threading.Channels; + +namespace BotSharp.Core.Infrastructures.Events; + +public class RedisSubscriber : IEventSubscriber +{ + private readonly IConnectionMultiplexer _redis; + private readonly ISubscriber _subscriber; + private readonly ILogger _logger; + + public RedisSubscriber(IConnectionMultiplexer redis, ILogger logger) + { + _redis = redis; + _logger = logger; + _subscriber = _redis.GetSubscriber(); + } + + public async Task SubscribeAsync(string channel, Func received) + { + await _subscriber.SubscribeAsync(channel, async (ch, message) => + { + _logger.LogInformation($"Received event from channel: {ch} message: {message}"); + await received(ch, message); + }); + } + + public async Task SubscribeAsync(string channel, string group, Func received) + { + var db = _redis.GetDatabase(); + + // Create the consumer group if it doesn't exist + try + { + await db.StreamCreateConsumerGroupAsync(channel, group, StreamPosition.NewMessages, createStream: true); + } + catch (RedisServerException ex) when (ex.Message.Contains("BUSYGROUP")) + { + // Group already exists, ignore the error + _logger.LogWarning($"Consumer group '{group}' already exists (caught exception)."); + } + catch (Exception ex) + { + _logger.LogError($"Error creating consumer group: '{group}' {ex.Message}"); + throw; + } + + while (true) + { + var entries = await db.StreamReadGroupAsync(channel, group, Environment.MachineName, count: 1); + foreach (var entry in entries) + { + _logger.LogInformation($"Consumer {Environment.MachineName} received: {channel} {entry.Values[0].Value}"); + await db.StreamAcknowledgeAsync(channel, group, entry.Id); + + try + { + await received(channel, entry.Values[0].Value); + + // Optionally delete the message to save space + await db.StreamDeleteAsync(channel, [entry.Id]); + } + catch (Exception ex) + { + _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}"); + } + } + + await Task.Delay(Random.Shared.Next(1, 11) * 100); + } + + } +} diff --git a/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs b/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs index d5d0753d..fa757e2f 100644 --- a/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs +++ b/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs @@ -317,8 +317,8 @@ public class UserService : IUserService new Claim("role", user.Role ?? UserRole.User), new Claim(JwtRegisteredClaimNames.Jti, Guid.NewGuid().ToString()), new Claim("phone", user.Phone ?? string.Empty), - new Claim("affiliateId", user.AffiliateId ?? string.Empty), - new Claim("employeeId", user.EmployeeId ?? string.Empty), + new Claim("affiliate_id", user.AffiliateId ?? string.Empty), + new Claim("employee_id", user.EmployeeId ?? string.Empty), new Claim("regionCode", user.RegionCode ?? "CN") }; diff --git a/src/Infrastructure/BotSharp.Core/Using.cs b/src/Infrastructure/BotSharp.Core/Using.cs index 8a0ca2af..9e9177ba 100644 --- a/src/Infrastructure/BotSharp.Core/Using.cs +++ b/src/Infrastructure/BotSharp.Core/Using.cs @@ -40,4 +40,5 @@ global using BotSharp.Core.Agents.Services; global using BotSharp.Core.Conversations.Services; global using BotSharp.Core.Infrastructures; global using BotSharp.Core.Users.Services; -global using Aspects.Cache; \ No newline at end of file +global using Aspects.Cache; +global using BotSharp.Abstraction.Infrastructures.Events; \ No newline at end of file diff --git a/src/Infrastructure/BotSharp.Core/data/agents/dfd9b46d-d00c-40af-8a75-3fbdc2b89869/templates/instruction.simulator.liquid b/src/Infrastructure/BotSharp.Core/data/agents/dfd9b46d-d00c-40af-8a75-3fbdc2b89869/templates/instruction.simulator.liquid index 7569f20c..a2af5fe2 100644 --- a/src/Infrastructure/BotSharp.Core/data/agents/dfd9b46d-d00c-40af-8a75-3fbdc2b89869/templates/instruction.simulator.liquid +++ b/src/Infrastructure/BotSharp.Core/data/agents/dfd9b46d-d00c-40af-8a75-3fbdc2b89869/templates/instruction.simulator.liquid @@ -4,7 +4,31 @@ Please take the content in the [REFERENCE CONVERSATION] section as a reference, ** You need to take a close look at the content in both [REFERENCE CONVERSATION] and [ONGOING CONVERSATION], and determine whether to generate a text message or stop the ongoing conversation. ** When you generate a message, please assume you are the user and reply in the user perceptive. ** Please do not generate or append a message with similar meaning that you have already mentioned in the [ONGOING CONVERSATION]. + + +================= +[ADDITIONAL INSTRUCTION] +{{ "\r\n" }} +{%- if additional_instruction != empty -%} +{{ additional_instruction }} +{%- endif -%} +{{ "\r\n" }} + + +================= +[CONVERSATION STOP CRITERIA] +{{ "\r\n" }} +{%- if stop_criteria != empty -%} +{{ stop_criteria }} +{%- else -%} ** If you see the assistant replies two or more than two similar messages in the [ONGOING CONVERSATION], please stop the conversation immediately. +{%- endif -%} +{{ "\r\n" }} + + +================= +[OUTPUT JSON FORMAT] + ** The output must be in JSON format: { "generated_message": the generated text message using the user tone, diff --git a/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs b/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs index efad6e6c..4d85167f 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs @@ -209,7 +209,7 @@ public static class BotSharpOpenApiExtensions app.UseSwagger(); - // if (env.IsDevelopment()) + if (env.IsDevelopment()) { IdentityModelEventSource.ShowPII = true; app.UseSwaggerUI();